Preserve fractional replay clocks during redrive

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

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);

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),
             ];
         }
 

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) {

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';

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);
+    }
 }