diff --git a/scripts/Support/DeferredChildAllLedger.php b/scripts/Support/DeferredChildAllLedger.php new file mode 100644 index 00000000..a3b53071 --- /dev/null +++ b/scripts/Support/DeferredChildAllLedger.php @@ -0,0 +1,179 @@ +> $initial */ + public function __construct(private readonly DeferredChildAllScenario $scenario, private readonly array $initial) + { + } + + /** + * @param list> $history + * @return list> + */ + public function validate(array $history, bool $parentFinished = false): array + { + $initial = $this->initial; + $runs = array_fill(0, $this->scenario->size, []); + $scheduled = array_fill(0, $this->scenario->size, false); + $usedRuns = array_column($initial, 'child_workflow_run_id'); + $usedLinks = array_column($initial, 'workflow_link_id'); + $parentClosed = false; + foreach ($history as $event) { + $type = $event['event_type'] ?? null; + DeferredChildAllScenario::check(is_string($type), 'History event type is not a string.'); + if ($type === 'WorkflowCompleted') { + DeferredChildAllScenario::check(!$parentClosed, 'Duplicate parent completion event.'); + $parentClosed = true; + } + if (!str_starts_with($type, 'Child')) { + continue; + } + DeferredChildAllScenario::check(!$parentClosed, 'Child event follows parent completion.'); + [$index, $payload] = $this->childEvent($type, $event, $initial); + if ($type === 'ChildWorkflowScheduled') { + DeferredChildAllScenario::check(!$scheduled[$index] && $runs[$index] === [], 'Child schedule is repeated or follows its start.'); + $scheduled[$index] = true; + continue; + } + DeferredChildAllScenario::check($scheduled[$index], 'Child run event precedes its schedule.'); + if ($type === 'ChildRunStarted') { + $next = $this->startedRun($payload, $initial[$index], $runs[$index], $usedRuns, $usedLinks); + if ($runs[$index] !== []) { + $usedRuns[] = $next['run_id']; + $usedLinks[] = $next['workflow_link_id']; + } + $runs[$index][] = $next; + continue; + } + $terminal = $this->terminalRun($type, $payload, $runs[$index]); + $runs[$index][array_key_last($runs[$index])] = $terminal; + } + $this->finishedChains($runs, $parentFinished, $parentClosed); + + return $runs; + } + + /** + * @param array $event + * @param list> $initial + * @return array{int, array} + */ + private function childEvent(string $type, array $event, array $initial): array + { + DeferredChildAllScenario::check(in_array($type, ['ChildWorkflowScheduled', 'ChildRunStarted', 'ChildRunCompleted', + 'ChildRunFailed', 'ChildRunCancelled', 'ChildRunTerminated'], true), 'Unqualified child event type.'); + $payload = $event['payload'] ?? null; + DeferredChildAllScenario::check(is_array($payload), 'Child event has no payload.'); + $sequence = $payload['sequence'] ?? null; + DeferredChildAllScenario::check(is_int($sequence) && $sequence >= 1 && $sequence <= $this->scenario->size, 'Child sequence is not an exact member integer.'); + $index = $sequence - 1; + DeferredChildAllScenario::check(($payload['child_workflow_type'] ?? null) === $this->scenario->childType(), 'Child workflow type changed.'); + foreach (['child_call_id', 'child_workflow_instance_id', 'parallel_group_id', 'parallel_group_path'] as $key) { + DeferredChildAllScenario::check(isset($payload[$key]) && $payload[$key] === $initial[$index][$key], 'Child event changed its original call/link/path.'); + } + // Schedules were already checked by identities(); the run transitions + // also need their exact nonempty identities before any state update. + self::runIdentity($payload); + + return [$index, $payload]; + } + + /** + * @param array $payload + * @return array{string, string} + */ + private static function runIdentity(array $payload): array + { + DeferredChildAllScenario::check(is_string($payload['child_workflow_run_id'] ?? null) && $payload['child_workflow_run_id'] !== '', 'Child event has no run identity.'); + DeferredChildAllScenario::check(is_string($payload['workflow_link_id'] ?? null) && $payload['workflow_link_id'] !== '', 'Child event has no link identity.'); + + return [$payload['child_workflow_run_id'], $payload['workflow_link_id']]; + } + + /** + * @param array $payload + * @param array $original + * @param list $chain + * @param list $usedRuns + * @param list $usedLinks + * @return array{run_id: string, workflow_link_id: string, terminal: ?string} + */ + private function startedRun(array $payload, array $original, array $chain, array $usedRuns, array $usedLinks): array + { + [$run, $link] = self::runIdentity($payload); + if ($chain === []) { + DeferredChildAllScenario::check($run === $original['child_workflow_run_id'] && $link === $original['workflow_link_id'] + && ($payload['child_run_number'] ?? null) === 1, 'Initial child run/link/number changed.'); + DeferredChildAllScenario::check(($payload['retry_of_child_workflow_run_id'] ?? null) === null + && ($payload['retry_attempt'] ?? null) === null, 'Initial child run claims a retry predecessor.'); + + return ['run_id' => $run, 'workflow_link_id' => $link, 'terminal' => null]; + } + $current = $chain[array_key_last($chain)]; + DeferredChildAllScenario::check($this->replacementMember($payload['sequence']) && count($chain) === 1 + && $current['terminal'] === null, 'Child run cannot replace a closed or unqualified member.'); + DeferredChildAllScenario::check(!in_array($run, $usedRuns, true) && !in_array($link, $usedLinks, true) + && ($payload['child_run_number'] ?? null) === 2, 'Replacement child run/link/number is not distinct.'); + // Retry carries explicit parent provenance. Continuation's predecessor + // is in child history, not an invented intermediate parent terminal. + $this->replacementProvenance($payload, $current['run_id']); + + return ['run_id' => $run, 'workflow_link_id' => $link, 'terminal' => null]; + } + + private function replacementMember(int $sequence): bool + { + return $sequence === 1 && in_array($this->scenario->case, ['retry', 'continue'], true); + } + + /** @param array $payload */ + private function replacementProvenance(array $payload, string $previousRun): void + { + if ($this->scenario->case === 'retry') { + DeferredChildAllScenario::check(($payload['retry_of_child_workflow_run_id'] ?? null) === $previousRun && ($payload['retry_attempt'] ?? null) === 2, + 'Replacement child transition does not match its case.'); + return; + } + DeferredChildAllScenario::check(($payload['retry_of_child_workflow_run_id'] ?? null) === null && ($payload['retry_attempt'] ?? null) === null, + 'Replacement child transition does not match its case.'); + } + + /** + * @param array $payload + * @param list $chain + * @return array{run_id: string, workflow_link_id: string, terminal: ?string} + */ + private function terminalRun(string $type, array $payload, array $chain): array + { + [$run, $link] = self::runIdentity($payload); + $current = $chain === [] ? null : $chain[array_key_last($chain)]; + DeferredChildAllScenario::check($current !== null && $current['terminal'] === null && $run === $current['run_id'] + && $link === $current['workflow_link_id'], 'Child terminal is repeated, late or belongs to another run/link.'); + DeferredChildAllScenario::check(($payload['child_run_number'] ?? null) === count($chain), 'Child terminal run number changed.'); + DeferredChildAllScenario::check(!$this->replacementMember($payload['sequence']) || count($chain) === 2, 'Replacement member resolved before its second run.'); + $expected = $payload['sequence'] === 1 && $this->scenario->failsEarly() ? $this->scenario->expected()['failure_type'] : 'ChildRunCompleted'; + DeferredChildAllScenario::check($type === $expected, 'Child terminal does not match its selected case.'); + $current['terminal'] = $type; + + return $current; + } + + /** @param list> $runs */ + private function finishedChains(array $runs, bool $parentFinished, bool $parentClosed): void + { + foreach ($runs as $index => $chain) { + DeferredChildAllScenario::check($chain !== [], 'Atomic child start has no initial run event.'); + if ($parentFinished && (!$this->scenario->failsEarly() || $index === 0)) { + DeferredChildAllScenario::check($chain[array_key_last($chain)]['terminal'] !== null, 'Finished parent has an unresolved required member.'); + } + } + DeferredChildAllScenario::check(!$parentFinished || $parentClosed, 'Finished parent has no canonical completion.'); + } + +} diff --git a/scripts/Support/DeferredChildAllScenario.php b/scripts/Support/DeferredChildAllScenario.php new file mode 100644 index 00000000..d919150c --- /dev/null +++ b/scripts/Support/DeferredChildAllScenario.php @@ -0,0 +1,263 @@ + */ + public readonly array $fixture; + + public readonly int $size; + + /** @var list */ + public readonly array $queues; + + /** @param array $queues */ + public function __construct(public readonly string $case, array $queues) + { + $bytes = (string) file_get_contents(dirname(__DIR__).'/fixtures/parallel-child-all.json'); + self::check(PHP_INT_SIZE === 8, 'The canonical int64 fixture requires 64-bit PHP.'); + self::check(hash('sha256', $bytes) === self::FIXTURE_SHA256, 'Canonical child-all fixture identity changed.'); + $this->fixture = json_decode($bytes, true, flags: JSON_THROW_ON_ERROR); + $sizes = $this->fixture['parallel_child_all']['cases'] + ['retry' => 2, 'continue' => 2]; + if (!isset($sizes[$case])) { + throw new InvalidArgumentException('Unknown deferred-child case.'); + } + $this->size = $sizes[$case]; + self::check(array_is_list($queues) && count($queues) === $this->size, 'One queue is required per child.'); + foreach ($queues as $queue) { + self::check(is_string($queue) && trim($queue) !== '' && strlen($queue) <= 191, 'Invalid child task queue.'); + } + self::check(count(array_unique($queues)) === $this->size, 'Child queues must be distinct for deterministic execution.'); + $this->queues = $queues; + } + + /** Handler for the canonical parent workflow. */ + public function parent(): Closure + { + return function (WorkflowContext $context, array $value): array { + $operations = []; + foreach ($this->queues as $index => $queue) { + $options = ['queue' => $queue]; + if ($this->case === 'retry' && $index === 0) { + $options['retry_policy'] = ['max_attempts' => 2, 'backoff_seconds' => [0]]; + } + $operations[] = $context->deferChildWorkflow($this->childType(), [['index' => $index, 'value' => $value]], $options); + } + try { + return ['results' => $context->all($operations)]; + } catch (ChildWorkflowFailed $error) { + return ['failure_class' => $error::class, 'failure_type' => $error->failureType, + 'workflow_type' => $error->workflowType, 'message' => $error->getMessage()]; + } + }; + } + + /** + * The retry fixture fails only its recorded initial run. This configuration + * is fixed before each worker starts and never reads a mutable attempt counter. + * + */ + public function child(?string $initialRunId = null): Closure + { + return function (WorkflowContext $context, array $value, int $stage = 0) use ($initialRunId): array { + if (($value['index'] ?? null) !== 0) { + return ['echo' => $value]; + } + if ($this->case === 'failed') { + throw new ChildAllProbeFailure($this->fixture['parallel_child_all']['failure_messages']['failed']); + } + if ($this->case === 'retry') { + self::check(is_string($initialRunId) && $initialRunId !== '', 'Retry requires the recorded initial child run.'); + if ($context->runId === $initialRunId) { + throw new ChildAllProbeFailure('retry group member'); + } + } + if ($this->case === 'continue' && $stage === 0) { + $context->continueAsNew([$value, 1]); + } + + return ['echo' => $value]; + }; + } + + public function parentType(): string + { + return $this->fixture['parallel_child_all']['parent_type']; + } + + public function childType(): string + { + return $this->fixture['parallel_child_all']['child_type']; + } + + /** @return list */ + public function order(): array + { + return match ($this->case) { + 'out_of_order' => [2, 0, 1], 'retry', 'continue' => [1, 0], default => range(0, $this->size - 1), + }; + } + + public function failsEarly(): bool + { + return in_array($this->case, ['failed', 'cancelled', 'terminated'], true); + } + + /** @return array */ + public function expected(): array + { + if ($this->failsEarly()) { + return ['failure_class' => ChildWorkflowFailed::class, + 'failure_type' => match ($this->case) { + 'failed' => 'ChildRunFailed', 'cancelled' => 'ChildRunCancelled', default => 'ChildRunTerminated', + }, 'workflow_type' => $this->childType(), + 'message' => $this->fixture['parallel_child_all']['failure_messages'][$this->case]]; + } + + return ['results' => array_map(fn (int $index): array => ['echo' => [ + 'index' => $index, 'value' => $this->fixture['input'], + ]], range(0, $this->size - 1))]; + } + + /** + * Bind all starts to the actual acknowledged commands. The private receipt + * supplies metadata only; no lease authority is included in returned identities. + * + * @param list> $history + * @param list> $commands + * @return list> + */ + public function identities(array $history, array $commands): array + { + $scheduled = array_values(array_filter($history, static fn (array $event): bool => $event['event_type'] === 'ChildWorkflowScheduled')); + self::check(count($scheduled) === $this->size && count($commands) === $this->size, 'Every member must start exactly once.'); + $identities = []; + foreach ($scheduled as $index => $event) { + $payload = $event['payload']; + $command = $commands[$index]; + self::check(($command['type'] ?? null) === 'start_child_workflow' + && ($command['queue'] ?? null) === $this->queues[$index] + && ($command['workflow_type'] ?? null) === $this->childType(), 'Unexpected child command or queue.'); + self::check(($payload['sequence'] ?? null) === $index + 1, 'Original child call sequence changed.'); + self::check(($payload['child_workflow_type'] ?? null) === $this->childType(), 'Original child workflow type changed.'); + $identity = []; + foreach (['child_call_id', 'workflow_link_id', 'child_workflow_instance_id', 'child_workflow_run_id'] as $key) { + self::check(is_string($payload[$key] ?? null) && $payload[$key] !== '', 'Missing canonical child identity.'); + $identity[$key] = $payload[$key]; + } + foreach (['parallel_group_id', 'parallel_group_path'] as $key) { + self::check(isset($payload[$key]) && $payload[$key] === ($command[$key] ?? null), 'Canonical child group path changed.'); + $identity[$key] = $payload[$key]; + } + $identities[] = ['sequence' => $index + 1, ...$identity]; + } + foreach (['child_call_id', 'workflow_link_id', 'child_workflow_instance_id', 'child_workflow_run_id'] as $key) { + self::check(count(array_unique(array_column($identities, $key))) === $this->size, 'Two members share one canonical identity.'); + } + + return $identities; + } + + /** + * @param list> $history + * @param list> $commands + * @param list> $initial + * @return list> + */ + public function assertStable(array $history, array $commands, array $initial, bool $parentFinished = false): array + { + self::check($this->identities($history, $commands) === $initial, 'Replacement changed a recorded initial identity.'); + + return (new DeferredChildAllLedger($this, $initial))->validate($history, $parentFinished); + } + + /** + * The parent continuation event intentionally has no predecessor field. + * Verify that relation from the replacement child's own canonical start. + * + * @param list> $history + * @param list> $predecessor + * @param array $original + * @param array{run_id: string, workflow_link_id: string, terminal: ?string} $replacement + * @param array{workflow_id: string, run_id: string} $parent + */ + public function assertReplacementHistory(array $history, array $original, array $replacement, array $parent, array $predecessor): void + { + self::check(in_array($this->case, ['retry', 'continue'], true), 'Unqualified replacement history.'); + $starts = array_values(array_filter($history, static fn (array $event): bool => ($event['event_type'] ?? null) === 'WorkflowStarted')); + self::check(count($starts) === 1, 'Replacement child must have exactly one canonical start.'); + $payload = $starts[0]['payload']; + foreach (['workflow_instance_id' => $original['child_workflow_instance_id'], 'workflow_run_id' => $replacement['run_id'], + 'child_call_id' => $original['child_call_id'], 'parent_workflow_instance_id' => $parent['workflow_id'], + 'parent_workflow_run_id' => $parent['run_id'], 'parent_sequence' => 1, 'workflow_type' => $this->childType()] as $key => $value) { + self::check(($payload[$key] ?? null) === $value, 'Replacement child start changed its parent/call/type identity.'); + } + if ($this->case === 'retry') { + self::check(($payload['retry_of_child_workflow_run_id'] ?? null) === $original['child_workflow_run_id'] + && ($payload['retry_attempt'] ?? null) === 2 && ($payload['continued_from_run_id'] ?? null) === null + && ($payload['workflow_link_id'] ?? null) === $replacement['workflow_link_id'], 'Replacement child has no exact retry predecessor/link.'); + $failed = array_filter($predecessor, static fn (array $event): bool => ($event['event_type'] ?? null) === 'WorkflowFailed'); + self::check(count($failed) === 1, 'Retry predecessor has no unique canonical failure.'); + } else { + self::check(($payload['continued_from_run_id'] ?? null) === $original['child_workflow_run_id'] + && ($payload['retry_of_child_workflow_run_id'] ?? null) === null && ($payload['retry_attempt'] ?? null) === null + && is_string($payload['workflow_link_id'] ?? null) && $payload['workflow_link_id'] !== '', 'Replacement child has no exact continuation predecessor.'); + self::check(is_string($original['workflow_link_id'] ?? null) && $original['workflow_link_id'] !== '' + && $payload['workflow_link_id'] !== $original['workflow_link_id'] + && $payload['workflow_link_id'] !== $replacement['workflow_link_id'], 'Continuation link reuses a parent-child link role.'); + $continued = array_values(array_filter($predecessor, static fn (array $event): bool => ($event['event_type'] ?? null) === 'WorkflowContinuedAsNew')); + self::check(count($continued) === 1, 'Continuation predecessor has no unique canonical handoff.'); + $handoff = $continued[0]['payload']; + self::check(($handoff['continued_to_run_id'] ?? null) === $replacement['run_id'] + && ($handoff['continued_to_run_number'] ?? null) === 2 && ($handoff['sequence'] ?? null) === 1 + && ($handoff['workflow_link_id'] ?? null) === $payload['workflow_link_id'], 'Continuation predecessor did not hand off to this child run/link.'); + } + } + + /** @param array $response */ + public static function assertEmptyPoll(array $response): void + { + self::check(array_key_exists('task', $response) && $response['task'] === null + && in_array($response['poll_status'] ?? null, ['empty', 'timeout'], true) + && ($response['reason'] ?? null) === null, 'Poll did not prove an empty parent queue.'); + } + + /** @param array $task */ + public static function assertParentTask(array $task, string $workflow, string $run, string $queue): void + { + self::check(($task['workflow_id'] ?? null) === $workflow && ($task['run_id'] ?? null) === $run + && ($task['task_queue'] ?? null) === $queue, 'Managed parent received another workflow, run or queue.'); + } + + /** @phpstan-assert true $condition */ + public static function check(bool $condition, string $message): void + { + if (!$condition) { + throw new ChildAllQualificationFailure($message); + } + } +} + +final class ChildAllProbeFailure extends RuntimeException +{ +} + +final class ChildAllQualificationFailure extends RuntimeException +{ +} diff --git a/scripts/Support/DeferredChildAllTurnTransport.php b/scripts/Support/DeferredChildAllTurnTransport.php new file mode 100644 index 00000000..47b3768b --- /dev/null +++ b/scripts/Support/DeferredChildAllTurnTransport.php @@ -0,0 +1,60 @@ +|null */ + private ?array $task = null; + + public function __construct( + private readonly BoundedTransport $http, + private readonly string $receipt, + private readonly Closure $completed, + ) { + } + + public function supportsBoundedRequests(): bool + { + return $this->http->supportsBoundedRequests(); + } + + public function send(string $method, string $uri, array $headers, ?array $body = null): ?array + { + return $this->observe($uri, $body, $this->http->send($method, $uri, $headers, $body)); + } + + public function sendBounded(string $method, string $uri, array $headers, ?array $body, int $timeoutSeconds): ?array + { + return $this->observe($uri, $body, $this->http->sendBounded($method, $uri, $headers, $body, $timeoutSeconds)); + } + + /** + * @param array|null $body + * @param array|null $response + * @return array|null + */ + private function observe(string $uri, ?array $body, ?array $response): ?array + { + if (str_ends_with($uri, '/workflow-tasks/poll') && is_array($response['task'] ?? null)) { + $this->task = array_intersect_key($response['task'], array_flip([ + 'task_id', 'workflow_id', 'run_id', 'task_queue', 'arguments', + ])); + } + if (preg_match('~/workflow-tasks/([^/]+)/complete$~', $uri, $match) === 1) { + DeferredChildAllScenario::check(($response['recorded'] ?? null) === true, 'Worker turn was not recorded.'); + DeferredChildAllScenario::check(($this->task['task_id'] ?? null) === rawurldecode($match[1]), 'Completion has no observed task.'); + $bytes = json_encode(['task' => $this->task, 'body' => $body, 'response' => $response], JSON_THROW_ON_ERROR); + DeferredChildAllScenario::check(file_put_contents($this->receipt, $bytes, LOCK_EX) === strlen($bytes), 'Could not retain the private turn receipt.'); + ($this->completed)(); + } + + return $response; + } +} diff --git a/scripts/deferred-child-all.md b/scripts/deferred-child-all.md new file mode 100644 index 00000000..1bc11ad6 --- /dev/null +++ b/scripts/deferred-child-all.md @@ -0,0 +1,79 @@ +# Deferred child barrier consumer adapter + +This source runner prepares the PHP consumer side of [#185](https://github.com/durable-workflow/sdk-php/issues/185) +and [Server #325](https://github.com/durable-workflow/server/issues/325). +It has not qualified an installed SDK or a native Server tuple. A future run +requires the published SDK, immutable native producer, runner revision, resource +budget and cleanup plan selected in those records. The autoload argument alone +does not attest a published package. Ordinary source checks cannot replace that +acceptance. + +`fixtures/parallel-child-all.json` is copied byte for byte from the Server source +identified by `DeferredChildAllScenario::SOURCE`; its checksum is enforced by the +adapter. The retained historical Server PHP observation uses an older SDK and +is not an expected installed result for this runner. The five canonical cases +are single-child success, ordered results after completion order `[2, 0, 1]`, and +an early child failure, cancellation or termination. Two additional controls +exercise a retry and continue-as-new. Those controls preserve the initial child +call, link, instance and run tuple in its original schedule. The second run has +a distinct parent-child link. Subsequent events must use that run/link and the +original call, instance and group path. Retry records its predecessor in parent +history. Continuation proves its predecessor through both child histories; the +continue-as-new link is separate from the new parent-child link. These shapes +follow the [accepted native source return](https://github.com/durable-workflow/server/issues/325#issuecomment-6665208804). + +The runner uses public `Client` and managed `Worker` APIs. It assigns a different +valid queue to each deferred child so one worker at a time can drive a fixed +completion order. Server's flat group contract permits independent member +queues. Each acknowledged turn exits, and the next turn starts in a new PHP +process without a shared replay cache. The replacement parent must return the +exact typed result or `ChildWorkflowFailed` shape without starting another child. +Duplicate parent and child completion attempts must be explicit no-ops or fenced +and leave complete canonical history unchanged. After an early failure closes +the parent under default abandon, successful siblings must add no parent history +or further ready task. The finite child ledger checks exact integer sequences, +workflow types, one start and at most one case-appropriate terminal per run. +Unknown child events, duplicate or conflicting terminals and events on closed +or replaced runs refuse, including malformed rows written by the first +completion. An early failure may leave a sibling started and unresolved under +abandon. Empty-queue observations require an explicit empty or +timeout poll with no refusal reason. Lifecycle or unavailable replies cannot +prove that the parent has no task. Both parent receipts bind the selected +workflow, run and queue. + +For a separately admitted disposable tuple, the command is: + +```sh +php scripts/qualify-deferred-child-all.php /consumer/vendor/autoload.php http://disposable-server:8080 +``` + +An optional final case name selects one control. The server must use the +synthetic `claimed-task-admin`, `claimed-task-client` and `claimed-task-worker` +roles from the existing child lifecycle qualification setup. The runner makes +synthetic namespaces and retains them for inspection until the outer disposable +server is removed. It does not run against production or select credentials. +Only sanitized workflow identities, queue names, event types and synthetic +results reach stdout. Private turn receipts contain completion lease authority, +are created with mode 0600 and are removed together with private worker logs in +each turn's `finally` block. Authentication headers and registration tokens are +not retained. Each child process has a 35-second turn deadline and bounded +termination; the controller checks a 600-second total budget. The admitted outer +runner must also enforce its own absolute resource and cleanup limits. + +This HTTP consumer adapter observes empty parent polls and two managed parent +completion turns. It cannot prove a database row count, competing native-node +serialization or cold database recovery. Server's cross-backend native tests own +those assertions. Started child handles inside `all`, nested, mixed, selected or +scoped groups and grouped conditions remain outside this adapter. The existing +`qualify-child-start-lifecycle.php` continues to own the separate sequential +started-handle and lifecycle policy cases. + +Offline source controls run with: + +```sh +vendor/bin/phpunit tests/DeferredChildAllAdapterTest.php +``` + +They use the actual SDK replayer and a managed worker with an inert transport, +including changed identity/path refusals and unrecorded receipt controls. They +make no network requests and do not invoke the runtime runner. diff --git a/scripts/fixtures/parallel-child-all.json b/scripts/fixtures/parallel-child-all.json new file mode 100644 index 00000000..038ae5e6 --- /dev/null +++ b/scripts/fixtures/parallel-child-all.json @@ -0,0 +1,51 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "review_state": "reviewed", + "workflow_type": "parity.v1.echo", + "input": { + "message": "concurrent waits λ", + "count": 9007199254740993, + "optional": null + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "concurrent waits λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "WorkflowCompleted" + ], + "id": "parallel-child-all", + "contract": "Worker-protocol flat child-only all barriers authored and cold-replayed by the real published PHP SDK: every child starts atomically, carries the original call/link/instance/run and path, and returns exact ordered typed results despite out-of-order completion. Success waits for every member; failure/cancel/terminate wakes the parent immediately as ChildWorkflowFailed. Duplicate parent/child completions add no starts or history. After a failed parent catches its outcome and closes under default abandon, later successful siblings add no parent resolution or wakeup. This worker-HTTP side contract is explicitly inapplicable to embedded execution. Native cold database replacement, retries/continuation, malformed and unqualified shape refusals, bounded inventory and rollback are covered separately by cross-backend native tests. Child select/nested/mixed/scoped/started-in-all remain refused.", + "parallel_child_all": { + "parent_type": "parity.v1.child_all_parent", + "child_type": "parity.v1.child_all_child", + "cases": { + "single": 1, + "out_of_order": 3, + "failed": 2, + "cancelled": 2, + "terminated": 2 + }, + "failure_messages": { + "failed": "child barrier failure λ", + "cancelled": "Workflow cancelled: child barrier closed λ", + "terminated": "Workflow terminated: child barrier closed λ" + } + } +} diff --git a/scripts/qualify-deferred-child-all.php b/scripts/qualify-deferred-child-all.php new file mode 100644 index 00000000..a7a09c8d --- /dev/null +++ b/scripts/qualify-deferred-child-all.php @@ -0,0 +1,299 @@ +requestShutdown(); + }); + $client = new Client($url, namespace: $namespace, workerToken: 'claimed-task-worker', transport: $transport); + $worker = new Worker($client, $queue, workerId: $workerId); + $worker->registerWorkflow($scenario->parentType(), $scenario->parent()); + $worker->registerWorkflow($scenario->childType(), $scenario->child($initialRun !== '' ? $initialRun : null)); + $worker->run(1); + exit(0); +} + +$deadline = microtime(true) + 600; + +function childAllBudget(): void +{ + global $deadline; + Scenario::check(microtime(true) < $deadline, 'Deferred-child adapter exceeded its absolute time budget.'); +} + +/** + * @param resource $process + * @phpstan-impure + */ +function childAllProcessRunning($process): bool +{ + return proc_get_status($process)['running']; +} + +/** @return list> */ +function childAllHistory(Client $client, string $workflow, string $run): array +{ + childAllBudget(); + $page = $client->workflowHistory($workflow, $run, 200); + Scenario::check(empty($page['next_page_token']), 'History exceeds the bounded complete page.'); + + return $page['events'] ?? $page['history_events'] ?? []; +} + +/** + * Exactly one worker exists at a time. Each turn has a fresh process/cache and + * exits after its completion acknowledgement, including an empty command turn. + * + * @return array + */ +function childAllTurn(Scenario $scenario, string $namespace, string $queue, string $initialRun = ''): array +{ + global $script, $autoload, $url, $deadline; + childAllBudget(); + $receipt = tempnam(sys_get_temp_dir(), 'php-child-all-turn-'); + $log = tempnam(sys_get_temp_dir(), 'php-child-all-log-'); + $process = null; + try { + Scenario::check(is_string($receipt) && is_string($log), 'Could not allocate private worker receipts.'); + $process = proc_open([ + PHP_BINARY, $script, $autoload, $url, '--worker', $scenario->case, $namespace, $queue, + json_encode($scenario->queues, JSON_THROW_ON_ERROR), $queue.'-'.bin2hex(random_bytes(4)), $receipt, $initialRun, + ], [0 => ['file', '/dev/null', 'r'], 1 => ['file', $log, 'a'], 2 => ['file', $log, 'a']], $pipes); + Scenario::check(is_resource($process), 'Could not start the managed worker.'); + $turnDeadline = min($deadline, microtime(true) + 35); + do { + $status = proc_get_status($process); + if (!$status['running']) { + Scenario::check($status['exitcode'] === 0, 'Managed worker exited unsuccessfully.'); + $bytes = file_get_contents($receipt); + Scenario::check(is_string($bytes) && $bytes !== '', 'Managed worker exited without a recorded turn.'); + + return json_decode($bytes, true, flags: JSON_THROW_ON_ERROR); + } + usleep(50_000); + } while (microtime(true) < $turnDeadline); + throw new ChildAllQualificationFailure('Managed worker exceeded its turn deadline.'); + } finally { + childAllStopProcess($process); + if (is_string($receipt)) { + @unlink($receipt); + } + if (is_string($log)) { + @unlink($log); + } + } +} + +/** @param resource|null $process */ +function childAllStopProcess($process): void +{ + if (!is_resource($process)) { + return; + } + if (childAllProcessRunning($process)) { + proc_terminate($process); + $stop = microtime(true) + 2; + while (childAllProcessRunning($process) && microtime(true) < $stop) { + usleep(50_000); + } + if (childAllProcessRunning($process)) { + proc_terminate($process, 9); + } + } + proc_close($process); +} + +/** A public poll observes readiness without running or retaining a parent claim. */ +function childAllNoParentTask(Client $worker, Scenario $scenario, string $queue): void +{ + childAllBudget(); + $id = $queue.'-observer-'.bin2hex(random_bytes(4)); + $registration = $worker->registerWorker($id, $queue, [$scenario->parentType()], []); + try { + $response = $worker->pollWorkflowTaskResponse($id, $queue, 0); + Scenario::assertEmptyPoll($response); + } finally { + $worker->deregisterWorkerRegistration($id, $registration['registration_token'] ?? null); + } +} + +/** + * @param array $receipt + * @param list $also + */ +function childAllDuplicate(Client $worker, Client $control, array $receipt, string $workflow, string $run, array $also = []): void +{ + $targets = [[$workflow, $run], ...$also]; + $before = array_map(static fn (array $target): array => childAllHistory($control, ...$target), $targets); + $body = $receipt['body']; + try { + $response = $worker->completeWorkflowTask($receipt['task']['task_id'], $body['lease_owner'], $body['workflow_task_attempt'], $body['commands']); + Scenario::check(($response['recorded'] ?? null) === false && ($response['reason'] ?? null) === 'already_completed', 'Duplicate completion was not an explicit no-op.'); + } catch (ServerException $error) { + Scenario::check($error->status === 409 && $error->reason === 'task_not_leased', 'Duplicate completion was not fenced.'); + } + foreach ($targets as $index => $target) { + Scenario::check(childAllHistory($control, ...$target) === $before[$index], 'Duplicate completion changed canonical history.'); + } +} + +/** + * @param array $identity + * @param array{workflow_id: string, run_id: string, queue: string} $parent + * @return array + */ +function childAllMember(Scenario $scenario, Client $worker, Client $control, string $namespace, int $index, array $identity, array $parent): array +{ + $receipt = childAllTurn($scenario, $namespace, $scenario->queues[$index], $identity['child_workflow_run_id']); + $task = $receipt['task']; + Scenario::check(($task['workflow_id'] ?? null) === $identity['child_workflow_instance_id'] + && ($task['task_queue'] ?? null) === $scenario->queues[$index], 'Managed child received a different member or queue.'); + $arguments = $control->payloadCodec()->decodeEnvelope($task['arguments']); + Scenario::check(($arguments[0] ?? null) === ['index' => $index, 'value' => $scenario->fixture['input']], 'Child arguments lost their exact values or types.'); + childAllDuplicate($worker, $control, $receipt, $task['workflow_id'], $task['run_id'], [[$parent['workflow_id'], $parent['run_id']]]); + + return $receipt; +} + +/** + * @param array $identity + * @param array{workflow_id: string, run_id: string, queue: string, commands: list>, identities: list>} $parent + * @return list + */ +function childAllRunMember(Scenario $scenario, Client $worker, Client $control, string $namespace, int $index, array $identity, array $parent): array +{ + if ($index === 0 && in_array($scenario->case, ['cancelled', 'terminated'], true)) { + if ($scenario->case === 'cancelled') { + $control->cancelWorkflow($identity['child_workflow_instance_id'], 'child barrier closed λ', $identity['child_workflow_run_id']); + } else { + $control->terminateWorkflow($identity['child_workflow_instance_id'], 'child barrier closed λ', $identity['child_workflow_run_id']); + } + + return []; + } + $receipt = childAllMember($scenario, $worker, $control, $namespace, $index, $identity, $parent); + Scenario::check($receipt['task']['run_id'] === $identity['child_workflow_run_id'], 'Initial child task used another run.'); + $runs = [$receipt['task']['run_id']]; + if ($index !== 0 || !in_array($scenario->case, ['retry', 'continue'], true)) { + return $runs; + } + Scenario::check(count($receipt['body']['commands']) === 1 + && $receipt['body']['commands'][0]['type'] === ($scenario->case === 'retry' ? 'fail_workflow' : 'continue_as_new'), 'Initial child turn did not request its selected replacement.'); + childAllNoParentTask($worker, $scenario, $parent['queue']); + $replacement = childAllMember($scenario, $worker, $control, $namespace, $index, $identity, $parent); + Scenario::check($replacement['task']['run_id'] !== $identity['child_workflow_run_id'], 'Child chain did not create a distinct second run.'); + $runs[] = $replacement['task']['run_id']; + $history = childAllHistory($control, $parent['workflow_id'], $parent['run_id']); + $chains = $scenario->assertStable($history, $parent['commands'], $parent['identities']); + $scenario->assertReplacementHistory(childAllHistory($control, $identity['child_workflow_instance_id'], $replacement['task']['run_id']), + $identity, $chains[0][1], ['workflow_id' => $parent['workflow_id'], 'run_id' => $parent['run_id']], + childAllHistory($control, $identity['child_workflow_instance_id'], $identity['child_workflow_run_id'])); + + return $runs; +} + +/** @return array */ +function childAllCase(string $case): array +{ + global $url; + $sizes = ['single' => 1, 'out_of_order' => 3, 'failed' => 2, 'cancelled' => 2, 'terminated' => 2, 'retry' => 2, 'continue' => 2]; + Scenario::check(isset($sizes[$case]), 'Unknown deferred-child case.'); + $suffix = bin2hex(random_bytes(6)); + $namespace = 'php-child-all-'.$suffix; + $parentQueue = 'parent-'.$suffix; + $queues = array_map(static fn (int $index): string => 'child-'.$suffix.'-'.$index, range(0, $sizes[$case] - 1)); + $scenario = new Scenario($case, $queues); + $admin = new Client($url, controlToken: 'claimed-task-admin'); + $admin->createNamespace($namespace); + $control = new Client($url, namespace: $namespace, controlToken: 'claimed-task-client'); + $worker = new Client($url, namespace: $namespace, workerToken: 'claimed-task-worker'); + $parent = $control->startWorkflow($scenario->parentType(), 'parent-'.$suffix, $parentQueue, [$scenario->fixture['input']]); + $run = (string) $parent->selectedRunId; + $start = childAllTurn($scenario, $namespace, $parentQueue); + Scenario::assertParentTask($start['task'], $parent->workflowId, $run, $parentQueue); + $prefix = childAllHistory($control, $parent->workflowId, $run); + $commands = $start['body']['commands']; + $identities = $scenario->identities($prefix, $commands); + $scenario->assertStable($prefix, $commands, $identities); + childAllDuplicate($worker, $control, $start, $parent->workflowId, $run); + childAllNoParentTask($worker, $scenario, $parentQueue); + $closed = null; + $memberRuns = []; + $parentContext = ['workflow_id' => $parent->workflowId, 'run_id' => $run, 'queue' => $parentQueue, + 'commands' => $commands, 'identities' => $identities]; + foreach ($scenario->order() as $offset => $index) { + childAllBudget(); + $runs = childAllRunMember($scenario, $worker, $control, $namespace, $index, $identities[$index], $parentContext); + if ($runs !== []) { + $memberRuns[$index] = $runs; + } + $history = childAllHistory($control, $parent->workflowId, $run); + $scenario->assertStable($history, $commands, $identities); + if ($closed !== null) { + Scenario::check($history === $closed, 'A late abandoned sibling mutated the closed parent.'); + childAllNoParentTask($worker, $scenario, $parentQueue); + continue; + } + $ready = $scenario->failsEarly() || $offset === $scenario->size - 1; + if (!$ready) { + childAllNoParentTask($worker, $scenario, $parentQueue); + continue; + } + $finish = childAllTurn($scenario, $namespace, $parentQueue); + Scenario::assertParentTask($finish['task'], $parent->workflowId, $run, $parentQueue); + $finishCommands = $finish['body']['commands']; + Scenario::check(count($finishCommands) === 1 && $finishCommands[0]['type'] === 'complete_workflow', 'Cold parent did not finish with one result.'); + $result = $control->payloadCodec()->decodeEnvelope($finishCommands[0]['result']); + Scenario::check($result === $scenario->expected() && $parent->result(5) === $scenario->expected(), 'Cold parent result changed declaration order, values or failure type.'); + childAllDuplicate($worker, $control, $finish, $parent->workflowId, $run); + $closed = childAllHistory($control, $parent->workflowId, $run); + $scenario->assertStable($closed, $commands, $identities, parentFinished: true); + childAllNoParentTask($worker, $scenario, $parentQueue); + } + Scenario::check($closed !== null, 'Parent never reached its expected terminal result.'); + if (in_array($case, ['retry', 'continue'], true)) { + $starts = array_values(array_filter($closed, static fn (array $event): bool => $event['event_type'] === 'ChildRunStarted' && $event['payload']['sequence'] === 1)); + Scenario::check(array_column(array_column($starts, 'payload'), 'child_workflow_run_id') === $memberRuns[0], 'Canonical child chain differs from the two managed runs.'); + } + + return ['case' => $case, 'namespace' => $namespace, 'workflow_id' => $parent->workflowId, 'run_id' => $run, + 'initial_children' => $identities, 'child_queues' => $queues, 'member_runs' => $memberRuns, + 'child_run_chains' => $scenario->assertStable($closed, $commands, $identities, parentFinished: true), + 'completion_order' => $scenario->order(), 'result' => $scenario->expected(), + 'events' => array_column($closed, 'event_type'), 'parent_completion_turns' => 2]; +} + +$started = gmdate(DATE_ATOM); +$cases = isset($argv[3]) ? [$argv[3]] : ['single', 'out_of_order', 'failed', 'cancelled', 'terminated', 'retry', 'continue']; +$results = []; +foreach ($cases as $case) { + $results[] = childAllCase($case); +} +echo json_encode(['fixture_source' => Scenario::SOURCE, 'fixture_sha256' => Scenario::FIXTURE_SHA256, + 'started_at' => $started, 'finished_at' => gmdate(DATE_ATOM), 'scenarios' => $results], JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT).PHP_EOL; diff --git a/tests/DeferredChildAllAdapterTest.php b/tests/DeferredChildAllAdapterTest.php new file mode 100644 index 00000000..0902c659 --- /dev/null +++ b/tests/DeferredChildAllAdapterTest.php @@ -0,0 +1,613 @@ +replay($scenario->parent(), [], [$scenario->fixture['input']], 'parents')->commands; + self::assertCount($size, $commands); + self::assertSame(9007199254740993, $scenario->fixture['input']['count']); + foreach ($commands as $index => $command) { + self::assertSame('start_child_workflow', $command['type']); + self::assertSame($scenario->childType(), $command['workflow_type']); + self::assertSame('child-'.$index, $command['queue']); + self::assertNotSame('started', $command['await'] ?? null); + self::assertCount(1, $command['parallel_group_path']); + self::assertSame('child', $command['parallel_group_path'][0]['parallel_group_kind']); + self::assertSame($index, $command['parallel_group_path'][0]['parallel_group_index']); + self::assertSame([['index' => $index, 'value' => $scenario->fixture['input']]], $codec->decodeEnvelope($command['arguments'])); + self::assertSame($case === 'retry' && $index === 0 ? ['max_attempts' => 2, 'backoff_seconds' => [0]] : null, $command['retry_policy'] ?? null); + } + $history = self::history($commands); + $identities = $scenario->identities($history, $commands); + $scenario->assertStable($history, $commands, $identities); + self::assertSame([], (new Replayer($codec))->replay($scenario->parent(), $history, [$scenario->fixture['input']], 'parents')->commands); + foreach ($scenario->order() as $offset => $index) { + $payload = $history[$index * 2]['payload']; + if ($index === 0 && in_array($case, ['retry', 'continue'], true)) { + $replacement = self::replacement($case, $payload); + $history[] = $replacement; + $scenario->assertStable($history, $commands, $identities); + $payload = $replacement['payload']; + } + $terminal = self::terminal($payload, $codec->envelope(['echo' => [ + 'index' => $index, 'value' => $scenario->fixture['input'], + ]])); + if ($scenario->failsEarly() && $offset === 0) { + $terminal['event_type'] = $scenario->expected()['failure_type']; + $terminal['payload']['message'] = $scenario->expected()['message']; + unset($terminal['payload']['result']); + } + $history[] = $terminal; + $scenario->assertStable($history, $commands, $identities); + $resumed = (new Replayer($codec))->replay($scenario->parent(), $history, [$scenario->fixture['input']], 'parents')->commands; + if (!$scenario->failsEarly() && $offset < $size - 1) { + self::assertSame([], $resumed); + } else { + self::assertCount(1, $resumed); + self::assertSame('complete_workflow', $resumed[0]['type']); + self::assertSame($scenario->expected(), $codec->decodeEnvelope($resumed[0]['result'])); + } + } + $history[] = ['event_type' => 'WorkflowCompleted', 'payload' => []]; + self::assertCount($size, $scenario->assertStable($history, $commands, $identities, parentFinished: true)); + } + + /** @return iterable */ + public static function cases(): iterable + { + yield 'single' => ['single', 1]; + yield 'out of order' => ['out_of_order', 3]; + yield 'failed' => ['failed', 2]; + yield 'cancelled' => ['cancelled', 2]; + yield 'terminated' => ['terminated', 2]; + yield 'retry extension' => ['retry', 2]; + yield 'continue extension' => ['continue', 2]; + } + + public function testRetryFailsOnlyItsFixedInitialRunAndReplacementReturnsTheSameTypedValue(): void + { + $scenario = self::scenario('retry', 2); + $codec = new AvroPayloadCodec(); + $input = ['index' => 0, 'value' => $scenario->fixture['input']]; + try { + (new Replayer($codec))->replay($scenario->child('initial'), [], [$input], 'child-0', ['run_id' => 'initial']); + self::fail('Initial retry run must fail.'); + } catch (ChildAllProbeFailure $error) { + self::assertSame('retry group member', $error->getMessage()); + } + $commands = (new Replayer($codec))->replay($scenario->child('initial'), [], [$input], 'child-0', ['run_id' => 'replacement'])->commands; + self::assertSame('complete_workflow', $commands[0]['type']); + self::assertSame(['echo' => $input], $codec->decodeEnvelope($commands[0]['result'])); + } + + public function testContinuePassesTheSameTypedArgumentsToItsSecondRun(): void + { + $scenario = self::scenario('continue', 2); + $codec = new AvroPayloadCodec(); + $input = ['index' => 0, 'value' => $scenario->fixture['input']]; + $first = (new Replayer($codec))->replay($scenario->child(), [], [$input], 'child-0')->commands; + self::assertSame('continue_as_new', $first[0]['type']); + $arguments = $codec->decodeEnvelope($first[0]['arguments']); + self::assertSame([$input, 1], $arguments); + $second = (new Replayer($codec))->replay($scenario->child(), [], $arguments, 'child-0')->commands; + self::assertSame(['echo' => $input], $codec->decodeEnvelope($second[0]['result'])); + } + + #[DataProvider('changedHistories')] + public function testIdentityGuardRejectsChangedOrDuplicatedProducerRows(string $mutation): void + { + $scenario = self::scenario('out_of_order', 3); + $codec = new AvroPayloadCodec(); + $commands = (new Replayer($codec))->replay($scenario->parent(), [], [$scenario->fixture['input']], 'parents')->commands; + $history = self::history($commands); + $identities = $scenario->identities($history, $commands); + $history[] = self::terminal($history[0]['payload'], $codec->envelope('done')); + switch ($mutation) { + case 'extra start': $history[] = $history[0]; break; + case 'duplicate run start': $history[] = $history[1]; break; + case 'missing run start': unset($history[1]); break; + case 'changed initial run': $history[0]['payload']['child_workflow_run_id'] = 'different'; break; + case 'shared call': $history[2]['payload']['child_call_id'] = $history[0]['payload']['child_call_id']; break; + case 'changed terminal call': $history[6]['payload']['child_call_id'] = 'different'; break; + case 'changed terminal link': $history[6]['payload']['workflow_link_id'] = 'different'; break; + case 'changed terminal instance': $history[6]['payload']['child_workflow_instance_id'] = 'different'; break; + case 'changed terminal run': $history[6]['payload']['child_workflow_run_id'] = 'different'; break; + case 'changed terminal path': $history[6]['payload']['parallel_group_path'][0]['parallel_group_index'] = 2; break; + case 'unknown member': $history[6]['payload']['sequence'] = 4; break; + } + $this->expectException(RuntimeException::class); + $scenario->assertStable($history, $commands, $identities); + } + + /** @return iterable */ + public static function changedHistories(): iterable + { + foreach (['extra start', 'duplicate run start', 'missing run start', 'changed initial run', 'shared call', 'changed terminal call', 'changed terminal link', + 'changed terminal instance', 'changed terminal run', 'changed terminal path', 'unknown member'] as $mutation) { + yield $mutation => [$mutation]; + } + } + + #[DataProvider('replacementCases')] + public function testReplacementRunKeepsInitialIdentityAndOriginalMemberPath(string $case): void + { + $scenario = self::scenario($case, 2); + $codec = new AvroPayloadCodec(); + $commands = (new Replayer($codec))->replay($scenario->parent(), [], [$scenario->fixture['input']], 'parents')->commands; + $history = self::history($commands); + $identities = $scenario->identities($history, $commands); + $next = self::replacement($case, $history[1]['payload']); + $history[] = $next; + $history[] = self::terminal($next['payload'], $codec->envelope('done')); + $chains = $scenario->assertStable($history, $commands, $identities); + self::assertSame('run-0', $scenario->identities($history, $commands)[0]['child_workflow_run_id']); + self::assertSame('link-0', $scenario->identities($history, $commands)[0]['workflow_link_id']); + self::assertSame('replacement-link', $chains[0][1]['workflow_link_id']); + self::assertSame('ChildRunCompleted', $chains[0][1]['terminal']); + } + + /** @return iterable */ + public static function replacementCases(): iterable + { + yield 'retry' => ['retry']; + yield 'continue' => ['continue']; + } + + #[DataProvider('replacementLedgerMutations')] + public function testReplacementLedgerRejectsReusedClosedOrUnprovenRuns(string $case, string $mutation): void + { + $scenario = self::scenario($case, 2); + $codec = new AvroPayloadCodec(); + $commands = (new Replayer($codec))->replay($scenario->parent(), [], [$scenario->fixture['input']], 'parents')->commands; + $history = self::history($commands); + $initial = $scenario->identities($history, $commands); + $next = self::replacement($case, $history[1]['payload']); + $terminal = self::terminal($next['payload'], $codec->envelope('done')); + switch ($mutation) { + case 'string replacement sequence': $next['payload']['sequence'] = '1'; break; + case 'wrong predecessor': $next['payload']['retry_of_child_workflow_run_id'] = 'other'; break; + case 'shared replacement link': $next['payload']['workflow_link_id'] = 'link-1'; break; + case 'wrong run number': $next['payload']['child_run_number'] = 1; break; + case 'initial terminal before replacement': $history[] = self::terminal($history[0]['payload'], $codec->envelope('done')); break; + } + $history[] = $next; + switch ($mutation) { + case 'duplicate replacement start': $history[] = $next; break; + case 'third run': + $third = $next; + $third['payload']['child_workflow_run_id'] = 'third'; + $third['payload']['workflow_link_id'] = 'third-link'; + $third['payload']['child_run_number'] = 3; + $history[] = $third; + break; + case 'event on replaced run': $history[] = self::terminal($history[0]['payload'], $codec->envelope('done')); break; + case 'wrong final link': $terminal['payload']['workflow_link_id'] = 'link-0'; $history[] = $terminal; break; + case 'repeat final terminal': $history[] = $terminal; $history[] = $terminal; break; + case 'replacement after terminal': + $history[] = $terminal; + $next['payload']['child_workflow_run_id'] = 'after-terminal'; + $next['payload']['workflow_link_id'] = 'after-terminal-link'; + $history[] = $next; + break; + } + $this->expectException(RuntimeException::class); + $scenario->assertStable($history, $commands, $initial); + } + + /** @return iterable */ + public static function replacementLedgerMutations(): iterable + { + foreach (['retry', 'continue'] as $case) { + foreach (['string replacement sequence', 'wrong predecessor', 'shared replacement link', 'wrong run number', + 'initial terminal before replacement', 'duplicate replacement start', 'third run', 'event on replaced run', + 'wrong final link', 'repeat final terminal', 'replacement after terminal'] as $mutation) { + yield $case.' '.$mutation => [$case, $mutation]; + } + } + } + + #[DataProvider('replacementCases')] + public function testChildHistoryProvesReplacementWithoutEquatingTheContinuationAndParentLinks(string $case): void + { + $proof = self::replacementProof($case); + self::scenario($case, 2)->assertReplacementHistory(...$proof); + self::assertSame($case === 'continue' ? 'continuation-link' : 'replacement-link', $proof[0][0]['payload']['workflow_link_id']); + if ($case === 'continue') { + self::assertNotSame($proof[1]['workflow_link_id'], $proof[0][0]['payload']['workflow_link_id']); + self::assertNotSame($proof[2]['workflow_link_id'], $proof[0][0]['payload']['workflow_link_id']); + } else { + self::assertSame($proof[2]['workflow_link_id'], $proof[0][0]['payload']['workflow_link_id']); + } + } + + #[DataProvider('parentChildLinkRoles')] + public function testContinuationRefusesCoherentTwoFieldParentChildLinkSubstitution(string $role): void + { + [$history, $original, $replacement, $parent, $predecessor] = self::replacementProof('continue'); + $link = $role === 'initial' ? $original['workflow_link_id'] : $replacement['workflow_link_id']; + $history[0]['payload']['workflow_link_id'] = $link; + $predecessor[0]['payload']['workflow_link_id'] = $link; + $this->expectException(RuntimeException::class); + self::scenario('continue', 2)->assertReplacementHistory($history, $original, $replacement, $parent, $predecessor); + } + + /** @return iterable */ + public static function parentChildLinkRoles(): iterable + { + yield 'initial parent-child link' => ['initial']; + yield 'replacement parent-child link' => ['replacement']; + } + + #[DataProvider('replacementProofMutations')] + public function testChildReplacementProofRefusesChangedPredecessorOrAuthority(string $case, string $mutation): void + { + [$history, $original, $replacement, $parent, $predecessor] = self::replacementProof($case); + switch ($mutation) { + case 'wrong parent run': $history[0]['payload']['parent_workflow_run_id'] = 'other'; break; + case 'wrong child call': $history[0]['payload']['child_call_id'] = 'other'; break; + case 'wrong workflow type': $history[0]['payload']['workflow_type'] = 'other'; break; + case 'string parent sequence': $history[0]['payload']['parent_sequence'] = '1'; break; + case 'duplicate child start': $history[] = $history[0]; break; + case 'missing predecessor terminal': $predecessor = []; break; + case 'duplicate predecessor terminal': $predecessor[] = $predecessor[0]; break; + case 'wrong predecessor': + $history[0]['payload'][$case === 'retry' ? 'retry_of_child_workflow_run_id' : 'continued_from_run_id'] = 'other'; + break; + case 'wrong predecessor link': + if ($case === 'retry') { + $history[0]['payload']['workflow_link_id'] = 'other'; + } else { + $predecessor[0]['payload']['workflow_link_id'] = 'replacement-link'; + } + break; + } + $this->expectException(RuntimeException::class); + self::scenario($case, 2)->assertReplacementHistory($history, $original, $replacement, $parent, $predecessor); + } + + /** @return iterable */ + public static function replacementProofMutations(): iterable + { + foreach (['retry', 'continue'] as $case) { + foreach (['wrong parent run', 'wrong child call', 'wrong workflow type', 'string parent sequence', + 'duplicate child start', 'missing predecessor terminal', 'duplicate predecessor terminal', 'wrong predecessor', 'wrong predecessor link'] as $mutation) { + yield $case.' '.$mutation => [$case, $mutation]; + } + } + } + + #[DataProvider('heldHistoryAdmissions')] + public function testHeldHistoryAdmissionsAreRefused(string $case, string $mutation): void + { + $scenario = self::scenario($case, $case === 'single' ? 1 : ($case === 'out_of_order' ? 3 : 2)); + $codec = new AvroPayloadCodec(); + $commands = (new Replayer($codec))->replay($scenario->parent(), [], [$scenario->fixture['input']], 'parents')->commands; + $history = self::history($commands); + $initial = $scenario->identities($history, $commands); + $terminal = self::terminal($history[0]['payload'], $codec->envelope('done')); + if ($scenario->failsEarly()) { + $terminal['event_type'] = $scenario->expected()['failure_type']; + } + $history[] = $terminal; + switch ($mutation) { + case 'duplicate terminal': $history[] = $terminal; break; + case 'conflicting terminal': + $terminal['event_type'] = $terminal['event_type'] === 'ChildRunCompleted' ? 'ChildRunFailed' : 'ChildRunCompleted'; + $history[] = $terminal; + break; + case 'unknown ChildRun type': $history[array_key_last($history)]['event_type'] = 'ChildRunAnything'; break; + case 'unknown Child type': $history[array_key_last($history)]['event_type'] = 'ChildWorkflowAnything'; break; + case 'string sequence': $history[array_key_last($history)]['payload']['sequence'] = '1'; break; + case 'leading-zero sequence': $history[array_key_last($history)]['payload']['sequence'] = '01'; break; + case 'float sequence': $history[array_key_last($history)]['payload']['sequence'] = 1.0; break; + case 'boolean sequence': $history[array_key_last($history)]['payload']['sequence'] = true; break; + case 'wrong workflow type': $history[array_key_last($history)]['payload']['child_workflow_type'] = 'other'; break; + case 'wrong run number': $history[array_key_last($history)]['payload']['child_run_number'] = '1'; break; + case 'post-parent-terminal event': + array_pop($history); + $history[] = ['event_type' => 'WorkflowCompleted', 'payload' => []]; + $history[] = $terminal; + break; + case 'post-terminal unknown event': + $terminal['event_type'] = 'ChildRunProgress'; + $history[] = $terminal; + break; + } + $this->expectException(RuntimeException::class); + $scenario->assertStable($history, $commands, $initial); + } + + /** @return iterable */ + public static function heldHistoryAdmissions(): iterable + { + foreach (['single', 'out_of_order', 'failed', 'cancelled', 'terminated'] as $case) { + foreach (['duplicate terminal', 'conflicting terminal'] as $mutation) { + yield $case.' '.$mutation => [$case, $mutation]; + } + } + foreach (['unknown ChildRun type', 'unknown Child type', 'string sequence', 'leading-zero sequence', + 'float sequence', 'boolean sequence', 'wrong workflow type', 'wrong run number', + 'post-parent-terminal event', 'post-terminal unknown event'] as $mutation) { + yield $mutation => ['out_of_order', $mutation]; + } + } + + #[DataProvider('abandonedCases')] + public function testFinishedEarlyFailurePreservesAStartedButAbandonedSibling(string $case): void + { + $scenario = self::scenario($case, 2); + $codec = new AvroPayloadCodec(); + $commands = (new Replayer($codec))->replay($scenario->parent(), [], [$scenario->fixture['input']], 'parents')->commands; + $history = self::history($commands); + $initial = $scenario->identities($history, $commands); + $terminal = self::terminal($history[0]['payload'], $codec->envelope('unused')); + $terminal['event_type'] = $scenario->expected()['failure_type']; + $history[] = $terminal; + $history[] = ['event_type' => 'WorkflowCompleted', 'payload' => []]; + $chains = $scenario->assertStable($history, $commands, $initial, parentFinished: true); + self::assertSame($scenario->expected()['failure_type'], $chains[0][0]['terminal']); + self::assertNull($chains[1][0]['terminal']); + self::assertSame('run-1', $chains[1][0]['run_id']); + } + + /** @return iterable */ + public static function abandonedCases(): iterable + { + foreach (['failed', 'cancelled', 'terminated'] as $case) { + yield $case => [$case]; + } + } + + public function testActualManagedWorkerStopsAfterItsRecordedTurnAndReceiptOmitsHeaders(): void + { + $scenario = self::scenario('single', 1); + $codec = new AvroPayloadCodec(); + $requests = []; + $wire = new ChildAllWire(static function (string $uri, ?array $body) use ($scenario, $codec, &$requests): array { + $requests[] = $uri; + if (str_ends_with($uri, '/workflow-tasks/poll')) { + return ['poll_status' => 'leased', 'task' => [ + 'task_id' => 'parent-task', 'workflow_id' => 'parent', 'run_id' => 'run', + 'workflow_type' => $scenario->parentType(), 'task_queue' => 'parents', + 'arguments' => $codec->envelope([$scenario->fixture['input']]), 'payload_codec' => 'avro', + 'history_events' => [], 'lease_owner' => 'worker', 'workflow_task_attempt' => 1, + 'registration_token' => 'unretained-registration', + ]]; + } + if (str_ends_with($uri, '/complete')) { + self::assertCount(1, $body['commands']); + return ['recorded' => true]; + } + if (str_ends_with($uri, '/parent-task/heartbeat')) { + return ['task_id' => 'parent-task', 'workflow_task_attempt' => 1, 'lease_owner' => 'worker', 'renewed' => true]; + } + if (str_ends_with($uri, '/register')) { + return ['registered' => true, 'worker_id' => 'worker', 'registration_token' => str_repeat('a', 32), 'heartbeat_interval_seconds' => 30, + 'server_capabilities' => ['worker_deregistration_fencing' => [ + 'schema' => 'durable-workflow.v2.worker-deregistration.v1', 'supported' => true, + 'endpoint' => '/worker/registrations/{workerId}/deregister', 'receipt_retention_seconds' => 600, + ]]]; + } + if (str_ends_with($uri, '/deregister')) { + return ['worker_id' => 'worker', 'registration_token' => str_repeat('a', 32), + 'outcome' => 'deregistered', 'recovered_workflow_task_count' => 0]; + } + + return ['task' => null, 'poll_status' => 'empty']; + }); + $receipt = tempnam(sys_get_temp_dir(), 'child-all-test-'); + self::assertIsString($receipt); + try { + $worker = null; + $transport = new DeferredChildAllTurnTransport($wire, $receipt, static function () use (&$worker): void { + /** @var Worker|null $worker */ + $worker?->requestShutdown(); + }); + $worker = new Worker(new Client('http://adapter.invalid', workerToken: 'unretained-worker-auth', transport: $transport), 'parents', workerId: 'worker'); + $worker->registerWorkflow($scenario->parentType(), $scenario->parent()); + $worker->run(0); + $bytes = (string) file_get_contents($receipt); + $value = json_decode($bytes, true, flags: JSON_THROW_ON_ERROR); + self::assertSame('parents', $value['task']['task_queue']); + self::assertSame('parent-task', $value['task']['task_id']); + self::assertSame('child-0', $value['body']['commands'][0]['queue']); + self::assertStringNotContainsString('unretained', $bytes); + self::assertCount(1, array_filter($requests, static fn (string $uri): bool => str_ends_with($uri, '/workflow-tasks/poll'))); + self::assertCount(1, array_filter($requests, static fn (string $uri): bool => str_ends_with($uri, '/deregister'))); + } finally { + unlink($receipt); + } + } + + #[DataProvider('unrecordedTurns')] + public function testUnrecordedOrUnboundCompletionCannotStopTheWorkerOrCreateAReceipt(bool $recorded, bool $poll): void + { + $receipt = tempnam(sys_get_temp_dir(), 'child-all-test-'); + $stopped = false; + $transport = new DeferredChildAllTurnTransport(new ChildAllWire(static fn (string $uri): array => str_ends_with($uri, '/poll') + ? ['task' => ['task_id' => 'task']] : ['recorded' => $recorded]), $receipt, + static function () use (&$stopped): void { $stopped = true; }); + try { + if ($poll) { + $transport->sendBounded('POST', 'http://adapter.invalid/workflow-tasks/poll', [], [], 1); + } + try { + $transport->sendBounded('POST', 'http://adapter.invalid/workflow-tasks/task/complete', [], ['commands' => []], 1); + self::fail('Unrecorded or unbound completion must refuse.'); + } catch (RuntimeException) { + self::assertFalse($stopped); + self::assertSame('', file_get_contents($receipt)); + } + } finally { + unlink($receipt); + } + } + + /** @return iterable */ + public static function unrecordedTurns(): iterable + { + yield 'not recorded' => [false, true]; + yield 'no matching poll' => [true, false]; + } + + public function testOnlyExplicitEmptyPollsProveNoParentTask(): void + { + foreach (['empty', 'timeout'] as $status) { + Scenario::assertEmptyPoll(['task' => null, 'poll_status' => $status]); + Scenario::assertEmptyPoll(['task' => null, 'poll_status' => $status, 'reason' => null]); + } + $this->addToAssertionCount(4); + } + + /** @param array $response */ + #[DataProvider('uncertainPolls')] + public function testUnavailableMalformedOrLeasedPollsCannotProveNoParentTask(array $response): void + { + $this->expectException(RuntimeException::class); + Scenario::assertEmptyPoll($response); + } + + /** @return iterable}> */ + public static function uncertainPolls(): iterable + { + foreach (['draining', 'stopped', 'stale_worker_registration', 'backend_lock_pressure', 'leased', 'future_status'] as $status) { + yield $status => [['task' => null, 'poll_status' => $status]]; + } + yield 'missing status' => [['task' => null]]; + yield 'missing task' => [['poll_status' => 'empty']]; + yield 'leased task' => [['task' => ['task_id' => 'task'], 'poll_status' => 'empty']]; + yield 'mixed terminal reason' => [['task' => null, 'poll_status' => 'empty', 'reason' => 'worker_draining']]; + yield 'unknown reason' => [['task' => null, 'poll_status' => 'timeout', 'reason' => 'future_reason']]; + } + + public function testParentReceiptBindsTheIntendedWorkflowRunAndQueue(): void + { + $task = ['workflow_id' => 'parent', 'run_id' => 'initial-run', 'task_queue' => 'parents']; + Scenario::assertParentTask($task, 'parent', 'initial-run', 'parents'); + $this->addToAssertionCount(1); + foreach (array_keys($task) as $key) { + try { + Scenario::assertParentTask(array_replace($task, [$key => 'other']), 'parent', 'initial-run', 'parents'); + self::fail('Another parent identity must refuse.'); + } catch (RuntimeException $error) { + self::assertSame('Managed parent received another workflow, run or queue.', $error->getMessage()); + } + } + } + + private static function scenario(string $case, int $size): Scenario + { + return new Scenario($case, array_map(static fn (int $index): string => 'child-'.$index, range(0, $size - 1))); + } + + /** + * @param list> $commands + * @return list> + */ + private static function history(array $commands): array + { + $history = []; + foreach ($commands as $index => $command) { + $metadata = array_filter($command, static fn (string $key): bool => str_starts_with($key, 'parallel_group_'), ARRAY_FILTER_USE_KEY); + $payload = ['sequence' => $index + 1, 'child_workflow_type' => $command['workflow_type'], + 'child_call_id' => 'call-'.$index, 'workflow_link_id' => 'link-'.$index, + 'child_workflow_instance_id' => 'child-'.$index, 'child_workflow_run_id' => 'run-'.$index, ...$metadata]; + $history[] = ['event_type' => 'ChildWorkflowScheduled', 'payload' => $payload]; + $history[] = ['event_type' => 'ChildRunStarted', 'payload' => [...$payload, 'child_run_number' => 1]]; + } + + return $history; + } + + /** + * @param array $payload + * @param array $result + * @return array + */ + private static function terminal(array $payload, array $result): array + { + return ['event_type' => 'ChildRunCompleted', 'payload' => [...$payload, + 'child_run_number' => $payload['child_run_number'] ?? 1, 'result' => $result]]; + } + + /** + * @param array $payload + * @return array + */ + private static function replacement(string $case, array $payload): array + { + return ['event_type' => 'ChildRunStarted', 'payload' => [...$payload, 'child_workflow_run_id' => 'replacement', + 'workflow_link_id' => 'replacement-link', 'child_run_number' => 2, + ...($case === 'retry' ? ['retry_of_child_workflow_run_id' => $payload['child_workflow_run_id'], 'retry_attempt' => 2] : [])]]; + } + + /** + * @return array{list>, array, array{run_id: string, workflow_link_id: string, terminal: ?string}, array{workflow_id: string, run_id: string}, list>} + */ + private static function replacementProof(string $case): array + { + $original = ['child_call_id' => 'call-0', 'child_workflow_instance_id' => 'child-0', 'child_workflow_run_id' => 'run-0', 'workflow_link_id' => 'link-0']; + $replacement = ['run_id' => 'replacement', 'workflow_link_id' => 'replacement-link', 'terminal' => 'ChildRunCompleted']; + $parent = ['workflow_id' => 'parent', 'run_id' => 'parent-run']; + $history = [['event_type' => 'WorkflowStarted', 'payload' => [ + 'workflow_instance_id' => 'child-0', 'workflow_run_id' => 'replacement', + 'child_call_id' => 'call-0', 'parent_workflow_instance_id' => 'parent', 'parent_workflow_run_id' => 'parent-run', + 'parent_sequence' => 1, 'workflow_type' => 'parity.v1.child_all_child', + ...($case === 'retry' ? ['retry_of_child_workflow_run_id' => 'run-0', 'retry_attempt' => 2, 'workflow_link_id' => 'replacement-link'] + : ['continued_from_run_id' => 'run-0', 'workflow_link_id' => 'continuation-link']), + ]]]; + $predecessor = $case === 'retry' ? [['event_type' => 'WorkflowFailed', 'payload' => []]] + : [['event_type' => 'WorkflowContinuedAsNew', 'payload' => [ + 'continued_to_run_id' => 'replacement', 'continued_to_run_number' => 2, 'sequence' => 1, 'workflow_link_id' => 'continuation-link', + ]]]; + + return [$history, $original, $replacement, $parent, $predecessor]; + } +} + +final class ChildAllWire implements BoundedTransport +{ + public function __construct(private readonly Closure $handler) + { + } + + public function supportsBoundedRequests(): bool + { + return true; + } + + public function send(string $method, string $uri, array $headers, ?array $body = null): ?array + { + return ($this->handler)($uri, $body); + } + + public function sendBounded(string $method, string $uri, array $headers, ?array $body, int $timeoutSeconds): ?array + { + return $this->send($method, $uri, $headers, $body); + } +}