From 9544950edcacec459a950a8e27c074bd9fb04684 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 17:35:46 +0000 Subject: [PATCH 1/6] Prepare managed-worker deferred child barrier adapter [skip ci] --- scripts/Support/DeferredChildAllScenario.php | 221 +++++++++++++ .../Support/DeferredChildAllTurnTransport.php | 60 ++++ scripts/deferred-child-all.md | 66 ++++ scripts/fixtures/parallel-child-all.json | 51 +++ scripts/qualify-deferred-child-all.php | 262 +++++++++++++++ tests/DeferredChildAllAdapterTest.php | 309 ++++++++++++++++++ 6 files changed, 969 insertions(+) create mode 100644 scripts/Support/DeferredChildAllScenario.php create mode 100644 scripts/Support/DeferredChildAllTurnTransport.php create mode 100644 scripts/deferred-child-all.md create mode 100644 scripts/fixtures/parallel-child-all.json create mode 100644 scripts/qualify-deferred-child-all.php create mode 100644 tests/DeferredChildAllAdapterTest.php diff --git a/scripts/Support/DeferredChildAllScenario.php b/scripts/Support/DeferredChildAllScenario.php new file mode 100644 index 00000000..333ebb7f --- /dev/null +++ b/scripts/Support/DeferredChildAllScenario.php @@ -0,0 +1,221 @@ + */ + 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) { + 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.'); + $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 + */ + public function assertStable(array $history, array $commands, array $initial): void + { + self::check($this->identities($history, $commands) === $initial, 'Replacement changed a recorded initial identity.'); + $runs = array_fill(0, $this->size, []); + foreach ($history as $event) { + if (!str_starts_with($event['event_type'], 'ChildRun')) { + continue; + } + $payload = $event['payload']; + $original = $initial[($payload['sequence'] ?? 0) - 1] ?? null; + self::check(is_array($original), 'Child event has no original member.'); + foreach (['child_call_id', 'workflow_link_id', 'child_workflow_instance_id', 'parallel_group_id', 'parallel_group_path'] as $key) { + self::check(isset($payload[$key]) && $payload[$key] === $original[$key], 'Child event changed its original call/link/path.'); + } + $canReplaceRun = in_array($this->case, ['retry', 'continue'], true) && $payload['sequence'] === 1; + self::check(is_string($payload['child_workflow_run_id'] ?? null) && $payload['child_workflow_run_id'] !== '', 'Child event has no run identity.'); + $index = $payload['sequence'] - 1; + $run = $payload['child_workflow_run_id']; + if ($event['event_type'] === 'ChildRunStarted') { + self::check(!in_array($run, $runs[$index], true), 'Duplicate child run start.'); + self::check($runs[$index] === [] ? $run === $original['child_workflow_run_id'] : $canReplaceRun && count($runs[$index]) === 1, + 'Unexpected replacement child run.'); + $runs[$index][] = $run; + } else { + self::check($runs[$index] !== [] && $run === $runs[$index][array_key_last($runs[$index])], 'Child event does not belong to its current recorded run.'); + } + } + foreach ($runs as $chain) { + self::check($chain !== [], 'Atomic child start has no initial run event.'); + } + } + + /** @phpstan-assert true $condition */ + public static function check(bool $condition, string $message): void + { + if (!$condition) { + throw new RuntimeException($message); + } + } +} + +final class ChildAllProbeFailure 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..b571ebf6 --- /dev/null +++ b/scripts/deferred-child-all.md @@ -0,0 +1,66 @@ +# 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 identity while a distinct second run finishes. + +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. + +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..9ecdf46e --- /dev/null +++ b/scripts/qualify-deferred-child-all.php @@ -0,0 +1,262 @@ +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 RuntimeException('Managed worker exceeded its turn deadline.'); + } finally { + if (is_resource($process)) { + 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); + } + if (is_string($receipt)) { + @unlink($receipt); + } + if (is_string($log)) { + @unlink($log); + } + } +} + +/** 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::check(array_key_exists('task', $response) && $response['task'] === null, 'Unexpected extra or premature parent task.'); + } 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 + * @return array + */ +function childAllMember(Scenario $scenario, Client $worker, Client $control, string $namespace, int $index, array $identity, string $parent, string $parentRun): 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, $parentRun]]); + + return $receipt; +} + +/** @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); + $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 = []; + foreach ($scenario->order() as $offset => $index) { + childAllBudget(); + $identity = $identities[$index]; + if ($index === 0 && in_array($case, ['cancelled', 'terminated'], true)) { + if ($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']); + } + } else { + $receipt = childAllMember($scenario, $worker, $control, $namespace, $index, $identity, $parent->workflowId, $run); + Scenario::check($receipt['task']['run_id'] === $identity['child_workflow_run_id'], 'Initial child task used another run.'); + $memberRuns[$index] = [$receipt['task']['run_id']]; + if ($index === 0 && in_array($case, ['retry', 'continue'], true)) { + childAllNoParentTask($worker, $scenario, $parentQueue); + $replacement = childAllMember($scenario, $worker, $control, $namespace, $index, $identity, $parent->workflowId, $run); + Scenario::check($replacement['task']['run_id'] !== $identity['child_workflow_run_id'], 'Child chain did not create a distinct second run.'); + $memberRuns[$index][] = $replacement['task']['run_id']; + } + } + $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); + $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); + 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, + '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..90e6f8da --- /dev/null +++ b/tests/DeferredChildAllAdapterTest.php @@ -0,0 +1,309 @@ +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) { + $terminal = self::terminal($history[$index * 2]['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'])); + } + } + } + + /** @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]; + } + } + + public function testReplacementRunKeepsInitialIdentityAndOriginalMemberPath(): void + { + $scenario = self::scenario('continue', 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 = $history[1]; + $next['payload']['child_workflow_run_id'] = 'replacement'; + $history[] = $next; + $history[] = self::terminal($next['payload'], $codec->envelope('done')); + $scenario->assertStable($history, $commands, $identities); + self::assertSame('run-0', $scenario->identities($history, $commands)[0]['child_workflow_run_id']); + } + + 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]; + } + + 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]; + } + + 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, 'result' => $result]]; + } +} + +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); + } +} From f5b4d607f97ac188c84a96ef42bf3bfa262e1f1a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 17:40:01 +0000 Subject: [PATCH 2/6] Bind parent observations to explicit empty polls and original run [skip ci] --- scripts/Support/DeferredChildAllScenario.php | 15 +++++++ scripts/deferred-child-all.md | 5 ++- scripts/qualify-deferred-child-all.php | 4 +- tests/DeferredChildAllAdapterTest.php | 45 ++++++++++++++++++++ 4 files changed, 67 insertions(+), 2 deletions(-) diff --git a/scripts/Support/DeferredChildAllScenario.php b/scripts/Support/DeferredChildAllScenario.php index 333ebb7f..58153671 100644 --- a/scripts/Support/DeferredChildAllScenario.php +++ b/scripts/Support/DeferredChildAllScenario.php @@ -207,6 +207,21 @@ public function assertStable(array $history, array $commands, array $initial): v } } + /** @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 { diff --git a/scripts/deferred-child-all.md b/scripts/deferred-child-all.md index b571ebf6..75c1dba9 100644 --- a/scripts/deferred-child-all.md +++ b/scripts/deferred-child-all.md @@ -26,7 +26,10 @@ 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. +or further ready task. 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: diff --git a/scripts/qualify-deferred-child-all.php b/scripts/qualify-deferred-child-all.php index 9ecdf46e..8d117bc1 100644 --- a/scripts/qualify-deferred-child-all.php +++ b/scripts/qualify-deferred-child-all.php @@ -129,7 +129,7 @@ function childAllNoParentTask(Client $worker, Scenario $scenario, string $queue) $registration = $worker->registerWorker($id, $queue, [$scenario->parentType()], []); try { $response = $worker->pollWorkflowTaskResponse($id, $queue, 0); - Scenario::check(array_key_exists('task', $response) && $response['task'] === null, 'Unexpected extra or premature parent task.'); + Scenario::assertEmptyPoll($response); } finally { $worker->deregisterWorkerRegistration($id, $registration['registration_token'] ?? null); } @@ -190,6 +190,7 @@ function childAllCase(string $case): array $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); @@ -231,6 +232,7 @@ function childAllCase(string $case): array 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']); diff --git a/tests/DeferredChildAllAdapterTest.php b/tests/DeferredChildAllAdapterTest.php index 90e6f8da..2d624712 100644 --- a/tests/DeferredChildAllAdapterTest.php +++ b/tests/DeferredChildAllAdapterTest.php @@ -251,6 +251,51 @@ public static function unrecordedTurns(): iterable 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))); From c688a73adffeb9e3cb2f2c13020541c4bd67b075 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 17:57:40 +0000 Subject: [PATCH 3/6] Close deferred-child history ledger and bind replacement links [skip ci] --- scripts/Support/DeferredChildAllScenario.php | 128 ++++++++-- scripts/deferred-child-all.md | 14 +- scripts/qualify-deferred-child-all.php | 9 +- tests/DeferredChildAllAdapterTest.php | 251 ++++++++++++++++++- 4 files changed, 375 insertions(+), 27 deletions(-) diff --git a/scripts/Support/DeferredChildAllScenario.php b/scripts/Support/DeferredChildAllScenario.php index 58153671..9d6eb7bd 100644 --- a/scripts/Support/DeferredChildAllScenario.php +++ b/scripts/Support/DeferredChildAllScenario.php @@ -152,6 +152,7 @@ public function identities(array $history, array $commands): array && ($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.'); @@ -174,36 +175,131 @@ public function identities(array $history, array $commands): array * @param list> $history * @param list> $commands * @param list> $initial + * @return list> */ - public function assertStable(array $history, array $commands, array $initial): void + 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.'); $runs = array_fill(0, $this->size, []); + $scheduled = array_fill(0, $this->size, false); + $usedRuns = array_column($initial, 'child_workflow_run_id'); + $usedLinks = array_column($initial, 'workflow_link_id'); + $parentClosed = false; foreach ($history as $event) { - if (!str_starts_with($event['event_type'], 'ChildRun')) { + $type = $event['event_type'] ?? null; + self::check(is_string($type), 'History event type is not a string.'); + if ($type === 'WorkflowCompleted') { + self::check(!$parentClosed, 'Duplicate parent completion event.'); + $parentClosed = true; + } + if (!str_starts_with($type, 'Child')) { continue; } - $payload = $event['payload']; - $original = $initial[($payload['sequence'] ?? 0) - 1] ?? null; - self::check(is_array($original), 'Child event has no original member.'); - foreach (['child_call_id', 'workflow_link_id', 'child_workflow_instance_id', 'parallel_group_id', 'parallel_group_path'] as $key) { + self::check(!$parentClosed, 'Child event follows parent completion.'); + self::check(in_array($type, ['ChildWorkflowScheduled', 'ChildRunStarted', 'ChildRunCompleted', + 'ChildRunFailed', 'ChildRunCancelled', 'ChildRunTerminated'], true), 'Unqualified child event type.'); + $payload = $event['payload'] ?? null; + self::check(is_array($payload), 'Child event has no payload.'); + $sequence = $payload['sequence'] ?? null; + self::check(is_int($sequence) && $sequence >= 1 && $sequence <= $this->size, 'Child sequence is not an exact member integer.'); + $index = $sequence - 1; + $original = $initial[$index]; + self::check(($payload['child_workflow_type'] ?? null) === $this->childType(), 'Child workflow type changed.'); + foreach (['child_call_id', 'child_workflow_instance_id', 'parallel_group_id', 'parallel_group_path'] as $key) { self::check(isset($payload[$key]) && $payload[$key] === $original[$key], 'Child event changed its original call/link/path.'); } - $canReplaceRun = in_array($this->case, ['retry', 'continue'], true) && $payload['sequence'] === 1; self::check(is_string($payload['child_workflow_run_id'] ?? null) && $payload['child_workflow_run_id'] !== '', 'Child event has no run identity.'); - $index = $payload['sequence'] - 1; + self::check(is_string($payload['workflow_link_id'] ?? null) && $payload['workflow_link_id'] !== '', 'Child event has no link identity.'); $run = $payload['child_workflow_run_id']; - if ($event['event_type'] === 'ChildRunStarted') { - self::check(!in_array($run, $runs[$index], true), 'Duplicate child run start.'); - self::check($runs[$index] === [] ? $run === $original['child_workflow_run_id'] : $canReplaceRun && count($runs[$index]) === 1, - 'Unexpected replacement child run.'); - $runs[$index][] = $run; - } else { - self::check($runs[$index] !== [] && $run === $runs[$index][array_key_last($runs[$index])], 'Child event does not belong to its current recorded run.'); + $link = $payload['workflow_link_id']; + if ($type === 'ChildWorkflowScheduled') { + self::check(!$scheduled[$index] && $runs[$index] === [], 'Child schedule is repeated or follows its start.'); + $scheduled[$index] = true; + continue; } + self::check($scheduled[$index], 'Child run event precedes its schedule.'); + $current = $runs[$index] === [] ? null : $runs[$index][array_key_last($runs[$index])]; + if ($type === 'ChildRunStarted') { + if ($current === null) { + self::check($run === $original['child_workflow_run_id'] && $link === $original['workflow_link_id'] + && ($payload['child_run_number'] ?? null) === 1, 'Initial child run/link/number changed.'); + self::check(($payload['retry_of_child_workflow_run_id'] ?? null) === null + && ($payload['retry_attempt'] ?? null) === null, 'Initial child run claims a retry predecessor.'); + } else { + self::check($index === 0 && in_array($this->case, ['retry', 'continue'], true) + && count($runs[$index]) === 1 && $current['terminal'] === null, 'Child run cannot replace a closed or unqualified member.'); + self::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.'); + // Native parent history records no intermediate terminal. + // Retry has explicit provenance; continuation has a new + // parent link but its predecessor is in child history. + self::check($this->case === 'retry' + ? ($payload['retry_of_child_workflow_run_id'] ?? null) === $current['run_id'] && ($payload['retry_attempt'] ?? null) === 2 + : ($payload['retry_of_child_workflow_run_id'] ?? null) === null && ($payload['retry_attempt'] ?? null) === null, + 'Replacement child transition does not match its case.'); + $usedRuns[] = $run; + $usedLinks[] = $link; + } + $runs[$index][] = ['run_id' => $run, 'workflow_link_id' => $link, 'terminal' => null]; + continue; + } + self::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.'); + self::check(($payload['child_run_number'] ?? null) === count($runs[$index]), 'Child terminal run number changed.'); + self::check($index !== 0 || !in_array($this->case, ['retry', 'continue'], true) + || count($runs[$index]) === 2, 'Replacement member resolved before its second run.'); + $expected = $index === 0 && $this->failsEarly() ? $this->expected()['failure_type'] : 'ChildRunCompleted'; + self::check($type === $expected, 'Child terminal does not match its selected case.'); + $runs[$index][array_key_last($runs[$index])]['terminal'] = $type; } - foreach ($runs as $chain) { + foreach ($runs as $index => $chain) { self::check($chain !== [], 'Atomic child start has no initial run event.'); + if ($parentFinished && (!$this->failsEarly() || $index === 0)) { + self::check($chain[array_key_last($chain)]['terminal'] !== null, 'Finished parent has an unresolved required member.'); + } + } + self::check(!$parentFinished || $parentClosed, 'Finished parent has no canonical completion.'); + + return $runs; + } + + /** + * 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.'); + $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.'); } } diff --git a/scripts/deferred-child-all.md b/scripts/deferred-child-all.md index 75c1dba9..1bc11ad6 100644 --- a/scripts/deferred-child-all.md +++ b/scripts/deferred-child-all.md @@ -15,7 +15,12 @@ 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 identity while a distinct second run finishes. +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 @@ -26,7 +31,12 @@ 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. Empty-queue observations require an explicit empty or +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. diff --git a/scripts/qualify-deferred-child-all.php b/scripts/qualify-deferred-child-all.php index 8d117bc1..b13304c7 100644 --- a/scripts/qualify-deferred-child-all.php +++ b/scripts/qualify-deferred-child-all.php @@ -213,10 +213,16 @@ function childAllCase(string $case): array Scenario::check($receipt['task']['run_id'] === $identity['child_workflow_run_id'], 'Initial child task used another run.'); $memberRuns[$index] = [$receipt['task']['run_id']]; if ($index === 0 && in_array($case, ['retry', 'continue'], true)) { + Scenario::check(count($receipt['body']['commands']) === 1 + && $receipt['body']['commands'][0]['type'] === ($case === 'retry' ? 'fail_workflow' : 'continue_as_new'), 'Initial child turn did not request its selected replacement.'); childAllNoParentTask($worker, $scenario, $parentQueue); $replacement = childAllMember($scenario, $worker, $control, $namespace, $index, $identity, $parent->workflowId, $run); Scenario::check($replacement['task']['run_id'] !== $identity['child_workflow_run_id'], 'Child chain did not create a distinct second run.'); $memberRuns[$index][] = $replacement['task']['run_id']; + $chains = $scenario->assertStable(childAllHistory($control, $parent->workflowId, $run), $commands, $identities); + $scenario->assertReplacementHistory(childAllHistory($control, $identity['child_workflow_instance_id'], $replacement['task']['run_id']), + $identity, $chains[0][1], ['workflow_id' => $parent->workflowId, 'run_id' => $run], + childAllHistory($control, $identity['child_workflow_instance_id'], $identity['child_workflow_run_id'])); } } $history = childAllHistory($control, $parent->workflowId, $run); @@ -239,7 +245,7 @@ function childAllCase(string $case): array 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); + $scenario->assertStable($closed, $commands, $identities, parentFinished: true); childAllNoParentTask($worker, $scenario, $parentQueue); } Scenario::check($closed !== null, 'Parent never reached its expected terminal result.'); @@ -250,6 +256,7 @@ function childAllCase(string $case): array 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]; } diff --git a/tests/DeferredChildAllAdapterTest.php b/tests/DeferredChildAllAdapterTest.php index 2d624712..db80ed0c 100644 --- a/tests/DeferredChildAllAdapterTest.php +++ b/tests/DeferredChildAllAdapterTest.php @@ -47,7 +47,14 @@ public function testManagedDefinitionsAuthorFlatChildrenAndColdReplayTheExactOut $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) { - $terminal = self::terminal($history[$index * 2]['payload'], $codec->envelope(['echo' => [ + $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) { @@ -66,6 +73,8 @@ public function testManagedDefinitionsAuthorFlatChildrenAndColdReplayTheExactOut 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 */ @@ -144,19 +153,210 @@ public static function changedHistories(): iterable } } - public function testReplacementRunKeepsInitialIdentityAndOriginalMemberPath(): void + #[DataProvider('replacementCases')] + public function testReplacementRunKeepsInitialIdentityAndOriginalMemberPath(string $case): void { - $scenario = self::scenario('continue', 2); + $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 = $history[1]; - $next['payload']['child_workflow_run_id'] = 'replacement'; + $next = self::replacement($case, $history[1]['payload']); $history[] = $next; $history[] = self::terminal($next['payload'], $codec->envelope('done')); - $scenario->assertStable($history, $commands, $identities); + $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']); + } + + #[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 @@ -314,7 +514,7 @@ private static function history(array $commands): array '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]; + $history[] = ['event_type' => 'ChildRunStarted', 'payload' => [...$payload, 'child_run_number' => 1]]; } return $history; @@ -327,7 +527,42 @@ private static function history(array $commands): array */ private static function terminal(array $payload, array $result): array { - return ['event_type' => 'ChildRunCompleted', 'payload' => [...$payload, 'result' => $result]]; + 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']; + $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]; } } From 88e134302db0b6bd2c8b77821072800efbb8f316 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 18:03:42 +0000 Subject: [PATCH 4/6] Keep continuation and parent-child link roles distinct [skip ci] --- scripts/Support/DeferredChildAllScenario.php | 3 +++ tests/DeferredChildAllAdapterTest.php | 26 +++++++++++++++++++- 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/scripts/Support/DeferredChildAllScenario.php b/scripts/Support/DeferredChildAllScenario.php index 9d6eb7bd..bf3d1d93 100644 --- a/scripts/Support/DeferredChildAllScenario.php +++ b/scripts/Support/DeferredChildAllScenario.php @@ -294,6 +294,9 @@ public function assertReplacementHistory(array $history, array $original, array 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']; diff --git a/tests/DeferredChildAllAdapterTest.php b/tests/DeferredChildAllAdapterTest.php index db80ed0c..0902c659 100644 --- a/tests/DeferredChildAllAdapterTest.php +++ b/tests/DeferredChildAllAdapterTest.php @@ -237,6 +237,30 @@ public function testChildHistoryProvesReplacementWithoutEquatingTheContinuationA $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')] @@ -547,7 +571,7 @@ private static function replacement(string $case, array $payload): array */ 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']; + $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' => [ From d7a70337f704f6985cb5dc4c0bc2d9879d9a34da Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 18:40:13 +0000 Subject: [PATCH 5/6] Keep child qualification state and worker steps focused --- scripts/Support/DeferredChildAllScenario.php | 199 +++++++++++++------ scripts/qualify-deferred-child-all.php | 106 ++++++---- 2 files changed, 206 insertions(+), 99 deletions(-) diff --git a/scripts/Support/DeferredChildAllScenario.php b/scripts/Support/DeferredChildAllScenario.php index bf3d1d93..f6d38264 100644 --- a/scripts/Support/DeferredChildAllScenario.php +++ b/scripts/Support/DeferredChildAllScenario.php @@ -74,20 +74,21 @@ public function parent(): Closure public function child(?string $initialRunId = null): Closure { return function (WorkflowContext $context, array $value, int $stage = 0) use ($initialRunId): array { - if (($value['index'] ?? null) === 0) { - 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]); + 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]; }; @@ -196,62 +197,138 @@ public function assertStable(array $history, array $commands, array $initial, bo continue; } self::check(!$parentClosed, 'Child event follows parent completion.'); - self::check(in_array($type, ['ChildWorkflowScheduled', 'ChildRunStarted', 'ChildRunCompleted', - 'ChildRunFailed', 'ChildRunCancelled', 'ChildRunTerminated'], true), 'Unqualified child event type.'); - $payload = $event['payload'] ?? null; - self::check(is_array($payload), 'Child event has no payload.'); - $sequence = $payload['sequence'] ?? null; - self::check(is_int($sequence) && $sequence >= 1 && $sequence <= $this->size, 'Child sequence is not an exact member integer.'); - $index = $sequence - 1; - $original = $initial[$index]; - self::check(($payload['child_workflow_type'] ?? null) === $this->childType(), 'Child workflow type changed.'); - foreach (['child_call_id', 'child_workflow_instance_id', 'parallel_group_id', 'parallel_group_path'] as $key) { - self::check(isset($payload[$key]) && $payload[$key] === $original[$key], 'Child event changed its original call/link/path.'); - } - self::check(is_string($payload['child_workflow_run_id'] ?? null) && $payload['child_workflow_run_id'] !== '', 'Child event has no run identity.'); - self::check(is_string($payload['workflow_link_id'] ?? null) && $payload['workflow_link_id'] !== '', 'Child event has no link identity.'); - $run = $payload['child_workflow_run_id']; - $link = $payload['workflow_link_id']; + [$index, $payload] = $this->childEvent($type, $event, $initial); if ($type === 'ChildWorkflowScheduled') { self::check(!$scheduled[$index] && $runs[$index] === [], 'Child schedule is repeated or follows its start.'); $scheduled[$index] = true; continue; } self::check($scheduled[$index], 'Child run event precedes its schedule.'); - $current = $runs[$index] === [] ? null : $runs[$index][array_key_last($runs[$index])]; if ($type === 'ChildRunStarted') { - if ($current === null) { - self::check($run === $original['child_workflow_run_id'] && $link === $original['workflow_link_id'] - && ($payload['child_run_number'] ?? null) === 1, 'Initial child run/link/number changed.'); - self::check(($payload['retry_of_child_workflow_run_id'] ?? null) === null - && ($payload['retry_attempt'] ?? null) === null, 'Initial child run claims a retry predecessor.'); - } else { - self::check($index === 0 && in_array($this->case, ['retry', 'continue'], true) - && count($runs[$index]) === 1 && $current['terminal'] === null, 'Child run cannot replace a closed or unqualified member.'); - self::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.'); - // Native parent history records no intermediate terminal. - // Retry has explicit provenance; continuation has a new - // parent link but its predecessor is in child history. - self::check($this->case === 'retry' - ? ($payload['retry_of_child_workflow_run_id'] ?? null) === $current['run_id'] && ($payload['retry_attempt'] ?? null) === 2 - : ($payload['retry_of_child_workflow_run_id'] ?? null) === null && ($payload['retry_attempt'] ?? null) === null, - 'Replacement child transition does not match its case.'); - $usedRuns[] = $run; - $usedLinks[] = $link; + $next = $this->startedRun($payload, $initial[$index], $runs[$index], $usedRuns, $usedLinks); + if ($runs[$index] !== []) { + $usedRuns[] = $next['run_id']; + $usedLinks[] = $next['workflow_link_id']; } - $runs[$index][] = ['run_id' => $run, 'workflow_link_id' => $link, 'terminal' => null]; + $runs[$index][] = $next; continue; } - self::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.'); - self::check(($payload['child_run_number'] ?? null) === count($runs[$index]), 'Child terminal run number changed.'); - self::check($index !== 0 || !in_array($this->case, ['retry', 'continue'], true) - || count($runs[$index]) === 2, 'Replacement member resolved before its second run.'); - $expected = $index === 0 && $this->failsEarly() ? $this->expected()['failure_type'] : 'ChildRunCompleted'; - self::check($type === $expected, 'Child terminal does not match its selected case.'); - $runs[$index][array_key_last($runs[$index])]['terminal'] = $type; + $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 + { + self::check(in_array($type, ['ChildWorkflowScheduled', 'ChildRunStarted', 'ChildRunCompleted', + 'ChildRunFailed', 'ChildRunCancelled', 'ChildRunTerminated'], true), 'Unqualified child event type.'); + $payload = $event['payload'] ?? null; + self::check(is_array($payload), 'Child event has no payload.'); + $sequence = $payload['sequence'] ?? null; + self::check(is_int($sequence) && $sequence >= 1 && $sequence <= $this->size, 'Child sequence is not an exact member integer.'); + $index = $sequence - 1; + self::check(($payload['child_workflow_type'] ?? null) === $this->childType(), 'Child workflow type changed.'); + foreach (['child_call_id', 'child_workflow_instance_id', 'parallel_group_id', 'parallel_group_path'] as $key) { + self::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 + { + self::check(is_string($payload['child_workflow_run_id'] ?? null) && $payload['child_workflow_run_id'] !== '', 'Child event has no run identity.'); + self::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 === []) { + self::check($run === $original['child_workflow_run_id'] && $link === $original['workflow_link_id'] + && ($payload['child_run_number'] ?? null) === 1, 'Initial child run/link/number changed.'); + self::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)]; + self::check($this->replacementMember($payload['sequence']) && count($chain) === 1 + && $current['terminal'] === null, 'Child run cannot replace a closed or unqualified member.'); + self::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->case, ['retry', 'continue'], true); + } + + /** @param array $payload */ + private function replacementProvenance(array $payload, string $previousRun): void + { + if ($this->case === 'retry') { + self::check(($payload['retry_of_child_workflow_run_id'] ?? null) === $previousRun && ($payload['retry_attempt'] ?? null) === 2, + 'Replacement child transition does not match its case.'); + return; + } + self::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)]; + self::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.'); + self::check(($payload['child_run_number'] ?? null) === count($chain), 'Child terminal run number changed.'); + self::check(!$this->replacementMember($payload['sequence']) || count($chain) === 2, 'Replacement member resolved before its second run.'); + $expected = $payload['sequence'] === 1 && $this->failsEarly() ? $this->expected()['failure_type'] : 'ChildRunCompleted'; + self::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) { self::check($chain !== [], 'Atomic child start has no initial run event.'); if ($parentFinished && (!$this->failsEarly() || $index === 0)) { @@ -259,8 +336,6 @@ public function assertStable(array $history, array $commands, array $initial, bo } } self::check(!$parentFinished || $parentClosed, 'Finished parent has no canonical completion.'); - - return $runs; } /** @@ -325,7 +400,7 @@ public static function assertParentTask(array $task, string $workflow, string $r public static function check(bool $condition, string $message): void { if (!$condition) { - throw new RuntimeException($message); + throw new ChildAllQualificationFailure($message); } } } @@ -333,3 +408,7 @@ public static function check(bool $condition, string $message): void final class ChildAllProbeFailure extends RuntimeException { } + +final class ChildAllQualificationFailure extends RuntimeException +{ +} diff --git a/scripts/qualify-deferred-child-all.php b/scripts/qualify-deferred-child-all.php index b13304c7..a7a09c8d 100644 --- a/scripts/qualify-deferred-child-all.php +++ b/scripts/qualify-deferred-child-all.php @@ -8,6 +8,7 @@ use DurableWorkflow\Client; use DurableWorkflow\Exception\ServerException; use DurableWorkflow\Qualification\DeferredChildAllScenario as Scenario; +use DurableWorkflow\Qualification\ChildAllQualificationFailure; use DurableWorkflow\Qualification\DeferredChildAllTurnTransport; use DurableWorkflow\Transport\Psr18Transport; use DurableWorkflow\Worker; @@ -97,21 +98,9 @@ function childAllTurn(Scenario $scenario, string $namespace, string $queue, stri } usleep(50_000); } while (microtime(true) < $turnDeadline); - throw new RuntimeException('Managed worker exceeded its turn deadline.'); + throw new ChildAllQualificationFailure('Managed worker exceeded its turn deadline.'); } finally { - if (is_resource($process)) { - 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); - } + childAllStopProcess($process); if (is_string($receipt)) { @unlink($receipt); } @@ -121,6 +110,25 @@ function childAllTurn(Scenario $scenario, string $namespace, string $queue, stri } } +/** @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 { @@ -157,9 +165,10 @@ function childAllDuplicate(Client $worker, Client $control, array $receipt, stri /** * @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, string $parent, string $parentRun): 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']; @@ -167,11 +176,48 @@ function childAllMember(Scenario $scenario, Client $worker, Client $control, str && ($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, $parentRun]]); + 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 { @@ -199,31 +245,13 @@ function childAllCase(string $case): array 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(); - $identity = $identities[$index]; - if ($index === 0 && in_array($case, ['cancelled', 'terminated'], true)) { - if ($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']); - } - } else { - $receipt = childAllMember($scenario, $worker, $control, $namespace, $index, $identity, $parent->workflowId, $run); - Scenario::check($receipt['task']['run_id'] === $identity['child_workflow_run_id'], 'Initial child task used another run.'); - $memberRuns[$index] = [$receipt['task']['run_id']]; - if ($index === 0 && in_array($case, ['retry', 'continue'], true)) { - Scenario::check(count($receipt['body']['commands']) === 1 - && $receipt['body']['commands'][0]['type'] === ($case === 'retry' ? 'fail_workflow' : 'continue_as_new'), 'Initial child turn did not request its selected replacement.'); - childAllNoParentTask($worker, $scenario, $parentQueue); - $replacement = childAllMember($scenario, $worker, $control, $namespace, $index, $identity, $parent->workflowId, $run); - Scenario::check($replacement['task']['run_id'] !== $identity['child_workflow_run_id'], 'Child chain did not create a distinct second run.'); - $memberRuns[$index][] = $replacement['task']['run_id']; - $chains = $scenario->assertStable(childAllHistory($control, $parent->workflowId, $run), $commands, $identities); - $scenario->assertReplacementHistory(childAllHistory($control, $identity['child_workflow_instance_id'], $replacement['task']['run_id']), - $identity, $chains[0][1], ['workflow_id' => $parent->workflowId, 'run_id' => $run], - childAllHistory($control, $identity['child_workflow_instance_id'], $identity['child_workflow_run_id'])); - } + $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); From ba7737d578b62fae1de0404c09d2616e5a6f7754 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 18:44:37 +0000 Subject: [PATCH 6/6] Separate child history reconstruction from scenario authoring --- scripts/Support/DeferredChildAllLedger.php | 179 +++++++++++++++++++ scripts/Support/DeferredChildAllScenario.php | 157 +--------------- 2 files changed, 182 insertions(+), 154 deletions(-) create mode 100644 scripts/Support/DeferredChildAllLedger.php 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 index f6d38264..d919150c 100644 --- a/scripts/Support/DeferredChildAllScenario.php +++ b/scripts/Support/DeferredChildAllScenario.php @@ -10,6 +10,8 @@ use InvalidArgumentException; use RuntimeException; +require_once __DIR__.'/DeferredChildAllLedger.php'; + /** Source adapter for Server's flat deferred-child-only all contract (#185). */ final class DeferredChildAllScenario { @@ -181,161 +183,8 @@ public function identities(array $history, array $commands): array 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.'); - $runs = array_fill(0, $this->size, []); - $scheduled = array_fill(0, $this->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; - self::check(is_string($type), 'History event type is not a string.'); - if ($type === 'WorkflowCompleted') { - self::check(!$parentClosed, 'Duplicate parent completion event.'); - $parentClosed = true; - } - if (!str_starts_with($type, 'Child')) { - continue; - } - self::check(!$parentClosed, 'Child event follows parent completion.'); - [$index, $payload] = $this->childEvent($type, $event, $initial); - if ($type === 'ChildWorkflowScheduled') { - self::check(!$scheduled[$index] && $runs[$index] === [], 'Child schedule is repeated or follows its start.'); - $scheduled[$index] = true; - continue; - } - self::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 - { - self::check(in_array($type, ['ChildWorkflowScheduled', 'ChildRunStarted', 'ChildRunCompleted', - 'ChildRunFailed', 'ChildRunCancelled', 'ChildRunTerminated'], true), 'Unqualified child event type.'); - $payload = $event['payload'] ?? null; - self::check(is_array($payload), 'Child event has no payload.'); - $sequence = $payload['sequence'] ?? null; - self::check(is_int($sequence) && $sequence >= 1 && $sequence <= $this->size, 'Child sequence is not an exact member integer.'); - $index = $sequence - 1; - self::check(($payload['child_workflow_type'] ?? null) === $this->childType(), 'Child workflow type changed.'); - foreach (['child_call_id', 'child_workflow_instance_id', 'parallel_group_id', 'parallel_group_path'] as $key) { - self::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 - { - self::check(is_string($payload['child_workflow_run_id'] ?? null) && $payload['child_workflow_run_id'] !== '', 'Child event has no run identity.'); - self::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 === []) { - self::check($run === $original['child_workflow_run_id'] && $link === $original['workflow_link_id'] - && ($payload['child_run_number'] ?? null) === 1, 'Initial child run/link/number changed.'); - self::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)]; - self::check($this->replacementMember($payload['sequence']) && count($chain) === 1 - && $current['terminal'] === null, 'Child run cannot replace a closed or unqualified member.'); - self::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->case, ['retry', 'continue'], true); - } - - /** @param array $payload */ - private function replacementProvenance(array $payload, string $previousRun): void - { - if ($this->case === 'retry') { - self::check(($payload['retry_of_child_workflow_run_id'] ?? null) === $previousRun && ($payload['retry_attempt'] ?? null) === 2, - 'Replacement child transition does not match its case.'); - return; - } - self::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)]; - self::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.'); - self::check(($payload['child_run_number'] ?? null) === count($chain), 'Child terminal run number changed.'); - self::check(!$this->replacementMember($payload['sequence']) || count($chain) === 2, 'Replacement member resolved before its second run.'); - $expected = $payload['sequence'] === 1 && $this->failsEarly() ? $this->expected()['failure_type'] : 'ChildRunCompleted'; - self::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) { - self::check($chain !== [], 'Atomic child start has no initial run event.'); - if ($parentFinished && (!$this->failsEarly() || $index === 0)) { - self::check($chain[array_key_last($chain)]['terminal'] !== null, 'Finished parent has an unresolved required member.'); - } - } - self::check(!$parentFinished || $parentClosed, 'Finished parent has no canonical completion.'); + return (new DeferredChildAllLedger($this, $initial))->validate($history, $parentFinished); } /**