diff --git a/regression-corpus-policy.json b/regression-corpus-policy.json index 2f415569..90f1a468 100644 --- a/regression-corpus-policy.json +++ b/regression-corpus-policy.json @@ -136,6 +136,15 @@ "content_patterns": [ "function\\s+workflowTaskHistory\\s*\\(" ] + }, + { + "glob": "src/Worker/BusinessReplayClock.php" + }, + { + "glob": "src/Worker/QueryContext.php" + }, + { + "glob": "src/Exception/ReplayClockUnavailable.php" } ] } diff --git a/scripts/ci/run-replay-regression-fixture.php b/scripts/ci/run-replay-regression-fixture.php index f50a3d4f..66ec146f 100644 --- a/scripts/ci/run-replay-regression-fixture.php +++ b/scripts/ci/run-replay-regression-fixture.php @@ -975,6 +975,21 @@ private static function workflow(string $workflowType): callable return $context->activity('Tests\\Fixtures\\V2\\TestRedriveFirstActivity', [$name, $token, $version]); }, + 'golden.business-replay-clock' => static function (WorkflowContext $context, mixed $name, mixed $seconds): array { + // Test-only compatibility observation: an old product has no accessor. + $read = static function () use ($context): ?string { + if (!method_exists($context, 'replayNow')) { return null; } + try { return $context->replayNow()->format('Y-m-d\TH:i:s.uP'); } + catch (\DurableWorkflow\Exception\ReplayClockUnavailable) { return null; } + }; + $times = [$read()]; + $context->activity('Tests\\Fixtures\\V2\\TestRedriveFirstActivity', [$name]); + $times[] = $read(); + $context->sleep((int) $seconds); + $times[] = $read(); + + return $times; + }, 'golden.redrive-reused-timer' => static function (WorkflowContext $context, mixed $name, mixed $seconds): mixed { $greeting = $context->activity('Tests\\Fixtures\\V2\\TestRedriveFirstActivity', [$name]); $context->sleep((int) $seconds); diff --git a/src/Exception/ReplayClockUnavailable.php b/src/Exception/ReplayClockUnavailable.php new file mode 100644 index 00000000..e0547e20 --- /dev/null +++ b/src/Exception/ReplayClockUnavailable.php @@ -0,0 +1,10 @@ +> */ + private array $bySequence = []; + + private ?DateTimeImmutable $time = null; + private bool $active = false; + private bool $redriveClaimed = false; + private ?string $sourceRun = null; + private ?int $resumeStep = null; + private ?DateTimeImmutable $redriveInserted = null; + + /** @param list> $history */ + public function __construct(private readonly array $history, private readonly string $workflowId, private readonly string $runId) + { + $starts = []; + foreach ($history as $index => $event) { + if (self::kind($event) === 'WorkflowStarted') { + $starts[] = $index; + } + $payload = $event['payload'] ?? null; + if (is_array($payload) && ((self::kind($event) === 'WorkflowStarted' && ($payload['recovery_kind'] ?? null) === 'redrive') + || (self::kind($event) === 'StartAccepted' && ($payload['outcome'] ?? null) === 'redriven'))) { + $this->redriveClaimed = true; + } + if (is_array($payload) && ($sequence = self::sequence($payload)) !== null) { + $this->bySequence[$sequence][] = $index; + } + } + if (count($starts) === 1) { + $this->initializeRedrive($starts[0], $workflowId, $runId); + } + } + + public function time(): DateTimeImmutable + { + if (!$this->active || $this->time === null) { + throw new ReplayClockUnavailable('Business replay time requires an active, validated consumed boundary.'); + } + + return $this->time; + } + + /** The engine leaves this scope before suspension teardown can run finally. + * @template TResult + * @param callable(): TResult $callback + * @return TResult + */ + public function invoke(callable $callback): mixed + { + $this->active = true; + try { + return $callback(); + } finally { + $this->active = false; + } + } + + /** Every resume resets authority, including unsupported or unrecorded calls. + * @param array|null $step The step actually matched by the executor. + */ + public function consume(?array $step): void + { + $this->time = null; + if (!self::identity($this->workflowId) || !self::identity($this->runId) + || ($this->redriveClaimed && $this->sourceRun === null) + || $step === null || !in_array($step['shape'] ?? null, ['activity', 'timer'], true) + || !is_int($step['sequence'] ?? null) || !is_int($step['resolution_order'] ?? null)) { + return; + } + $index = $step['resolution_order']; + $event = $this->history[$index] ?? null; + if (!is_array($event) || !$this->scalarBoundary($index, $step['shape'], $step['sequence'])) { + return; + } + $payload = $event['payload']; + $reusedPrefix = $this->sourceRun !== null && $this->resumeStep !== null && $step['sequence'] < $this->resumeStep; + if (in_array(self::kind($event), ['ActivityCompleted', 'TimerFired'], true) + && ($reusedPrefix || array_key_exists('reused_recorded_at', $payload) || array_key_exists('reused_from_run_id', $payload))) { + $this->time = $this->reusedTime($payload, $step['shape'], $step['sequence']); + } elseif (!$reusedPrefix) { + $this->time = self::recordedTime($event); + } + } + + /** @param array $payload */ + private function reusedTime(array $payload, string $family, int $sequence): ?DateTimeImmutable + { + if ($this->sourceRun === null || ($payload['reused_from_run_id'] ?? null) !== $this->sourceRun + || $this->resumeStep === null || $sequence >= $this->resumeStep) { + return null; + } + if ($family === 'activity' && (!self::identity($payload['reused_activity_execution_id'] ?? null) + || $payload['reused_activity_execution_id'] === $payload['activity_execution_id'])) { + return null; + } + $original = self::parseTime($payload['reused_recorded_at'] ?? null); + + return $original !== null && $this->redriveInserted !== null && $original <= $this->redriveInserted ? $original : null; + } + + private function initializeRedrive(int $index, string $workflowId, string $runId): void + { + $payload = $this->history[$index]['payload'] ?? null; + if (!is_array($payload) || ($payload['recovery_kind'] ?? null) !== 'redrive' + || !$this->startPrefix($index, $payload)) { + return; + } + if (!$this->redriveIdentity($payload, $workflowId, $runId) || !$this->orderedHistory()) { + return; + } + $inserted = self::recordedTime($this->history[$index]); + $original = self::parseTime($payload['replayed_started_at'] ?? null); + if ($inserted === null || $original === null || $original > $inserted) { + return; + } + $this->sourceRun = $payload['continued_from_run_id']; + $this->resumeStep = $payload['resume_step_sequence']; + $this->redriveInserted = $inserted; + $this->time = $original; + } + + /** @param array $payload */ + private function redriveIdentity(array $payload, string $workflowId, string $runId): bool + { + foreach (['workflow_instance_id', 'workflow_run_id', 'workflow_command_id', 'continued_from_run_id', 'workflow_link_id'] as $key) { + if (!self::identity($payload[$key] ?? null)) { + return false; + } + } + + return $workflowId !== '' && $runId !== '' && $payload['workflow_instance_id'] === $workflowId + && $payload['workflow_run_id'] === $runId && $payload['continued_from_run_id'] !== $runId + && is_int($payload['resume_step_sequence'] ?? null) && $payload['resume_step_sequence'] >= 1; + } + + private function orderedHistory(): bool + { + foreach ($this->history as $position => $event) { + if (($event['sequence'] ?? null) !== $position + 1) { + return false; + } + } + + return true; + } + + /** The shared StartAccepted prefix proves correlation, never producer identity. + * @param array $payload + */ + private function startPrefix(int $index, array $payload): bool + { + if ($index === 0) { + return true; + } + $accepted = $index === 1 && self::kind($this->history[0]) === 'StartAccepted' + ? ($this->history[0]['payload'] ?? null) : null; + if (!is_array($accepted) || ($accepted['outcome'] ?? null) !== 'redriven') { + return false; + } + + return $this->acceptedIdentity($accepted, $payload); + } + + /** @param array $accepted + * @param array $payload + */ + private function acceptedIdentity(array $accepted, array $payload): bool + { + foreach (['workflow_instance_id', 'workflow_run_id', 'workflow_command_id'] as $key) { + if (!self::identity($payload[$key] ?? null) || ($accepted[$key] ?? null) !== $payload[$key]) { + return false; + } + } + + return true; + } + + private function scalarBoundary(int $index, string $family, int $sequence): bool + { + $event = $this->history[$index]; + $payload = $event['payload'] ?? null; + if (!is_array($payload) || self::sequence($payload) !== $sequence) { + return false; + } + $terminals = $family === 'activity' ? self::ACTIVITY_TERMINALS : ['TimerFired']; + $resource = $family === 'activity' ? 'activity_execution_id' : 'timer_id'; + $identity = $payload[$resource] ?? null; + if (!in_array(self::kind($event), $terminals, true) || !self::identity($identity)) { + return false; + } + return $this->scalarLifecycle($index, $family, $sequence, $resource, $identity, $terminals, $payload); + } + + /** @param list $terminals + * @param array $payload + */ + private function scalarLifecycle(int $index, string $family, int $sequence, string $resource, string $identity, array $terminals, array $payload): bool + { + $opening = $family === 'activity' ? 'ActivityScheduled' : 'TimerScheduled'; + $allowed = [$opening, ...$terminals]; + if ($family === 'activity') { + $allowed[] = 'ActivityStarted'; + } + $positions = []; + foreach ($this->bySequence[$sequence] ?? [] as $position) { + $row = $this->history[$position]; + $body = $row['payload']; + $kind = self::kind($row); + if ($position > $index || !in_array($kind, $allowed, true) || ($body[$resource] ?? null) !== $identity + || !$this->scalarRow($body, $family)) { + return false; + } + $positions[$kind][] = $position; + } + $openings = $positions[$opening] ?? []; + $starts = $positions['ActivityStarted'] ?? []; + $closures = array_merge(...array_values(array_intersect_key($positions, array_flip($terminals)))); + $reusedActivity = $family === 'activity' && self::kind($this->history[$index]) === 'ActivityCompleted' + && array_key_exists('reused_recorded_at', $payload) && $this->sourceRun !== null; + + return $this->scalarOrder($openings, $starts, $closures, $index, $reusedActivity); + } + + /** @param list $openings + * @param list $starts + * @param list $closures + */ + private function scalarOrder(array $openings, array $starts, array $closures, int $index, bool $reusedActivity): bool + { + if ($closures !== [$index] || count($starts) > 1 || ($starts !== [] && ($openings === [] || $starts[0] < $openings[0]))) { + return false; + } + + return (count($openings) === 1 && $openings[0] < $index) || ($reusedActivity && $openings === [] && $starts === []); + } + + /** @param array $body */ + private function scalarRow(array $body, string $family): bool + { + if (!in_array($body['execution_mode'] ?? null, [null, 'remote'], true) || ($body['local_activity'] ?? false) !== false + || ($body['parallel_group'] ?? null) !== null || !in_array($body['parallel_group_path'] ?? null, [null, []], true) + || ($body['selection_group'] ?? null) !== null || ($body['selection_group_id'] ?? null) !== null + || ($body['parallel_group_id'] ?? null) !== null || ($body['parallel_group_base_sequence'] ?? null) !== null + || ($body['parallel_group_size'] ?? null) !== null || ($body['cancellation_cleanup'] ?? null) !== null + || !in_array($body['cancellation_scope_id'] ?? null, [null, 'root'], true)) { + return false; + } + + return $family !== 'timer' || (in_array($body['timer_kind'] ?? null, [null, 'sleep', 'durable_timer'], true) + && !array_key_exists('condition_wait_id', $body) && !array_key_exists('signal_wait_id', $body)); + } + + /** @param array $event */ + private static function kind(array $event): ?string + { + if (array_key_exists('event_type', $event) && array_key_exists('type', $event) && $event['event_type'] !== $event['type']) { + return null; + } + $kind = $event['event_type'] ?? $event['type'] ?? null; + + return is_string($kind) && $kind !== '' ? $kind : null; + } + + /** @param array $payload */ + private static function sequence(array $payload): ?int + { + $value = $payload['sequence'] ?? null; + if (!is_int($value) || $value < 1 || (array_key_exists('workflow_sequence', $payload) && $payload['workflow_sequence'] !== $value)) { + return null; + } + + return $value; + } + + private static function identity(mixed $value): bool + { + return is_string($value) && trim($value) !== ''; + } + + /** Public-history timestamp is authoritative when present, including invalid values. + * @param array $event + */ + private static function recordedTime(array $event): ?DateTimeImmutable + { + return self::parseTime(array_key_exists('timestamp', $event) ? $event['timestamp'] : ($event['recorded_at'] ?? null)); + } + + private static function parseTime(mixed $value): ?DateTimeImmutable + { + if (!is_string($value) || preg_match('/\A\d{4}-\d{2}-\d{2}T(?:[01]\d|2[0-3]):[0-5]\d:[0-5]\d(?:\.\d{1,6})?(?:Z|[+-](?:[01]\d|2[0-3]):[0-5]\d)\z/', $value) !== 1) { + return null; + } + try { + $time = new DateTimeImmutable($value); + } catch (\Exception) { + return null; + } + $errors = DateTimeImmutable::getLastErrors(); + return $errors !== false && ($errors['warning_count'] !== 0 || $errors['error_count'] !== 0) + ? null : $time->setTimezone(new DateTimeZone('UTC')); + } +} diff --git a/src/Worker/QueryContext.php b/src/Worker/QueryContext.php index ba474816..2af7fb9b 100644 --- a/src/Worker/QueryContext.php +++ b/src/Worker/QueryContext.php @@ -4,9 +4,18 @@ namespace DurableWorkflow\Worker; +use DateTimeImmutable; +use DurableWorkflow\Exception\ReplayClockUnavailable; + /** Immutable committed state supplied to query and update handlers. */ final class QueryContext { + /** Query and update clocks require a separately validated consumed prefix. */ + public function replayNow(): DateTimeImmutable + { + throw new ReplayClockUnavailable('This callback has no validated consumed replay-time prefix.'); + } + /** * @param list> $history * @param array $task diff --git a/src/Worker/Replayer.php b/src/Worker/Replayer.php index f56a0630..a7cd0834 100644 --- a/src/Worker/Replayer.php +++ b/src/Worker/Replayer.php @@ -173,7 +173,8 @@ public function replay( $allowScopeAuthoring = $allowCancellationScopeAuthoring && ($cancellation->request === null || $replayCommittedCancellationScopes); $context = null; - $execution = new Fiber(ExecutionTelemetry::bind(function () use ($handler, $history, $input, $task, $cancellation, $scopes, $localActivityExecutor, $prepareLocalActivities, $prepareLocalActivityGroups, $localActivityCancellationPolicies, $allowScopeAuthoring, $allowScopedPreparedLocalActivities, &$context): mixed { + $businessReplayClock = new BusinessReplayClock($history, (string) ($task['workflow_id'] ?? ''), (string) ($task['run_id'] ?? '')); + $execution = new Fiber(ExecutionTelemetry::bind(function () use ($handler, $history, $input, $task, $businessReplayClock, $cancellation, $scopes, $localActivityExecutor, $prepareLocalActivities, $prepareLocalActivityGroups, $localActivityCancellationPolicies, $allowScopeAuthoring, $allowScopedPreparedLocalActivities, &$context): mixed { $current = Fiber::getCurrent(); if ($current === null) { throw new LogicException('Workflow execution did not start inside its Fiber.'); @@ -195,6 +196,7 @@ public function replay( $allowScopeAuthoring, $scopes->openings !== [], $allowScopedPreparedLocalActivities, + $businessReplayClock, ); try { @@ -209,7 +211,7 @@ public function replay( $commands = []; /** @var array $versionDecisions */ $versionDecisions = []; - $suspended = $execution->start(); + $suspended = $businessReplayClock->invoke($execution->start(...)); $nextSequence = 1; $cancellationConsumed = false; $consumedScopeDeliveries = []; @@ -225,7 +227,7 @@ public function replay( } if ($suspended instanceof WorkflowCommand && !isset($cancelledConditionSequences[$nextSequence]) && ($conditionResult = $this->unrecordedConditionResult($suspended, $steps[$stepCursor] ?? null)) !== null) { - $suspended = $execution->resume($conditionResult); + $suspended = $this->resumeExecution($execution, $businessReplayClock, $conditionResult); continue; } if (!$cancellationConsumed && $cancellation->delivery?->sequence === $nextSequence @@ -263,7 +265,7 @@ public function replay( $nextSequence = $scopeSequence + $delivery->sequenceSpan; $consumedScopeDeliveries[$scopeSequence] = true; $context->observeCancellationReplayTime($scopeDelivery['event']); - $suspended = $execution->throw($context->deliveredScopeCascade(array_map( + $suspended = $this->throwIntoExecution($execution, $businessReplayClock, $context->deliveredScopeCascade(array_map( static fn (array $state): ScopedCancellationContext => $state['context'], $scopeDelivery['states'], ), array_map(static fn (array $state): array => CancellationScopeTimerCleanup::snapshot( $state['context'], $state['authority_deadline_at'], $scopeDelivery['event'], @@ -336,7 +338,7 @@ public function replay( $cancellationConsumed = true; $context->observeCancellationReplayTime($cancellation->deliveryIndex === null ? null : ($history[$cancellation->deliveryIndex] ?? null)); - $suspended = $execution->throw($context->deliveredCancellation($delivery->requestId, $cancellation->request->context)); + $suspended = $this->throwIntoExecution($execution, $businessReplayClock, $context->deliveredCancellation($delivery->requestId, $cancellation->request->context)); continue; } if ($delivery === null && $boundary !== null && !$context->isCancellationShielded() @@ -554,7 +556,7 @@ public function replay( $selectionCancellations, ); $this->advanceCancellationClock($context, $history, [$selectionResolutionOrders[(string) $groupId]]); - $suspended = $execution->resume($selection); + $suspended = $this->resumeExecution($execution, $businessReplayClock, $selection); continue; } return $this->result($commands, $context); @@ -562,7 +564,7 @@ public function replay( if ($failure !== null) { $this->advanceCancellationClock($context, $history, array_values(array_filter($resolvedOrders, static fn (int $order): bool => $order <= $failure['resolution_order']))); - $suspended = $execution->throw($failure['exception']); + $suspended = $this->throwIntoExecution($execution, $businessReplayClock, $failure['exception']); continue; } if ($preparedCalls !== []) { @@ -576,7 +578,7 @@ public function replay( ksort($results); $this->advanceCancellationClock($context, $history, $resolvedOrders); - $suspended = $execution->resume($suspended->nestedResults(array_values($results))); + $suspended = $this->resumeExecution($execution, $businessReplayClock, $suspended->nestedResults(array_values($results))); continue; } @@ -593,8 +595,8 @@ public function replay( } $this->advanceCancellationClock($context, $history, $resolution['resolution_orders']); $suspended = $resolution['failure'] instanceof Throwable - ? $execution->throw($resolution['failure']) - : $execution->resume($resolution['value']); + ? $this->throwIntoExecution($execution, $businessReplayClock, $resolution['failure']) + : $this->resumeExecution($execution, $businessReplayClock, $resolution['value']); continue; } @@ -603,7 +605,7 @@ public function replay( if ($this->selectionCancellationForHandle($suspended->handle, $selectionCancellations) === null) { $commands[] = $suspended->toWire(); } - $suspended = $execution->resume(null); + $suspended = $this->resumeExecution($execution, $businessReplayClock, null); continue; } @@ -630,7 +632,7 @@ public function replay( } ++$stepCursor; ++$nextSequence; - $suspended = $execution->resume($step['value']); + $suspended = $this->resumeExecution($execution, $businessReplayClock, $step['value']); continue; } if ($suspended->type === 'continue_as_new') { @@ -653,7 +655,7 @@ public function replay( $decision = $versionDecisions[$changeId] ?? null; if ($decision !== null) { $this->assertVersionDecisionCompatible($changeId, $decision, $suspended); - $suspended = $execution->resume($suspended->versionResult($decision['version'])); + $suspended = $this->resumeExecution($execution, $businessReplayClock, $suspended->versionResult($decision['version'])); continue; } @@ -672,7 +674,7 @@ public function replay( 'version' => $version, 'sequence' => $step['sequence'], ]; - $suspended = $execution->resume($suspended->versionResult($version)); + $suspended = $this->resumeExecution($execution, $businessReplayClock, $suspended->versionResult($version)); continue; } if ($step['detail'] !== $changeId) { @@ -698,7 +700,7 @@ public function replay( 'version' => $version, 'sequence' => $step['sequence'], ]; - $suspended = $execution->resume($suspended->versionResult($version)); + $suspended = $this->resumeExecution($execution, $businessReplayClock, $suspended->versionResult($version)); continue; } @@ -715,7 +717,7 @@ public function replay( 'version' => $version, 'sequence' => null, ]; - $suspended = $execution->resume($suspended->versionResult($version)); + $suspended = $this->resumeExecution($execution, $businessReplayClock, $suspended->versionResult($version)); continue; } @@ -733,7 +735,7 @@ public function replay( 'version' => $version, 'sequence' => null, ]; - $suspended = $execution->resume($suspended->versionResult($version)); + $suspended = $this->resumeExecution($execution, $businessReplayClock, $suspended->versionResult($version)); continue; } @@ -750,8 +752,8 @@ public function replay( } $this->advanceCancellationClock($context, $history, $this->blockingResolutionOrders($awaited)); $suspended = $awaited['failure'] instanceof Throwable - ? $execution->throw($awaited['failure']) - : $execution->resume($awaited['value']); + ? $this->throwIntoExecution($execution, $businessReplayClock, $awaited['failure']) + : $this->resumeExecution($execution, $businessReplayClock, $awaited['value']); continue; } @@ -781,7 +783,7 @@ public function replay( throw new NonDeterministicWorkflow('A committed child start lacks its child call, instance or run identity.', $step['sequence'], reason: 'child_start_identity_missing'); } - $suspended = $execution->resume(new ChildWorkflowHandle($context, $step['sequence'], $identity)); + $suspended = $this->resumeExecution($execution, $businessReplayClock, new ChildWorkflowHandle($context, $step['sequence'], $identity)); continue; } if ($suspended->type === 'open_condition_wait') { @@ -794,12 +796,12 @@ public function replay( if ($step['resolved']) { $this->advanceCancellationClock($context, $history, $this->blockingResolutionOrders($step)); $consumedConditionTail = $this->conditionTailAfterConsumption($step, $suspended, $step['value'] === true); - $suspended = $execution->resume($step['value']); + $suspended = $this->resumeExecution($execution, $businessReplayClock, $step['value']); continue; } if ($suspended->conditionSatisfied()) { $consumedConditionTail = $this->conditionTailAfterConsumption($step, $suspended, true); - $suspended = $execution->resume(true); + $suspended = $this->resumeExecution($execution, $businessReplayClock, true); continue; } @@ -828,7 +830,7 @@ public function replay( } if ($step['failure'] instanceof Throwable) { try { - $suspended = $execution->throw($step['failure']); + $suspended = $this->throwIntoExecution($execution, $businessReplayClock, $step['failure'], $step); } catch (Throwable $failure) { $activityId = $failure instanceof ActivityFailed && is_array($failure->failure) ? ($failure->failure['activity_execution_id'] ?? null) @@ -845,7 +847,7 @@ public function replay( throw $failure; } } else { - $suspended = $execution->resume($step['value']); + $suspended = $this->resumeExecution($execution, $businessReplayClock, $step['value'], $step); } continue; } @@ -875,9 +877,9 @@ public function replay( ++$nextSequence; try { if ($suspended->localResult instanceof Throwable) { - $suspended = $execution->throw($suspended->localResult); + $suspended = $this->throwIntoExecution($execution, $businessReplayClock, $suspended->localResult); } else { - $suspended = $execution->resume($suspended->localResult); + $suspended = $this->resumeExecution($execution, $businessReplayClock, $suspended->localResult); } } catch (Throwable $failure) { return $this->result($commands, $context, $failure); @@ -894,7 +896,7 @@ public function replay( } ++$nextSequence; if (in_array($suspended->type, ['record_side_effect', 'upsert_memo', 'upsert_search_attributes'], true)) { - $suspended = $execution->resume($suspended->localResult); + $suspended = $this->resumeExecution($execution, $businessReplayClock, $suspended->localResult); continue; } @@ -2516,6 +2518,28 @@ private function advanceCancellationClock(WorkflowContext $context, array $histo } } + /** + * @param Fiber $execution + * @param array|null $step + */ + private function resumeExecution(Fiber $execution, BusinessReplayClock $clock, mixed $value, ?array $step = null): mixed + { + $clock->consume($step); + + return $clock->invoke(static fn (): mixed => $execution->resume($value)); + } + + /** + * @param Fiber $execution + * @param array|null $step + */ + private function throwIntoExecution(Fiber $execution, BusinessReplayClock $clock, Throwable $failure, ?array $step = null): mixed + { + $clock->consume($step); + + return $clock->invoke(static fn (): mixed => $execution->throw($failure)); + } + /** * @param array> $selectionCancellations * @return array|null diff --git a/src/Worker/WorkflowContext.php b/src/Worker/WorkflowContext.php index 0fca43c4..8497f4f4 100644 --- a/src/Worker/WorkflowContext.php +++ b/src/Worker/WorkflowContext.php @@ -6,6 +6,7 @@ use DurableWorkflow\Codec\PayloadCodec; use DurableWorkflow\Exception\NonDeterministicWorkflow; +use DurableWorkflow\Exception\ReplayClockUnavailable; use DurableWorkflow\Exception\WorkflowCancelled; use DurableWorkflow\Model\WorkflowStreamAppendItem; use DurableWorkflow\Model\ServiceOperationOptions; @@ -19,6 +20,18 @@ /** Straight-line deterministic operations available while a workflow Fiber is replayed. */ final class WorkflowContext { + /** The recorded time at this invocation's consumed replay boundary. */ + public function replayNow(): DateTimeImmutable + { + if ($this->execution === null || Fiber::getCurrent() !== $this->execution || $this->isCapturing() + || $this->cancellationRequested || $this->deliveredCancellationRequestId !== null || $this->cancellationContext() !== null + || $this->businessReplayClock === null) { + throw new ReplayClockUnavailable('Business replay time requires an active, validated consumed boundary.'); + } + + return $this->businessReplayClock->time(); + } + public const MESSAGE_STREAM_SIGNAL = '__durable_workflow_message_stream'; public const MESSAGE_STREAM_SCHEMA = 'durable-workflow.v2.message-stream.message'; @@ -121,6 +134,7 @@ public function __construct( private readonly bool $allowCancellationScopeAuthoring = false, private bool $hasAuthoredCancellationScopes = false, private readonly bool $allowScopedPreparedLocalActivities = false, + private readonly ?BusinessReplayClock $businessReplayClock = null, ) { $this->execution = $execution; $this->loadMessageStreamMessages(); diff --git a/tests/BusinessReplayClockTest.php b/tests/BusinessReplayClockTest.php new file mode 100644 index 00000000..70513d48 --- /dev/null +++ b/tests/BusinessReplayClockTest.php @@ -0,0 +1,475 @@ +activity(1, self::FIRST), + ...self::timer(2, '2026-11-06T00:00:02.123456+00:00'), + ...self::timer(3, '2026-11-06T00:00:04.123456+00:00')]; + $handler = static function (WorkflowContext $context): array { + $context->activity('anchor'); + $start = $context->replayNow(); + self::assertInstanceOf(DateTimeImmutable::class, $start); + self::assertSame($start, $context->replayNow()); + $deadline = $start->modify('+4 seconds'); + $times = [$start->format('Y-m-d\TH:i:s.uP')]; + while ($context->replayNow() < $deadline) { + $context->sleep(2); + $times[] = $context->replayNow()->format('Y-m-d\TH:i:s.uP'); + } + + return $times; + }; + $first = $this->replayResult($handler, $history); + $cold = $this->replayResult($handler, $history); + + self::assertSame([self::FIRST, '2026-11-06T00:00:02.123456+00:00', '2026-11-06T00:00:04.123456+00:00'], $first); + self::assertSame($first, $cold); + } + + #[DataProvider('nextInstants')] + public function testEqualBackwardAndOffsetInstantsAreNotClamped(string $input, string $expected): void + { + $observed = $this->replayResult(static function (WorkflowContext $context): array { + $context->activity('anchor'); + $first = $context->replayNow(); + $context->activity('anchor'); + + return [$first->format('Y-m-d\TH:i:s.uP'), $context->replayNow()->format('Y-m-d\TH:i:s.uP')]; + }, [...$this->activity(1, self::FIRST), ...$this->activity(2, $input)]); + + self::assertSame([self::FIRST, $expected], $observed); + } + + public static function nextInstants(): array + { + return [ + 'equal' => [self::FIRST, self::FIRST], + 'backward' => ['2026-11-05T23:59:59.000001Z', '2026-11-05T23:59:59.000001+00:00'], + 'offset' => ['2026-11-06T01:00:01.987654+01:00', '2026-11-06T00:00:01.987654+00:00'], + ]; + } + + #[DataProvider('invalidTimes')] + public function testInvalidPresentTimestampCannotFallBackToRecordedAt(mixed $timestamp): void + { + $history = $this->activity(1, self::FIRST); + $history[1]['timestamp'] = $timestamp; + + $this->expectException(ReplayClockUnavailable::class); + $this->replayResult(static function (WorkflowContext $context): string { + $context->activity('anchor'); + + return $context->replayNow()->format('c'); + }, $history); + } + + public static function invalidTimes(): array + { + return array_map(static fn (mixed $value): array => [$value], [ + null, 123, '', '2026-02-30T00:00:00Z', '2026-11-06T24:00:00Z', + '2026-11-06T00:00:00.1234567Z', '2026-11-06T00:00:00+24:00', + '2026-11-06 00:00:00', 'now', + ]); + } + + public function testOnlyTheConsumedBoundarySuppliesTime(): void + { + $history = [...$this->activity(1, self::FIRST), ...$this->activity(2, 'invalid future time')]; + $observed = $this->replayResult(static function (WorkflowContext $context): array { + $context->activity('anchor'); + $before = $context->replayNow()->format('Y-m-d\TH:i:s.uP'); + $context->activity('anchor'); + try { $context->replayNow(); } catch (ReplayClockUnavailable) { return [$before, 'unavailable']; } + + self::fail('The invalid consumed boundary must clear availability.'); + }, $history); + + self::assertSame([self::FIRST, 'unavailable'], $observed); + } + + public function testARecordedSideEffectCannotRetainTimeAndANewScalarMayRecover(): void + { + $history = [...$this->activity(1, self::FIRST), [ + 'event_type' => 'SideEffectRecorded', 'recorded_at' => '2026-11-06T00:00:02Z', + 'payload' => ['sequence' => 2, 'result' => (new AvroPayloadCodec())->envelope('fixed')], + ], ...$this->activity(3, '2026-11-06T00:00:03Z')]; + $observed = $this->replayResult(static function (WorkflowContext $context): array { + $context->activity('anchor'); + $context->sideEffect(static fn (): string => 'fixed'); + try { $context->replayNow(); } catch (ReplayClockUnavailable) { + $context->activity('anchor'); + + return ['unavailable', $context->replayNow()->format('Y-m-d\TH:i:s.uP')]; + } + self::fail('An inline result cannot retain the preceding clock.'); + }, $history); + + self::assertSame(['unavailable', '2026-11-06T00:00:03.000000+00:00'], $observed); + } + + public function testANewInlineSideEffectCannotRetainThePrecedingTime(): void + { + $this->expectException(ReplayClockUnavailable::class); + $this->replayResult(static function (WorkflowContext $context): string { + $context->activity('anchor'); + $context->sideEffect(static fn (): string => 'fixed'); + + return $context->replayNow()->format('c'); + }, $this->activity(1, self::FIRST)); + } + + #[DataProvider('activityFailureTypes')] + public function testFailureDeliveryKeepsItsOwnTimeAndOriginalError(string $eventType): void + { + $history = $this->activity(1, self::FIRST); + $history[1]['event_type'] = $eventType; + $history[1]['payload']['message'] = 'original activity failure'; + $history[1]['payload']['reused_recorded_at'] = '2026-01-01T00:00:00Z'; + $observed = $this->replayResult(static function (WorkflowContext $context): array { + try { $context->activity('anchor'); } catch (ActivityFailed $error) { + return [$error->getMessage(), $context->replayNow()->format('Y-m-d\TH:i:s.uP')]; + } + self::fail('The original activity failure must still be thrown.'); + }, $history); + + self::assertSame(['original activity failure', self::FIRST], $observed); + } + + public static function activityFailureTypes(): array + { + return [['ActivityFailed'], ['ActivityTimedOut'], ['ActivityCancelled']]; + } + + public function testLocalExecutorAndItsInlineResultCannotBorrowBusinessTime(): void + { + $retained = null; + $insideUnavailable = false; + $codec = new AvroPayloadCodec(); + $result = (new Replayer($codec))->replay( + static function (WorkflowContext $context) use (&$retained): string { + $context->activity('anchor'); + $retained = $context; + $context->localActivity('local-anchor'); + try { $context->replayNow(); } catch (ReplayClockUnavailable) { return 'unavailable'; } + self::fail('An inline local result cannot retain its preceding business instant.'); + }, + $this->activity(1, self::FIRST), [], 'php-workers', ['workflow_id' => 'workflow', 'run_id' => 'run'], + static function () use (&$retained, &$insideUnavailable): array { + try { $retained->replayNow(); } catch (ReplayClockUnavailable) { $insideUnavailable = true; } + + return ['outcome' => 'completed', 'result' => 'fixed', + 'attempts' => [['attempt_number' => 1, 'outcome' => 'completed', 'heartbeats' => []]]]; + }, + ); + + self::assertTrue($insideUnavailable); + self::assertSame('unavailable', $codec->decodeEnvelope($result->commands[array_key_last($result->commands)]['result'])); + } + + public function testRetainedContextCannotBeReadAfterScopeExit(): void + { + $retained = null; + $this->replayResult(static function (WorkflowContext $context) use (&$retained): string { + $context->activity('anchor'); + $retained = $context; + + return $context->replayNow()->format('c'); + }, $this->activity(1, self::FIRST)); + + self::assertInstanceOf(WorkflowContext::class, $retained); + $this->expectException(ReplayClockUnavailable::class); + $retained->replayNow(); + } + + #[DataProvider('ambiguousActivityHistories')] + public function testUnprovedActivityLifecyclesDoNotSupplyTime(string $mutation): void + { + $history = $this->activity(1, self::FIRST); + switch ($mutation) { + case 'missing-resource': unset($history[1]['payload']['activity_execution_id']); break; + case 'different-resource': $history[1]['payload']['activity_execution_id'] = 'other-activity'; break; + case 'coerced-sequence': + foreach ($history as &$event) { $event['payload']['sequence'] = '1'; } + unset($event); + break; + case 'duplicate-terminal': $history[] = $history[1]; break; + case 'conflicting-terminal': + $history[] = [...$history[1], 'event_type' => 'ActivityFailed']; + break; + case 'local-mode': $history[1]['payload']['execution_mode'] = 'local'; break; + case 'local-flag': $history[1]['payload']['local_activity'] = true; break; + case 'missing-opening': array_shift($history); break; + case 'duplicate-opening': array_unshift($history, $history[0]); break; + case 'started-before-opening': + array_unshift($history, [...$history[0], 'event_type' => 'ActivityStarted']); + break; + case 'partial-reuse': $history[1]['payload']['reused_from_run_id'] = 'source'; break; + } + + $this->expectException($mutation === 'local-mode' ? NonDeterministicWorkflow::class : ReplayClockUnavailable::class); + $this->replayResult(static function (WorkflowContext $context): string { + $context->activity('anchor'); + + return $context->replayNow()->format('c'); + }, $history); + } + + public static function ambiguousActivityHistories(): array + { + return array_map(static fn (string $mutation): array => [$mutation], [ + 'missing-resource', 'different-resource', 'coerced-sequence', 'duplicate-terminal', + 'conflicting-terminal', 'local-mode', 'local-flag', 'missing-opening', 'duplicate-opening', + 'started-before-opening', 'partial-reuse', + ]); + } + + public function testNestedInvocationCannotBorrowParentTimeAndAnErrorRestoresParentScope(): void + { + $failure = new \LogicException('original nested error'); + $observed = $this->replayResult(function (WorkflowContext $parent) use ($failure): array { + $parent->activity('anchor'); + $before = $parent->replayNow(); + try { + $this->replayResult(static function (WorkflowContext $nested) use ($parent, $failure): never { + $nested->activity('anchor'); + self::assertSame('2026-11-06T01:00:00.000000+00:00', $nested->replayNow()->format('Y-m-d\TH:i:s.uP')); + try { + $parent->replayNow(); + self::fail('A different workflow Fiber cannot read the parent clock.'); + } catch (ReplayClockUnavailable) { + throw $failure; + } + }, $this->activity(1, '2026-11-06T01:00:00Z')); + self::fail('The nested error must propagate unchanged.'); + } catch (\LogicException $error) { + self::assertSame($failure, $error); + } + + return [$before->format('Y-m-d\TH:i:s.uP'), $parent->replayNow()->format('Y-m-d\TH:i:s.uP')]; + }, $this->activity(1, self::FIRST)); + + self::assertSame([self::FIRST, self::FIRST], $observed); + } + + public function testExecutorPredicateCannotBorrowTheWorkflowClock(): void + { + $this->expectException(ReplayClockUnavailable::class); + $this->replayResult(static function (WorkflowContext $context): bool { + $context->activity('anchor'); + + return $context->waitCondition(static fn (): bool => $context->replayNow() instanceof DateTimeImmutable, 'clock-condition'); + }, $this->activity(1, self::FIRST)); + } + + public function testGroupCaptureCallbackCannotBorrowTheWorkflowClock(): void + { + $this->expectException(ReplayClockUnavailable::class); + $this->replayResult(static function (WorkflowContext $context): array { + $context->activity('anchor'); + + return $context->all([static fn () => $context->replayNow()]); + }, $this->activity(1, self::FIRST)); + } + + public function testForceClosedFinallyHasNoActiveBusinessClock(): void + { + $observed = null; + (new Replayer(new AvroPayloadCodec()))->replay( + static function (WorkflowContext $context) use (&$observed): void { + $context->activity('anchor'); + try { + $context->sleep(5); + } finally { + try { + $observed = $context->replayNow()->format('c'); + } catch (ReplayClockUnavailable) { + $observed = 'unavailable'; + } + } + }, + $this->activity(1, self::FIRST), [], 'php-workers', + ['workflow_id' => 'workflow', 'run_id' => 'run'], + ); + gc_collect_cycles(); + + self::assertSame('unavailable', $observed); + } + + public function testExplicitRedriveEntryAndReusedStepsUseOriginalTimes(): void + { + $fixture = self::redriveFixture(); + $times = $this->replayResult(self::redriveHandler(), $fixture['history'], $fixture['workflow_tasks'][0]); + + self::assertSame(['2026-11-03T00:00:00.000000+00:00', '2026-11-03T00:00:00.000000+00:00', '2026-11-03T00:00:06.000000+00:00'], $times); + } + + public function testCorrelatedStartAcceptedPrefixDoesNotChangeRedriveTimes(): void + { + $fixture = self::redriveFixture(); + $start = $fixture['history'][0]; + foreach ($fixture['history'] as &$event) { ++$event['sequence']; } + unset($event); + array_unshift($fixture['history'], [ + 'sequence' => 1, 'event_type' => 'StartAccepted', 'recorded_at' => $start['recorded_at'], + 'payload' => ['outcome' => 'redriven', ...array_intersect_key($start['payload'], + array_flip(['workflow_instance_id', 'workflow_run_id', 'workflow_command_id']))], + ]); + + self::assertSame($this->replayResult(self::redriveHandler(), self::redriveFixture()['history'], $fixture['workflow_tasks'][0]), + $this->replayResult(self::redriveHandler(), $fixture['history'], $fixture['workflow_tasks'][0])); + } + + #[DataProvider('invalidRedriveEvidence')] + public function testUncorrelatedRedriveEvidenceRemainsUnavailable(string $mutation): void + { + $fixture = self::redriveFixture(); + $expected = ['2026-11-03T00:00:00.000000+00:00', '2026-11-03T00:00:00.000000+00:00', '2026-11-03T00:00:06.000000+00:00']; + switch ($mutation) { + case 'wrong-source': $fixture['history'][1]['payload']['reused_from_run_id'] = 'other-source'; $expected[1] = null; break; + case 'missing-original-activity': unset($fixture['history'][1]['payload']['reused_activity_execution_id']); $expected[1] = null; break; + case 'same-original-activity': + $fixture['history'][1]['payload']['reused_activity_execution_id'] = $fixture['history'][1]['payload']['activity_execution_id']; + $expected[1] = null; break; + case 'missing-timer-source': unset($fixture['history'][3]['payload']['reused_from_run_id']); $expected[2] = null; break; + case 'missing-timer-time': unset($fixture['history'][3]['payload']['reused_recorded_at']); $expected[2] = null; break; + case 'missing-reuse-pair': + unset($fixture['history'][3]['payload']['reused_from_run_id'], $fixture['history'][3]['payload']['reused_recorded_at']); + $expected[2] = null; break; + case 'future-original-time': $fixture['history'][3]['payload']['reused_recorded_at'] = '2026-11-03T00:06:00Z'; $expected[2] = null; break; + case 'unbound-resume': $fixture['history'][0]['payload']['resume_step_sequence'] = '3'; $expected = [null, null, null]; break; + case 'at-resume-step': $fixture['history'][0]['payload']['resume_step_sequence'] = 2; $expected[2] = null; break; + case 'wrong-run': $fixture['history'][0]['payload']['workflow_run_id'] = 'other-run'; $expected = [null, null, null]; break; + case 'ordinary-start': unset($fixture['history'][0]['payload']['recovery_kind']); $expected = [null, null, null]; break; + } + + self::assertSame($expected, $this->replayResult(self::redriveHandler(), $fixture['history'], $fixture['workflow_tasks'][0])); + } + + public static function invalidRedriveEvidence(): array + { + return array_map(static fn (string $mutation): array => [$mutation], [ + 'wrong-source', 'missing-original-activity', 'same-original-activity', 'missing-timer-source', + 'missing-timer-time', 'missing-reuse-pair', 'future-original-time', 'unbound-resume', 'at-resume-step', 'wrong-run', 'ordinary-start', + ]); + } + + public function testLegacyCancellationFlagKeepsBusinessTimeUnavailable(): void + { + $this->expectException(ReplayClockUnavailable::class); + $this->replayResult(static function (WorkflowContext $context): string { + $context->activity('anchor'); + self::assertTrue($context->isCancellationRequested()); + + return $context->replayNow()->format('c'); + }, $this->activity(1, self::FIRST), ['workflow_id' => 'workflow', 'run_id' => 'run', 'cancel_requested' => true]); + } + + private static function redriveFixture(): array + { + return json_decode((string) file_get_contents(__DIR__.'/fixtures/replay-regressions/business-replay-clock-canonical-redrive.json'), true, flags: JSON_THROW_ON_ERROR); + } + + private static function redriveHandler(): callable + { + return static function (WorkflowContext $context): array { + $read = static function () use ($context): ?string { + try { return $context->replayNow()->format('Y-m-d\TH:i:s.uP'); } + catch (ReplayClockUnavailable) { return null; } + }; + $times = [$read()]; + $context->activity('Tests\\Fixtures\\V2\\TestRedriveFirstActivity', ['Taylor']); + $times[] = $read(); + $context->sleep(5); + $times[] = $read(); + + return $times; + }; + } + + /** @return list> */ + private function activity(int $sequence, string $timestamp): array + { + $payload = ['sequence' => $sequence, 'activity_execution_id' => 'activity-'.$sequence, 'activity_type' => 'anchor']; + + return [ + ['event_type' => 'ActivityScheduled', 'payload' => $payload], + ['event_type' => 'ActivityCompleted', 'recorded_at' => $timestamp, + 'payload' => [...$payload, 'result' => (new AvroPayloadCodec())->envelope('fixed')]], + ]; + } + + /** @return list> */ + private static function timer(int $sequence, string $timestamp): array + { + $payload = ['sequence' => $sequence, 'timer_id' => 'timer-'.$sequence, 'delay_seconds' => 2]; + + return [ + ['event_type' => 'TimerScheduled', 'payload' => $payload], + ['event_type' => 'TimerFired', 'recorded_at' => $timestamp, 'payload' => $payload], + ]; + } + + /** @param list> $history */ + private function replayResult(callable $handler, array $history, array $task = ['workflow_id' => 'workflow', 'run_id' => 'run']): mixed + { + $codec = new AvroPayloadCodec(); + $result = (new Replayer($codec))->replay($handler, $history, [], 'php-workers', $task); + + return $codec->decodeEnvelope($result->commands[array_key_last($result->commands)]['result']); + } + + public function testDirectContextHasNoBusinessReplayTime(): void + { + $context = new WorkflowContext('workflow', 'run', [], new AvroPayloadCodec()); + + $this->expectException(ReplayClockUnavailable::class); + $context->replayNow(); + } + + public function testQuerySnapshotDoesNotGuessAConsumedPrefix(): void + { + $context = new QueryContext('workflow', 'run', [[ + 'event_type' => 'ActivityCompleted', + 'recorded_at' => '2026-11-06T00:00:00.123456+00:00', + 'payload' => ['sequence' => 1, 'activity_execution_id' => 'activity'], + ]], []); + + $this->expectException(ReplayClockUnavailable::class); + $context->replayNow(); + } + + public function testOrdinaryEntryDoesNotSubstituteHistoryInsertionTime(): void + { + $this->expectException(ReplayClockUnavailable::class); + (new Replayer(new AvroPayloadCodec()))->replay( + static fn (WorkflowContext $context) => $context->replayNow(), + [[ + 'event_type' => 'WorkflowStarted', + 'recorded_at' => '2026-11-06T00:00:00.123456+00:00', + 'payload' => ['workflow_run_id' => 'run'], + ]], + [], + 'php-workers', + ['workflow_id' => 'workflow', 'run_id' => 'run'], + ); + } +} diff --git a/tests/Support/RedriveFixtureRegistry.php b/tests/Support/RedriveFixtureRegistry.php index 49825759..81f996df 100644 --- a/tests/Support/RedriveFixtureRegistry.php +++ b/tests/Support/RedriveFixtureRegistry.php @@ -23,6 +23,7 @@ final class RedriveFixtureRegistry * @var array> */ public const SUCCESSORS = [ + 'business-replay-clock-canonical-redrive' => ['ActivityCompleted', 'TimerScheduled', 'TimerFired'], 'redrive-reused-upserts-clock' => ['MemoUpserted', 'SearchAttributesUpserted'], 'redrive-successor-reused-upserts-cancellation-clock' => ['MemoUpserted', 'SearchAttributesUpserted'], 'redrive-reused-signal-wait-clock' => ['SignalWaitOpened', 'SignalReceived', 'SignalApplied'], diff --git a/tests/Support/ReplayRegressionFixture.php b/tests/Support/ReplayRegressionFixture.php index 85dad64b..1ddd8fc2 100644 --- a/tests/Support/ReplayRegressionFixture.php +++ b/tests/Support/ReplayRegressionFixture.php @@ -675,6 +675,21 @@ private static function workflow(string $workflowType): callable return $context->activity('Tests\\Fixtures\\V2\\TestRedriveFirstActivity', [$name, $token, $version]); }, + 'golden.business-replay-clock' => static function (WorkflowContext $context, mixed $name, mixed $seconds): array { + // Test-only compatibility observation: an old product has no accessor. + $read = static function () use ($context): ?string { + if (!method_exists($context, 'replayNow')) { return null; } + try { return $context->replayNow()->format('Y-m-d\TH:i:s.uP'); } + catch (\DurableWorkflow\Exception\ReplayClockUnavailable) { return null; } + }; + $times = [$read()]; + $context->activity('Tests\\Fixtures\\V2\\TestRedriveFirstActivity', [$name]); + $times[] = $read(); + $context->sleep((int) $seconds); + $times[] = $read(); + + return $times; + }, 'golden.redrive-reused-timer' => static function (WorkflowContext $context, mixed $name, mixed $seconds): mixed { $greeting = $context->activity('Tests\\Fixtures\\V2\\TestRedriveFirstActivity', [$name]); $context->sleep((int) $seconds); diff --git a/tests/fixtures/replay-regressions/business-replay-clock-canonical-redrive.json b/tests/fixtures/replay-regressions/business-replay-clock-canonical-redrive.json new file mode 100644 index 00000000..67aa1652 --- /dev/null +++ b/tests/fixtures/replay-regressions/business-replay-clock-canonical-redrive.json @@ -0,0 +1,100 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "php-business-replay-clock-canonical-redrive", + "protocol_version": "1.20", + "bindings": [ + "php" + ], + "workflow": { + "type": "golden.business-replay-clock", + "input": [ + "Taylor", + 5 + ] + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_class": "Tests\\Fixtures\\V2\\TestRedriveTimerWorkflow", + "workflow_type": "golden.business-replay-clock", + "workflow_instance_id": "regression-corpus-redrive-successor-timer", + "workflow_run_id": "regression-corpus-run", + "workflow_command_id": "01JBREDRIVECOMMAND00000000", + "continued_from_run_id": "01JBREDRIVESOURCERUN000000", + "resume_step_sequence": 3, + "recovery_kind": "redrive", + "execution_deadline_at": null, + "run_deadline_at": null, + "replayed_started_at": "2026-11-03T00:00:00+00:00", + "workflow_link_id": "01JBREDRIVELINK00000000000" + } + }, + { + "sequence": 2, + "event_type": "ActivityCompleted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "activity_execution_id": "01JBREDRIVESUCCESSORACT000", + "activity_class": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "sequence": 1, + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro", + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_activity_execution_id": "01JBREDRIVESOURCEACTIVITY0", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 3, + "event_type": "TimerScheduled", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "timer_id": "01JBREDRIVESUCCESSORTIMER0", + "sequence": 2, + "delay_seconds": 5, + "fire_at": "2026-11-03T00:00:05.000000Z" + } + }, + { + "sequence": 4, + "event_type": "TimerFired", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "timer_id": "01JBREDRIVESUCCESSORTIMER0", + "sequence": 2, + "delay_seconds": 5, + "fire_at": "2026-11-03T00:00:05.000000Z", + "fired_at": "2026-11-03T00:00:06.000000Z", + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:06+00:00" + } + } + ], + "workflow_tasks": [ + { + "workflow_id": "regression-corpus-redrive-successor-timer", + "run_id": "regression-corpus-run" + } + ], + "expected": { + "workflow_tasks": [ + { + "command_sequence": [ + { + "type": "complete_workflow", + "result": [ + "2026-11-03T00:00:00.000000+00:00", + "2026-11-03T00:00:00.000000+00:00", + "2026-11-03T00:00:06.000000+00:00" + ] + } + ] + } + ] + } +}