From 3fce8dad55a1e6abde4d4175186715cd5db66912 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 12:18:05 +0000 Subject: [PATCH 01/11] test: preserve datetime search attribute instants and precision --- .../V2/SearchAttributeValueFilterTest.php | 26 ++++++++++- .../V2/WorkflowSearchAttributeValueTest.php | 46 +++++++++++++++++++ 2 files changed, 71 insertions(+), 1 deletion(-) diff --git a/tests/Unit/V2/SearchAttributeValueFilterTest.php b/tests/Unit/V2/SearchAttributeValueFilterTest.php index 84663e86d..36af5c965 100644 --- a/tests/Unit/V2/SearchAttributeValueFilterTest.php +++ b/tests/Unit/V2/SearchAttributeValueFilterTest.php @@ -33,10 +33,34 @@ public static function typedValues(): iterable yield 'boolean' => [true, 'value_bool']; yield 'integer' => [42, 'value_int']; yield 'float' => [2.5, 'value_float']; - yield 'datetime' => [new DateTimeImmutable('2026-09-04T12:00:00Z'), 'value_datetime']; yield 'long string' => [str_repeat('x', 256), 'value_string']; } + #[DataProvider('datetimeValues')] + public function testDatetimeBindingsPreserveUtcAndMicroseconds(string $input): void + { + $value = new DateTimeImmutable($input); + $query = $this->query('sqlite'); + SearchAttributeValueFilter::apply($query, $value); + + $this->assertStringContainsString('"value_datetime" = ?', $query->toSql()); + $this->assertSame( + ['2026-10-08 10:34:56.123456'], + $query->getQuery()->getConnection()->prepareBindings($query->getBindings()), + ); + $this->assertSame($input, $value->format('Y-m-d\TH:i:s.uP')); + } + + /** + * @return iterable + */ + public static function datetimeValues(): iterable + { + yield 'positive offset' => ['2026-10-08T12:34:56.123456+02:00']; + yield 'negative offset' => ['2026-10-08T06:34:56.123456-04:00']; + yield 'UTC control' => ['2026-10-08T10:34:56.123456+00:00']; + } + public function testShortStringsMatchKeywordsOrKeywordListMembersOnSqlite(): void { $query = $this->query('sqlite'); diff --git a/tests/Unit/V2/WorkflowSearchAttributeValueTest.php b/tests/Unit/V2/WorkflowSearchAttributeValueTest.php index 728ba096d..f3fc7a228 100644 --- a/tests/Unit/V2/WorkflowSearchAttributeValueTest.php +++ b/tests/Unit/V2/WorkflowSearchAttributeValueTest.php @@ -124,6 +124,52 @@ public function testItCoercesCarbonAndNativeDatetimes(): void $this->assertSame('2026-09-04 13:00:00.654321', $nativeAttribute->getValue()?->format('Y-m-d H:i:s.u')); } + #[DataProvider('offsetDatetimes')] + public function testDatetimeStoragePreservesTheInstantWithoutMutatingInput(mixed $value): void + { + $before = $value instanceof \DateTimeInterface ? $value->format('Y-m-d\TH:i:s.uP') : $value; + $attribute = new WorkflowSearchAttribute(); + $attribute->setTypedValue($value, WorkflowSearchAttribute::TYPE_DATETIME); + + $this->assertSame('2026-10-08 10:34:56.123456', $attribute->getAttributes()['value_datetime']); + $this->assertSame('2026-10-08T10:34:56.123456Z', $attribute->getValue()?->toISOString()); + $this->assertSame( + $before, + $value instanceof \DateTimeInterface ? $value->format('Y-m-d\TH:i:s.uP') : $value, + ); + } + + /** + * @return iterable + */ + public static function offsetDatetimes(): iterable + { + yield 'positive offset text' => ['2026-10-08T12:34:56.123456+02:00']; + yield 'negative offset text' => ['2026-10-08T06:34:56.123456-04:00']; + yield 'mutable Carbon' => [Carbon::parse('2026-10-08T12:34:56.123456+02:00')]; + yield 'immutable Carbon' => [\Carbon\CarbonImmutable::parse('2026-10-08T06:34:56.123456-04:00')]; + yield 'mutable native datetime' => [new \DateTime('2026-10-08T12:34:56.123456+02:00')]; + yield 'immutable native datetime' => [new DateTimeImmutable('2026-10-08T06:34:56.123456-04:00')]; + yield 'UTC control' => ['2026-10-08T10:34:56.123456Z']; + } + + public function testStoredUtcDatetimeDoesNotAdoptTheApplicationTimezone(): void + { + $previous = date_default_timezone_get(); + try { + date_default_timezone_set('America/New_York'); + $attribute = new WorkflowSearchAttribute(); + $attribute->setRawAttributes([ + 'type' => WorkflowSearchAttribute::TYPE_DATETIME, + 'value_datetime' => '2026-10-08 10:34:56.123456', + ]); + + $this->assertSame('2026-10-08T10:34:56.123456Z', $attribute->getValue()?->toISOString()); + } finally { + date_default_timezone_set($previous); + } + } + public function testItRejectsAnInvalidDeclaredType(): void { $this->expectException(InvalidArgumentException::class); From 42eb97e03e22d092cde727f4a18888e8f5243e93 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 12:24:38 +0000 Subject: [PATCH 02/11] fix: preserve UTC instants and precision in datetime attributes --- src/V2/Models/WorkflowSearchAttribute.php | 3 ++- src/V2/Support/SearchAttributeValueFilter.php | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/src/V2/Models/WorkflowSearchAttribute.php b/src/V2/Models/WorkflowSearchAttribute.php index 88a615c61..6dc899ab8 100644 --- a/src/V2/Models/WorkflowSearchAttribute.php +++ b/src/V2/Models/WorkflowSearchAttribute.php @@ -10,6 +10,7 @@ use InvalidArgumentException; use Workflow\Traits\ResolvesStorageConnection; use Workflow\V2\Support\ConfiguredV2Models; +use Workflow\V2\Support\UtcScheduleTimestamp; /** * Workflow Search Attribute - Typed indexed metadata. @@ -97,7 +98,7 @@ class WorkflowSearchAttribute extends Model 'value_int' => 'integer', 'value_float' => 'float', 'value_bool' => 'boolean', - 'value_datetime' => 'datetime', + 'value_datetime' => UtcScheduleTimestamp::class, 'value_keyword_list' => 'array', 'upserted_at_sequence' => 'integer', 'inherited_from_parent' => 'boolean', diff --git a/src/V2/Support/SearchAttributeValueFilter.php b/src/V2/Support/SearchAttributeValueFilter.php index c5d38dfce..da006cd08 100644 --- a/src/V2/Support/SearchAttributeValueFilter.php +++ b/src/V2/Support/SearchAttributeValueFilter.php @@ -30,7 +30,7 @@ public static function apply(Builder $query, mixed $value): Builder } if ($value instanceof \DateTimeInterface) { - return $query->where('value_datetime', $value); + return $query->where('value_datetime', UtcScheduleTimestamp::databaseValue($value)); } if (is_string($value) && mb_strlen($value) <= 255) { From 774e4f9eb6cabaceb752886062712372ffd86301 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 12:47:37 +0000 Subject: [PATCH 03/11] style: format prepared datetime binding assertion --- tests/Unit/V2/SearchAttributeValueFilterTest.php | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/Unit/V2/SearchAttributeValueFilterTest.php b/tests/Unit/V2/SearchAttributeValueFilterTest.php index 36af5c965..f0012d2bf 100644 --- a/tests/Unit/V2/SearchAttributeValueFilterTest.php +++ b/tests/Unit/V2/SearchAttributeValueFilterTest.php @@ -46,7 +46,9 @@ public function testDatetimeBindingsPreserveUtcAndMicroseconds(string $input): v $this->assertStringContainsString('"value_datetime" = ?', $query->toSql()); $this->assertSame( ['2026-10-08 10:34:56.123456'], - $query->getQuery()->getConnection()->prepareBindings($query->getBindings()), + $query->getQuery() + ->getConnection() + ->prepareBindings($query->getBindings()), ); $this->assertSame($input, $value->format('Y-m-d\TH:i:s.uP')); } From 9e569d89183073ecb88751fe1922e65edbab336a Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 13:02:42 +0000 Subject: [PATCH 04/11] test: persist and filter offset datetime attributes through SQL --- tests/Feature/V2/V2SearchAttributeTest.php | 60 ++++++++++++++++++++++ 1 file changed, 60 insertions(+) diff --git a/tests/Feature/V2/V2SearchAttributeTest.php b/tests/Feature/V2/V2SearchAttributeTest.php index 269a14cb9..2293e486d 100644 --- a/tests/Feature/V2/V2SearchAttributeTest.php +++ b/tests/Feature/V2/V2SearchAttributeTest.php @@ -4,6 +4,8 @@ namespace Tests\Feature\V2; +use DateTimeImmutable; +use PHPUnit\Framework\Attributes\DataProvider; use Tests\Fixtures\V2\TestGreetingActivity; use Tests\Fixtures\V2\TestSearchAttributeWorkflow; use Tests\TestCase; @@ -11,11 +13,69 @@ use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowRunSummary; +use Workflow\V2\Models\WorkflowSearchAttribute; +use Workflow\V2\Support\SearchAttributeUpsertService; +use Workflow\V2\Support\SearchAttributeValueFilter; +use Workflow\V2\Support\UpsertSearchAttributesCall; use Workflow\V2\Support\VisibilityFilters; use Workflow\V2\WorkflowStub; final class V2SearchAttributeTest extends TestCase { + #[DataProvider('datetimeInstants')] + public function testDatetimeUpsertsPreserveTheInstantThroughStorageAndFiltering(string $value): void + { + WorkflowStub::fake(); + WorkflowStub::mock(TestGreetingActivity::class, 'Hello, Taylor!'); + $workflow = WorkflowStub::make(TestSearchAttributeWorkflow::class, 'sa-datetime'); + $workflow->start('Taylor'); + $run = WorkflowRun::query()->findOrFail($workflow->runId()); + + $this->app->make(SearchAttributeUpsertService::class)->upsert( + $run, + new UpsertSearchAttributesCall(['happened_at' => $value]), + 0, + attributeTypes: ['happened_at' => WorkflowSearchAttribute::TYPE_DATETIME], + ); + + $attribute = WorkflowSearchAttribute::query() + ->where('workflow_run_id', $run->id) + ->where('key', 'happened_at') + ->firstOrFail(); + $this->assertSame('2026-10-08 10:34:56.123456', $attribute->getRawOriginal('value_datetime')); + + $previous = date_default_timezone_get(); + + try { + date_default_timezone_set('America/New_York'); + $this->assertSame('2026-10-08T10:34:56.123456Z', $attribute->fresh()?->getValue()?->toISOString()); + } finally { + date_default_timezone_set($previous); + } + + $query = static fn () => WorkflowSearchAttribute::query() + ->where('workflow_run_id', $run->id) + ->where('key', 'happened_at'); + $this->assertSame(1, SearchAttributeValueFilter::apply( + $query(), + new DateTimeImmutable('2026-10-08T10:34:56.123456Z'), + )->count()); + $this->assertSame(0, SearchAttributeValueFilter::apply( + $query(), + new DateTimeImmutable('2026-10-08T10:34:56.123455Z'), + )->count()); + } + + /** + * @return iterable + */ + public static function datetimeInstants(): iterable + { + yield 'positive offset' => ['2026-10-08T12:34:56.123456+02:00']; + yield 'negative offset' => ['2026-10-08T06:34:56.123456-04:00']; + yield 'UTC control' => ['2026-10-08T10:34:56.123456Z']; + } + public function testWorkflowCanUpsertSearchAttributes(): void { WorkflowStub::fake(); From 34305a25ff7de5ca6651ce7aa2ea58f593ec713a Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 13:16:50 +0000 Subject: [PATCH 05/11] style: format datetime upsert fixture arguments --- tests/Feature/V2/V2SearchAttributeTest.php | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/tests/Feature/V2/V2SearchAttributeTest.php b/tests/Feature/V2/V2SearchAttributeTest.php index 2293e486d..711bc4e50 100644 --- a/tests/Feature/V2/V2SearchAttributeTest.php +++ b/tests/Feature/V2/V2SearchAttributeTest.php @@ -33,9 +33,13 @@ public function testDatetimeUpsertsPreserveTheInstantThroughStorageAndFiltering( $this->app->make(SearchAttributeUpsertService::class)->upsert( $run, - new UpsertSearchAttributesCall(['happened_at' => $value]), + new UpsertSearchAttributesCall([ + 'happened_at' => $value, + ]), 0, - attributeTypes: ['happened_at' => WorkflowSearchAttribute::TYPE_DATETIME], + attributeTypes: [ + 'happened_at' => WorkflowSearchAttribute::TYPE_DATETIME, + ], ); $attribute = WorkflowSearchAttribute::query() From 31755ae53c9cd8152d8074d8e871045b0aad1f9c Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 13:44:28 +0000 Subject: [PATCH 06/11] test: require explicit history-backed datetime projection repair --- .../V2SearchAttributeDatetimeRepairTest.php | 207 ++++++++++++++++++ tests/Feature/V2/V2SearchAttributeTest.php | 2 + .../V2/SearchAttributeValueFilterTest.php | 2 + .../V2/WorkflowSearchAttributeValueTest.php | 2 + 4 files changed, 213 insertions(+) create mode 100644 tests/Feature/V2/V2SearchAttributeDatetimeRepairTest.php diff --git a/tests/Feature/V2/V2SearchAttributeDatetimeRepairTest.php b/tests/Feature/V2/V2SearchAttributeDatetimeRepairTest.php new file mode 100644 index 000000000..a7f5055a0 --- /dev/null +++ b/tests/Feature/V2/V2SearchAttributeDatetimeRepairTest.php @@ -0,0 +1,207 @@ +legacyRun(['Offset' => $original]); + $business = $this->businessSnapshot($run); + $before = $this->attributes($run); + + $this->assertSame(0, $this->command($run)); + $preview = json_decode(Artisan::output(), true, flags: JSON_THROW_ON_ERROR); + $this->assertTrue($preview['dry_run']); + $this->assertSame(1, $preview['would_repair']); + $this->assertSame(0, $preview['repaired']); + $this->assertSame($before, $this->attributes($run)); + + $this->assertSame(0, $this->command($run, ['--apply' => true])); + $applied = json_decode(Artisan::output(), true, flags: JSON_THROW_ON_ERROR); + $this->assertSame(1, $applied['repaired']); + $this->assertSame( + '2026-10-08T10:34:56.123456Z', + $run->searchAttributes()->firstOrFail()->getValue()?->toISOString(), + ); + $this->assertSame($business, $this->businessSnapshot($run)); + + $repaired = $this->attributes($run); + $this->assertSame(0, $this->command($run, ['--apply' => true])); + $repeat = json_decode(Artisan::output(), true, flags: JSON_THROW_ON_ERROR); + $this->assertSame(0, $repeat['repaired']); + $this->assertSame(1, $repeat['already_correct']); + $this->assertSame($repaired, $this->attributes($run)); + $this->assertSame($business, $this->businessSnapshot($run)); + } + + /** + * @return iterable + */ + public static function offsetDatetimes(): iterable + { + yield 'positive offset' => ['2026-10-08T12:34:56.123456+02:00']; + yield 'negative offset' => ['2026-10-08T06:34:56.123456-04:00']; + yield 'next local day' => ['2026-10-09T00:34:56.123456+14:00']; + yield 'previous local day' => ['2026-10-07T22:34:56.123456-12:00']; + } + + public function testMissingTimezoneRefusesTheRunWithoutGuessingOrPartialWrites(): void + { + $run = $this->legacyRun([ + 'Recoverable' => '2026-10-08T12:34:56.123456+02:00', + 'Unknown' => '2026-10-08T12:34:56.123456', + ]); + $before = $this->attributes($run); + $business = $this->businessSnapshot($run); + + $this->assertSame(1, $this->command($run, ['--apply' => true])); + $this->assertSame($before, $this->attributes($run)); + $this->assertSame($business, $this->businessSnapshot($run)); + + $this->assertSame(0, $this->command($run, ['--apply' => true, '--key' => ['Recoverable']])); + $rows = $run->searchAttributes()->get()->keyBy('key'); + $this->assertSame('2026-10-08 10:34:56.123456', $rows['Recoverable']->getRawOriginal('value_datetime')); + $this->assertSame('2026-10-08 12:34:56.123456', $rows['Unknown']->getRawOriginal('value_datetime')); + } + + public function testLaterCanonicalUpsertFencesAStaleProjection(): void + { + $run = $this->legacyRun(['Offset' => '2026-10-08T12:34:56.123456+02:00']); + WorkflowHistoryEvent::record($run, HistoryEventType::SearchAttributesUpserted, [ + 'sequence' => 8, + 'attributes' => ['Offset' => '2026-10-09T10:00:00Z'], + 'attribute_types' => ['Offset' => WorkflowSearchAttribute::TYPE_DATETIME], + 'merged' => ['Offset' => '2026-10-09T10:00:00Z'], + ]); + $before = $this->attributes($run); + $business = $this->businessSnapshot($run); + + $this->assertSame(1, $this->command($run, ['--apply' => true])); + $this->assertSame($before, $this->attributes($run)); + $this->assertSame($business, $this->businessSnapshot($run)); + } + + public function testProjectionFailureRollsBackEveryAttributeWrite(): void + { + $run = $this->legacyRun([ + 'First' => '2026-10-08T12:34:56.123456+02:00', + 'Second' => '2026-10-08T06:34:56.123456-04:00', + ]); + $before = $this->attributes($run); + $business = $this->businessSnapshot($run); + $this->mock(HistoryProjectionMaintenanceRole::class, static function (MockInterface $mock): void { + $mock->shouldReceive('projectRun')->once()->andThrow(new RuntimeException('Projection write failed')); + }); + + $this->assertSame(1, $this->command($run, ['--apply' => true])); + $this->assertSame($before, $this->attributes($run)); + $this->assertSame($business, $this->businessSnapshot($run)); + } + + public function testNamespaceAndPrunedHistoryRefusalsDoNotMutateRows(): void + { + $run = $this->legacyRun(['Offset' => '2026-10-08T12:34:56.123456+02:00']); + $before = $this->attributes($run); + $this->assertSame(1, $this->command($run, ['--namespace' => 'other', '--apply' => true])); + $this->assertSame($before, $this->attributes($run)); + + $run->forceFill(['details_pruned_at' => now()])->save(); + $business = $this->businessSnapshot($run); + $this->assertSame(1, $this->command($run, ['--apply' => true])); + $this->assertSame($before, $this->attributes($run)); + $this->assertSame($business, $this->businessSnapshot($run)); + } + + /** + * @param array $originals + */ + private function legacyRun(array $originals): WorkflowRun + { + WorkflowStub::fake(); + WorkflowStub::mock(TestGreetingActivity::class, 'Hello, Taylor!'); + $workflow = WorkflowStub::make(TestGreetingWorkflow::class, 'datetime-repair'); + $workflow->start('Taylor'); + $run = WorkflowRun::query()->findOrFail($workflow->runId()); + + foreach ($originals as $key => $original) { + $attribute = new WorkflowSearchAttribute([ + 'workflow_run_id' => $run->id, + 'workflow_instance_id' => $run->workflow_instance_id, + 'key' => $key, + 'upserted_at_sequence' => 7, + ]); + $attribute->setTypedValue($original, WorkflowSearchAttribute::TYPE_DATETIME); + $attribute->save(); + // Reproduce the original wall-clock SQL representation without altering history. + $attribute->getConnection()->table($attribute->getTable()) + ->where('id', $attribute->id) + ->update(['value_datetime' => (new DateTimeImmutable($original))->format('Y-m-d H:i:s.u')]); + } + + WorkflowHistoryEvent::record($run, HistoryEventType::SearchAttributesUpserted, [ + 'sequence' => 7, + 'attributes' => $originals, + 'attribute_types' => array_fill_keys(array_keys($originals), WorkflowSearchAttribute::TYPE_DATETIME), + 'merged' => $originals, + ]); + + return $run; + } + + /** + * @param array $options + */ + private function command(WorkflowRun $run, array $options = []): int + { + return Artisan::call('workflow:v2:repair-search-datetimes', array_merge([ + 'run' => $run->id, + '--namespace' => $run->namespace, + '--json' => true, + ], $options)); + } + + /** + * @return list> + */ + private function attributes(WorkflowRun $run): array + { + return $run->getConnection()->table('workflow_search_attributes') + ->where('workflow_run_id', $run->id)->orderBy('key')->get() + ->map(static fn (object $row): array => (array) $row)->all(); + } + + /** + * @return array>> + */ + private function businessSnapshot(WorkflowRun $run): array + { + $snapshot = []; + + foreach (['workflow_runs', 'workflow_tasks', 'workflow_history_events', 'workflow_commands'] as $table) { + $snapshot[$table] = $run->getConnection()->table($table) + ->where($table === 'workflow_runs' ? 'id' : 'workflow_run_id', $run->id) + ->orderBy('id')->get()->map(static fn (object $row): array => (array) $row)->all(); + } + + return $snapshot; + } +} diff --git a/tests/Feature/V2/V2SearchAttributeTest.php b/tests/Feature/V2/V2SearchAttributeTest.php index 711bc4e50..f915779f8 100644 --- a/tests/Feature/V2/V2SearchAttributeTest.php +++ b/tests/Feature/V2/V2SearchAttributeTest.php @@ -77,6 +77,8 @@ public static function datetimeInstants(): iterable { yield 'positive offset' => ['2026-10-08T12:34:56.123456+02:00']; yield 'negative offset' => ['2026-10-08T06:34:56.123456-04:00']; + yield 'next local day' => ['2026-10-09T00:34:56.123456+14:00']; + yield 'previous local day' => ['2026-10-07T22:34:56.123456-12:00']; yield 'UTC control' => ['2026-10-08T10:34:56.123456Z']; } diff --git a/tests/Unit/V2/SearchAttributeValueFilterTest.php b/tests/Unit/V2/SearchAttributeValueFilterTest.php index f0012d2bf..108d29101 100644 --- a/tests/Unit/V2/SearchAttributeValueFilterTest.php +++ b/tests/Unit/V2/SearchAttributeValueFilterTest.php @@ -60,6 +60,8 @@ public static function datetimeValues(): iterable { yield 'positive offset' => ['2026-10-08T12:34:56.123456+02:00']; yield 'negative offset' => ['2026-10-08T06:34:56.123456-04:00']; + yield 'next local day' => ['2026-10-09T00:34:56.123456+14:00']; + yield 'previous local day' => ['2026-10-07T22:34:56.123456-12:00']; yield 'UTC control' => ['2026-10-08T10:34:56.123456+00:00']; } diff --git a/tests/Unit/V2/WorkflowSearchAttributeValueTest.php b/tests/Unit/V2/WorkflowSearchAttributeValueTest.php index f3fc7a228..429008e71 100644 --- a/tests/Unit/V2/WorkflowSearchAttributeValueTest.php +++ b/tests/Unit/V2/WorkflowSearchAttributeValueTest.php @@ -150,6 +150,8 @@ public static function offsetDatetimes(): iterable yield 'immutable Carbon' => [\Carbon\CarbonImmutable::parse('2026-10-08T06:34:56.123456-04:00')]; yield 'mutable native datetime' => [new \DateTime('2026-10-08T12:34:56.123456+02:00')]; yield 'immutable native datetime' => [new DateTimeImmutable('2026-10-08T06:34:56.123456-04:00')]; + yield 'next local day' => ['2026-10-09T00:34:56.123456+14:00']; + yield 'previous local day' => ['2026-10-07T22:34:56.123456-12:00']; yield 'UTC control' => ['2026-10-08T10:34:56.123456Z']; } From 874255a44b38f07282bea37c561d401c0e1b7b5b Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 14:06:33 +0000 Subject: [PATCH 07/11] feat: repair datetime projections from retained typed history --- .../V2RepairSearchDatetimesCommand.php | 92 +++++++ src/Providers/WorkflowServiceProvider.php | 2 + .../Support/SearchAttributeDatetimeRepair.php | 260 ++++++++++++++++++ .../V2SearchAttributeDatetimeRepairTest.php | 96 +++++-- 4 files changed, 426 insertions(+), 24 deletions(-) create mode 100644 src/Commands/V2RepairSearchDatetimesCommand.php create mode 100644 src/V2/Support/SearchAttributeDatetimeRepair.php diff --git a/src/Commands/V2RepairSearchDatetimesCommand.php b/src/Commands/V2RepairSearchDatetimesCommand.php new file mode 100644 index 000000000..0a39838ed --- /dev/null +++ b/src/Commands/V2RepairSearchDatetimesCommand.php @@ -0,0 +1,92 @@ +whereKey((string) $this->argument('run')); + $namespace = $this->option('namespace'); + + if ($namespace !== null) { + if (! is_string($namespace) || $namespace === '') { + throw new InvalidArgumentException('An explicit namespace must be a non-empty string.'); + } + + $query->where('namespace', $namespace); + } + + $keys = $this->option('key'); + + if (! is_array($keys)) { + throw new InvalidArgumentException('Datetime keys must be a list of strings.'); + } + + foreach ($keys as $key) { + if (! is_string($key) || $key === '') { + throw new InvalidArgumentException('Each selected datetime key must be a non-empty string.'); + } + } + + $keys = array_values(array_unique($keys)); + $report = $repair->repair($query->firstOrFail(), (bool) $this->option('apply'), $keys); + + if ((bool) $this->option('json')) { + $this->line(json_encode($report, JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR)); + } else { + $this->table(['Run', 'Examined', 'Would repair', 'Repaired', 'Already correct'], [[ + $report['run_id'], $report['examined'], $report['would_repair'], + $report['repaired'], $report['already_correct'], + ]]); + } + + return self::SUCCESS; + } catch (ModelNotFoundException) { + return $this->failure('The selected run was not found in the requested namespace.', 'refused'); + } catch (DomainException|InvalidArgumentException $exception) { + return $this->failure($exception->getMessage(), 'refused'); + } catch (Throwable) { + return $this->failure( + 'Repair could not be verified. Inspect the selected projections before retrying.', + 'unverified', + ); + } + } + + private function failure(string $message, string $status): int + { + if ((bool) $this->option('json')) { + $this->line(json_encode([ + 'status' => $status, + 'message' => $message, + ], JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR)); + } else { + $this->error($message); + } + + return self::FAILURE; + } +} diff --git a/src/Providers/WorkflowServiceProvider.php b/src/Providers/WorkflowServiceProvider.php index 3322a7484..d20f4a748 100644 --- a/src/Providers/WorkflowServiceProvider.php +++ b/src/Providers/WorkflowServiceProvider.php @@ -19,6 +19,7 @@ use Workflow\Commands\V2HistoryImportCommand; use Workflow\Commands\V2RebuildProjectionsCommand; use Workflow\Commands\V2RepairPassCommand; +use Workflow\Commands\V2RepairSearchDatetimesCommand; use Workflow\Commands\V2ReplayConformanceCommand; use Workflow\Commands\V2ReplaySimulateCommand; use Workflow\Commands\V2ReplayVerifyCommand; @@ -189,6 +190,7 @@ public function boot(): void V2HistoryExportCommand::class, V2HistoryImportCommand::class, V2RepairPassCommand::class, + V2RepairSearchDatetimesCommand::class, V2RebuildProjectionsCommand::class, V2ReplayConformanceCommand::class, V2ReplaySimulateCommand::class, diff --git a/src/V2/Support/SearchAttributeDatetimeRepair.php b/src/V2/Support/SearchAttributeDatetimeRepair.php new file mode 100644 index 000000000..5051107b2 --- /dev/null +++ b/src/V2/Support/SearchAttributeDatetimeRepair.php @@ -0,0 +1,260 @@ + $keys + * @return array{run_id: string, namespace: ?string, dry_run: bool, examined: int, would_repair: int, repaired: int, already_correct: int} + */ + public function repair(WorkflowRun $selectedRun, bool $apply = false, array $keys = []): array + { + return $selectedRun->getConnection() + ->transaction(function () use ($selectedRun, $apply, $keys): array { + $query = $selectedRun->newQuery() + ->whereKey($selectedRun->id) + ->where('namespace', $selectedRun->namespace); + + if ($apply) { + $query->lockForUpdate(); + } + + $run = $query->firstOrFail(); + + if ($run->details_pruned_at !== null) { + throw new DomainException('Retained history is unavailable for this pruned run.'); + } + + $attributeQuery = $run->searchAttributes() + ->where('type', WorkflowSearchAttribute::TYPE_DATETIME); + + if ($keys !== []) { + $attributeQuery->whereIn('key', $keys); + } + + if ($apply) { + $attributeQuery->lockForUpdate(); + } + + $attributes = $attributeQuery->orderBy('key') + ->get(); + $selectedKeys = array_map('strval', $attributes->pluck('key')->all()); + + if (array_diff($keys, $selectedKeys) !== []) { + throw new DomainException('A selected key is not a retained datetime attribute on this run.'); + } + + $latest = $this->latestValues($run, $selectedKeys); + $plan = []; + $alreadyCorrect = 0; + + foreach ($attributes as $attribute) { + $canonical = $this->canonicalValue($attribute, $latest[$attribute->key] ?? null); + + if ($this->storedValue($attribute) === $canonical) { + $alreadyCorrect++; + + continue; + } + + $plan[] = [$attribute, $canonical]; + } + + $report = [ + 'run_id' => (string) $run->id, + 'namespace' => $run->namespace === null ? null : (string) $run->namespace, + 'dry_run' => ! $apply, + 'examined' => $attributes->count(), + 'would_repair' => count($plan), + 'repaired' => 0, + 'already_correct' => $alreadyCorrect, + ]; + + if (! $apply || $plan === []) { + return $report; + } + + foreach ($plan as [$attribute, $canonical]) { + $changed = $attribute->getConnection() + ->table($attribute->getTable()) + ->where('id', $attribute->id) + ->where('workflow_run_id', $run->id) + ->where('type', WorkflowSearchAttribute::TYPE_DATETIME) + ->where('upserted_at_sequence', $attribute->upserted_at_sequence) + ->update([ + 'value_datetime' => $canonical, + ]); + + if ($changed !== 1) { + throw new DomainException('The selected datetime projection changed before repair.'); + } + } + + $this->projectionRole->projectRun($run); + $report['repaired'] = count($plan); + + return $report; + }); + } + + /** + * Read newest events first, retaining only selected scalar evidence rather + * than a run's entire history. Physical event order identifies the latest + * value; payload.sequence identifies its original workflow command. + * + * @param list $keys + * @return array + */ + private function latestValues(WorkflowRun $run, array $keys): array + { + $wanted = array_fill_keys($keys, true); + $found = []; + + if ($wanted === []) { + return $found; + } + + $events = $run->historyEvents() + ->where('event_type', HistoryEventType::SearchAttributesUpserted->value) + ->lazyByIdDesc(10, 'sequence'); + + foreach ($events as $event) { + $payload = $event->payload; + + if (! is_array($payload) || ! is_array($payload['attributes'] ?? null)) { + continue; + } + + $types = is_array($payload['attribute_types'] ?? null) ? $payload['attribute_types'] : []; + + foreach ($payload['attributes'] as $key => $value) { + if (! isset($wanted[$key])) { + continue; + } + + $found[$key] = [ + 'sequence' => $payload['sequence'] ?? null, + 'type' => $types[$key] ?? null, + 'value' => $this->retainedValue($value), + ]; + unset($wanted[$key]); + } + + if ($wanted === []) { + break; + } + } + + return $found; + } + + private function retainedValue(mixed $value): int|string|null + { + if (is_int($value)) { + return $value; + } + + if (! is_string($value) || strlen($value) > WorkflowSearchAttribute::MAX_STRING_LENGTH) { + return null; + } + + return $value; + } + + /** + * @param array{sequence: mixed, type: mixed, value: mixed}|null $evidence + */ + private function canonicalValue(WorkflowSearchAttribute $attribute, ?array $evidence): string + { + if ($evidence === null || ! is_int($evidence['sequence']) + || $evidence['sequence'] !== $attribute->upserted_at_sequence + || $evidence['type'] !== WorkflowSearchAttribute::TYPE_DATETIME) { + throw new DomainException(sprintf( + 'Attribute [%s] has no matching latest typed history evidence.', + $attribute->key, + )); + } + + $original = $this->absoluteDatetime($evidence['value']); + + if ($original === null) { + throw new DomainException(sprintf( + 'Attribute [%s] lacks a retained absolute value with an unambiguous timezone.', + $attribute->key, + )); + } + + return UtcScheduleTimestamp::databaseValue($original); + } + + private function absoluteDatetime(mixed $value): ?DateTimeImmutable + { + if (is_int($value)) { + return new DateTimeImmutable('@' . $value); + } + + if (! is_string($value)) { + return null; + } + + $parsed = date_parse($value); + + if ($parsed['warning_count'] !== 0 || $parsed['error_count'] !== 0 || isset($parsed['relative'])) { + return null; + } + + foreach (['year', 'month', 'day', 'hour', 'minute', 'second'] as $field) { + if (! is_int($parsed[$field])) { + return null; + } + } + + $numericOffset = ($parsed['zone_type'] ?? null) === 1; + $explicitUtc = in_array(strtoupper($parsed['tz_abbr'] ?? $parsed['tz_id'] ?? ''), ['UTC', 'GMT', 'Z'], true); + + if (! $numericOffset && ! $explicitUtc) { + return null; + } + + return new DateTimeImmutable($value); + } + + private function storedValue(WorkflowSearchAttribute $attribute): ?string + { + $raw = $attribute->getRawOriginal('value_datetime'); + + if (! is_string($raw) || preg_match('/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(?:\.\d{1,6})?$/D', $raw) !== 1) { + return null; + } + + try { + $value = new DateTimeImmutable($raw, new DateTimeZone('UTC')); + $errors = DateTimeImmutable::getLastErrors(); + + if ($errors !== false && ($errors['warning_count'] !== 0 || $errors['error_count'] !== 0)) { + return null; + } + + return $value->format('Y-m-d H:i:s.u'); + } catch (Throwable) { + return null; + } + } +} diff --git a/tests/Feature/V2/V2SearchAttributeDatetimeRepairTest.php b/tests/Feature/V2/V2SearchAttributeDatetimeRepairTest.php index a7f5055a0..31d96a89e 100644 --- a/tests/Feature/V2/V2SearchAttributeDatetimeRepairTest.php +++ b/tests/Feature/V2/V2SearchAttributeDatetimeRepairTest.php @@ -24,7 +24,9 @@ final class V2SearchAttributeDatetimeRepairTest extends TestCase #[DataProvider('offsetDatetimes')] public function testPreviewApplyAndRepeatedApplyPreserveBusinessState(string $original): void { - $run = $this->legacyRun(['Offset' => $original]); + $run = $this->legacyRun([ + 'Offset' => $original, + ]); $business = $this->businessSnapshot($run); $before = $this->attributes($run); @@ -35,17 +37,23 @@ public function testPreviewApplyAndRepeatedApplyPreserveBusinessState(string $or $this->assertSame(0, $preview['repaired']); $this->assertSame($before, $this->attributes($run)); - $this->assertSame(0, $this->command($run, ['--apply' => true])); + $this->assertSame(0, $this->command($run, [ + '--apply' => true, + ])); $applied = json_decode(Artisan::output(), true, flags: JSON_THROW_ON_ERROR); $this->assertSame(1, $applied['repaired']); $this->assertSame( '2026-10-08T10:34:56.123456Z', - $run->searchAttributes()->firstOrFail()->getValue()?->toISOString(), + $run->searchAttributes() + ->firstOrFail() + ->getValue()?->toISOString(), ); $this->assertSame($business, $this->businessSnapshot($run)); $repaired = $this->attributes($run); - $this->assertSame(0, $this->command($run, ['--apply' => true])); + $this->assertSame(0, $this->command($run, [ + '--apply' => true, + ])); $repeat = json_decode(Artisan::output(), true, flags: JSON_THROW_ON_ERROR); $this->assertSame(0, $repeat['repaired']); $this->assertSame(1, $repeat['already_correct']); @@ -73,29 +81,46 @@ public function testMissingTimezoneRefusesTheRunWithoutGuessingOrPartialWrites() $before = $this->attributes($run); $business = $this->businessSnapshot($run); - $this->assertSame(1, $this->command($run, ['--apply' => true])); + $this->assertSame(1, $this->command($run, [ + '--apply' => true, + ])); $this->assertSame($before, $this->attributes($run)); $this->assertSame($business, $this->businessSnapshot($run)); - $this->assertSame(0, $this->command($run, ['--apply' => true, '--key' => ['Recoverable']])); - $rows = $run->searchAttributes()->get()->keyBy('key'); + $this->assertSame(0, $this->command($run, [ + '--apply' => true, + '--key' => ['Recoverable'], + ])); + $rows = $run->searchAttributes() + ->get() + ->keyBy('key'); $this->assertSame('2026-10-08 10:34:56.123456', $rows['Recoverable']->getRawOriginal('value_datetime')); $this->assertSame('2026-10-08 12:34:56.123456', $rows['Unknown']->getRawOriginal('value_datetime')); } public function testLaterCanonicalUpsertFencesAStaleProjection(): void { - $run = $this->legacyRun(['Offset' => '2026-10-08T12:34:56.123456+02:00']); + $run = $this->legacyRun([ + 'Offset' => '2026-10-08T12:34:56.123456+02:00', + ]); WorkflowHistoryEvent::record($run, HistoryEventType::SearchAttributesUpserted, [ 'sequence' => 8, - 'attributes' => ['Offset' => '2026-10-09T10:00:00Z'], - 'attribute_types' => ['Offset' => WorkflowSearchAttribute::TYPE_DATETIME], - 'merged' => ['Offset' => '2026-10-09T10:00:00Z'], + 'attributes' => [ + 'Offset' => '2026-10-09T10:00:00Z', + ], + 'attribute_types' => [ + 'Offset' => WorkflowSearchAttribute::TYPE_DATETIME, + ], + 'merged' => [ + 'Offset' => '2026-10-09T10:00:00Z', + ], ]); $before = $this->attributes($run); $business = $this->businessSnapshot($run); - $this->assertSame(1, $this->command($run, ['--apply' => true])); + $this->assertSame(1, $this->command($run, [ + '--apply' => true, + ])); $this->assertSame($before, $this->attributes($run)); $this->assertSame($business, $this->businessSnapshot($run)); } @@ -109,24 +134,37 @@ public function testProjectionFailureRollsBackEveryAttributeWrite(): void $before = $this->attributes($run); $business = $this->businessSnapshot($run); $this->mock(HistoryProjectionMaintenanceRole::class, static function (MockInterface $mock): void { - $mock->shouldReceive('projectRun')->once()->andThrow(new RuntimeException('Projection write failed')); + $mock->shouldReceive('projectRun') + ->once() + ->andThrow(new RuntimeException('Projection write failed')); }); - $this->assertSame(1, $this->command($run, ['--apply' => true])); + $this->assertSame(1, $this->command($run, [ + '--apply' => true, + ])); $this->assertSame($before, $this->attributes($run)); $this->assertSame($business, $this->businessSnapshot($run)); } public function testNamespaceAndPrunedHistoryRefusalsDoNotMutateRows(): void { - $run = $this->legacyRun(['Offset' => '2026-10-08T12:34:56.123456+02:00']); + $run = $this->legacyRun([ + 'Offset' => '2026-10-08T12:34:56.123456+02:00', + ]); $before = $this->attributes($run); - $this->assertSame(1, $this->command($run, ['--namespace' => 'other', '--apply' => true])); + $this->assertSame(1, $this->command($run, [ + '--namespace' => 'other', + '--apply' => true, + ])); $this->assertSame($before, $this->attributes($run)); - $run->forceFill(['details_pruned_at' => now()])->save(); + $run->forceFill([ + 'details_pruned_at' => now(), + ])->save(); $business = $this->businessSnapshot($run); - $this->assertSame(1, $this->command($run, ['--apply' => true])); + $this->assertSame(1, $this->command($run, [ + '--apply' => true, + ])); $this->assertSame($before, $this->attributes($run)); $this->assertSame($business, $this->businessSnapshot($run)); } @@ -152,9 +190,12 @@ private function legacyRun(array $originals): WorkflowRun $attribute->setTypedValue($original, WorkflowSearchAttribute::TYPE_DATETIME); $attribute->save(); // Reproduce the original wall-clock SQL representation without altering history. - $attribute->getConnection()->table($attribute->getTable()) + $attribute->getConnection() + ->table($attribute->getTable()) ->where('id', $attribute->id) - ->update(['value_datetime' => (new DateTimeImmutable($original))->format('Y-m-d H:i:s.u')]); + ->update([ + 'value_datetime' => (new DateTimeImmutable($original))->format('Y-m-d H:i:s.u'), + ]); } WorkflowHistoryEvent::record($run, HistoryEventType::SearchAttributesUpserted, [ @@ -184,9 +225,13 @@ private function command(WorkflowRun $run, array $options = []): int */ private function attributes(WorkflowRun $run): array { - return $run->getConnection()->table('workflow_search_attributes') - ->where('workflow_run_id', $run->id)->orderBy('key')->get() - ->map(static fn (object $row): array => (array) $row)->all(); + return $run->getConnection() + ->table('workflow_search_attributes') + ->where('workflow_run_id', $run->id) + ->orderBy('key') + ->get() + ->map(static fn (object $row): array => (array) $row) + ->all(); } /** @@ -199,7 +244,10 @@ private function businessSnapshot(WorkflowRun $run): array foreach (['workflow_runs', 'workflow_tasks', 'workflow_history_events', 'workflow_commands'] as $table) { $snapshot[$table] = $run->getConnection()->table($table) ->where($table === 'workflow_runs' ? 'id' : 'workflow_run_id', $run->id) - ->orderBy('id')->get()->map(static fn (object $row): array => (array) $row)->all(); + ->orderBy('id') + ->get() + ->map(static fn (object $row): array => (array) $row) + ->all(); } return $snapshot; From db050e63ba6fbdffaff99dfcd8a6bd61e2ccf38a Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 14:16:22 +0000 Subject: [PATCH 08/11] docs: explain explicit datetime projection recovery --- CHANGELOG.md | 7 +++ .../projection-timestamp-storage.md | 51 +++++++++++++++++++ 2 files changed, 58 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5cc3ad333..c385e4f70 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## Unreleased + +- Preserve UTC instants and microseconds in datetime search attributes and + embedded visibility predicates, including values with non-UTC offsets. +- Add `workflow:v2:repair-search-datetimes` to preview or explicitly repair + selected datetime projections from matching retained, typed history. + ## 2.11.4 - 2026-10-23 - Preserve whole-valued floats and signed zero in CLI history-export JSON, diff --git a/docs/architecture/projection-timestamp-storage.md b/docs/architecture/projection-timestamp-storage.md index 102ddf1b4..48f0fd5c4 100644 --- a/docs/architecture/projection-timestamp-storage.md +++ b/docs/architecture/projection-timestamp-storage.md @@ -23,3 +23,54 @@ views show the intended instants and that operator projection warnings clear. The rebuild changes derived rows only. It cannot recover an offset already lost from authoritative history; investigate such rows under the history timestamp procedure before interpreting a rebuilt projection as corrected. + +## Datetime search attributes + +Datetime search attributes also store UTC instants with microsecond precision. +An input such as `2026-10-08T12:34:56.123456+02:00` represents +`2026-10-08T10:34:56.123456Z`. Offset conversion must happen before writing the +timezone-free column and before binding a visibility predicate. Hydration uses +UTC even when the application's default timezone differs. + +Earlier writers could store the input's local wall time while retaining its +offset in the `SearchAttributesUpserted` history event. Upgrading the writer +prevents new errors but does not identify or rewrite those existing rows. +Never infer an original offset from a stored wall time. + +The unreleased `workflow:v2:repair-search-datetimes` command derives a repair +from the latest retained, explicitly typed upsert for each selected key. Back +up the database and upgrade every writer before inspecting affected runs: + +```sh +php artisan workflow:v2:repair-search-datetimes RUN_ID --namespace=production --json +php artisan workflow:v2:repair-search-datetimes RUN_ID --namespace=production --key=SubmittedAt --json +``` + +The default is a preview. `would_repair` reports the number of selected rows +whose stored value differs from their history-backed UTC value. Review the +selected run and keys, then explicitly apply the same selection: + +```sh +php artisan workflow:v2:repair-search-datetimes RUN_ID --namespace=production --key=SubmittedAt --apply --json +``` + +The command locks the selected run and attributes, validates all selected keys +before writing, and updates the run summary in the same storage transaction. +It preserves history, business commands, run/task identities and leases. A +repeat apply reports zero repairs when the selected values already match. +The command operates on one exact run. `--namespace` adds an equality guard; +omit it only when the exact run ID is sufficient, including legacy null +namespaces. Repeated `--key` options restrict the operation to those keys. + +Pruned history, missing or stale typed evidence, and timestamps without an +unambiguous timezone are refused. A selected key must still be a datetime +attribute with the matching original upsert sequence. One unprovable key +refuses the entire selection. Inspect the original history and select only +proven keys rather than guessing a timezone. An unexpected storage error is +reported as `unverified`; inspect the rows before retrying because the command +cannot promise the outcome of an uncertain database commit. + +This command repairs derived search attributes and their summary. It does not +restore pruned history, change authoritative input, or correct the separate +Server visibility-query binding path. That path requires its corresponding +Server fix. From 1667dd18381577bc5936075877ec09ff6e420173 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 14:38:14 +0000 Subject: [PATCH 09/11] test: retain datetime projection recovery in the history corpus --- .../V2EmbeddedReplayRegressionCorpusTest.php | 51 +++++++++++++++++++ ...time-search-projection-history-repair.json | 37 ++++++++++++++ 2 files changed, 88 insertions(+) create mode 100644 tests/Fixtures/V2/ReplayRegression/datetime-search-projection-history-repair.json diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index a1544bfe0..d1544ba03 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -4,6 +4,7 @@ namespace Tests\Feature\V2; +use DateTimeImmutable; use Illuminate\Queue\WorkerOptions; use Illuminate\Support\Carbon; use Illuminate\Support\Facades\Artisan; @@ -228,6 +229,11 @@ public function testFixturesExecuteThroughDeclaredReplayConsumers(): void continue; } + if ($fixture['id'] === 'datetime-search-projection-history-repair') { + $this->assertDatetimeSearchProjectionRepairPreservesItsOriginal($fixture); + continue; + } + if ($fixture['id'] === 'repeated-version-legacy-clock-boundaries') { $this->assertRepeatedVersionClockBoundaries($fixture); continue; @@ -1017,6 +1023,51 @@ private function assertSignalResumedMixedGroupCommandSequenceFixture(array $fixt $this->assertCount(3, $scheduled['created_task_ids']); } + /** + * @param array $fixture + */ + private function assertDatetimeSearchProjectionRepairPreservesItsOriginal(array $fixture): void + { + $run = $this->createRunFromFixture($fixture); + $payload = $fixture['history'][1]['payload']; + $original = $payload['attributes']['SubmittedAt']; + $attribute = new WorkflowSearchAttribute([ + 'workflow_run_id' => $run->id, + 'workflow_instance_id' => $run->workflow_instance_id, + 'key' => 'SubmittedAt', + 'upserted_at_sequence' => $payload['sequence'], + ]); + $attribute->setTypedValue($original, WorkflowSearchAttribute::TYPE_DATETIME); + $attribute->save(); + $connection = $run->getConnection(); + $connection->table('workflow_search_attributes')->where('id', $attribute->id)->update([ + 'value_datetime' => (new DateTimeImmutable($original))->format('Y-m-d H:i:s.u'), + ]); + $history = $run->historyEvents()->orderBy('sequence')->get()->toArray(); + $tasks = $connection->table('workflow_tasks')->where('workflow_run_id', $run->id)->get()->all(); + $options = [ + 'run' => $run->id, + '--namespace' => $run->namespace, + '--json' => true, + ]; + + $this->assertSame(0, Artisan::call('workflow:v2:repair-search-datetimes', $options)); + $preview = json_decode(Artisan::output(), true, flags: JSON_THROW_ON_ERROR); + $this->assertSame(1, $preview['would_repair']); + $this->assertSame(0, $preview['repaired']); + $this->assertSame(0, Artisan::call('workflow:v2:repair-search-datetimes', [ + ...$options, + '--apply' => true, + ])); + $this->assertSame( + '2026-10-08T10:34:56.123456Z', + $attribute->fresh()->getValue()?->toISOString(), + ); + $this->assertSame($history, $run->historyEvents()->orderBy('sequence')->get()->toArray()); + $tasksAfter = $connection->table('workflow_tasks')->where('workflow_run_id', $run->id)->get()->all(); + $this->assertEquals($tasks, $tasksAfter); + } + /** * @param array $fixture */ diff --git a/tests/Fixtures/V2/ReplayRegression/datetime-search-projection-history-repair.json b/tests/Fixtures/V2/ReplayRegression/datetime-search-projection-history-repair.json new file mode 100644 index 000000000..ded0fe954 --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/datetime-search-projection-history-repair.json @@ -0,0 +1,37 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "datetime-search-projection-history-repair", + "protocol_version": "1.0", + "bindings": ["php"], + "consumers": ["workflow-fiber-runner", "workflow-executor"], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestGoldenReplayWorkflow", + "arguments": ["single-activity"], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-10-08T10:34:50.000000Z" + }, + { + "sequence": 2, + "event_type": "SearchAttributesUpserted", + "payload": { + "sequence": 7, + "attributes": {"SubmittedAt": "2026-10-09T00:34:56.123456+14:00"}, + "attribute_types": {"SubmittedAt": "datetime"}, + "merged": {"SubmittedAt": "2026-10-09T00:34:56.123456+14:00"} + }, + "recorded_at": "2026-10-08T10:34:56.123456Z" + } + ], + "expected": { + "completed": false, + "result": null, + "commands": [{"type": "schedule_activity"}] + } +} From e70766646c2904993d7c132646393c4f71e31f73 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 14:45:12 +0000 Subject: [PATCH 10/11] refactor: separate datetime repair selection and application --- .../V2RepairSearchDatetimesCommand.php | 12 +- .../Support/SearchAttributeDatetimeRepair.php | 149 ++++++++++-------- .../V2EmbeddedReplayRegressionCorpusTest.php | 28 ++-- 3 files changed, 112 insertions(+), 77 deletions(-) diff --git a/src/Commands/V2RepairSearchDatetimesCommand.php b/src/Commands/V2RepairSearchDatetimesCommand.php index 0a39838ed..f9c2bcfc9 100644 --- a/src/Commands/V2RepairSearchDatetimesCommand.php +++ b/src/Commands/V2RepairSearchDatetimesCommand.php @@ -26,6 +26,8 @@ final class V2RepairSearchDatetimesCommand extends Command public function handle(SearchAttributeDatetimeRepair $repair): int { + $status = self::SUCCESS; + try { $model = ConfiguredV2Models::resolve('run_model', WorkflowRun::class); $query = $model::query()->whereKey((string) $this->argument('run')); @@ -62,18 +64,18 @@ public function handle(SearchAttributeDatetimeRepair $repair): int $report['repaired'], $report['already_correct'], ]]); } - - return self::SUCCESS; } catch (ModelNotFoundException) { - return $this->failure('The selected run was not found in the requested namespace.', 'refused'); + $status = $this->failure('The selected run was not found in the requested namespace.', 'refused'); } catch (DomainException|InvalidArgumentException $exception) { - return $this->failure($exception->getMessage(), 'refused'); + $status = $this->failure($exception->getMessage(), 'refused'); } catch (Throwable) { - return $this->failure( + $status = $this->failure( 'Repair could not be verified. Inspect the selected projections before retrying.', 'unverified', ); } + + return $status; } private function failure(string $message, string $status): int diff --git a/src/V2/Support/SearchAttributeDatetimeRepair.php b/src/V2/Support/SearchAttributeDatetimeRepair.php index 5051107b2..6b222a51a 100644 --- a/src/V2/Support/SearchAttributeDatetimeRepair.php +++ b/src/V2/Support/SearchAttributeDatetimeRepair.php @@ -7,6 +7,7 @@ use DateTimeImmutable; use DateTimeZone; use DomainException; +use Illuminate\Database\Eloquent\Collection; use Throwable; use Workflow\V2\Contracts\HistoryProjectionMaintenanceRole; use Workflow\V2\Enums\HistoryEventType; @@ -28,39 +29,10 @@ public function repair(WorkflowRun $selectedRun, bool $apply = false, array $key { return $selectedRun->getConnection() ->transaction(function () use ($selectedRun, $apply, $keys): array { - $query = $selectedRun->newQuery() - ->whereKey($selectedRun->id) - ->where('namespace', $selectedRun->namespace); - - if ($apply) { - $query->lockForUpdate(); - } - - $run = $query->firstOrFail(); - - if ($run->details_pruned_at !== null) { - throw new DomainException('Retained history is unavailable for this pruned run.'); - } - - $attributeQuery = $run->searchAttributes() - ->where('type', WorkflowSearchAttribute::TYPE_DATETIME); - - if ($keys !== []) { - $attributeQuery->whereIn('key', $keys); - } - - if ($apply) { - $attributeQuery->lockForUpdate(); - } - - $attributes = $attributeQuery->orderBy('key') - ->get(); + $run = $this->selectedRun($selectedRun, $apply); + $attributes = $this->selectedAttributes($run, $keys, $apply); $selectedKeys = array_map('strval', $attributes->pluck('key')->all()); - if (array_diff($keys, $selectedKeys) !== []) { - throw new DomainException('A selected key is not a retained datetime attribute on this run.'); - } - $latest = $this->latestValues($run, $selectedKeys); $plan = []; $alreadyCorrect = 0; @@ -91,29 +63,84 @@ public function repair(WorkflowRun $selectedRun, bool $apply = false, array $key return $report; } - foreach ($plan as [$attribute, $canonical]) { - $changed = $attribute->getConnection() - ->table($attribute->getTable()) - ->where('id', $attribute->id) - ->where('workflow_run_id', $run->id) - ->where('type', WorkflowSearchAttribute::TYPE_DATETIME) - ->where('upserted_at_sequence', $attribute->upserted_at_sequence) - ->update([ - 'value_datetime' => $canonical, - ]); - - if ($changed !== 1) { - throw new DomainException('The selected datetime projection changed before repair.'); - } - } - - $this->projectionRole->projectRun($run); + $this->applyPlan($run, $plan); $report['repaired'] = count($plan); return $report; }); } + private function selectedRun(WorkflowRun $selectedRun, bool $lock): WorkflowRun + { + $query = $selectedRun->newQuery() + ->whereKey($selectedRun->id) + ->where('namespace', $selectedRun->namespace); + + if ($lock) { + $query->lockForUpdate(); + } + + $run = $query->firstOrFail(); + + if ($run->details_pruned_at !== null) { + throw new DomainException('Retained history is unavailable for this pruned run.'); + } + + return $run; + } + + /** + * @param list $keys + * @return Collection + */ + private function selectedAttributes(WorkflowRun $run, array $keys, bool $lock): Collection + { + $query = $run->searchAttributes() + ->where('type', WorkflowSearchAttribute::TYPE_DATETIME); + + if ($keys !== []) { + $query->whereIn('key', $keys); + } + + if ($lock) { + $query->lockForUpdate(); + } + + $attributes = $query->orderBy('key') + ->get(); + $selectedKeys = array_map('strval', $attributes->pluck('key')->all()); + + if (array_diff($keys, $selectedKeys) !== []) { + throw new DomainException('A selected key is not a retained datetime attribute on this run.'); + } + + return $attributes; + } + + /** + * @param list $plan + */ + private function applyPlan(WorkflowRun $run, array $plan): void + { + foreach ($plan as [$attribute, $canonical]) { + $changed = $attribute->getConnection() + ->table($attribute->getTable()) + ->where('id', $attribute->id) + ->where('workflow_run_id', $run->id) + ->where('type', WorkflowSearchAttribute::TYPE_DATETIME) + ->where('upserted_at_sequence', $attribute->upserted_at_sequence) + ->update([ + 'value_datetime' => $canonical, + ]); + + if ($changed !== 1) { + throw new DomainException('The selected datetime projection changed before repair.'); + } + } + + $this->projectionRole->projectRun($run); + } + /** * Read newest events first, retaining only selected scalar evidence rather * than a run's entire history. Physical event order identifies the latest @@ -210,30 +237,31 @@ private function absoluteDatetime(mixed $value): ?DateTimeImmutable return new DateTimeImmutable('@' . $value); } - if (! is_string($value)) { - return null; + if (is_string($value) && $this->hasAbsoluteDateParts($value)) { + return new DateTimeImmutable($value); } + return null; + } + + private function hasAbsoluteDateParts(string $value): bool + { $parsed = date_parse($value); if ($parsed['warning_count'] !== 0 || $parsed['error_count'] !== 0 || isset($parsed['relative'])) { - return null; + return false; } foreach (['year', 'month', 'day', 'hour', 'minute', 'second'] as $field) { if (! is_int($parsed[$field])) { - return null; + return false; } } $numericOffset = ($parsed['zone_type'] ?? null) === 1; $explicitUtc = in_array(strtoupper($parsed['tz_abbr'] ?? $parsed['tz_id'] ?? ''), ['UTC', 'GMT', 'Z'], true); - if (! $numericOffset && ! $explicitUtc) { - return null; - } - - return new DateTimeImmutable($value); + return $numericOffset || $explicitUtc; } private function storedValue(WorkflowSearchAttribute $attribute): ?string @@ -248,11 +276,8 @@ private function storedValue(WorkflowSearchAttribute $attribute): ?string $value = new DateTimeImmutable($raw, new DateTimeZone('UTC')); $errors = DateTimeImmutable::getLastErrors(); - if ($errors !== false && ($errors['warning_count'] !== 0 || $errors['error_count'] !== 0)) { - return null; - } - - return $value->format('Y-m-d H:i:s.u'); + return $errors !== false && ($errors['warning_count'] !== 0 || $errors['error_count'] !== 0) + ? null : $value->format('Y-m-d H:i:s.u'); } catch (Throwable) { return null; } diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index d1544ba03..93749b3da 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -1040,11 +1040,19 @@ private function assertDatetimeSearchProjectionRepairPreservesItsOriginal(array $attribute->setTypedValue($original, WorkflowSearchAttribute::TYPE_DATETIME); $attribute->save(); $connection = $run->getConnection(); - $connection->table('workflow_search_attributes')->where('id', $attribute->id)->update([ - 'value_datetime' => (new DateTimeImmutable($original))->format('Y-m-d H:i:s.u'), - ]); - $history = $run->historyEvents()->orderBy('sequence')->get()->toArray(); - $tasks = $connection->table('workflow_tasks')->where('workflow_run_id', $run->id)->get()->all(); + $connection->table('workflow_search_attributes') + ->where('id', $attribute->id) + ->update([ + 'value_datetime' => (new DateTimeImmutable($original))->format('Y-m-d H:i:s.u'), + ]); + $history = $run->historyEvents() + ->orderBy('sequence') + ->get() + ->toArray(); + $tasks = $connection->table('workflow_tasks') + ->where('workflow_run_id', $run->id) + ->get() + ->all(); $options = [ 'run' => $run->id, '--namespace' => $run->namespace, @@ -1059,12 +1067,12 @@ private function assertDatetimeSearchProjectionRepairPreservesItsOriginal(array ...$options, '--apply' => true, ])); - $this->assertSame( - '2026-10-08T10:34:56.123456Z', - $attribute->fresh()->getValue()?->toISOString(), - ); + $this->assertSame('2026-10-08T10:34:56.123456Z', $attribute->fresh() ->getValue()?->toISOString()); $this->assertSame($history, $run->historyEvents()->orderBy('sequence')->get()->toArray()); - $tasksAfter = $connection->table('workflow_tasks')->where('workflow_run_id', $run->id)->get()->all(); + $tasksAfter = $connection->table('workflow_tasks') + ->where('workflow_run_id', $run->id) + ->get() + ->all(); $this->assertEquals($tasks, $tasksAfter); } From 6e53b067c2f36ed86bd0a19219364d2ccbc1be52 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 15:16:16 +0000 Subject: [PATCH 11/11] release: prepare Workflow 2.12.0 datetime recovery --- CHANGELOG.md | 2 +- composer.json | 2 +- docs/architecture/projection-timestamp-storage.md | 3 ++- 3 files changed, 4 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c385e4f70..dc585b400 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ # Changelog -## Unreleased +## 2.12.0 - 2026-10-23 - Preserve UTC instants and microseconds in datetime search attributes and embedded visibility predicates, including values with non-UTC offsets. diff --git a/composer.json b/composer.json index 9b73cbf5f..dd72463f3 100644 --- a/composer.json +++ b/composer.json @@ -87,7 +87,7 @@ "dev-main": "2.0.x-dev" }, "durable-workflow": { - "product-train": "2.11.4", + "product-train": "2.12.0", "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json", "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json" }, diff --git a/docs/architecture/projection-timestamp-storage.md b/docs/architecture/projection-timestamp-storage.md index 48f0fd5c4..1edc2c66d 100644 --- a/docs/architecture/projection-timestamp-storage.md +++ b/docs/architecture/projection-timestamp-storage.md @@ -37,7 +37,8 @@ offset in the `SearchAttributesUpserted` history event. Upgrading the writer prevents new errors but does not identify or rewrite those existing rows. Never infer an original offset from a stored wall time. -The unreleased `workflow:v2:repair-search-datetimes` command derives a repair +The `workflow:v2:repair-search-datetimes` command, added in Workflow 2.12.0, +derives a repair from the latest retained, explicitly typed upsert for each selected key. Back up the database and upgrade every writer before inspecting affected runs: