diff --git a/scripts/ci/run-replay-regression-fixture.php b/scripts/ci/run-replay-regression-fixture.php index 9b67b6b3..8c561447 100644 --- a/scripts/ci/run-replay-regression-fixture.php +++ b/scripts/ci/run-replay-regression-fixture.php @@ -1005,6 +1005,13 @@ private static function workflow(string $workflowType): callable return ['failure' => $message ?? 'missing', 'failed_time' => $failedTime, 'time' => $time]; }, + 'golden.business-replay-clock-group-condition' => static function (WorkflowContext $context): array { + $values = $context->all([$context->deferCondition(static fn (): bool => true, 'clock-condition'), $context->deferTimer(2)]); + try { $time = $context->replayNow()->format(self::CLOCK_FORMAT); } + catch (\DurableWorkflow\Exception\ReplayClockUnavailable) { $time = null; } + + return ['values' => $values, 'time' => $time]; + }, 'golden.business-replay-clock-group-child' => static function (WorkflowContext $context): array { $values = $context->all([$context->deferChildWorkflow('clock-child'), $context->deferTimer(2)]); try { $time = $context->replayNow()->format(self::CLOCK_FORMAT); } diff --git a/src/Worker/BusinessReplayClock.php b/src/Worker/BusinessReplayClock.php index 6c8b5579..f8f2699a 100644 --- a/src/Worker/BusinessReplayClock.php +++ b/src/Worker/BusinessReplayClock.php @@ -145,14 +145,16 @@ private function groupMemberTime(array $step, array $path, Closure $pathMatches) $family = $step['shape'] ?? null; $index = $step['resolution_order'] ?? null; $sequence = $step['sequence'] ?? null; - if (!in_array($family, ['activity', 'timer', self::CHILD_SHAPE], true) || !is_int($index) || !is_int($sequence) + if (!in_array($family, ['activity', 'timer', self::CHILD_SHAPE, 'condition_wait'], true) || !is_int($index) || !is_int($sequence) || !isset($this->history[$index]) || $path === []) { return null; } $rowMatches = fn (array $body, string $kind): bool => $this->groupRow($body, $kind, $path, $pathMatches); - $matches = $family === self::CHILD_SHAPE - ? ChildReplayClockEvidence::matches($this->history, $this->bySequence[$sequence] ?? [], $index, $sequence, $rowMatches) - : $this->scalarBoundary($index, $family, $sequence, $rowMatches); + $matches = match ($family) { + self::CHILD_SHAPE => ChildReplayClockEvidence::matches($this->history, $this->bySequence[$sequence] ?? [], $index, $sequence, $rowMatches), + 'condition_wait' => ConditionReplayClockEvidence::matches($this->history, $index, $sequence, $rowMatches), + default => $this->scalarBoundary($index, $family, $sequence, $rowMatches), + }; if (!$matches) { return null; } diff --git a/src/Worker/ConditionReplayClockEvidence.php b/src/Worker/ConditionReplayClockEvidence.php index fbe4f8a0..32a02b19 100644 --- a/src/Worker/ConditionReplayClockEvidence.php +++ b/src/Worker/ConditionReplayClockEvidence.php @@ -4,13 +4,17 @@ namespace DurableWorkflow\Worker; +use Closure; + /** @internal Proves one consumed physical condition wait, including its timeout pair. */ final class ConditionReplayClockEvidence { private const TERMINALS = ['ConditionWaitSatisfied', 'ConditionWaitTimedOut']; - /** @param list> $history */ - public static function matches(array $history, int $index, int $sequence): bool + /** @param list> $history + * @param (Closure(array, string): bool)|null $rowMatches + */ + public static function matches(array $history, int $index, int $sequence, ?Closure $rowMatches = null): bool { $event = $history[$index]; $payload = $event['payload'] ?? null; @@ -18,7 +22,7 @@ public static function matches(array $history, int $index, int $sequence): bool if (!is_array($payload) || !ReplayClockHistory::identity($waitId) || ReplayClockHistory::sequence($payload) !== $sequence) { return false; } - $rows = self::linkedRows($history, $sequence, $waitId); + $rows = self::linkedRows($history, $sequence, $waitId, $rowMatches ?? ReplayClockHistory::scalarRow(...)); return $rows !== null && self::lifecycle($history, $rows, $index); } @@ -41,9 +45,10 @@ private static function lifecycle(array $history, array $rows, int $index): bool } /** @param list> $history + * @param Closure(array, string): bool $rowMatches * @return array>|null */ - private static function linkedRows(array $history, int $sequence, string $waitId): ?array + private static function linkedRows(array $history, int $sequence, string $waitId, Closure $rowMatches): ?array { $positions = []; $allowed = ['ConditionWaitOpened', ...self::TERMINALS, 'TimerScheduled', 'TimerFired', 'TimerCancelled']; @@ -55,8 +60,13 @@ private static function linkedRows(array $history, int $sequence, string $waitId || (ReplayClockHistory::legacySequenceClaim($body) !== $sequence && ($body['condition_wait_id'] ?? null) !== $waitId)) { continue; } + // Core's condition-timeout timers carry the wait identity, without + // the owning condition's parallel path. The complete timer pair + // must remain scalar and correlate with that validated wait. + $rowQualified = str_starts_with((string) $kind, 'Timer') + ? ReplayClockHistory::scalarRow($body, 'condition_wait') : $rowMatches($body, 'condition_wait'); if (!in_array($kind, $allowed, true) || ReplayClockHistory::sequence($body) !== $sequence - || ($body['condition_wait_id'] ?? null) !== $waitId || !ReplayClockHistory::scalarRow($body, 'condition_wait') + || ($body['condition_wait_id'] ?? null) !== $waitId || !$rowQualified || ($body['cancellation_scope'] ?? null) !== null) { return null; } diff --git a/tests/ConditionGroupBusinessReplayClockTest.php b/tests/ConditionGroupBusinessReplayClockTest.php new file mode 100644 index 00000000..12dd96cf --- /dev/null +++ b/tests/ConditionGroupBusinessReplayClockTest.php @@ -0,0 +1,186 @@ +replay($handler, $history)); + self::assertSame($expected, $this->replay($handler, $history)); + } + + public static function conditionOutcomes(): array + { + return [['ConditionWaitSatisfied', false], ['ConditionWaitSatisfied', true], ['ConditionWaitTimedOut', true]]; + } + + public function testConditionAndSiblingUseTheirOwnConsumedTimes(): void + { + $handler = self::handler(); + $history = self::history($handler, times: [self::FIRST, '2026-11-06T00:00:00.124500Z']); + // A later, unconsumed condition is not part of the first group's fold. + $history[] = ['event_type' => 'ConditionWaitSatisfied', 'recorded_at' => '2099-01-01T00:00:00Z', + 'payload' => ['sequence' => 3, 'condition_wait_id' => 'future']]; + + self::assertSame([[true, null], '2026-11-06T00:00:00.124500+00:00'], $this->replay($handler, $history)); + } + + public function testTwoConditionsCannotBorrowOnePhysicalWait(): void + { + $handler = self::handler(true); + $history = self::history($handler); + foreach ($history as &$event) { + if (($event['payload']['sequence'] ?? null) === 2) { + $event['payload']['condition_wait_id'] = 'wait-1'; + } + } + unset($event); + + self::assertSame([[true, true], null], $this->replay($handler, $history)); + } + + public function testNestedConditionGroupsUseTheAuthoredLeafPaths(): void + { + $handler = static function (WorkflowContext $context): array { + $values = $context->all([static fn (): array => $context->all([ + $context->deferCondition(static fn (): bool => true, 'clock-condition', 5), $context->deferTimer(2), + ]), $context->deferTimer(2)]); + + return [$values, self::read($context)]; + }; + + self::assertSame([[[true, null], null], self::FIRST], $this->replay($handler, + self::history($handler, times: [self::FIRST, self::SECOND, self::SECOND]))); + } + + #[DataProvider('unprovedRows')] + public function testUnprovedConditionLifecycleCannotSupplyTheGroupClock(string $mutation): void + { + $handler = self::handler(); + $history = self::history($handler, backed: true); + switch ($mutation) { + case 'terminal-path': $history[3]['payload']['parallel_group_id'] = 'foreign'; break; + case 'terminal-scope': $history[3]['payload']['cancellation_scope_id'] = 'foreign'; break; + case 'terminal-local': $history[3]['payload']['execution_mode'] = 'local'; break; + case 'foreign-wait': $history[3]['payload']['condition_wait_id'] = 'foreign'; break; + case 'foreign-definition': $history[3]['payload']['condition_definition_fingerprint'] = 'foreign'; break; + case 'foreign-occurrence': $history[3]['payload']['condition_wait_occurrence_id'] = 'foreign'; break; + case 'timer-resource': $history[2]['payload']['timer_id'] = 'foreign'; break; + case 'timer-kind': $history[2]['payload']['timer_kind'] = 'signal_timeout'; break; + case 'timer-path': $history[1]['payload']['parallel_group_id'] = 'foreign'; break; + case 'duplicate-timer': array_splice($history, 2, 0, [$history[1]]); break; + case 'duplicate-terminal': $history[] = $history[3]; break; + case 'coerced-duplicate': + $duplicate = $history[3]; $duplicate['payload']['sequence'] = '1'; $history[] = $duplicate; break; + case 'bad-time': $history[3]['recorded_at'] = '2026-02-30T00:00:00Z'; break; + } + + if (in_array($mutation, ['terminal-path', 'terminal-scope'], true)) { + $this->expectException($mutation === 'terminal-path' ? NonDeterministicWorkflow::class : WorkflowClaimAborted::class); + } + self::assertSame([[true, null], null], $this->replay($handler, $history)); + } + + public static function unprovedRows(): array + { + return array_map(static fn (string $mutation): array => [$mutation], ['terminal-path', 'terminal-scope', + 'terminal-local', 'foreign-wait', 'foreign-definition', 'foreign-occurrence', 'timer-resource', + 'timer-kind', 'timer-path', 'duplicate-timer', 'duplicate-terminal', 'coerced-duplicate', 'bad-time']); + } + + public function testACompletedGroupCannotLeaveItsClockAvailableOutsideExecution(): void + { + $retained = null; + $handler = static function (WorkflowContext $context) use (&$retained): ?string { + $retained = $context; + $context->all([$context->deferCondition(static fn (): bool => true, 'clock-condition', 5), $context->deferTimer(2)]); + + return self::read($context); + }; + self::assertSame(self::FIRST, $this->replay($handler, self::history(self::handler()))); + $this->expectException(ReplayClockUnavailable::class); + $retained->replayNow(); + } + + private static function handler(bool $conditionsOnly = false): callable + { + return static function (WorkflowContext $context) use ($conditionsOnly): array { + $values = $context->all([$context->deferCondition(static fn (): bool => true, 'clock-condition', 5), + $conditionsOnly ? $context->deferCondition(static fn (): bool => true, 'clock-condition', 5) : $context->deferTimer(2)]); + + return [$values, self::read($context)]; + }; + } + + private static function read(WorkflowContext $context): ?string + { + try { return $context->replayNow()->format(self::FORMAT); } + catch (ReplayClockUnavailable) { return null; } + } + + /** @param list $times + * @return list> + */ + private static function history(callable $handler, string $terminal = 'ConditionWaitSatisfied', bool $backed = false, + array $times = [self::FIRST, self::SECOND]): array + { + $commands = (new Replayer(new AvroPayloadCodec()))->replay($handler, [], [], 'php-workers')->commands; + $history = []; + foreach ($commands as $offset => $command) { + $sequence = $offset + 1; + $metadata = array_intersect_key($command, array_flip(['parallel_group_id', 'parallel_group_kind', 'parallel_group_mode', + 'parallel_group_base_sequence', 'parallel_group_size', 'parallel_group_index', 'parallel_group_path'])); + $body = ['sequence' => $sequence]; + if ($command['type'] === 'open_condition_wait') { + $body += ['condition_wait_id' => 'wait-'.$sequence, 'condition_key' => $command['condition_key'], + 'condition_definition_fingerprint' => $command['condition_definition_fingerprint'], 'timeout_seconds' => 5]; + $history[] = ['event_type' => 'ConditionWaitOpened', 'payload' => $body + $metadata]; + if ($backed) { + // Core's timeout writer has wait identity but no parallel metadata. + $timer = $body + ['timer_id' => 'timer-'.$sequence, 'timer_kind' => 'condition_timeout', 'delay_seconds' => 5]; + $history[] = ['event_type' => 'TimerScheduled', 'payload' => $timer]; + $history[] = ['event_type' => $terminal === 'ConditionWaitTimedOut' ? 'TimerFired' : 'TimerCancelled', + 'recorded_at' => $times[$offset], 'payload' => $timer]; + $body['timer_id'] = $timer['timer_id']; + } + $history[] = ['event_type' => $terminal, 'recorded_at' => $times[$offset], 'payload' => $body + $metadata]; + } else { + $body += ['timer_id' => 'timer-'.$sequence, 'delay_seconds' => 2]; + $history[] = ['event_type' => 'TimerScheduled', 'payload' => $body + $metadata]; + $history[] = ['event_type' => 'TimerFired', 'recorded_at' => $times[$offset], 'payload' => $body + $metadata]; + } + } + + return $history; + } + + /** @param list> $history */ + private function replay(callable $handler, array $history): mixed + { + $codec = new AvroPayloadCodec(); + $result = (new Replayer($codec))->replay($handler, $history, [], 'php-workers', ['workflow_id' => 'parent', 'run_id' => '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 9d8d9e5c..20294555 100644 --- a/tests/Support/ReplayRegressionFixture.php +++ b/tests/Support/ReplayRegressionFixture.php @@ -690,6 +690,13 @@ private static function workflow(string $workflowType): callable return ['failure' => $message ?? 'missing', 'failed_time' => $failedTime, 'time' => $time]; }, + 'golden.business-replay-clock-group-condition' => static function (WorkflowContext $context): array { + $values = $context->all([$context->deferCondition(static fn (): bool => true, 'clock-condition'), $context->deferTimer(2)]); + try { $time = $context->replayNow()->format('Y-m-d\TH:i:s.uP'); } + catch (\DurableWorkflow\Exception\ReplayClockUnavailable) { $time = null; } + + return ['values' => $values, 'time' => $time]; + }, 'golden.business-replay-clock-group-child' => static function (WorkflowContext $context): array { $values = $context->all([$context->deferChildWorkflow('clock-child'), $context->deferTimer(2)]); try { $time = $context->replayNow()->format('Y-m-d\TH:i:s.uP'); } diff --git a/tests/fixtures/replay-regressions/business-replay-clock-group-condition.json b/tests/fixtures/replay-regressions/business-replay-clock-group-condition.json new file mode 100644 index 00000000..d5ef97b8 --- /dev/null +++ b/tests/fixtures/replay-regressions/business-replay-clock-group-condition.json @@ -0,0 +1,139 @@ +{ + "$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-group-condition", + "protocol_version": "1.20", + "bindings": [ + "php" + ], + "workflow": { + "type": "golden.business-replay-clock-group-condition", + "input": [] + }, + "history": [ + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-05T23:59:58Z", + "payload": { + "sequence": 1, + "parallel_group_id": "parallel-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ], + "condition_wait_id": "clock-wait-1", + "condition_key": "clock-condition", + "condition_definition_fingerprint": "dwphp2:sha256:04cd1ecb5cb316516273dd0f29ef8fd313c87e1a32be60e839db44adbb9f9d1b" + }, + "sequence": 1 + }, + { + "event_type": "TimerScheduled", + "recorded_at": "2026-11-05T23:59:58Z", + "payload": { + "sequence": 2, + "parallel_group_id": "parallel-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ], + "timer_id": "clock-group-timer-2", + "delay_seconds": 2, + "fire_at": "2026-11-06T00:00:00Z" + }, + "sequence": 2 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-06T00:00:00.123400Z", + "payload": { + "sequence": 1, + "parallel_group_id": "parallel-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ], + "condition_wait_id": "clock-wait-1", + "condition_key": "clock-condition", + "condition_definition_fingerprint": "dwphp2:sha256:04cd1ecb5cb316516273dd0f29ef8fd313c87e1a32be60e839db44adbb9f9d1b" + }, + "sequence": 3 + }, + { + "event_type": "TimerFired", + "recorded_at": "2026-11-06T00:00:00.123499Z", + "payload": { + "sequence": 2, + "parallel_group_id": "parallel-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ], + "timer_id": "clock-group-timer-2", + "delay_seconds": 2, + "fire_at": "2026-11-06T00:00:00Z" + }, + "sequence": 4 + } + ], + "workflow_tasks": [ + { + "workflow_id": "clock-parent", + "run_id": "clock-parent-run" + } + ], + "expected": { + "workflow_tasks": [ + { + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "values": [ + true, + null + ], + "time": "2026-11-06T00:00:00.123400+00:00" + } + } + ] + } + ] + } +}