diff --git a/CHANGELOG.md b/CHANGELOG.md index e44897b98..66406cde4 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. + - 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/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 be73b2360..12d6cdc49 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; @@ -595,7 +596,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']); @@ -656,7 +660,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()); @@ -724,7 +731,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']); @@ -1660,6 +1670,110 @@ 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(); + 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(); + $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(); + $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; + } + $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(); + } + } + public function testRedriveReusedTimerCorpusDoesNotWaitAgain(): void { $path = self::FIXTURE_DIR . '/redrive-reused-timer-clock.json'; diff --git a/tests/Feature/V2/V2FailedRunRedriveTest.php b/tests/Feature/V2/V2FailedRunRedriveTest.php index 50e44934b..ce01d5a39 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,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->assertSame( - $original->recorded_at?->toIso8601String(), - $reused->payload['reused_recorded_at'] ?? null, - ); + $this->assertReplayClock($original->recorded_at, $reused->payload['reused_recorded_at'] ?? null); } $expectedTimes = [ @@ -675,7 +674,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 +721,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 +747,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,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->assertSame( - $original->recorded_at?->toIso8601String(), - $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. @@ -1328,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->assertSame( - $original->recorded_at?->toIso8601String(), - $reused->payload['reused_recorded_at'] ?? null - ); + $this->assertReplayClock($original->recorded_at, $reused->payload['reused_recorded_at'] ?? null); } } @@ -1378,8 +1371,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,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->assertSame( - $sourceFired->recorded_at?->toIso8601String(), - $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(); @@ -2124,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->assertSame( - $original->recorded_at?->toIso8601String(), - $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); @@ -2679,4 +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); + } } 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..00f5b5f51 --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/redrive-reused-fractional-clock.json @@ -0,0 +1,108 @@ +{ + "$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": [ + { + "type": "complete_workflow" + } + ] + }, + "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; + } +}