diff --git a/regression-corpus-policy.json b/regression-corpus-policy.json index 90f1a468..adcf4787 100644 --- a/regression-corpus-policy.json +++ b/regression-corpus-policy.json @@ -145,6 +145,12 @@ }, { "glob": "src/Exception/ReplayClockUnavailable.php" + }, + { + "glob": "src/Worker/ReplayClockHistory.php" + }, + { + "glob": "src/Worker/ChildReplayClockEvidence.php" } ] } diff --git a/scripts/ci/run-replay-regression-fixture.php b/scripts/ci/run-replay-regression-fixture.php index 66ec146f..c70a4792 100644 --- a/scripts/ci/run-replay-regression-fixture.php +++ b/scripts/ci/run-replay-regression-fixture.php @@ -975,6 +975,18 @@ private static function workflow(string $workflowType): callable return $context->activity('Tests\\Fixtures\\V2\\TestRedriveFirstActivity', [$name, $token, $version]); }, + 'golden.business-replay-clock-child' => static function (WorkflowContext $context, int $count = 4): array { + $outcomes = []; + for ($index = 0; $index < $count; ++$index) { + try { $value = $context->childWorkflow('clock-child'); } + catch (\DurableWorkflow\Exception\ChildWorkflowFailed $failure) { $value = $failure->failureType; } + try { $time = $context->replayNow()->format('Y-m-d\TH:i:s.uP'); } + catch (\DurableWorkflow\Exception\ReplayClockUnavailable) { $time = null; } + $outcomes[] = ['outcome' => $value, 'time' => $time]; + } + + return $outcomes; + }, '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 { diff --git a/src/Worker/BusinessReplayClock.php b/src/Worker/BusinessReplayClock.php index 563be9ad..41c3e694 100644 --- a/src/Worker/BusinessReplayClock.php +++ b/src/Worker/BusinessReplayClock.php @@ -5,7 +5,6 @@ namespace DurableWorkflow\Worker; use DateTimeImmutable; -use DateTimeZone; use DurableWorkflow\Exception\ReplayClockUnavailable; /** @internal Business time is separate from cancellation's monotonic budget clock. */ @@ -13,6 +12,8 @@ final class BusinessReplayClock { private const ACTIVITY_TERMINALS = ['ActivityCompleted', 'ActivityFailed', 'ActivityTimedOut', 'ActivityCancelled']; + private const CHILD_SHAPE = 'child_workflow'; + /** @var array> */ private array $bySequence = []; @@ -28,15 +29,15 @@ public function __construct(private readonly array $history, private readonly st { $starts = []; foreach ($history as $index => $event) { - if (self::kind($event) === 'WorkflowStarted') { + if (ReplayClockHistory::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'))) { + if (is_array($payload) && ((ReplayClockHistory::kind($event) === 'WorkflowStarted' && ($payload['recovery_kind'] ?? null) === 'redrive') + || (ReplayClockHistory::kind($event) === 'StartAccepted' && ($payload['outcome'] ?? null) === 'redriven'))) { $this->redriveClaimed = true; } - if (is_array($payload) && ($sequence = self::sequence($payload)) !== null) { + if (is_array($payload) && ($sequence = ReplayClockHistory::legacySequenceClaim($payload)) !== null) { $this->bySequence[$sequence][] = $index; } } @@ -75,27 +76,33 @@ public function invoke(callable $callback): mixed public function consume(?array $step): void { $this->time = null; - if (!self::identity($this->workflowId) || !self::identity($this->runId) + if (!ReplayClockHistory::identity($this->workflowId) || !ReplayClockHistory::identity($this->runId) || ($this->redriveClaimed && $this->sourceRun === null) - || $step === null || !in_array($step['shape'] ?? null, ['activity', 'timer'], true) + || $step === null || !in_array($step['shape'] ?? null, ['activity', 'timer', self::CHILD_SHAPE], 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'])) { + if (!is_array($event) || !$this->matchesBoundary($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) + if (in_array(ReplayClockHistory::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); + $this->time = ReplayClockHistory::recordedTime($event); } } + private function matchesBoundary(int $index, string $family, int $sequence): bool + { + return $family === self::CHILD_SHAPE + ? ChildReplayClockEvidence::matches($this->history, $this->bySequence[$sequence] ?? [], $index, $sequence) : $this->scalarBoundary($index, $family, $sequence); + } + /** @param array $payload */ private function reusedTime(array $payload, string $family, int $sequence): ?DateTimeImmutable { @@ -103,11 +110,11 @@ private function reusedTime(array $payload, string $family, int $sequence): ?Dat || $this->resumeStep === null || $sequence >= $this->resumeStep) { return null; } - if ($family === 'activity' && (!self::identity($payload['reused_activity_execution_id'] ?? null) + if ($family === 'activity' && (!ReplayClockHistory::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); + $original = ReplayClockHistory::parseTime($payload['reused_recorded_at'] ?? null); return $original !== null && $this->redriveInserted !== null && $original <= $this->redriveInserted ? $original : null; } @@ -122,8 +129,8 @@ private function initializeRedrive(int $index, string $workflowId, string $runId if (!$this->redriveIdentity($payload, $workflowId, $runId) || !$this->orderedHistory()) { return; } - $inserted = self::recordedTime($this->history[$index]); - $original = self::parseTime($payload['replayed_started_at'] ?? null); + $inserted = ReplayClockHistory::recordedTime($this->history[$index]); + $original = ReplayClockHistory::parseTime($payload['replayed_started_at'] ?? null); if ($inserted === null || $original === null || $original > $inserted) { return; } @@ -137,7 +144,7 @@ private function initializeRedrive(int $index, string $workflowId, string $runId 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)) { + if (!ReplayClockHistory::identity($payload[$key] ?? null)) { return false; } } @@ -166,7 +173,7 @@ private function startPrefix(int $index, array $payload): bool if ($index === 0) { return true; } - $accepted = $index === 1 && self::kind($this->history[0]) === 'StartAccepted' + $accepted = $index === 1 && ReplayClockHistory::kind($this->history[0]) === 'StartAccepted' ? ($this->history[0]['payload'] ?? null) : null; if (!is_array($accepted) || ($accepted['outcome'] ?? null) !== 'redriven') { return false; @@ -181,7 +188,7 @@ private function startPrefix(int $index, array $payload): bool 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]) { + if (!ReplayClockHistory::identity($payload[$key] ?? null) || ($accepted[$key] ?? null) !== $payload[$key]) { return false; } } @@ -193,13 +200,13 @@ 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) { + if (!is_array($payload) || ReplayClockHistory::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)) { + if (!in_array(ReplayClockHistory::kind($event), $terminals, true) || !ReplayClockHistory::identity($identity)) { return false; } return $this->scalarLifecycle($index, $family, $sequence, $resource, $identity, $terminals, $payload); @@ -219,9 +226,10 @@ private function scalarLifecycle(int $index, string $family, int $sequence, stri 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)) { + $kind = ReplayClockHistory::kind($row); + if ($position > $index || ReplayClockHistory::sequence($body) !== $sequence + || !in_array($kind, $allowed, true) || ($body[$resource] ?? null) !== $identity + || !ReplayClockHistory::scalarRow($body, $family)) { return false; } $positions[$kind][] = $position; @@ -229,7 +237,7 @@ private function scalarLifecycle(int $index, string $family, int $sequence, stri $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' + $reusedActivity = $family === 'activity' && ReplayClockHistory::kind($this->history[$index]) === 'ActivityCompleted' && array_key_exists('reused_recorded_at', $payload) && $this->sourceRun !== null; return $this->scalarOrder($openings, $starts, $closures, $index, $reusedActivity); @@ -247,70 +255,4 @@ private function scalarOrder(array $openings, array $starts, array $closures, in 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/ChildReplayClockEvidence.php b/src/Worker/ChildReplayClockEvidence.php new file mode 100644 index 00000000..424a0815 --- /dev/null +++ b/src/Worker/ChildReplayClockEvidence.php @@ -0,0 +1,69 @@ +> $history + * @param list $indices + */ + public static function matches(array $history, array $indices, int $index, int $sequence): bool + { + $event = $history[$index]; + $payload = $event['payload'] ?? null; + $identity = is_array($payload) ? self::childIdentity($payload) : null; + if ($identity === null || ReplayClockHistory::sequence($payload) !== $sequence || !in_array(ReplayClockHistory::kind($event), self::CHILD_TERMINALS, true)) { + return false; + } + $positions = []; + $allowed = ['ChildWorkflowScheduled', 'ChildRunStarted', ...self::CHILD_TERMINALS]; + foreach ($indices as $position) { + $row = $history[$position]; + $body = $row['payload']; + $kind = ReplayClockHistory::kind($row); + if ($position > $index || ReplayClockHistory::sequence($body) !== $sequence + || !in_array($kind, $allowed, true) || self::childIdentity($body) !== $identity + || !ReplayClockHistory::scalarRow($body, 'child_workflow')) { + return false; + } + $positions[$kind][] = $position; + } + $closures = array_merge(...array_values(array_intersect_key($positions, array_flip(self::CHILD_TERMINALS)))); + + $openings = $positions['ChildWorkflowScheduled'] ?? []; + $starts = $positions['ChildRunStarted'] ?? []; + + return $closures === [$index] && count($openings) === 1 && $openings[0] < $index && count($starts) <= 1 + && ($starts === [] || ($starts[0] > $openings[0] && $starts[0] < $index)); + } + + /** @param array $payload + * @return array|null + */ + private static function childIdentity(array $payload): ?array + { + $identity = []; + foreach (self::CHILD_IDENTITY as $field) { + $value = $payload[$field] ?? null; + if (!ReplayClockHistory::identity($value)) { + return null; + } + $identity[$field] = $value; + } + foreach (['workflow_instance_id' => 'child_workflow_instance_id', 'workflow_run_id' => 'child_workflow_run_id'] as $alias => $field) { + if (array_key_exists($alias, $payload) && $payload[$alias] !== $identity[$field]) { + return null; + } + } + + return $identity; + } +} diff --git a/src/Worker/ReplayClockHistory.php b/src/Worker/ReplayClockHistory.php new file mode 100644 index 00000000..580688ee --- /dev/null +++ b/src/Worker/ReplayClockHistory.php @@ -0,0 +1,88 @@ + $body */ + public static 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 */ + public 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 */ + public 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; + } + + /** Index every sequence the legacy resolver can claim, solely to refuse ambiguous evidence. + * @param array $payload + */ + public static function legacySequenceClaim(array $payload): ?int + { + $value = $payload['sequence'] ?? $payload['workflow_sequence'] ?? null; + + return is_numeric($value) && (int) $value > 0 ? (int) $value : null; + } + + public 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 + */ + public static function recordedTime(array $event): ?DateTimeImmutable + { + return self::parseTime(array_key_exists('timestamp', $event) ? $event['timestamp'] : ($event['recorded_at'] ?? null)); + } + + public 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/Replayer.php b/src/Worker/Replayer.php index a7cd0834..acab884d 100644 --- a/src/Worker/Replayer.php +++ b/src/Worker/Replayer.php @@ -752,8 +752,8 @@ public function replay( } $this->advanceCancellationClock($context, $history, $this->blockingResolutionOrders($awaited)); $suspended = $awaited['failure'] instanceof Throwable - ? $this->throwIntoExecution($execution, $businessReplayClock, $awaited['failure']) - : $this->resumeExecution($execution, $businessReplayClock, $awaited['value']); + ? $this->throwIntoExecution($execution, $businessReplayClock, $awaited['failure'], $awaited) + : $this->resumeExecution($execution, $businessReplayClock, $awaited['value'], $awaited); continue; } diff --git a/tests/BusinessReplayClockTest.php b/tests/BusinessReplayClockTest.php index 70513d48..3b40ffe3 100644 --- a/tests/BusinessReplayClockTest.php +++ b/tests/BusinessReplayClockTest.php @@ -242,6 +242,31 @@ public static function ambiguousActivityHistories(): array ]); } + #[DataProvider('coercedTerminalFamilies')] + public function testANumericStringDuplicateCannotDisappearFromTheClockProof(string $family): void + { + $history = $family === 'activity' ? $this->activity(1, self::FIRST) : self::timer(1, self::FIRST); + $duplicate = $history[1]; + $duplicate['payload']['sequence'] = '1'; + $history[] = $duplicate; + + $this->expectException(ReplayClockUnavailable::class); + $this->replayResult(static function (WorkflowContext $context) use ($family): string { + if ($family === 'activity') { + self::assertSame('fixed', $context->activity('anchor')); + } else { + $context->sleep(2); + } + + return $context->replayNow()->format('c'); + }, $history); + } + + public static function coercedTerminalFamilies(): array + { + return [['activity'], ['timer']]; + } + public function testNestedInvocationCannotBorrowParentTimeAndAnErrorRestoresParentScope(): void { $failure = new \LogicException('original nested error'); diff --git a/tests/ChildBusinessReplayClockTest.php b/tests/ChildBusinessReplayClockTest.php new file mode 100644 index 00000000..df83a674 --- /dev/null +++ b/tests/ChildBusinessReplayClockTest.php @@ -0,0 +1,200 @@ +childWorkflow('clock-child'); + } catch (ChildWorkflowFailed $failure) { + $outcome = [$failure->failureType, $failure->getMessage()]; + } + + return [$outcome, $context->replayNow()->format('Y-m-d\TH:i:s.uP')]; + }; + $history = self::history($type); + $expectedOutcome = $type === 'ChildRunCompleted' ? 'child-value' : [$type, 'original child failure']; + + self::assertSame([$expectedOutcome, self::TERMINAL_TIME], $this->replayResult($handler, $history)); + self::assertSame($this->replayResult($handler, $history), $this->replayResult($handler, $history)); + } + + public static function terminalTypes(): array + { + return [['ChildRunCompleted'], ['ChildRunFailed'], ['ChildRunCancelled'], ['ChildRunTerminated']]; + } + + public function testStartedHandleCannotReadAFutureTerminalButAwaitConsumesIt(): void + { + $observed = $this->replayResult(static function (WorkflowContext $context): array { + $child = $context->startChildWorkflow('clock-child'); + try { $context->replayNow(); } catch (ReplayClockUnavailable) { $started = 'unavailable'; } + $result = $child->await(); + + return [$started ?? 'borrowed', $result, $context->replayNow()->format('Y-m-d\TH:i:s.uP')]; + }, self::history('ChildRunCompleted')); + + self::assertSame(['unavailable', 'child-value', self::TERMINAL_TIME], $observed); + } + + public function testScheduledAndTerminalAreSufficientWithoutAStartedRow(): void + { + $history = self::history('ChildRunCompleted'); + array_splice($history, 1, 1); + + self::assertSame(self::TERMINAL_TIME, $this->replayResult(self::timeHandler(), $history)); + } + + #[DataProvider('terminalInstants')] + public function testTheExactParentInstantIsPreserved(string $input, string $expected): void + { + $history = self::history('ChildRunCompleted'); + $history[2]['recorded_at'] = $input; + + self::assertSame($expected, $this->replayResult(self::timeHandler(), $history)); + } + + public static function terminalInstants(): array + { + return [ + ['2026-11-06T01:00:00Z', '2026-11-06T01:00:00.000000+00:00'], + ['2026-11-06T00:59:59.000001Z', '2026-11-06T00:59:59.000001+00:00'], + ['2026-11-06T03:00:00.654321+01:00', '2026-11-06T02:00:00.654321+00:00'], + ]; + } + + public function testOnlyTheConsumedChildCanSupplyTime(): void + { + $history = self::history('ChildRunCompleted'); + $second = self::history('ChildRunCompleted'); + foreach ($second as &$event) { + $event['payload']['sequence'] = 2; + $event['payload']['child_call_id'] = 'second-call'; + $event['payload']['child_workflow_instance_id'] = 'second-child'; + $event['payload']['child_workflow_run_id'] = 'second-run'; + $event['recorded_at'] = 'invalid future clock'; + } + unset($event); + $result = $this->replayResult(static function (WorkflowContext $context): array { + $context->childWorkflow('clock-child'); + $first = $context->replayNow()->format('Y-m-d\TH:i:s.uP'); + $context->childWorkflow('clock-child'); + try { $context->replayNow(); } catch (ReplayClockUnavailable) { return [$first, 'unavailable']; } + self::fail('Invalid consumed time must not retain the prior child clock.'); + }, [...$history, ...$second]); + + self::assertSame([self::TERMINAL_TIME, 'unavailable'], $result); + } + + public function testChildCallsDoNotCreateATimeWithoutAnActiveCallback(): void + { + $retained = null; + $this->replayResult(static function (WorkflowContext $context) use (&$retained): string { + $context->childWorkflow('clock-child'); + $retained = $context; + + return $context->replayNow()->format('c'); + }, self::history('ChildRunCompleted')); + + $this->expectException(ReplayClockUnavailable::class); + $retained->replayNow(); + } + + private static function timeHandler(): callable + { + return static function (WorkflowContext $context): string { + $context->childWorkflow('clock-child'); + + return $context->replayNow()->format('Y-m-d\TH:i:s.uP'); + }; + } + + #[DataProvider('unprovedLifecycles')] + public function testUnprovedChildIdentityAndLifecycleCannotSupplyTime(string $mutation): void + { + $history = self::history('ChildRunCompleted'); + switch ($mutation) { + case 'missing-call': unset($history[2]['payload']['child_call_id']); break; + case 'wrong-call': $history[2]['payload']['child_call_id'] = 'other-call'; break; + case 'wrong-run': $history[2]['payload']['child_workflow_run_id'] = 'other-run'; break; + case 'wrong-instance': $history[2]['payload']['child_workflow_instance_id'] = 'other-instance'; break; + case 'alias-only': + $history[2]['payload']['workflow_run_id'] = $history[2]['payload']['child_workflow_run_id']; + unset($history[2]['payload']['child_workflow_run_id']); break; + case 'conflicting-alias': $history[2]['payload']['workflow_run_id'] = 'other-run'; break; + case 'coerced-sequence': foreach ($history as &$row) { $row['payload']['sequence'] = '1'; } unset($row); break; + case 'coerced-duplicate': + $duplicate = $history[2]; $duplicate['payload']['sequence'] = '1'; $history[] = $duplicate; break; + case 'duplicate-terminal': $history[] = $history[2]; break; + case 'conflicting-terminal': $history[] = [...$history[2], 'event_type' => 'ChildRunFailed']; break; + case 'missing-opening': array_shift($history); break; + case 'duplicate-opening': array_unshift($history, $history[0]); break; + case 'duplicate-started': array_splice($history, 1, 0, [$history[1]]); break; + case 'started-after-terminal': [$history[1], $history[2]] = [$history[2], $history[1]]; break; + case 'replacement-run': + $replacement = $history[1]; + $replacement['payload']['child_workflow_run_id'] = 'replacement'; + array_splice($history, 2, 0, [$replacement]); + $history[3]['payload']['child_workflow_run_id'] = 'replacement'; break; + case 'bad-time': $history[2]['recorded_at'] = '2026-02-30T00:00:00Z'; break; + } + + $this->expectException(ReplayClockUnavailable::class); + $this->replayResult(static function (WorkflowContext $context): string { + self::assertSame('child-value', $context->childWorkflow('clock-child')); + + return $context->replayNow()->format('c'); + }, $history); + } + + public static function unprovedLifecycles(): array + { + return array_map(static fn (string $mutation): array => [$mutation], [ + 'missing-call', 'wrong-call', 'wrong-run', 'wrong-instance', 'alias-only', 'conflicting-alias', 'coerced-sequence', + 'coerced-duplicate', 'duplicate-terminal', 'conflicting-terminal', 'missing-opening', 'duplicate-opening', + 'duplicate-started', 'started-after-terminal', 'replacement-run', 'bad-time', + ]); + } + + /** @return list> */ + private static function history(string $terminal): array + { + $payload = ['sequence' => 1, 'child_call_id' => 'call', 'workflow_link_id' => 'call', + 'child_workflow_instance_id' => 'child-instance', 'child_workflow_run_id' => 'child-run', + 'child_workflow_type' => 'clock-child']; + + return [ + ['event_type' => 'ChildWorkflowScheduled', 'payload' => $payload], + ['event_type' => 'ChildRunStarted', 'recorded_at' => '2026-11-06T01:00:00Z', 'payload' => $payload], + ['event_type' => $terminal, 'recorded_at' => self::TERMINAL_TIME, + 'payload' => [...$payload, 'closed_at' => '1990-01-01T00:00:00Z', + 'reused_recorded_at' => '2099-01-01T00:00:00Z', + 'message' => 'original child failure', 'result' => (new AvroPayloadCodec())->envelope('child-value')]], + ]; + } + + /** @param list> $history */ + private function replayResult(callable $handler, array $history): mixed + { + $codec = new AvroPayloadCodec(); + $result = (new Replayer($codec))->replay($handler, $history, [], 'php-workers', ['workflow_id' => 'parent', 'run_id' => 'parent-run']); + + return $codec->decodeEnvelope($result->commands[array_key_last($result->commands)]['result']); + } +} diff --git a/tests/Support/ReplayRegressionFixture.php b/tests/Support/ReplayRegressionFixture.php index 1ddd8fc2..15f3d9eb 100644 --- a/tests/Support/ReplayRegressionFixture.php +++ b/tests/Support/ReplayRegressionFixture.php @@ -675,6 +675,18 @@ private static function workflow(string $workflowType): callable return $context->activity('Tests\\Fixtures\\V2\\TestRedriveFirstActivity', [$name, $token, $version]); }, + 'golden.business-replay-clock-child' => static function (WorkflowContext $context, int $count = 4): array { + $outcomes = []; + for ($index = 0; $index < $count; ++$index) { + try { $value = $context->childWorkflow('clock-child'); } + catch (\DurableWorkflow\Exception\ChildWorkflowFailed $failure) { $value = $failure->failureType; } + try { $time = $context->replayNow()->format('Y-m-d\TH:i:s.uP'); } + catch (\DurableWorkflow\Exception\ReplayClockUnavailable) { $time = null; } + $outcomes[] = ['outcome' => $value, 'time' => $time]; + } + + return $outcomes; + }, '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 { diff --git a/tests/fixtures/replay-regressions/business-replay-clock-child-coerced-duplicate.json b/tests/fixtures/replay-regressions/business-replay-clock-child-coerced-duplicate.json new file mode 100644 index 00000000..79075cac --- /dev/null +++ b/tests/fixtures/replay-regressions/business-replay-clock-child-coerced-duplicate.json @@ -0,0 +1,153 @@ +{ + "$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-child-coerced-duplicate", + "protocol_version": "1.20", + "bindings": [ + "php" + ], + "workflow": { + "type": "golden.business-replay-clock-child", + "input": [ + 2 + ] + }, + "history": [ + { + "sequence": 1, + "event_type": "ChildWorkflowScheduled", + "recorded_at": "2026-11-06T03:00:01.123456+00:00", + "payload": { + "sequence": 1, + "child_call_id": "child-call-1", + "workflow_link_id": "child-call-1", + "child_workflow_instance_id": "child-instance-1", + "child_workflow_run_id": "child-run-1", + "child_workflow_type": "clock-child" + } + }, + { + "sequence": 2, + "event_type": "ChildRunStarted", + "recorded_at": "2026-11-06T03:00:01.123456+00:00", + "payload": { + "sequence": 1, + "child_call_id": "child-call-1", + "workflow_link_id": "child-call-1", + "child_workflow_instance_id": "child-instance-1", + "child_workflow_run_id": "child-run-1", + "child_workflow_type": "clock-child" + } + }, + { + "sequence": 3, + "event_type": "ChildRunCompleted", + "recorded_at": "2026-11-06T03:00:01.123456+00:00", + "payload": { + "sequence": 1, + "child_call_id": "child-call-1", + "workflow_link_id": "child-call-1", + "child_workflow_instance_id": "child-instance-1", + "child_workflow_run_id": "child-run-1", + "child_workflow_type": "clock-child", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoWY2hpbGQtdmFsdWU=" + }, + "payload_codec": "avro", + "message": "source child failure" + } + }, + { + "sequence": 4, + "event_type": "ChildRunCompleted", + "recorded_at": "2026-11-06T03:00:01.123456+00:00", + "payload": { + "sequence": "1", + "child_call_id": "child-call-1", + "workflow_link_id": "child-call-1", + "child_workflow_instance_id": "child-instance-1", + "child_workflow_run_id": "child-run-1", + "child_workflow_type": "clock-child", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoWY2hpbGQtdmFsdWU=" + }, + "payload_codec": "avro", + "message": "source child failure" + } + }, + { + "sequence": 5, + "event_type": "ChildWorkflowScheduled", + "recorded_at": "2026-11-06T03:00:02.123456+00:00", + "payload": { + "sequence": 2, + "child_call_id": "child-call-2", + "workflow_link_id": "child-call-2", + "child_workflow_instance_id": "child-instance-2", + "child_workflow_run_id": "child-run-2", + "child_workflow_type": "clock-child" + } + }, + { + "sequence": 6, + "event_type": "ChildRunStarted", + "recorded_at": "2026-11-06T03:00:02.123456+00:00", + "payload": { + "sequence": 2, + "child_call_id": "child-call-2", + "workflow_link_id": "child-call-2", + "child_workflow_instance_id": "child-instance-2", + "child_workflow_run_id": "child-run-2", + "child_workflow_type": "clock-child" + } + }, + { + "sequence": 7, + "event_type": "ChildRunFailed", + "recorded_at": "2026-11-06T03:00:02.123456+00:00", + "payload": { + "sequence": 2, + "child_call_id": "child-call-2", + "workflow_link_id": "child-call-2", + "child_workflow_instance_id": "child-instance-2", + "child_workflow_run_id": "child-run-2", + "child_workflow_type": "clock-child", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoWY2hpbGQtdmFsdWU=" + }, + "payload_codec": "avro", + "message": "source child failure" + } + } + ], + "workflow_tasks": [ + { + "workflow_id": "clock-parent", + "run_id": "clock-parent-run" + } + ], + "expected": { + "workflow_tasks": [ + { + "command_sequence": [ + { + "type": "complete_workflow", + "result": [ + { + "outcome": "child-value", + "time": null + }, + { + "outcome": "ChildRunFailed", + "time": "2026-11-06T03:00:02.123456+00:00" + } + ] + } + ] + } + ] + } +} diff --git a/tests/fixtures/replay-regressions/business-replay-clock-child-results.json b/tests/fixtures/replay-regressions/business-replay-clock-child-results.json new file mode 100644 index 00000000..108b3651 --- /dev/null +++ b/tests/fixtures/replay-regressions/business-replay-clock-child-results.json @@ -0,0 +1,230 @@ +{ + "$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-child-results", + "protocol_version": "1.20", + "bindings": [ + "php" + ], + "workflow": { + "type": "golden.business-replay-clock-child", + "input": [] + }, + "history": [ + { + "sequence": 1, + "event_type": "ChildWorkflowScheduled", + "recorded_at": "2026-11-06T03:00:01.123456+00:00", + "payload": { + "sequence": 1, + "child_call_id": "child-call-1", + "workflow_link_id": "child-call-1", + "child_workflow_instance_id": "child-instance-1", + "child_workflow_run_id": "child-run-1", + "child_workflow_type": "clock-child" + } + }, + { + "sequence": 2, + "event_type": "ChildRunStarted", + "recorded_at": "2026-11-06T03:00:01.123456+00:00", + "payload": { + "sequence": 1, + "child_call_id": "child-call-1", + "workflow_link_id": "child-call-1", + "child_workflow_instance_id": "child-instance-1", + "child_workflow_run_id": "child-run-1", + "child_workflow_type": "clock-child" + } + }, + { + "sequence": 3, + "event_type": "ChildRunCompleted", + "recorded_at": "2026-11-06T03:00:01.123456+00:00", + "payload": { + "sequence": 1, + "child_call_id": "child-call-1", + "workflow_link_id": "child-call-1", + "child_workflow_instance_id": "child-instance-1", + "child_workflow_run_id": "child-run-1", + "child_workflow_type": "clock-child", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoWY2hpbGQtdmFsdWU=" + }, + "payload_codec": "avro", + "message": "source child failure" + } + }, + { + "sequence": 4, + "event_type": "ChildWorkflowScheduled", + "recorded_at": "2026-11-06T03:00:02.123456+00:00", + "payload": { + "sequence": 2, + "child_call_id": "child-call-2", + "workflow_link_id": "child-call-2", + "child_workflow_instance_id": "child-instance-2", + "child_workflow_run_id": "child-run-2", + "child_workflow_type": "clock-child" + } + }, + { + "sequence": 5, + "event_type": "ChildRunStarted", + "recorded_at": "2026-11-06T03:00:02.123456+00:00", + "payload": { + "sequence": 2, + "child_call_id": "child-call-2", + "workflow_link_id": "child-call-2", + "child_workflow_instance_id": "child-instance-2", + "child_workflow_run_id": "child-run-2", + "child_workflow_type": "clock-child" + } + }, + { + "sequence": 6, + "event_type": "ChildRunFailed", + "recorded_at": "2026-11-06T03:00:02.123456+00:00", + "payload": { + "sequence": 2, + "child_call_id": "child-call-2", + "workflow_link_id": "child-call-2", + "child_workflow_instance_id": "child-instance-2", + "child_workflow_run_id": "child-run-2", + "child_workflow_type": "clock-child", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoWY2hpbGQtdmFsdWU=" + }, + "payload_codec": "avro", + "message": "source child failure" + } + }, + { + "sequence": 7, + "event_type": "ChildWorkflowScheduled", + "recorded_at": "2026-11-06T03:00:03.123456+00:00", + "payload": { + "sequence": 3, + "child_call_id": "child-call-3", + "workflow_link_id": "child-call-3", + "child_workflow_instance_id": "child-instance-3", + "child_workflow_run_id": "child-run-3", + "child_workflow_type": "clock-child" + } + }, + { + "sequence": 8, + "event_type": "ChildRunStarted", + "recorded_at": "2026-11-06T03:00:03.123456+00:00", + "payload": { + "sequence": 3, + "child_call_id": "child-call-3", + "workflow_link_id": "child-call-3", + "child_workflow_instance_id": "child-instance-3", + "child_workflow_run_id": "child-run-3", + "child_workflow_type": "clock-child" + } + }, + { + "sequence": 9, + "event_type": "ChildRunCancelled", + "recorded_at": "2026-11-06T03:00:03.123456+00:00", + "payload": { + "sequence": 3, + "child_call_id": "child-call-3", + "workflow_link_id": "child-call-3", + "child_workflow_instance_id": "child-instance-3", + "child_workflow_run_id": "child-run-3", + "child_workflow_type": "clock-child", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoWY2hpbGQtdmFsdWU=" + }, + "payload_codec": "avro", + "message": "source child failure" + } + }, + { + "sequence": 10, + "event_type": "ChildWorkflowScheduled", + "recorded_at": "2026-11-06T03:00:04.123456+00:00", + "payload": { + "sequence": 4, + "child_call_id": "child-call-4", + "workflow_link_id": "child-call-4", + "child_workflow_instance_id": "child-instance-4", + "child_workflow_run_id": "child-run-4", + "child_workflow_type": "clock-child" + } + }, + { + "sequence": 11, + "event_type": "ChildRunStarted", + "recorded_at": "2026-11-06T03:00:04.123456+00:00", + "payload": { + "sequence": 4, + "child_call_id": "child-call-4", + "workflow_link_id": "child-call-4", + "child_workflow_instance_id": "child-instance-4", + "child_workflow_run_id": "child-run-4", + "child_workflow_type": "clock-child" + } + }, + { + "sequence": 12, + "event_type": "ChildRunTerminated", + "recorded_at": "2026-11-06T03:00:04.123456+00:00", + "payload": { + "sequence": 4, + "child_call_id": "child-call-4", + "workflow_link_id": "child-call-4", + "child_workflow_instance_id": "child-instance-4", + "child_workflow_run_id": "child-run-4", + "child_workflow_type": "clock-child", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoWY2hpbGQtdmFsdWU=" + }, + "payload_codec": "avro", + "message": "source child failure" + } + } + ], + "workflow_tasks": [ + { + "workflow_id": "clock-parent", + "run_id": "clock-parent-run" + } + ], + "expected": { + "workflow_tasks": [ + { + "command_sequence": [ + { + "type": "complete_workflow", + "result": [ + { + "outcome": "child-value", + "time": "2026-11-06T03:00:01.123456+00:00" + }, + { + "outcome": "ChildRunFailed", + "time": "2026-11-06T03:00:02.123456+00:00" + }, + { + "outcome": "ChildRunCancelled", + "time": "2026-11-06T03:00:03.123456+00:00" + }, + { + "outcome": "ChildRunTerminated", + "time": "2026-11-06T03:00:04.123456+00:00" + } + ] + } + ] + } + ] + } +}