From 89ced9fc5f7149365c6a930119a12818430227a9 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 21:45:19 +0000 Subject: [PATCH 01/13] Preserve fractional persisted replay clocks in redrive --- CHANGELOG.md | 4 + .../Support/DefaultWorkflowControlPlane.php | 2 +- src/V2/Support/FailedRunRedrivePlan.php | 12 +- src/V2/Support/RedriveReplayClock.php | 10 ++ .../V2EmbeddedReplayRegressionCorpusTest.php | 99 +++++++++++++++++ .../redrive-reused-fractional-clock.json | 104 ++++++++++++++++++ .../V2/TestRedriveFractionalClockWorkflow.php | 46 ++++++++ 7 files changed, 270 insertions(+), 7 deletions(-) create mode 100644 tests/Fixtures/V2/ReplayRegression/redrive-reused-fractional-clock.json create mode 100644 tests/Fixtures/V2/TestRedriveFractionalClockWorkflow.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 8706e54be..49935f328 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## Unreleased +- Failed-run redrive preserves fractional seconds in replay clocks for the + original run start and reused steps. Repeated redrive keeps previously + recorded clock strings unchanged. + - Failed-run redrive accepts a verified linear activity retry chain. A retried completed prefix is reused and only the failed boundary step runs again. Substituted attempt/task identities refuse with diff --git a/src/V2/Support/DefaultWorkflowControlPlane.php b/src/V2/Support/DefaultWorkflowControlPlane.php index 3d9f160d9..c2ff14488 100644 --- a/src/V2/Support/DefaultWorkflowControlPlane.php +++ b/src/V2/Support/DefaultWorkflowControlPlane.php @@ -268,7 +268,7 @@ public function redrive(string $instanceId, string $failedRunId, array $options ]), 'replayed_started_at' => is_string($startedPayload['replayed_started_at'] ?? null) ? $startedPayload['replayed_started_at'] - : $source->started_at?->toIso8601String(), + : RedriveReplayClock::formatRecordedAt($source->started_at), 'workflow_link_id' => $link->id, ], ), null, $command); diff --git a/src/V2/Support/FailedRunRedrivePlan.php b/src/V2/Support/FailedRunRedrivePlan.php index 50ee2a3fc..140c05fa2 100644 --- a/src/V2/Support/FailedRunRedrivePlan.php +++ b/src/V2/Support/FailedRunRedrivePlan.php @@ -343,7 +343,7 @@ public static function forRun(WorkflowRun $run): array $reusedSteps[] = [ 'event_type' => $marker->event_type, 'payload' => $markerPayload, - 'recorded_at' => $marker->recorded_at?->toIso8601String(), + 'recorded_at' => RedriveReplayClock::formatRecordedAt($marker->recorded_at), ]; continue; @@ -462,7 +462,7 @@ public static function forRun(WorkflowRun $run): array $reusedSteps[] = [ 'event_type' => HistoryEventType::ActivityCompleted, 'payload' => $payload, - 'recorded_at' => $completion->recorded_at?->toIso8601String(), + 'recorded_at' => RedriveReplayClock::formatRecordedAt($completion->recorded_at), ]; } @@ -914,7 +914,7 @@ private static function updateBundles( $bundleEvents[] = [ 'event_type' => $event->event_type, 'payload' => $eventPayload, - 'recorded_at' => $event->recorded_at?->toIso8601String(), + 'recorded_at' => RedriveReplayClock::formatRecordedAt($event->recorded_at), ]; } @@ -1031,7 +1031,7 @@ private static function firedTimerStep(array $events): ?array return [ 'event_type' => HistoryEventType::TimerFired, 'payload' => $firedPayload, - 'recorded_at' => $fired->recorded_at?->toIso8601String(), + 'recorded_at' => RedriveReplayClock::formatRecordedAt($fired->recorded_at), 'scheduled_payload' => $scheduledPayload, ]; } @@ -1182,7 +1182,7 @@ private static function signalSteps(array $events, WorkflowRun $run, array $prov $signalEvents[] = [ 'event_type' => $event->event_type, 'payload' => $eventPayload, - 'recorded_at' => $event->recorded_at?->toIso8601String(), + 'recorded_at' => RedriveReplayClock::formatRecordedAt($event->recorded_at), ]; } @@ -1325,7 +1325,7 @@ private static function completedChildStep(array $events, WorkflowRun $run, arra $childEvents[] = [ 'event_type' => $event->event_type, 'payload' => $eventPayload, - 'recorded_at' => $event->recorded_at?->toIso8601String(), + 'recorded_at' => RedriveReplayClock::formatRecordedAt($event->recorded_at), ]; } diff --git a/src/V2/Support/RedriveReplayClock.php b/src/V2/Support/RedriveReplayClock.php index ef6cc5767..bffab553b 100644 --- a/src/V2/Support/RedriveReplayClock.php +++ b/src/V2/Support/RedriveReplayClock.php @@ -12,6 +12,16 @@ final class RedriveReplayClock { + public static function formatRecordedAt(?CarbonInterface $time): ?string + { + if ($time === null) { + return null; + } + return $time->format('u') === '000000' + ? $time->toIso8601String() + : $time->format('Y-m-d\TH:i:s.uP'); + } + public static function startedAt(WorkflowRun $run): ?CarbonInterface { if ($run->run_number <= 1) { diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 615b3faaf..64a603133 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -23,6 +23,7 @@ use Tests\Fixtures\V2\TestLocalActivityWorkflow; use Tests\Fixtures\V2\TestParallelGreetingChildrenWorkflow; use Tests\Fixtures\V2\TestQueryReplayGuardActivity; +use Tests\Fixtures\V2\TestRedriveFractionalClockWorkflow; use Tests\Fixtures\V2\TestRepeatedPatchClockWorkflow; use Tests\Fixtures\V2\TestSinglePatchReplayWorkflow; use Tests\Fixtures\V2\TestTimerWorkflow; @@ -1616,6 +1617,104 @@ public function testRedriveTypedLineageCorpusLinksBothRunsFromHistory(): void } } + public function testFractionalRedriveClocksSurviveRepeatedNativeAdmissionAndColdReplay(): void + { + $path = self::FIXTURE_DIR . '/redrive-reused-fractional-clock.json'; + if (! is_file($path)) { + $this->markTestSkipped('The fractional redrive clock evidence is absent.'); + } + $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + $scenario = $fixture['redrive_fractional_clock']; + self::stopWorkers(); + config([ + 'queue.default' => 'sync', + 'queue.connections.sync.driver' => 'sync', + 'workflows.v2.testing.redrive_failures_before_success' => $scenario['redrives'], + ]); + Cache::forget('test:redrive:first-calls'); + Cache::forget('test:redrive:second-calls'); + $this->clearWorkflowState(); + WorkflowStub::fake(); + Carbon::setTestNow(Carbon::parse($scenario['source_times']['started_at'])); + $stub = WorkflowStub::make($fixture['workflow']['type'], 'regression-corpus-fractional-clock'); + $stub->start(...$fixture['workflow']['arguments']); + Carbon::setTestNow(Carbon::parse($scenario['source_times']['timer_fired_at'])); + $timer = WorkflowTask::query() + ->where('workflow_run_id', $stub->runId()) + ->where('task_type', TaskType::Timer->value) + ->sole(); + $this->app->call([new RunTimerTask($timer->id), 'handle']); + $source = WorkflowRun::query()->findOrFail($stub->runId()); + $this->assertSame(RunStatus::Failed, $source->status); + // Give the consumed activity a distinct persisted instant. The real + // admission, timer and failure lifecycle ran before this clock mutation. + WorkflowHistoryEvent::query() + ->where('workflow_run_id', $source->id) + ->where('event_type', HistoryEventType::ActivityCompleted) + ->sole() + ->forceFill([ + 'recorded_at' => Carbon::parse($scenario['source_times']['activity_completed_at']), + ])->save(); + $originalHistory = $source->historyEvents()->get()->toArray(); + $this->assertSame( + $scenario['expected_times'], + (new QueryStateReplayer())->query($source->fresh(), 'recordedTimes', []), + ); + $predecessor = $source; + $firstSerializedClocks = null; + for ($attempt = 1; $attempt <= $scenario['redrives']; ++$attempt) { + Carbon::setTestNow(now()->addMinutes($scenario['redrive_after_minutes'])); + $result = app(WorkflowControlPlane::class)->redrive($source->workflow_instance_id, $predecessor->id); + $this->assertTrue($result['accepted'], (string) $result['reason']); + $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']); + $this->assertSame( + $attempt === $scenario['redrives'] ? RunStatus::Completed : RunStatus::Failed, + $successor->status, + ); + $this->assertSame( + $scenario['expected_times'], + (new QueryStateReplayer())->query($successor->fresh(), 'recordedTimes', []), + ); + $replayed = (new WorkflowReplayer())->replay($successor->fresh())->workflow; + $this->assertInstanceOf(TestRedriveFractionalClockWorkflow::class, $replayed); + $this->assertSame($scenario['expected_times'], $replayed->recordedTimes()); + $started = $successor->historyEvents() + ->where('event_type', HistoryEventType::WorkflowStarted) + ->sole(); + $activity = $successor->historyEvents() + ->where('event_type', HistoryEventType::ActivityCompleted) + ->orderBy('sequence') + ->firstOrFail(); + $fired = $successor->historyEvents() + ->where('event_type', HistoryEventType::TimerFired) + ->sole(); + $serializedClocks = [ + $started->payload['replayed_started_at'], + $activity->payload['reused_recorded_at'], + $fired->payload['reused_recorded_at'], + ]; + if ($firstSerializedClocks !== null) { + $this->assertSame($firstSerializedClocks, $serializedClocks); + } + $firstSerializedClocks ??= $serializedClocks; + $this->assertSame(0, WorkflowTimer::query() + ->where('workflow_run_id', $successor->id) + ->count()); + $this->assertSame(0, WorkflowTask::query() + ->where('workflow_run_id', $successor->id) + ->where('task_type', TaskType::Timer->value) + ->count()); + $this->assertSame($originalHistory, $source->historyEvents()->get()->toArray()); + $predecessor = $successor; + } + $output = $predecessor->workflowOutput(); + unset($output['result']); + $this->assertSame($scenario['expected_times'], $output); + $this->assertSame(1, (int) Cache::get('test:redrive:first-calls')); + $this->assertSame(3, (int) Cache::get('test:redrive:second-calls')); + Carbon::setTestNow(); + } + public function testRedriveReusedTimerCorpusDoesNotWaitAgain(): void { $path = self::FIXTURE_DIR . '/redrive-reused-timer-clock.json'; diff --git a/tests/Fixtures/V2/ReplayRegression/redrive-reused-fractional-clock.json b/tests/Fixtures/V2/ReplayRegression/redrive-reused-fractional-clock.json new file mode 100644 index 000000000..1ce4b4eaa --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/redrive-reused-fractional-clock.json @@ -0,0 +1,104 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-reused-fractional-clock", + "protocol_version": "1.0", + "bindings": [ + "php" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestRedriveFractionalClockWorkflow", + "arguments": [ + "Taylor", + 5 + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": { + "recovery_kind": "redrive", + "resume_step_sequence": 3, + "replayed_started_at": "2026-11-03T00:00:00.123456+00:00" + }, + "recorded_at": "2026-11-03T00:05:00+00:00" + }, + { + "sequence": 2, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro", + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00.234567+00:00" + }, + "recorded_at": "2026-11-03T00:05:00+00:00" + }, + { + "sequence": 3, + "event_type": "TimerScheduled", + "payload": { + "timer_id": "01JBREDRIVESUCCESSORTIMER0", + "sequence": 2, + "delay_seconds": 5, + "fire_at": "2026-11-03T00:00:05.000000Z" + }, + "recorded_at": "2026-11-03T00:05:00+00:00" + }, + { + "sequence": 4, + "event_type": "TimerFired", + "payload": { + "timer_id": "01JBREDRIVESUCCESSORTIMER0", + "sequence": 2, + "delay_seconds": 5, + "fired_at": "2026-11-03T00:00:06.000000Z", + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:06.345678+00:00" + }, + "recorded_at": "2026-11-03T00:05:00+00:00" + }, + { + "sequence": 5, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 3, + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity", + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro" + }, + "recorded_at": "2026-11-03T00:05:01.456789+00:00" + } + ], + "expected": { + "completed": true, + "result": { + "result": "Hello, Taylor!", + "started_at": "1793664000.123456", + "activity_completed_at": "1793664000.234567", + "timer_fired_at": "1793664006.345678" + }, + "commands": [] + }, + "consumers": [ + "workflow-fiber-runner" + ], + "redrive_fractional_clock": { + "source_times": { + "started_at": "2026-11-03T00:00:00.123456+00:00", + "activity_completed_at": "2026-11-03T00:00:00.234567+00:00", + "timer_fired_at": "2026-11-03T00:00:06.345678+00:00" + }, + "redrive_after_minutes": 5, + "redrives": 2, + "expected_times": { + "started_at": "1793664000.123456", + "activity_completed_at": "1793664000.234567", + "timer_fired_at": "1793664006.345678" + } + } +} diff --git a/tests/Fixtures/V2/TestRedriveFractionalClockWorkflow.php b/tests/Fixtures/V2/TestRedriveFractionalClockWorkflow.php new file mode 100644 index 000000000..15ca11c1e --- /dev/null +++ b/tests/Fixtures/V2/TestRedriveFractionalClockWorkflow.php @@ -0,0 +1,46 @@ + + */ + private array $recordedTimes = []; + + /** + * @return array + */ + public function handle(string $name, int $seconds): array + { + $this->recordedTimes['started_at'] = now()->format('U.u'); + $greeting = activity(TestRedriveFirstActivity::class, $name); + $this->recordedTimes['activity_completed_at'] = now()->format('U.u'); + timer($seconds); + $this->recordedTimes['timer_fired_at'] = now()->format('U.u'); + return [ + 'result' => activity(TestRedriveSecondActivity::class, $greeting), + ...$this->recordedTimes, + ]; + } + + /** + * @return array + */ + #[QueryMethod] + public function recordedTimes(): array + { + return $this->recordedTimes; + } +} From d8d57ffd2f733cc474116739f3f3baa600249617 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 21:52:59 +0000 Subject: [PATCH 02/13] Keep whole-second corpus coverage beside fractional clock controls --- .../V2EmbeddedReplayRegressionCorpusTest.php | 168 ++++++++++-------- .../redrive-reused-fractional-clock.json | 6 +- 2 files changed, 96 insertions(+), 78 deletions(-) diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 64a603133..342afb50d 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -552,7 +552,10 @@ public function testRedriveReusedMarkersCorpusKeepsTheirStepsAndTime(): void ->firstOrFail(); $this->assertSame($step['sequence'], $reused->payload['sequence']); $this->assertSame($sourceRunId, $reused->payload['reused_from_run_id'] ?? null); - $this->assertSame($sourceTimes[$step['sequence']], $reused->payload['reused_recorded_at'] ?? null); + $this->assertSame( + $sourceTimes[$step['sequence']], + Carbon::parse($reused->payload['reused_recorded_at'])->toIso8601String(), + ); } $output = $successor->workflowOutput(); $this->assertSame($scenario['version'], $output['version']); @@ -613,7 +616,10 @@ public function testRedriveReusedUpsertsCorpusKeepsTheirStepsRowsAndTime(): void ->sole(); $this->assertSame($step['sequence'], $reused->payload['sequence']); $this->assertSame($sourceRunId, $reused->payload['reused_from_run_id'] ?? null); - $this->assertSame($sourceTimes[$step['sequence']], $reused->payload['reused_recorded_at'] ?? null); + $this->assertSame( + $sourceTimes[$step['sequence']], + Carbon::parse($reused->payload['reused_recorded_at'])->toIso8601String(), + ); } $this->assertEquals($scenario['memo'], $successor->typedMemos()); $this->assertEquals($scenario['search_attributes'], $successor->typedSearchAttributes()); @@ -681,7 +687,10 @@ public function testRedriveReusedSignalWaitCorpusKeepsItsValueAndTime(): void ); foreach ($reused as $event) { $this->assertSame($sourceRunId, $event->payload['reused_from_run_id'] ?? null); - $this->assertSame($sourceTimes[$event->event_type->value], $event->payload['reused_recorded_at'] ?? null); + $this->assertSame( + $sourceTimes[$event->event_type->value], + Carbon::parse($event->payload['reused_recorded_at'])->toIso8601String(), + ); } $output = $successor->workflowOutput(); $this->assertSame($scenario['signal_value'], $output['approved_by']); @@ -1635,84 +1644,89 @@ public function testFractionalRedriveClocksSurviveRepeatedNativeAdmissionAndCold Cache::forget('test:redrive:second-calls'); $this->clearWorkflowState(); WorkflowStub::fake(); - Carbon::setTestNow(Carbon::parse($scenario['source_times']['started_at'])); - $stub = WorkflowStub::make($fixture['workflow']['type'], 'regression-corpus-fractional-clock'); - $stub->start(...$fixture['workflow']['arguments']); - Carbon::setTestNow(Carbon::parse($scenario['source_times']['timer_fired_at'])); - $timer = WorkflowTask::query() - ->where('workflow_run_id', $stub->runId()) - ->where('task_type', TaskType::Timer->value) - ->sole(); - $this->app->call([new RunTimerTask($timer->id), 'handle']); - $source = WorkflowRun::query()->findOrFail($stub->runId()); - $this->assertSame(RunStatus::Failed, $source->status); - // Give the consumed activity a distinct persisted instant. The real - // admission, timer and failure lifecycle ran before this clock mutation. - WorkflowHistoryEvent::query() - ->where('workflow_run_id', $source->id) - ->where('event_type', HistoryEventType::ActivityCompleted) - ->sole() - ->forceFill([ - 'recorded_at' => Carbon::parse($scenario['source_times']['activity_completed_at']), - ])->save(); - $originalHistory = $source->historyEvents()->get()->toArray(); - $this->assertSame( - $scenario['expected_times'], - (new QueryStateReplayer())->query($source->fresh(), 'recordedTimes', []), - ); - $predecessor = $source; - $firstSerializedClocks = null; - for ($attempt = 1; $attempt <= $scenario['redrives']; ++$attempt) { - Carbon::setTestNow(now()->addMinutes($scenario['redrive_after_minutes'])); - $result = app(WorkflowControlPlane::class)->redrive($source->workflow_instance_id, $predecessor->id); - $this->assertTrue($result['accepted'], (string) $result['reason']); - $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']); - $this->assertSame( - $attempt === $scenario['redrives'] ? RunStatus::Completed : RunStatus::Failed, - $successor->status, - ); - $this->assertSame( - $scenario['expected_times'], - (new QueryStateReplayer())->query($successor->fresh(), 'recordedTimes', []), - ); - $replayed = (new WorkflowReplayer())->replay($successor->fresh())->workflow; - $this->assertInstanceOf(TestRedriveFractionalClockWorkflow::class, $replayed); - $this->assertSame($scenario['expected_times'], $replayed->recordedTimes()); - $started = $successor->historyEvents() - ->where('event_type', HistoryEventType::WorkflowStarted) + try { + Carbon::setTestNow(Carbon::parse($scenario['source_times']['started_at'])); + $stub = WorkflowStub::make($fixture['workflow']['type'], 'regression-corpus-fractional-clock'); + $stub->start(...$fixture['workflow']['arguments']); + Carbon::setTestNow(Carbon::parse($scenario['source_times']['timer_fired_at'])); + $timer = WorkflowTask::query() + ->where('workflow_run_id', $stub->runId()) + ->where('task_type', TaskType::Timer->value) ->sole(); - $activity = $successor->historyEvents() + $this->app->call([new RunTimerTask($timer->id), 'handle']); + $source = WorkflowRun::query()->findOrFail($stub->runId()); + $this->assertSame(RunStatus::Failed, $source->status); + // Give the consumed activity a distinct persisted instant. The real + // admission, timer and failure lifecycle ran before this clock mutation. + WorkflowHistoryEvent::query() + ->where('workflow_run_id', $source->id) ->where('event_type', HistoryEventType::ActivityCompleted) - ->orderBy('sequence') - ->firstOrFail(); - $fired = $successor->historyEvents() - ->where('event_type', HistoryEventType::TimerFired) - ->sole(); - $serializedClocks = [ - $started->payload['replayed_started_at'], - $activity->payload['reused_recorded_at'], - $fired->payload['reused_recorded_at'], - ]; - if ($firstSerializedClocks !== null) { - $this->assertSame($firstSerializedClocks, $serializedClocks); + ->sole() + ->forceFill([ + 'recorded_at' => Carbon::parse($scenario['source_times']['activity_completed_at']), + ])->save(); + $originalHistory = $source->historyEvents()->get()->toArray(); + $predecessor = $source; + $firstSerializedClocks = null; + for ($attempt = 1; $attempt <= $scenario['redrives']; ++$attempt) { + Carbon::setTestNow(now()->addMinutes($scenario['redrive_after_minutes'])); + $result = app(WorkflowControlPlane::class)->redrive($source->workflow_instance_id, $predecessor->id); + $this->assertTrue($result['accepted'], (string) $result['reason']); + $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']); + $this->assertSame( + $attempt === $scenario['redrives'] ? RunStatus::Completed : RunStatus::Failed, + $successor->status, + ); + if ($successor->status === RunStatus::Completed) { + $this->assertSame( + $scenario['expected_times'], + (new QueryStateReplayer())->query($successor->fresh(), 'recordedTimes', []), + ); + $replayed = (new WorkflowReplayer())->replay($successor->fresh())->workflow; + $this->assertInstanceOf(TestRedriveFractionalClockWorkflow::class, $replayed); + $this->assertSame($scenario['expected_times'], $replayed->recordedTimes()); + } + $started = $successor->historyEvents() + ->where('event_type', HistoryEventType::WorkflowStarted) + ->sole(); + $activity = $successor->historyEvents() + ->where('event_type', HistoryEventType::ActivityCompleted) + ->orderBy('sequence') + ->firstOrFail(); + $fired = $successor->historyEvents() + ->where('event_type', HistoryEventType::TimerFired) + ->sole(); + $serializedClocks = [ + $started->payload['replayed_started_at'], + $activity->payload['reused_recorded_at'], + $fired->payload['reused_recorded_at'], + ]; + $this->assertSame(array_values($scenario['expected_times']), array_map( + static fn (string $clock): string => Carbon::parse($clock)->format('U.u'), + $serializedClocks, + )); + if ($firstSerializedClocks !== null) { + $this->assertSame($firstSerializedClocks, $serializedClocks); + } + $firstSerializedClocks ??= $serializedClocks; + $this->assertSame(0, WorkflowTimer::query() + ->where('workflow_run_id', $successor->id) + ->count()); + $this->assertSame(0, WorkflowTask::query() + ->where('workflow_run_id', $successor->id) + ->where('task_type', TaskType::Timer->value) + ->count()); + $this->assertSame($originalHistory, $source->historyEvents()->get()->toArray()); + $predecessor = $successor; } - $firstSerializedClocks ??= $serializedClocks; - $this->assertSame(0, WorkflowTimer::query() - ->where('workflow_run_id', $successor->id) - ->count()); - $this->assertSame(0, WorkflowTask::query() - ->where('workflow_run_id', $successor->id) - ->where('task_type', TaskType::Timer->value) - ->count()); - $this->assertSame($originalHistory, $source->historyEvents()->get()->toArray()); - $predecessor = $successor; + $output = $predecessor->workflowOutput(); + unset($output['result']); + $this->assertSame($scenario['expected_times'], $output); + $this->assertSame(1, (int) Cache::get('test:redrive:first-calls')); + $this->assertSame(3, (int) Cache::get('test:redrive:second-calls')); + } finally { + Carbon::setTestNow(); } - $output = $predecessor->workflowOutput(); - unset($output['result']); - $this->assertSame($scenario['expected_times'], $output); - $this->assertSame(1, (int) Cache::get('test:redrive:first-calls')); - $this->assertSame(3, (int) Cache::get('test:redrive:second-calls')); - Carbon::setTestNow(); } public function testRedriveReusedTimerCorpusDoesNotWaitAgain(): void diff --git a/tests/Fixtures/V2/ReplayRegression/redrive-reused-fractional-clock.json b/tests/Fixtures/V2/ReplayRegression/redrive-reused-fractional-clock.json index 1ce4b4eaa..00f5b5f51 100644 --- a/tests/Fixtures/V2/ReplayRegression/redrive-reused-fractional-clock.json +++ b/tests/Fixtures/V2/ReplayRegression/redrive-reused-fractional-clock.json @@ -82,7 +82,11 @@ "activity_completed_at": "1793664000.234567", "timer_fired_at": "1793664006.345678" }, - "commands": [] + "commands": [ + { + "type": "complete_workflow" + } + ] }, "consumers": [ "workflow-fiber-runner" From 3cdab091541a1327c6173b1c1e25df0a2670233e Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 22:08:09 +0000 Subject: [PATCH 03/13] Assert full persisted instants in native redrive controls --- tests/Feature/V2/V2FailedRunRedriveTest.php | 41 +++++++++++++-------- 1 file changed, 25 insertions(+), 16 deletions(-) diff --git a/tests/Feature/V2/V2FailedRunRedriveTest.php b/tests/Feature/V2/V2FailedRunRedriveTest.php index 50e44934b..c8c29b224 100644 --- a/tests/Feature/V2/V2FailedRunRedriveTest.php +++ b/tests/Feature/V2/V2FailedRunRedriveTest.php @@ -4,6 +4,8 @@ namespace Tests\Feature\V2; +use Carbon\CarbonInterface; +use Illuminate\Support\Carbon; use Illuminate\Support\Facades\Cache; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; @@ -553,8 +555,8 @@ public function testRedriveReusesSideEffectsAndVersionMarkersAtTheirRecordedTime $this->assertSame($sourceMarker->payload['version'], $reusedMarker->payload['version']); foreach ([[$sourceSideEffect, $reusedSideEffect], [$sourceMarker, $reusedMarker]] as [$original, $reused]) { $this->assertSame($sourceRunId, $reused->payload['reused_from_run_id'] ?? null); - $this->assertSame( - $original->recorded_at?->toIso8601String(), + $this->assertReplayClock( + $original->recorded_at, $reused->payload['reused_recorded_at'] ?? null, ); } @@ -675,7 +677,7 @@ public function testRedriveReusesWorkflowCodeMemoAndSearchAttributeUpserts(): vo ->sole(); $payload = $reused->payload; $this->assertSame($sourceRunId, $payload['reused_from_run_id'] ?? null); - $this->assertSame($original->recorded_at?->toIso8601String(), $payload['reused_recorded_at'] ?? null); + $this->assertReplayClock($original->recorded_at, $payload['reused_recorded_at'] ?? null); unset($payload['reused_from_run_id'], $payload['reused_recorded_at']); // The source's task and command are its recording context, not the step. $expected = $original->payload; @@ -722,7 +724,7 @@ public function testARedriveSuccessorWithUpsertsCanBeRedrivenAgain(): void ->where('workflow_run_id', $firstRunId) ->where('event_type', HistoryEventType::MemoUpserted) ->sole() - ->recorded_at?->toIso8601String(); + ->recorded_at; $second = app(WorkflowControlPlane::class)->redrive('redrive-upserts-chain', $firstRunId); $this->assertTrue($second['accepted'], (string) $second['reason']); @@ -748,8 +750,8 @@ public function testARedriveSuccessorWithUpsertsCanBeRedrivenAgain(): void ->sole(); // A step reused twice keeps the run it came from and its first time. $this->assertSame($secondRun->id, $reusedMemo->payload['reused_from_run_id']); - $this->assertSame($memoRecordedAt, $reusedMemo->payload['reused_recorded_at']); - $this->assertSame($memoRecordedAt, $thirdRun->workflowOutput()['time_after_memo']); + $this->assertReplayClock($memoRecordedAt, $reusedMemo->payload['reused_recorded_at']); + $this->assertSame($memoRecordedAt?->toIso8601String(), $thirdRun->workflowOutput()['time_after_memo']); } public function testRedriveReusesAnAppliedSignalWaitAtItsRecordedTime(): void @@ -800,8 +802,8 @@ public function testRedriveReusesAnAppliedSignalWaitAtItsRecordedTime(): void foreach ($reused as $event) { $original = $sourceEvents->firstWhere('event_type', $event->event_type); $this->assertSame($sourceRunId, $event->payload['reused_from_run_id'] ?? null); - $this->assertSame( - $original->recorded_at?->toIso8601String(), + $this->assertReplayClock( + $original->recorded_at, $event->payload['reused_recorded_at'] ?? null, ); } @@ -1328,8 +1330,8 @@ public function testRedriveReappliesAnUpdateTheSourceApplied(): void $original = $this->sourceEvent($source, $type); $reused = $this->sourceEvent($successor, $type); $this->assertSame($source->id, $reused->payload['reused_from_run_id'] ?? null); - $this->assertSame( - $original->recorded_at?->toIso8601String(), + $this->assertReplayClock( + $original->recorded_at, $reused->payload['reused_recorded_at'] ?? null ); } @@ -1378,8 +1380,8 @@ public function testAnUpdateAndSignalWaitRedriveAcrossAChain(): void // The immediate source, the original run and its first time. $this->assertSame($secondRun->id, $events[0]->payload['reused_from_run_id'], $type->value); $this->assertSame($first->id, $events[0]->payload['workflow_run_id'] ?? $first->id, $type->value); - $this->assertSame( - $original->recorded_at?->toIso8601String(), + $this->assertReplayClock( + $original->recorded_at, $events[0]->payload['reused_recorded_at'], $type->value, ); @@ -2010,8 +2012,8 @@ public function testRedriveReusesAFiredTimerAtItsRecordedTime(): void $this->assertSame($reusedScheduled->payload['timer_id'], $reusedFired->payload['timer_id']); $this->assertNotSame($sourceFired->payload['timer_id'], $reusedFired->payload['timer_id']); $this->assertSame($sourceRunId, $reusedFired->payload['reused_from_run_id'] ?? null); - $this->assertSame( - $sourceFired->recorded_at?->toIso8601String(), + $this->assertReplayClock( + $sourceFired->recorded_at, $reusedFired->payload['reused_recorded_at'] ?? null, ); @@ -2124,8 +2126,8 @@ public function testRedriveReusesACompletedChildWorkflowAtItsRecordedTime(): voi foreach ($reused as $index => $event) { $original = $sourceChildEvents[$index]; $this->assertSame($immediateSource, $event->payload['reused_from_run_id'] ?? null); - $this->assertSame( - $original->recorded_at?->toIso8601String(), + $this->assertReplayClock( + $original->recorded_at, $event->payload['reused_recorded_at'] ?? null, ); $this->assertSame($childRunId, $event->payload['child_workflow_run_id']); @@ -2679,4 +2681,11 @@ private function assertRedriveAction( $this->assertArrayHasKey('redrive_blocked_reason', $describe['actions']); $this->assertSame($reason, $describe['actions']['redrive_blocked_reason'], "describe reason for {$runId}"); } + private function assertReplayClock(?CarbonInterface $expected, mixed $actual, string $message = ''): void + { + $this->assertNotNull($expected, $message); + $this->assertIsString($actual, $message); + $this->assertSame($expected->format('U.u'), Carbon::parse($actual)->format('U.u'), $message); + } + } From a990a42e463cc07cf84f40393670d8412115ea23 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 22:23:49 +0000 Subject: [PATCH 04/13] Format fractional clock regression controls --- .../V2EmbeddedReplayRegressionCorpusTest.php | 11 ++++---- tests/Feature/V2/V2FailedRunRedriveTest.php | 27 +++++-------------- 2 files changed, 12 insertions(+), 26 deletions(-) diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 342afb50d..91272d64f 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -1665,7 +1665,9 @@ public function testFractionalRedriveClocksSurviveRepeatedNativeAdmissionAndCold ->forceFill([ 'recorded_at' => Carbon::parse($scenario['source_times']['activity_completed_at']), ])->save(); - $originalHistory = $source->historyEvents()->get()->toArray(); + $originalHistory = $source->historyEvents() + ->get() + ->toArray(); $predecessor = $source; $firstSerializedClocks = null; for ($attempt = 1; $attempt <= $scenario['redrives']; ++$attempt) { @@ -1682,7 +1684,8 @@ public function testFractionalRedriveClocksSurviveRepeatedNativeAdmissionAndCold $scenario['expected_times'], (new QueryStateReplayer())->query($successor->fresh(), 'recordedTimes', []), ); - $replayed = (new WorkflowReplayer())->replay($successor->fresh())->workflow; + $replayed = (new WorkflowReplayer())->replay($successor->fresh()) +->workflow; $this->assertInstanceOf(TestRedriveFractionalClockWorkflow::class, $replayed); $this->assertSame($scenario['expected_times'], $replayed->recordedTimes()); } @@ -1709,9 +1712,7 @@ public function testFractionalRedriveClocksSurviveRepeatedNativeAdmissionAndCold $this->assertSame($firstSerializedClocks, $serializedClocks); } $firstSerializedClocks ??= $serializedClocks; - $this->assertSame(0, WorkflowTimer::query() - ->where('workflow_run_id', $successor->id) - ->count()); + $this->assertSame(0, WorkflowTimer::query() ->where('workflow_run_id', $successor->id) ->count()); $this->assertSame(0, WorkflowTask::query() ->where('workflow_run_id', $successor->id) ->where('task_type', TaskType::Timer->value) diff --git a/tests/Feature/V2/V2FailedRunRedriveTest.php b/tests/Feature/V2/V2FailedRunRedriveTest.php index c8c29b224..ce01d5a39 100644 --- a/tests/Feature/V2/V2FailedRunRedriveTest.php +++ b/tests/Feature/V2/V2FailedRunRedriveTest.php @@ -555,10 +555,7 @@ public function testRedriveReusesSideEffectsAndVersionMarkersAtTheirRecordedTime $this->assertSame($sourceMarker->payload['version'], $reusedMarker->payload['version']); foreach ([[$sourceSideEffect, $reusedSideEffect], [$sourceMarker, $reusedMarker]] as [$original, $reused]) { $this->assertSame($sourceRunId, $reused->payload['reused_from_run_id'] ?? null); - $this->assertReplayClock( - $original->recorded_at, - $reused->payload['reused_recorded_at'] ?? null, - ); + $this->assertReplayClock($original->recorded_at, $reused->payload['reused_recorded_at'] ?? null); } $expectedTimes = [ @@ -802,10 +799,7 @@ public function testRedriveReusesAnAppliedSignalWaitAtItsRecordedTime(): void foreach ($reused as $event) { $original = $sourceEvents->firstWhere('event_type', $event->event_type); $this->assertSame($sourceRunId, $event->payload['reused_from_run_id'] ?? null); - $this->assertReplayClock( - $original->recorded_at, - $event->payload['reused_recorded_at'] ?? null, - ); + $this->assertReplayClock($original->recorded_at, $event->payload['reused_recorded_at'] ?? null); } $this->assertSame(0, WorkflowSignal::query()->where('workflow_run_id', $successor->id)->count()); // The consumed message stays consumed for the successor. @@ -1330,10 +1324,7 @@ public function testRedriveReappliesAnUpdateTheSourceApplied(): void $original = $this->sourceEvent($source, $type); $reused = $this->sourceEvent($successor, $type); $this->assertSame($source->id, $reused->payload['reused_from_run_id'] ?? null); - $this->assertReplayClock( - $original->recorded_at, - $reused->payload['reused_recorded_at'] ?? null - ); + $this->assertReplayClock($original->recorded_at, $reused->payload['reused_recorded_at'] ?? null); } } @@ -2012,10 +2003,7 @@ public function testRedriveReusesAFiredTimerAtItsRecordedTime(): void $this->assertSame($reusedScheduled->payload['timer_id'], $reusedFired->payload['timer_id']); $this->assertNotSame($sourceFired->payload['timer_id'], $reusedFired->payload['timer_id']); $this->assertSame($sourceRunId, $reusedFired->payload['reused_from_run_id'] ?? null); - $this->assertReplayClock( - $sourceFired->recorded_at, - $reusedFired->payload['reused_recorded_at'] ?? null, - ); + $this->assertReplayClock($sourceFired->recorded_at, $reusedFired->payload['reused_recorded_at'] ?? null); $expectedTime = $sourceFired->recorded_at?->toIso8601String(); $output = $successor->workflowOutput(); @@ -2126,10 +2114,7 @@ public function testRedriveReusesACompletedChildWorkflowAtItsRecordedTime(): voi foreach ($reused as $index => $event) { $original = $sourceChildEvents[$index]; $this->assertSame($immediateSource, $event->payload['reused_from_run_id'] ?? null); - $this->assertReplayClock( - $original->recorded_at, - $event->payload['reused_recorded_at'] ?? null, - ); + $this->assertReplayClock($original->recorded_at, $event->payload['reused_recorded_at'] ?? null); $this->assertSame($childRunId, $event->payload['child_workflow_run_id']); $this->assertSame($original->payload['workflow_link_id'], $event->payload['workflow_link_id']); $this->assertArrayNotHasKey('task', $event->payload); @@ -2681,11 +2666,11 @@ private function assertRedriveAction( $this->assertArrayHasKey('redrive_blocked_reason', $describe['actions']); $this->assertSame($reason, $describe['actions']['redrive_blocked_reason'], "describe reason for {$runId}"); } + private function assertReplayClock(?CarbonInterface $expected, mixed $actual, string $message = ''): void { $this->assertNotNull($expected, $message); $this->assertIsString($actual, $message); $this->assertSame($expected->format('U.u'), Carbon::parse($actual)->format('U.u'), $message); } - } From d6a1962f6c9dfd6de009e76c815da96b9835d41f Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 22:26:23 +0000 Subject: [PATCH 05/13] Preserve scheduled timer deadlines on older MariaDB schemas --- CHANGELOG.md | 6 + .../Support/ConditionOccurrenceDeadline.php | 6 +- ...erve_workflow_timer_fire_at_on_mariadb.php | 43 +++++++ .../V2/V2TimerDeadlineMigrationTest.php | 106 ++++++++++++++++++ tests/Feature/V2/V2WorkflowTaskBridgeTest.php | 35 +++++- 5 files changed, 190 insertions(+), 6 deletions(-) create mode 100644 src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php create mode 100644 tests/Feature/V2/V2TimerDeadlineMigrationTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index e44897b98..96380fb10 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ ## Unreleased +- **Run migrations on MariaDB.** Timer status updates preserve the scheduled + deadline on older MariaDB schemas. The forward migration removes the implicit + timestamp update without rewriting retained rows. Condition occurrence + reopens use the original scheduled history deadline, including when an older + cancelled timer row has already drifted (#1139). + - Experimental domain entity references for top-level arguments of embedded queued and local activities (#996). Register an entity class, deterministic adapter and authorized resolver. History records a portable reference and diff --git a/src/V2/Support/ConditionOccurrenceDeadline.php b/src/V2/Support/ConditionOccurrenceDeadline.php index 703c579c5..de850b56e 100644 --- a/src/V2/Support/ConditionOccurrenceDeadline.php +++ b/src/V2/Support/ConditionOccurrenceDeadline.php @@ -119,12 +119,14 @@ public static function resolve(WorkflowRun $run, array $command): array ->whereKey($first['timer_id']) ->first(); if (! $timer instanceof WorkflowTimer || (int) $timer->sequence !== $first['sequence'] - || ! $timer->fire_at instanceof CarbonInterface) { + || ! $first['deadline_at'] instanceof CarbonInterface) { throw new ConditionOccurrenceDeadlineConflict('condition_wait_anchor_unavailable'); } return [ - 'deadline_at' => $timer->fire_at, + // The scheduled history retains the first deadline even if an + // older MariaDB schema rewrote the cancelled timer row's fire_at. + 'deadline_at' => $first['deadline_at'], 'keep_open' => $keepOpen, ]; } diff --git a/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php b/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php new file mode 100644 index 000000000..68a85c0a7 --- /dev/null +++ b/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php @@ -0,0 +1,43 @@ +getConnection()); + if (! $connection instanceof MySqlConnection || ! $connection->isMaria()) { + return; + } + + $column = $connection->selectOne( + 'SELECT EXTRA AS extra FROM information_schema.columns ' + . 'WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?', + [$connection->getTablePrefix() . 'workflow_run_timers', 'fire_at'], + ); + if ($column === null || ! str_contains(strtolower((string) $column->extra), 'on update')) { + return; + } + + $table = $connection->getQueryGrammar()->wrapTable('workflow_run_timers'); + $this->withSchemaLockBudget(static function () use ($connection, $table): void { + // Older MariaDB adds ON UPDATE to the first non-null TIMESTAMP. + // Keep its existing insert default, precision and stored values. + $connection->statement( + "ALTER TABLE {$table} MODIFY fire_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), ALGORITHM=INPLACE, LOCK=NONE", + ); + }); + } + + public function down(): void + { + // Older runtime code can use the corrected column. Reinstating an + // automatic deadline rewrite would corrupt retained timer evidence. + } +}; diff --git a/tests/Feature/V2/V2TimerDeadlineMigrationTest.php b/tests/Feature/V2/V2TimerDeadlineMigrationTest.php new file mode 100644 index 000000000..9a1ddff0a --- /dev/null +++ b/tests/Feature/V2/V2TimerDeadlineMigrationTest.php @@ -0,0 +1,106 @@ +create([ + 'workflow_run_id' => (string) Str::ulid(), + 'sequence' => 1, + 'status' => TimerStatus::Pending, + 'delay_seconds' => 20, + 'fire_at' => '2026-11-08 04:00:20.234567', + ]); + foreach ([TimerStatus::Cancelled, TimerStatus::Fired] as $status) { + $timer->update(['status' => $status]); + $this->assertSame('2026-11-08 04:00:20.234567', $timer->fresh()->fire_at->format('Y-m-d H:i:s.u')); + } + } + + public function testItPreservesStoredValuesAndCanRunAgainWithoutRestoringAutomaticUpdates(): void + { + $connection = DB::connection(); + if (! $connection instanceof MySqlConnection || ! $connection->isMaria()) { + $before = Schema::getColumns('workflow_run_timers'); + $this->migration()->up(); + $this->migration()->down(); + $this->assertSame($before, Schema::getColumns('workflow_run_timers')); + return; + } + + $table = $connection->getQueryGrammar()->wrapTable('workflow_run_timers'); + $connection->statement( + "ALTER TABLE {$table} MODIFY fire_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)", + ); + $id = (string) Str::ulid(); + DB::table('workflow_run_timers')->insert([ + 'id' => $id, + 'workflow_run_id' => (string) Str::ulid(), + 'sequence' => 1, + 'status' => 'pending', + 'delay_seconds' => 20, + 'fire_at' => '2026-11-08 04:00:20.234567', + ]); + DB::table('workflow_run_timers')->where('id', $id)->update(['status' => 'cancelled']); + $drifted = DB::table('workflow_run_timers')->where('id', $id)->value('fire_at'); + $this->assertNotSame('2026-11-08 04:00:20.234567', $drifted); + $migration = $this->migration(); + $this->assertFalse($migration->withinTransaction); + $lockBudget = $connection->selectOne('SELECT @@SESSION.lock_wait_timeout AS value')->value; + $migration->up(); + $migration->up(); + $migration->down(); + $this->assertSame($lockBudget, $connection->selectOne('SELECT @@SESSION.lock_wait_timeout AS value')->value); + $this->assertSame($drifted, DB::table('workflow_run_timers')->where('id', $id)->value('fire_at')); + DB::table('workflow_run_timers')->where('id', $id)->update(['status' => 'fired']); + $this->assertSame($drifted, DB::table('workflow_run_timers')->where('id', $id)->value('fire_at')); + } + + public function testItUsesTheConfiguredConnectionAndItsTablePrefix(): void + { + $connection = DB::connection(); + if (! $connection instanceof MySqlConnection || ! $connection->isMaria()) { + $this->markTestSkipped('MariaDB-specific timestamp schema correction.'); + } + + $original = config('workflows.storage.connection'); + config(['database.connections.timer_deadline_migration' => [ + ...$connection->getConfig(), + 'prefix' => 'timer_deadline_', + ]]); + $other = DB::connection('timer_deadline_migration'); + $table = $other->getQueryGrammar()->wrapTable('workflow_run_timers'); + try { + $other->statement("CREATE TABLE {$table} (id INT PRIMARY KEY, status VARCHAR(20), fire_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6))"); + $other->insert("INSERT INTO {$table} VALUES (1, 'pending', '2026-11-08 04:00:20.234567')"); + $before = Schema::getColumns('workflow_run_timers'); + config(['workflows.storage.connection' => 'timer_deadline_migration']); + $this->migration()->up(); + $other->update("UPDATE {$table} SET status = 'cancelled' WHERE id = 1"); + $this->assertSame('2026-11-08 04:00:20.234567', $other->selectOne("SELECT fire_at FROM {$table} WHERE id = 1")->fire_at); + $this->assertSame($before, Schema::getColumns('workflow_run_timers')); + } finally { + config(['workflows.storage.connection' => $original]); + $other->statement("DROP TABLE IF EXISTS {$table}"); + DB::purge('timer_deadline_migration'); + } + } + + private function migration(): WorkflowMigration + { + return require dirname(__DIR__, 3) . '/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php'; + } +} diff --git a/tests/Feature/V2/V2WorkflowTaskBridgeTest.php b/tests/Feature/V2/V2WorkflowTaskBridgeTest.php index 18a3e1bab..6c582180a 100644 --- a/tests/Feature/V2/V2WorkflowTaskBridgeTest.php +++ b/tests/Feature/V2/V2WorkflowTaskBridgeTest.php @@ -8660,7 +8660,11 @@ public function testCompleteOpenConditionWaitWithTimeoutSchedulesConditionTimeou } #[DataProvider('conditionOccurrenceDeadlinePolicies')] - public function testFalseWakeUsesTheSelectedOccurrenceDeadlinePolicy(bool $anchored, int $deadlineSeconds): void + public function testFalseWakeUsesTheSelectedOccurrenceDeadlinePolicy( + bool $anchored, + int $deadlineSeconds, + bool $driftedTimer = false, + ): void { $origin = Carbon::parse('2026-11-08T04:00:00.000000Z'); Carbon::setTestNow($origin); @@ -8684,6 +8688,14 @@ public function testFalseWakeUsesTheSelectedOccurrenceDeadlinePolicy(bool $ancho ->sole(); $this->assertTrue($first->fire_at->equalTo($origin->copy()->addSeconds(20))); + if ($driftedTimer) { + // A historical MariaDB ON UPDATE column could already have + // changed this projection. History still owns the anchor. + DB::table('workflow_run_timers')->where('id', $first->id)->update([ + 'fire_at' => $origin->copy()->addHours(1)->format('Y-m-d H:i:s.u'), + ]); + } + // S is the signal/replay clock. R is three seconds later; the // worker's remaining 15 seconds must not add those three to A. Carbon::setTestNow($origin->copy()->addSeconds(5)); @@ -8733,6 +8745,7 @@ public static function conditionOccurrenceDeadlinePolicies(): iterable { yield 'legacy keeps per-open scheduling' => [false, 23]; yield 'anchored false wake keeps its first wall deadline' => [true, 20]; + yield 'recorded anchor survives a drifted physical timer' => [true, 20, true]; } public function testOccurrenceDeadlineConflictsRollBackTheWholeBatchAndKeepTheLease(): void @@ -8866,13 +8879,21 @@ public function testPastOccurrenceDeadlineIsDueAtItsAnchorAndNeverFiresInline(): } } - public function testOccurrenceDeadlineRefusesAnUnavailableCanonicalTimer(): void + #[DataProvider('unavailableOccurrenceDeadlineAnchors')] + public function testOccurrenceDeadlineRefusesAnUnavailableCanonicalTimer(bool $missingTimer): void { $run = $this->createWaitingRun(); $command = self::occurrenceDeadlineCommand(); $this->assertTrue($this->bridge->complete($this->createLeasedTask($run)->id, [$command])['completed']); - $run->timers() - ->delete(); + if ($missingTimer) { + $run->timers()->delete(); + } else { + $scheduled = $run->historyEvents() + ->where('event_type', HistoryEventType::TimerScheduled->value)->sole(); + $payload = $scheduled->payload; + unset($payload['fire_at']); + $scheduled->update(['payload' => $payload]); + } $signal = $this->recordReceivedSignal($run); $task = $this->createLeasedTask($run); $task->forceFill([ @@ -8889,6 +8910,12 @@ public function testOccurrenceDeadlineRefusesAnUnavailableCanonicalTimer(): void $this->assertSame(TaskStatus::Leased, $task->fresh()->status); } + public static function unavailableOccurrenceDeadlineAnchors(): iterable + { + yield 'canonical timer missing' => [true]; + yield 'recorded schedule deadline missing' => [false]; + } + public function testOccurrenceDeadlineRefusesALateOptInToALegacyChain(): void { $run = $this->createWaitingRun(); From 48b8bfa842fc2b0a4b340baee507f1f3209079b7 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 22:27:44 +0000 Subject: [PATCH 06/13] Format timer deadline migration controls --- ...erve_workflow_timer_fire_at_on_mariadb.php | 3 +- .../V2/V2TimerDeadlineMigrationTest.php | 62 +++++++++++++------ tests/Feature/V2/V2WorkflowTaskBridgeTest.php | 14 +++-- 3 files changed, 55 insertions(+), 24 deletions(-) diff --git a/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php b/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php index 68a85c0a7..b90c07df2 100644 --- a/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php +++ b/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php @@ -25,7 +25,8 @@ public function up(): void return; } - $table = $connection->getQueryGrammar()->wrapTable('workflow_run_timers'); + $table = $connection->getQueryGrammar() + ->wrapTable('workflow_run_timers'); $this->withSchemaLockBudget(static function () use ($connection, $table): void { // Older MariaDB adds ON UPDATE to the first non-null TIMESTAMP. // Keep its existing insert default, precision and stored values. diff --git a/tests/Feature/V2/V2TimerDeadlineMigrationTest.php b/tests/Feature/V2/V2TimerDeadlineMigrationTest.php index 9a1ddff0a..2b069894b 100644 --- a/tests/Feature/V2/V2TimerDeadlineMigrationTest.php +++ b/tests/Feature/V2/V2TimerDeadlineMigrationTest.php @@ -25,7 +25,9 @@ public function testStatusChangesPreserveTheExactScheduledInstant(): void 'fire_at' => '2026-11-08 04:00:20.234567', ]); foreach ([TimerStatus::Cancelled, TimerStatus::Fired] as $status) { - $timer->update(['status' => $status]); + $timer->update([ + 'status' => $status, + ]); $this->assertSame('2026-11-08 04:00:20.234567', $timer->fresh()->fire_at->format('Y-m-d H:i:s.u')); } } @@ -35,13 +37,16 @@ public function testItPreservesStoredValuesAndCanRunAgainWithoutRestoringAutomat $connection = DB::connection(); if (! $connection instanceof MySqlConnection || ! $connection->isMaria()) { $before = Schema::getColumns('workflow_run_timers'); - $this->migration()->up(); - $this->migration()->down(); + $this->migration() + ->up(); + $this->migration() + ->down(); $this->assertSame($before, Schema::getColumns('workflow_run_timers')); return; } - $table = $connection->getQueryGrammar()->wrapTable('workflow_run_timers'); + $table = $connection->getQueryGrammar() + ->wrapTable('workflow_run_timers'); $connection->statement( "ALTER TABLE {$table} MODIFY fire_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)", ); @@ -54,18 +59,23 @@ public function testItPreservesStoredValuesAndCanRunAgainWithoutRestoringAutomat 'delay_seconds' => 20, 'fire_at' => '2026-11-08 04:00:20.234567', ]); - DB::table('workflow_run_timers')->where('id', $id)->update(['status' => 'cancelled']); + DB::table('workflow_run_timers')->where('id', $id)->update([ + 'status' => 'cancelled', + ]); $drifted = DB::table('workflow_run_timers')->where('id', $id)->value('fire_at'); $this->assertNotSame('2026-11-08 04:00:20.234567', $drifted); $migration = $this->migration(); $this->assertFalse($migration->withinTransaction); - $lockBudget = $connection->selectOne('SELECT @@SESSION.lock_wait_timeout AS value')->value; + $lockBudget = $connection->selectOne('SELECT @@SESSION.lock_wait_timeout AS value') +->value; $migration->up(); $migration->up(); $migration->down(); $this->assertSame($lockBudget, $connection->selectOne('SELECT @@SESSION.lock_wait_timeout AS value')->value); $this->assertSame($drifted, DB::table('workflow_run_timers')->where('id', $id)->value('fire_at')); - DB::table('workflow_run_timers')->where('id', $id)->update(['status' => 'fired']); + DB::table('workflow_run_timers')->where('id', $id)->update([ + 'status' => 'fired', + ]); $this->assertSame($drifted, DB::table('workflow_run_timers')->where('id', $id)->value('fire_at')); } @@ -77,23 +87,36 @@ public function testItUsesTheConfiguredConnectionAndItsTablePrefix(): void } $original = config('workflows.storage.connection'); - config(['database.connections.timer_deadline_migration' => [ - ...$connection->getConfig(), - 'prefix' => 'timer_deadline_', - ]]); + config([ + 'database.connections.timer_deadline_migration' => [ + ...$connection->getConfig(), + 'prefix' => 'timer_deadline_', + ], + ]); $other = DB::connection('timer_deadline_migration'); - $table = $other->getQueryGrammar()->wrapTable('workflow_run_timers'); + $table = $other->getQueryGrammar() + ->wrapTable('workflow_run_timers'); try { - $other->statement("CREATE TABLE {$table} (id INT PRIMARY KEY, status VARCHAR(20), fire_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6))"); + $other->statement( + "CREATE TABLE {$table} (id INT PRIMARY KEY, status VARCHAR(20), fire_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6))" + ); $other->insert("INSERT INTO {$table} VALUES (1, 'pending', '2026-11-08 04:00:20.234567')"); $before = Schema::getColumns('workflow_run_timers'); - config(['workflows.storage.connection' => 'timer_deadline_migration']); - $this->migration()->up(); + config([ + 'workflows.storage.connection' => 'timer_deadline_migration', + ]); + $this->migration() + ->up(); $other->update("UPDATE {$table} SET status = 'cancelled' WHERE id = 1"); - $this->assertSame('2026-11-08 04:00:20.234567', $other->selectOne("SELECT fire_at FROM {$table} WHERE id = 1")->fire_at); + $this->assertSame( + '2026-11-08 04:00:20.234567', + $other->selectOne("SELECT fire_at FROM {$table} WHERE id = 1")->fire_at + ); $this->assertSame($before, Schema::getColumns('workflow_run_timers')); } finally { - config(['workflows.storage.connection' => $original]); + config([ + 'workflows.storage.connection' => $original, + ]); $other->statement("DROP TABLE IF EXISTS {$table}"); DB::purge('timer_deadline_migration'); } @@ -101,6 +124,9 @@ public function testItUsesTheConfiguredConnectionAndItsTablePrefix(): void private function migration(): WorkflowMigration { - return require dirname(__DIR__, 3) . '/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php'; + return require dirname( + __DIR__, + 3 + ) . '/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php'; } } diff --git a/tests/Feature/V2/V2WorkflowTaskBridgeTest.php b/tests/Feature/V2/V2WorkflowTaskBridgeTest.php index 6c582180a..5d4983a22 100644 --- a/tests/Feature/V2/V2WorkflowTaskBridgeTest.php +++ b/tests/Feature/V2/V2WorkflowTaskBridgeTest.php @@ -8664,8 +8664,7 @@ public function testFalseWakeUsesTheSelectedOccurrenceDeadlinePolicy( bool $anchored, int $deadlineSeconds, bool $driftedTimer = false, - ): void - { + ): void { $origin = Carbon::parse('2026-11-08T04:00:00.000000Z'); Carbon::setTestNow($origin); @@ -8692,7 +8691,9 @@ public function testFalseWakeUsesTheSelectedOccurrenceDeadlinePolicy( // A historical MariaDB ON UPDATE column could already have // changed this projection. History still owns the anchor. DB::table('workflow_run_timers')->where('id', $first->id)->update([ - 'fire_at' => $origin->copy()->addHours(1)->format('Y-m-d H:i:s.u'), + 'fire_at' => $origin->copy() + ->addHours(1) + ->format('Y-m-d H:i:s.u'), ]); } @@ -8886,13 +8887,16 @@ public function testOccurrenceDeadlineRefusesAnUnavailableCanonicalTimer(bool $m $command = self::occurrenceDeadlineCommand(); $this->assertTrue($this->bridge->complete($this->createLeasedTask($run)->id, [$command])['completed']); if ($missingTimer) { - $run->timers()->delete(); + $run->timers() + ->delete(); } else { $scheduled = $run->historyEvents() ->where('event_type', HistoryEventType::TimerScheduled->value)->sole(); $payload = $scheduled->payload; unset($payload['fire_at']); - $scheduled->update(['payload' => $payload]); + $scheduled->update([ + 'payload' => $payload, + ]); } $signal = $this->recordReceivedSignal($run); $task = $this->createLeasedTask($run); From 12fca2ab435435b1faa84756b8d2045a84dcf155 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 22:29:07 +0000 Subject: [PATCH 07/13] Keep migration assertion formatting stable --- tests/Feature/V2/V2TimerDeadlineMigrationTest.php | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/Feature/V2/V2TimerDeadlineMigrationTest.php b/tests/Feature/V2/V2TimerDeadlineMigrationTest.php index 2b069894b..137163263 100644 --- a/tests/Feature/V2/V2TimerDeadlineMigrationTest.php +++ b/tests/Feature/V2/V2TimerDeadlineMigrationTest.php @@ -110,7 +110,8 @@ public function testItUsesTheConfiguredConnectionAndItsTablePrefix(): void $other->update("UPDATE {$table} SET status = 'cancelled' WHERE id = 1"); $this->assertSame( '2026-11-08 04:00:20.234567', - $other->selectOne("SELECT fire_at FROM {$table} WHERE id = 1")->fire_at + $other->selectOne("SELECT fire_at FROM {$table} WHERE id = 1") +->fire_at ); $this->assertSame($before, Schema::getColumns('workflow_run_timers')); } finally { From 2560388cd5d32215019d9fdefa00c7e58ca8af8d Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 22:41:12 +0000 Subject: [PATCH 08/13] Fence timer migration to the exact legacy timestamp metadata --- ...erve_workflow_timer_fire_at_on_mariadb.php | 10 +++- .../V2/V2TimerDeadlineMigrationTest.php | 55 +++++++++++++++++++ 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php b/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php index b90c07df2..fd9cdb10d 100644 --- a/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php +++ b/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php @@ -17,13 +17,21 @@ public function up(): void } $column = $connection->selectOne( - 'SELECT EXTRA AS extra FROM information_schema.columns ' + 'SELECT COLUMN_TYPE AS column_type, IS_NULLABLE AS is_nullable, COLUMN_DEFAULT AS default_value, ' + . 'DATETIME_PRECISION AS datetime_precision, EXTRA AS extra FROM information_schema.columns ' . 'WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?', [$connection->getTablePrefix() . 'workflow_run_timers', 'fire_at'], ); if ($column === null || ! str_contains(strtolower((string) $column->extra), 'on update')) { return; } + if (strtolower((string) $column->column_type) !== 'timestamp(6)' + || $column->is_nullable !== 'NO' + || strtolower((string) $column->default_value) !== 'current_timestamp(6)' + || (int) $column->datetime_precision !== 6 + || strtolower((string) $column->extra) !== 'on update current_timestamp(6)') { + throw new RuntimeException('workflow_timer_fire_at_legacy_shape_unavailable'); + } $table = $connection->getQueryGrammar() ->wrapTable('workflow_run_timers'); diff --git a/tests/Feature/V2/V2TimerDeadlineMigrationTest.php b/tests/Feature/V2/V2TimerDeadlineMigrationTest.php index 137163263..799724292 100644 --- a/tests/Feature/V2/V2TimerDeadlineMigrationTest.php +++ b/tests/Feature/V2/V2TimerDeadlineMigrationTest.php @@ -8,6 +8,8 @@ use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Schema; use Illuminate\Support\Str; +use PHPUnit\Framework\Attributes\DataProvider; +use RuntimeException; use Tests\SchemaTestCase; use Workflow\Support\WorkflowMigration; use Workflow\V2\Enums\TimerStatus; @@ -66,11 +68,17 @@ public function testItPreservesStoredValuesAndCanRunAgainWithoutRestoringAutomat $this->assertNotSame('2026-11-08 04:00:20.234567', $drifted); $migration = $this->migration(); $this->assertFalse($migration->withinTransaction); + $before = $this->columnMetadata(); $lockBudget = $connection->selectOne('SELECT @@SESSION.lock_wait_timeout AS value') ->value; $migration->up(); $migration->up(); $migration->down(); + $after = $this->columnMetadata(); + $this->assertSame('on update current_timestamp(6)', $before['extra']); + $this->assertSame('', $after['extra']); + unset($before['extra'], $after['extra']); + $this->assertSame($before, $after); $this->assertSame($lockBudget, $connection->selectOne('SELECT @@SESSION.lock_wait_timeout AS value')->value); $this->assertSame($drifted, DB::table('workflow_run_timers')->where('id', $id)->value('fire_at')); DB::table('workflow_run_timers')->where('id', $id)->update([ @@ -130,4 +138,51 @@ private function migration(): WorkflowMigration 3 ) . '/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php'; } + + #[DataProvider('unsupportedAutomaticTimestampShapes')] + public function testUnexpectedLegacyMetadataRefusesBeforeAnyWrite(string $definition): void + { + $connection = DB::connection(); + if (! $connection instanceof MySqlConnection || ! $connection->isMaria()) { + $this->markTestSkipped('MariaDB-specific timestamp schema correction.'); + } + $table = $connection->getQueryGrammar()->wrapTable('workflow_run_timers'); + $connection->statement("ALTER TABLE {$table} MODIFY fire_at {$definition}"); + $before = $this->columnMetadata(); + $queries = []; + $connection->listen(static function ($query) use (&$queries): void { + $queries[] = $query->sql; + }); + try { + $this->migration()->up(); + $this->fail('An unsupported automatic timestamp shape must refuse.'); + } catch (RuntimeException $exception) { + $this->assertSame('workflow_timer_fire_at_legacy_shape_unavailable', $exception->getMessage()); + } + $this->assertSame($before, $this->columnMetadata()); + $this->assertSame([], array_values(array_filter( + $queries, + static fn (string $sql): bool => preg_match('/^\s*(alter|update|insert|delete|set)\b/i', $sql) === 1, + ))); + } + + public static function unsupportedAutomaticTimestampShapes(): iterable + { + yield 'custom default' => ["TIMESTAMP(6) NOT NULL DEFAULT '2026-11-08 04:00:20.234567' ON UPDATE CURRENT_TIMESTAMP(6)"]; + yield 'nullable column' => ['TIMESTAMP(6) NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)']; + yield 'different precision' => ['TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3)']; + yield 'different type' => ['DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)']; + } + + /** @return array */ + private function columnMetadata(): array + { + $connection = DB::connection(); + return (array) $connection->selectOne( + 'SELECT COLUMN_TYPE AS column_type, IS_NULLABLE AS is_nullable, COLUMN_DEFAULT AS default_value, ' + . 'DATETIME_PRECISION AS datetime_precision, EXTRA AS extra FROM information_schema.columns ' + . 'WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?', + [$connection->getTablePrefix() . 'workflow_run_timers', 'fire_at'], + ); + } } From f494118c4faa60369b605fa82b80e250c408c1a1 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 22:48:47 +0000 Subject: [PATCH 09/13] Format exact timer metadata guards and controls --- .../V2/V2TimerDeadlineMigrationTest.php | 34 ++++++++++++------- 1 file changed, 21 insertions(+), 13 deletions(-) diff --git a/tests/Feature/V2/V2TimerDeadlineMigrationTest.php b/tests/Feature/V2/V2TimerDeadlineMigrationTest.php index 799724292..b12db8c3a 100644 --- a/tests/Feature/V2/V2TimerDeadlineMigrationTest.php +++ b/tests/Feature/V2/V2TimerDeadlineMigrationTest.php @@ -131,14 +131,6 @@ public function testItUsesTheConfiguredConnectionAndItsTablePrefix(): void } } - private function migration(): WorkflowMigration - { - return require dirname( - __DIR__, - 3 - ) . '/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php'; - } - #[DataProvider('unsupportedAutomaticTimestampShapes')] public function testUnexpectedLegacyMetadataRefusesBeforeAnyWrite(string $definition): void { @@ -146,7 +138,8 @@ public function testUnexpectedLegacyMetadataRefusesBeforeAnyWrite(string $defini if (! $connection instanceof MySqlConnection || ! $connection->isMaria()) { $this->markTestSkipped('MariaDB-specific timestamp schema correction.'); } - $table = $connection->getQueryGrammar()->wrapTable('workflow_run_timers'); + $table = $connection->getQueryGrammar() + ->wrapTable('workflow_run_timers'); $connection->statement("ALTER TABLE {$table} MODIFY fire_at {$definition}"); $before = $this->columnMetadata(); $queries = []; @@ -154,7 +147,8 @@ public function testUnexpectedLegacyMetadataRefusesBeforeAnyWrite(string $defini $queries[] = $query->sql; }); try { - $this->migration()->up(); + $this->migration() + ->up(); $this->fail('An unsupported automatic timestamp shape must refuse.'); } catch (RuntimeException $exception) { $this->assertSame('workflow_timer_fire_at_legacy_shape_unavailable', $exception->getMessage()); @@ -168,13 +162,27 @@ public function testUnexpectedLegacyMetadataRefusesBeforeAnyWrite(string $defini public static function unsupportedAutomaticTimestampShapes(): iterable { - yield 'custom default' => ["TIMESTAMP(6) NOT NULL DEFAULT '2026-11-08 04:00:20.234567' ON UPDATE CURRENT_TIMESTAMP(6)"]; + yield 'custom default' => [ + "TIMESTAMP(6) NOT NULL DEFAULT '2026-11-08 04:00:20.234567' ON UPDATE CURRENT_TIMESTAMP(6)", + ]; yield 'nullable column' => ['TIMESTAMP(6) NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)']; - yield 'different precision' => ['TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3)']; + yield 'different precision' => [ + 'TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3)', + ]; yield 'different type' => ['DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)']; } - /** @return array */ + private function migration(): WorkflowMigration + { + return require dirname( + __DIR__, + 3 + ) . '/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php'; + } + + /** + * @return array + */ private function columnMetadata(): array { $connection = DB::connection(); From 83c605c788946afdc1b5be31eef51ca0e69d9ded Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 23:27:38 +0000 Subject: [PATCH 10/13] Keep locking parent history reads on the existing run index --- CHANGELOG.md | 4 ++ src/V2/Support/ParallelChildGroup.php | 31 +++++++++-- .../V2ServiceModeParallelChildBarrierTest.php | 53 ++++++++++++++++++- 3 files changed, 83 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e44897b98..c6f115fa2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## Unreleased +- Concurrent child completion keeps parent-history row locks on the existing + per-run index on MySQL and MariaDB, avoiding an unrelated sibling history + row in the parent barrier lock cycle. No new index or retry is introduced. + - Experimental domain entity references for top-level arguments of embedded queued and local activities (#996). Register an entity class, deterministic adapter and authorized resolver. History records a portable reference and diff --git a/src/V2/Support/ParallelChildGroup.php b/src/V2/Support/ParallelChildGroup.php index ad1a47b93..ead6f5a56 100644 --- a/src/V2/Support/ParallelChildGroup.php +++ b/src/V2/Support/ParallelChildGroup.php @@ -4,6 +4,8 @@ namespace Workflow\V2\Support; +use Illuminate\Database\Eloquent\Collection; +use Illuminate\Database\Schema\Blueprint; use Workflow\V2\Enums\ActivityStatus; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; @@ -732,10 +734,10 @@ private static function persistSelectionWinner( $events = $parentRun->historyEvents(); if ($lockHistoryForUpdate) { - $events->lockForUpdate(); + $parentRun->setRelation('historyEvents', self::lockedParentHistory($parentRun)); + } else { + $parentRun->setRelation('historyEvents', $events->get()); } - - $parentRun->setRelation('historyEvents', $events->get()); $groupId = $metadata['parallel_group_id']; $memberBase = self::intValue($metadata['selection_member_base_sequence'] ?? null) ?? ($metadata['parallel_group_base_sequence'] + $metadata['parallel_group_index']); @@ -800,7 +802,7 @@ private static function groupCompletedSuccessfully( if ($lockHistoryForUpdate) { // Observe the resolution event committed by the previous holder of // the parent lock even when this transaction has an older snapshot. - $parentRun->setRelation('historyEvents', $parentRun->historyEvents() ->lockForUpdate() ->get()); + $parentRun->setRelation('historyEvents', self::lockedParentHistory($parentRun)); } $activitiesBySequence = collect(RunActivityView::activitiesForRun($parentRun, decodePayloads: false)) @@ -876,6 +878,27 @@ private static function groupCompletedSuccessfully( return true; } + /** + * @return Collection + */ + private static function lockedParentHistory(WorkflowRun $parentRun): Collection + { + $events = $parentRun->historyEvents(); + $model = $events->getRelated(); + $connection = $model->getConnection(); + if (in_array($connection->getDriverName(), ['mysql', 'mariadb'], true)) { + // A table scan can lock an uncommitted sibling event while that + // sibling waits for our parent-run lock. Use the migration's + // per-run index, including Laravel's configured table/index prefix. + $connection->getSchemaBuilder(); + $index = (new Blueprint($connection, $model->getTable())) + ->unique(['workflow_run_id', 'sequence'])->index; + $events->forceIndex($index); + } + return $events->lockForUpdate() + ->get(); + } + /** * @param array $payload * @return array{ diff --git a/tests/Feature/V2/V2ServiceModeParallelChildBarrierTest.php b/tests/Feature/V2/V2ServiceModeParallelChildBarrierTest.php index 29e20aaf6..36cfae242 100644 --- a/tests/Feature/V2/V2ServiceModeParallelChildBarrierTest.php +++ b/tests/Feature/V2/V2ServiceModeParallelChildBarrierTest.php @@ -299,12 +299,50 @@ public function testConcurrentChildCompletionsSerializeOneBarrierWakeOnParentRun $children = []; $results = []; + $completionGates = []; + $parentRunId = $parentRun->id; DB::purge(); try { foreach ($childTasks as $childTask) { - $children[] = $this->forkDatabaseOperation(static function () use ($childTask): array { + $gate = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); + $this->assertIsArray($gate); + stream_set_timeout($gate[0], 15); + stream_set_timeout($gate[1], 15); + $completionGates[] = $gate[0]; + $children[] = $this->forkDatabaseOperation(static function () use ( + $childTask, + $parentRunId, + $gate, + ): array { + fclose($gate[0]); + $appended = false; + $parentLocked = false; + DB::listen(static function (QueryExecuted $query) use ( + $childTask, + $parentRunId, + $gate, + &$appended, + &$parentLocked, + ): void { + if (! $appended && str_starts_with(strtolower($query->sql), 'insert') + && str_contains($query->sql, 'workflow_history_events') + && in_array($childTask->workflow_run_id, $query->bindings, true) + && in_array(HistoryEventType::WorkflowCompleted->value, $query->bindings, true)) { + $appended = true; + fwrite($gate[1], "appended\n"); + if (fgets($gate[1]) !== "continue\n") { + throw new RuntimeException('Child completion gate was not released.'); + } + } + if ($appended && ! $parentLocked && str_contains($query->sql, 'workflow_runs') + && str_contains(strtolower($query->sql), 'for update') + && in_array($parentRunId, $query->bindings, true)) { + $parentLocked = true; + fwrite($gate[1], "parent-locked\n"); + } + }); /** @var WorkflowTaskBridge $bridge */ $bridge = app(WorkflowTaskBridge::class); @@ -315,6 +353,7 @@ public function testConcurrentChildCompletionsSerializeOneBarrierWakeOnParentRun ]), ]]); }); + fclose($gate[1]); } foreach ($children as $child) { @@ -325,12 +364,24 @@ public function testConcurrentChildCompletionsSerializeOneBarrierWakeOnParentRun $this->releaseDatabaseOperation($child); } + // Keep both child history writes uncommitted. The first child + // then owns the parent while the sibling still holds its row. + foreach ($completionGates as $gate) { + $this->assertSame("appended\n", fgets($gate)); + } + fwrite($completionGates[0], "continue\n"); + $this->assertSame("parent-locked\n", fgets($completionGates[0])); + fwrite($completionGates[1], "continue\n"); + $results = array_map(fn (array $child): array => $this->awaitDatabaseOperation($child), $children); $children = []; } finally { foreach ($children as $child) { $this->terminateDatabaseOperation($child); } + foreach ($completionGates as $gate) { + fclose($gate); + } } DB::reconnect(); From 12467c291d6b35e92067e0acfaf71baf028ef778 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 23:42:17 +0000 Subject: [PATCH 11/13] Isolate frozen-clock watchdog recovery from live queue workers --- tests/Feature/V2/V2WorkflowTest.php | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/Feature/V2/V2WorkflowTest.php b/tests/Feature/V2/V2WorkflowTest.php index f16827786..99a31a88d 100644 --- a/tests/Feature/V2/V2WorkflowTest.php +++ b/tests/Feature/V2/V2WorkflowTest.php @@ -6914,6 +6914,9 @@ public function testTaskWatchdogLeavesRowOnlyRunningActivityFallbackDiagnosticOn public function testTaskWatchdogReclaimsExpiredActivityTaskLease(): void { + // The manual watchdog pass owns the frozen clock. Queue workers + // use wall time and can otherwise reclaim this attempt first. + self::stopWorkers(); Queue::fake(); Carbon::setTestNow('2026-10-21 08:00:00.123456'); From fdfbb30c35d2083e914739d9d3f9557bfd3b4a7c Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 00:37:46 +0000 Subject: [PATCH 12/13] Fix MariaDB schema refusal and lock test exception types --- ...3_000200_preserve_workflow_timer_fire_at_on_mariadb.php | 2 +- tests/Feature/V2/V2ActivityLeaseRepairLockTest.php | 7 +++++-- tests/Feature/V2/V2TimerDeadlineMigrationTest.php | 4 ++-- 3 files changed, 8 insertions(+), 5 deletions(-) diff --git a/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php b/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php index fd9cdb10d..b9c40c305 100644 --- a/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php +++ b/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php @@ -30,7 +30,7 @@ public function up(): void || strtolower((string) $column->default_value) !== 'current_timestamp(6)' || (int) $column->datetime_precision !== 6 || strtolower((string) $column->extra) !== 'on update current_timestamp(6)') { - throw new RuntimeException('workflow_timer_fire_at_legacy_shape_unavailable'); + throw new UnexpectedValueException('workflow_timer_fire_at_legacy_shape_unavailable'); } $table = $connection->getQueryGrammar() diff --git a/tests/Feature/V2/V2ActivityLeaseRepairLockTest.php b/tests/Feature/V2/V2ActivityLeaseRepairLockTest.php index 0f4343c0a..baae54716 100644 --- a/tests/Feature/V2/V2ActivityLeaseRepairLockTest.php +++ b/tests/Feature/V2/V2ActivityLeaseRepairLockTest.php @@ -4,6 +4,7 @@ namespace Tests\Feature\V2; +use Illuminate\Database\DeadlockException; use Illuminate\Database\QueryException; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; @@ -69,11 +70,13 @@ public function testLeaseRepairDoesNotWaitForAnAttemptWhileHoldingItsRun(): void $run = WorkflowRun::query()->lockForUpdate()->findOrFail($task->workflow_run_id); TaskRepair::recoverExistingTask($task->fresh(), $run); }); - } catch (QueryException $error) { - $failure = $error; + } catch (QueryException|DeadlockException $error) { + // Laravel wraps MariaDB NOWAIT contention in nested transactions. + $failure = $error instanceof DeadlockException ? $error->getPrevious() : $error; } $elapsed = (hrtime(true) - $started) / 1e9; $this->assertInstanceOf(QueryException::class, $failure); + $this->assertStringContainsString('for update nowait', $failure->getSql()); $this->assertLessThan( 2.0, $elapsed, diff --git a/tests/Feature/V2/V2TimerDeadlineMigrationTest.php b/tests/Feature/V2/V2TimerDeadlineMigrationTest.php index b12db8c3a..370bbe3a9 100644 --- a/tests/Feature/V2/V2TimerDeadlineMigrationTest.php +++ b/tests/Feature/V2/V2TimerDeadlineMigrationTest.php @@ -9,8 +9,8 @@ use Illuminate\Support\Facades\Schema; use Illuminate\Support\Str; use PHPUnit\Framework\Attributes\DataProvider; -use RuntimeException; use Tests\SchemaTestCase; +use UnexpectedValueException; use Workflow\Support\WorkflowMigration; use Workflow\V2\Enums\TimerStatus; use Workflow\V2\Models\WorkflowTimer; @@ -150,7 +150,7 @@ public function testUnexpectedLegacyMetadataRefusesBeforeAnyWrite(string $defini $this->migration() ->up(); $this->fail('An unsupported automatic timestamp shape must refuse.'); - } catch (RuntimeException $exception) { + } catch (UnexpectedValueException $exception) { $this->assertSame('workflow_timer_fire_at_legacy_shape_unavailable', $exception->getMessage()); } $this->assertSame($before, $this->columnMetadata()); From 34a631f543c78836de0a1879d69bd4369b7ba555 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 01:46:56 +0000 Subject: [PATCH 13/13] Preserve MariaDB service dispatch timestamps and cover migration guards --- CHANGELOG.md | 7 +- src/Support/WorkflowMigration.php | 38 +++ ...erve_workflow_timer_fire_at_on_mariadb.php | 36 +-- ...rvice_dispatch_available_at_on_mariadb.php | 23 ++ ...2ServiceDispatchTimestampMigrationTest.php | 133 ++++++++++ .../TimerDeadlineMetadataMigrationTest.php | 236 ++++++++++++++++++ 6 files changed, 438 insertions(+), 35 deletions(-) create mode 100644 src/migrations/2026_11_14_000100_preserve_service_dispatch_available_at_on_mariadb.php create mode 100644 tests/Feature/V2/V2ServiceDispatchTimestampMigrationTest.php create mode 100644 tests/Unit/migrations/TimerDeadlineMetadataMigrationTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 5842f43a5..0f8ba759a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,9 +2,10 @@ ## Unreleased -- **Run migrations on MariaDB.** Timer status updates preserve the scheduled - deadline on older MariaDB schemas. The forward migration removes the implicit - timestamp update without rewriting retained rows. Condition occurrence +- **Run migrations on MariaDB.** Timer status and service-dispatch metadata + updates preserve their scheduled timestamps on older MariaDB schemas. The + forward migrations remove implicit timestamp updates without rewriting + retained rows. Condition occurrence reopens use the original scheduled history deadline, including when an older cancelled timer row has already drifted (#1139). diff --git a/src/Support/WorkflowMigration.php b/src/Support/WorkflowMigration.php index 9e0f6b899..b854e6950 100644 --- a/src/Support/WorkflowMigration.php +++ b/src/Support/WorkflowMigration.php @@ -6,8 +6,10 @@ use Illuminate\Database\Connection; use Illuminate\Database\Migrations\Migration; +use Illuminate\Database\MySqlConnection; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Schema; +use UnexpectedValueException; abstract class WorkflowMigration extends Migration { @@ -16,6 +18,42 @@ public function getConnection(): ?string return config('workflows.storage.connection') ?? $this->connection; } + protected function preserveMariaDbScheduledTimestamp(string $table, string $column, string $refusal): void + { + $connection = DB::connection($this->getConnection()); + if (! $connection instanceof MySqlConnection || ! $connection->isMaria()) { + return; + } + + // Older MariaDB implicitly rewrites its first non-null TIMESTAMP. + // Remove only that behavior, preserving the rest of its definition. + $metadata = $connection->selectOne( + 'SELECT COLUMN_TYPE AS column_type, IS_NULLABLE AS is_nullable, COLUMN_DEFAULT AS default_value, ' + . 'DATETIME_PRECISION AS datetime_precision, EXTRA AS extra FROM information_schema.columns ' + . 'WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?', + [$connection->getTablePrefix() . $table, $column], + ); + if ($metadata === null || ! str_contains(strtolower((string) $metadata->extra), 'on update')) { + return; + } + if (strtolower((string) $metadata->column_type) !== 'timestamp(6)' + || $metadata->is_nullable !== 'NO' + || strtolower((string) $metadata->default_value) !== 'current_timestamp(6)' + || (int) $metadata->datetime_precision !== 6 + || strtolower((string) $metadata->extra) !== 'on update current_timestamp(6)') { + throw new UnexpectedValueException($refusal); + } + + $grammar = $connection->getQueryGrammar(); + $wrappedTable = $grammar->wrapTable($table); + $wrappedColumn = $grammar->wrap($column); + $this->withSchemaLockBudget(static function () use ($connection, $wrappedTable, $wrappedColumn): void { + $connection->statement( + "ALTER TABLE {$wrappedTable} MODIFY {$wrappedColumn} TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), ALGORITHM=INPLACE, LOCK=NONE", + ); + }); + } + /** * Run a short schema change with a one-second lock wait. * diff --git a/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php b/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php index b9c40c305..e0c36b5b6 100644 --- a/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php +++ b/src/migrations/2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php @@ -2,8 +2,6 @@ declare(strict_types=1); -use Illuminate\Database\MySqlConnection; -use Illuminate\Support\Facades\DB; use Workflow\Support\WorkflowMigration; return new class() extends WorkflowMigration { @@ -11,37 +9,11 @@ public function up(): void { - $connection = DB::connection($this->getConnection()); - if (! $connection instanceof MySqlConnection || ! $connection->isMaria()) { - return; - } - - $column = $connection->selectOne( - 'SELECT COLUMN_TYPE AS column_type, IS_NULLABLE AS is_nullable, COLUMN_DEFAULT AS default_value, ' - . 'DATETIME_PRECISION AS datetime_precision, EXTRA AS extra FROM information_schema.columns ' - . 'WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?', - [$connection->getTablePrefix() . 'workflow_run_timers', 'fire_at'], + $this->preserveMariaDbScheduledTimestamp( + 'workflow_run_timers', + 'fire_at', + 'workflow_timer_fire_at_legacy_shape_unavailable', ); - if ($column === null || ! str_contains(strtolower((string) $column->extra), 'on update')) { - return; - } - if (strtolower((string) $column->column_type) !== 'timestamp(6)' - || $column->is_nullable !== 'NO' - || strtolower((string) $column->default_value) !== 'current_timestamp(6)' - || (int) $column->datetime_precision !== 6 - || strtolower((string) $column->extra) !== 'on update current_timestamp(6)') { - throw new UnexpectedValueException('workflow_timer_fire_at_legacy_shape_unavailable'); - } - - $table = $connection->getQueryGrammar() - ->wrapTable('workflow_run_timers'); - $this->withSchemaLockBudget(static function () use ($connection, $table): void { - // Older MariaDB adds ON UPDATE to the first non-null TIMESTAMP. - // Keep its existing insert default, precision and stored values. - $connection->statement( - "ALTER TABLE {$table} MODIFY fire_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), ALGORITHM=INPLACE, LOCK=NONE", - ); - }); } public function down(): void diff --git a/src/migrations/2026_11_14_000100_preserve_service_dispatch_available_at_on_mariadb.php b/src/migrations/2026_11_14_000100_preserve_service_dispatch_available_at_on_mariadb.php new file mode 100644 index 000000000..bf570bf9d --- /dev/null +++ b/src/migrations/2026_11_14_000100_preserve_service_dispatch_available_at_on_mariadb.php @@ -0,0 +1,23 @@ +preserveMariaDbScheduledTimestamp( + 'workflow_service_call_dispatches', + 'available_at', + 'workflow_service_dispatch_available_at_legacy_shape_unavailable', + ); + } + + public function down(): void + { + // Preserve scheduling evidence rather than restore automatic rewrites. + } +}; diff --git a/tests/Feature/V2/V2ServiceDispatchTimestampMigrationTest.php b/tests/Feature/V2/V2ServiceDispatchTimestampMigrationTest.php new file mode 100644 index 000000000..08943126b --- /dev/null +++ b/tests/Feature/V2/V2ServiceDispatchTimestampMigrationTest.php @@ -0,0 +1,133 @@ +dispatch(); + $dispatch->update([ + 'dispatch_attempted_at' => now(), + 'repair_available_at' => now() + ->addSeconds(5), + ]); + $this->assertSame('2026-10-11 00:00:00.123456', $dispatch->fresh()->available_at->format('Y-m-d H:i:s.u')); + } + + public function testUpgradePreservesValuesAndAllOtherMetadataAndDoesNotRestoreAutomaticUpdates(): void + { + $connection = DB::connection(); + if (! $connection instanceof MySqlConnection || ! $connection->isMaria()) { + $this->markTestSkipped('MariaDB-specific automatic timestamp correction.'); + } + $table = $connection->getQueryGrammar() + ->wrapTable('workflow_service_call_dispatches'); + $connection->statement( + "ALTER TABLE {$table} MODIFY available_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)" + ); + $dispatch = $this->dispatch(); + $before = $dispatch->fresh() + ->getRawOriginal(); + $metadata = $this->columnMetadata(); + $this->migration() + ->up(); + $after = $this->columnMetadata(); + $this->assertSame('on update current_timestamp(6)', $metadata['extra']); + $this->assertSame('', $after['extra']); + unset($metadata['extra'], $after['extra']); + $this->assertSame($metadata, $after); + $this->assertSame($before, $dispatch->fresh()->getRawOriginal()); + $this->migration() + ->up(); + $this->migration() + ->down(); + $dispatch->update([ + 'dispatch_attempted_at' => now(), + ]); + $this->assertSame('2026-10-11 00:00:00.123456', $dispatch->fresh()->available_at->format('Y-m-d H:i:s.u')); + } + + public function testConfiguredConnectionAndPrefixSelectOnlyTheOwnedDispatchTable(): void + { + $connection = DB::connection(); + if (! $connection instanceof MySqlConnection || ! $connection->isMaria()) { + $this->markTestSkipped('MariaDB-specific automatic timestamp correction.'); + } + $original = config('workflows.storage.connection'); + config([ + 'database.connections.dispatch_timestamp' => array_replace($connection->getConfig(), [ + 'prefix' => 'timestamp_', + ]), + ]); + $other = DB::connection('dispatch_timestamp'); + $table = $other->getQueryGrammar() + ->wrapTable('workflow_service_call_dispatches'); + $before = Schema::getColumns('workflow_service_call_dispatches'); + try { + $other->statement( + "CREATE TABLE {$table} (id INT PRIMARY KEY, state VARCHAR(32), available_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6))" + ); + $other->insert("INSERT INTO {$table} VALUES (1, 'scheduled', '2026-10-11 00:00:00.123456')"); + config([ + 'workflows.storage.connection' => 'dispatch_timestamp', + ]); + $this->migration() + ->up(); + $other->update("UPDATE {$table} SET state = 'claimed' WHERE id = 1"); + $this->assertSame( + '2026-10-11 00:00:00.123456', + $other->selectOne("SELECT available_at FROM {$table} WHERE id = 1") +->available_at + ); + $this->assertSame($before, Schema::getColumns('workflow_service_call_dispatches')); + } finally { + config([ + 'workflows.storage.connection' => $original, + ]); + $other->statement("DROP TABLE IF EXISTS {$table}"); + DB::purge('dispatch_timestamp'); + } + } + + private function dispatch(): WorkflowServiceCallDispatch + { + return WorkflowServiceCallDispatch::query()->create([ + 'service_call_id' => (string) Str::ulid(), + 'state' => 'scheduled', + 'dispatch_options' => [], + 'attempt' => 1, + 'max_attempts' => 1, + 'available_at' => '2026-10-11 00:00:00.123456', + 'claim_generation' => 0, + ]); + } + + private function columnMetadata(): array + { + $connection = DB::connection(); + + return (array) $connection->selectOne( + 'SELECT COLUMN_TYPE AS column_type, IS_NULLABLE AS is_nullable, COLUMN_DEFAULT AS default_value, ' + . 'DATETIME_PRECISION AS datetime_precision, EXTRA AS extra FROM information_schema.columns ' + . 'WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?', + [$connection->getTablePrefix() . 'workflow_service_call_dispatches', 'available_at'], + ); + } + + private function migration(): WorkflowMigration + { + return require dirname(__DIR__, 3) + . '/src/migrations/2026_11_14_000100_preserve_service_dispatch_available_at_on_mariadb.php'; + } +} diff --git a/tests/Unit/migrations/TimerDeadlineMetadataMigrationTest.php b/tests/Unit/migrations/TimerDeadlineMetadataMigrationTest.php new file mode 100644 index 000000000..b581e344c --- /dev/null +++ b/tests/Unit/migrations/TimerDeadlineMetadataMigrationTest.php @@ -0,0 +1,236 @@ +assertInstanceOf(DatabaseManager::class, $manager); + $this->databaseManager = $manager; + config([ + 'workflows.storage.connection' => 'timer_metadata', + ]); + } + + protected function tearDown(): void + { + DB::swap($this->databaseManager); + parent::tearDown(); + } + + public function testOtherDriversAndMysqlDoNotInspectOrChangeTheSchema(): void + { + foreach ([$this->createMock(Connection::class), $this->createMock(MySqlConnection::class)] as $connection) { + if ($connection instanceof MySqlConnection) { + $connection->method('isMaria') + ->willReturn(false); + } + $connection->expects($this->never()) + ->method('selectOne'); + $connection->expects($this->never()) + ->method('statement'); + DB::shouldReceive('connection')->with('timer_metadata')->twice()->andReturn($connection); + $this->migration() + ->up(); + $this->migration(true) + ->up(); + } + } + + #[DataProvider('alreadySafeMetadata')] + public function testMissingOrNonautomaticColumnsRemainUntouched(?object $column): void + { + $connection = $this->mariaConnection(); + $connection->expects($this->exactly(2)) + ->method('selectOne') + ->willReturn($column); + $connection->expects($this->never()) + ->method('statement'); + $this->migration() + ->up(); + $this->migration() + ->down(); + $this->migration(true) + ->up(); + $this->migration(true) + ->down(); + } + + public static function alreadySafeMetadata(): iterable + { + yield 'missing column' => [null]; + yield 'corrected timestamp' => [ + (object) [ + 'extra' => '', + ]]; + yield 'unrelated metadata' => [ + (object) [ + 'extra' => 'DEFAULT_GENERATED', + ]]; + } + + #[DataProvider('unsupportedMetadata')] + /** + * @param array $change + */ + public function testUnsupportedAutomaticMetadataCannotChangeRowsSchemaOrSession( + array $change, + bool $serviceDispatch + ): void { + $connection = $this->mariaConnection(); + $connection->expects($this->once()) + ->method('selectOne') + ->willReturn((object) array_replace(self::legacyMetadata(), $change)); + $connection->expects($this->never()) + ->method('statement'); + $this->expectException(UnexpectedValueException::class); + $this->expectExceptionMessage($serviceDispatch + ? 'workflow_service_dispatch_available_at_legacy_shape_unavailable' + : 'workflow_timer_fire_at_legacy_shape_unavailable'); + $this->migration($serviceDispatch) + ->up(); + } + + public static function unsupportedMetadata(): iterable + { + $changes = [ + 'type' => [ + 'column_type' => 'datetime(6)', + ], + 'nullable' => [ + 'is_nullable' => 'YES', + ], + 'custom default' => [ + 'default_value' => '2026-11-08 04:00:20.234567', + ], + 'precision' => [ + 'datetime_precision' => 3, + ], + 'additional automatic behavior' => [ + 'extra' => 'on update current_timestamp(6) DEFAULT_GENERATED', + ], + ]; + foreach ($changes as $name => $change) { + yield 'timer ' . $name => [$change, false]; + yield 'service dispatch ' . $name => [$change, true]; + } + } + + #[DataProvider('schemaOutcomes')] + public function testKnownShapeUsesConfiguredPrefixAndRestoresTheLockBudget( + bool $ddlFails, + bool $serviceDispatch + ): void { + $table = $serviceDispatch ? 'workflow_service_call_dispatches' : 'workflow_run_timers'; + $column = $serviceDispatch ? 'available_at' : 'fire_at'; + $connection = $this->mariaConnection(); + $connection->expects($this->exactly(2)) + ->method('selectOne') + ->willReturnCallback(function (string $sql, array $bindings = []) use ($table, $column): object { + if (str_contains($sql, 'information_schema.columns')) { + $this->assertSame(['tenant_' . $table, $column], $bindings); + + return (object) self::legacyMetadata(); + } + $this->assertSame('SELECT @@SESSION.lock_wait_timeout AS lock_wait_seconds', $sql); + + return (object) [ + 'lock_wait_seconds' => 37, + ]; + }); + $statements = []; + $connection->expects($this->exactly(3)) + ->method('statement') + ->willReturnCallback(static function (string $sql) use (&$statements, $ddlFails): bool { + $statements[] = $sql; + if ($ddlFails && str_starts_with($sql, 'ALTER TABLE')) { + throw new PDOException('schema lock refused'); + } + + return true; + }); + try { + $this->migration($serviceDispatch) + ->up(); + $this->assertFalse($ddlFails); + } catch (PDOException $exception) { + $this->assertTrue($ddlFails); + $this->assertSame('schema lock refused', $exception->getMessage()); + } + $this->assertSame([ + 'SET SESSION lock_wait_timeout = 1', + "ALTER TABLE `tenant_{$table}` MODIFY `{$column}` TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), ALGORITHM=INPLACE, LOCK=NONE", + 'SET SESSION lock_wait_timeout = 37', + ], $statements); + $this->migration($serviceDispatch) + ->down(); + } + + public static function schemaOutcomes(): iterable + { + yield 'timer corrected' => [false, false]; + yield 'timer schema lock refused' => [true, false]; + yield 'service dispatch corrected' => [false, true]; + yield 'service dispatch schema lock refused' => [true, true]; + } + + private function mariaConnection(): MySqlConnection&MockObject + { + $connection = $this->createMock(MySqlConnection::class); + $connection->method('isMaria') + ->willReturn(true); + $connection->method('getDriverName') + ->willReturn('mysql'); + $connection->method('getTablePrefix') + ->willReturn('tenant_'); + $connection->method('getQueryGrammar') + ->willReturn(new MySqlGrammar($connection)); + $connection->expects($this->never()) + ->method('getPdo'); + DB::shouldReceive('connection')->with('timer_metadata')->andReturn($connection); + + return $connection; + } + + /** + * @return array{column_type: string, is_nullable: string, default_value: string, datetime_precision: int, extra: string} + */ + private static function legacyMetadata(): array + { + return [ + 'column_type' => 'timestamp(6)', + 'is_nullable' => 'NO', + 'default_value' => 'current_timestamp(6)', + 'datetime_precision' => 6, + 'extra' => 'on update current_timestamp(6)', + ]; + } + + private function migration(bool $serviceDispatch = false): WorkflowMigration + { + $file = $serviceDispatch + ? '2026_11_14_000100_preserve_service_dispatch_available_at_on_mariadb.php' + : '2026_11_13_000200_preserve_workflow_timer_fire_at_on_mariadb.php'; + + return require dirname(__DIR__, 3) . '/src/migrations/' . $file; + } +}