diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b2b4937..0ef990a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,20 @@ project follows [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Added + +- `WorkflowContext::startChildWorkflow()` starts a child and returns a + `ChildWorkflowHandle` once the runtime has committed the start (#185). The + handle's `identity()` gives the child call ID, the child workflow ID and the + initial run ID from canonical history while the child keeps running, and + `await()` returns its result or throws `ChildWorkflowFailed`, following the + same call through child continue-as-new and retries. The start sends + `start_child_workflow` with `await: "started"` and needs a runtime that + advertises `child_workflow_start_turn`; without it the worker abandons the + task before submitting anything. `childWorkflow()`, `deferChildWorkflow()` + and group APIs are unchanged, and `childWorkflow()` refuses an `await` + option. + ## [3.9.4] - 2026-11-04 ### Fixed diff --git a/docs/portal/build/workflows-activities.md b/docs/portal/build/workflows-activities.md index b8137a12..f60ee6f8 100644 --- a/docs/portal/build/workflows-activities.md +++ b/docs/portal/build/workflows-activities.md @@ -274,6 +274,106 @@ The barrier is fail-fast. The first failure in durable history order is thrown i On retry or worker restart, replay validates the exact group ID, base sequence, size, leaf index, and nested path. Pending or completed members keep their original sequence and are not scheduled again. Changing a dynamic fan-out count, order, operation kind, or nesting while an open run can reach the group raises `NonDeterministicWorkflow` before any replacement commands leave the worker. +## Start a child and keep its identity + +`childWorkflow()` returns the child's result, so the parent waits until the +child finishes. When the parent needs the child's identity first, for example +to record it or notify someone, start the child with `startChildWorkflow()`: + +```php +use DurableWorkflow\Worker\ParentClosePolicy; + +$child = $context->startChildWorkflow('billing.reconcile', [$accountId], [ + 'parent_close_policy' => ParentClosePolicy::Abandon, +]); + +$identity = $child->identity(); +$context->activity('notify.reconcile-started', [$identity->workflowId]); + +$summary = $child->await(); +``` + +The turn that calls `startChildWorkflow()` ends after the start command. The +runtime commits the child's start and gives the parent a new turn while the +child keeps running, so the handle returns only for a committed child. +`identity()` reads canonical history: + +| Property | Meaning | +| --- | --- | +| `callId` | The parent's call for this child. It stays the same through child continue-as-new and retries. | +| `workflowId` | The child workflow instance ID. | +| `initialRunId` | The first run of the child. A later continued or retried run does not change it. | + +`await()` returns the child's result or throws `ChildWorkflowFailed`. It +follows the same child call through continue-as-new and retries, and sends no +command of its own. Replay and a replacement worker rebuild the same handle +from history without starting another child. + +Parent-close and cancellation policies, timeouts and retries work as they do +for `childWorkflow()`. The handle supports only `identity()` and `await()`: +it cannot signal, query, update, cancel or terminate the child. Await it in the +workflow execution that started it. `startChildWorkflow()` cannot run inside an +`all()` or `parallel()` group; use `deferChildWorkflow()` there. + +This needs a runtime that advertises `child_workflow_start_turn`. Native Server +3.0.x and PHP Server 2.x do not. On those, the worker abandons the task with +`child_workflow_start_turn_unavailable` before it submits anything, so no child +starts. A cancellation-scope checkpoint keeps the parent's claim, so the +runtime refuses a start there with `child_start_turn_not_at_checkpoint`. + +### Run the service-mode child example + +The complete [worker](https://github.com/durable-workflow/sdk-php/blob/main/examples/child-workflow-worker.php) +and [client](https://github.com/durable-workflow/sdk-php/blob/main/examples/child-workflow-client.php) +use the quickstart's [bootstrap](https://github.com/durable-workflow/sdk-php/blob/main/examples/bootstrap.php). +Copy those three files beside the `vendor/` directory of your Composer project. +Use an SDK release containing `startChildWorkflow()` and a Server advertising +`child_workflow_start_turn`. The same runtime URL and namespace must be available +to both processes. Set `DURABLE_WORKFLOW_RUNTIME_URL`, +`DURABLE_WORKFLOW_NAMESPACE`, and `DURABLE_WORKFLOW_TASK_QUEUE` in both terminals. +Set `DURABLE_WORKFLOW_WORKER_TOKEN` in the worker terminal and +`DURABLE_WORKFLOW_CLIENT_TOKEN` in the client terminal, using their respective +role credentials. + +Start the worker, then run a client command in the second terminal: + +```bash +php child-workflow-worker.php +``` + +```bash +php child-workflow-client.php await +php child-workflow-client.php await reject +php child-workflow-client.php abandon +php child-workflow-client.php cancel +php child-workflow-client.php terminate +``` + +Each parent first demonstrates a sequential result. It then starts a child with a committed handle and records +the identity in an activity. The activity may be retried, so an application +replacing its sample output with an external write should make that write +idempotent by `call_id`. + +`await` returns the typed greeting. `await reject` catches `ChildWorkflowFailed` +and returns its recorded failure. The other modes close the parent while its +child remains open: `abandon` allows it to complete, `cancel` applies legacy +terminal cancellation, and `terminate` terminates it. The client prints the +parent result and the child's observed status. Cancellation through the handle +is unsupported. Cooperative cleanup uses the separate opt-in cancellation APIs +and `ParentClosePolicy::RequestCancellation`, as described in the cancellation +guide. Initial run IDs remain snapshots through retry and continue-as-new; +instance-oriented client operations target the child's current run unless the +caller explicitly selects a run. + +`deferChildWorkflow()` prepares an operation without starting it or providing a +child identity. Join it at an `all()` or `parallel()` barrier on a runtime that +supports child members in those groups, as shown in the concurrency examples +above. Native Server currently refuses child members with +`development_parallel_group_unqualified`. +[Server #325](https://github.com/durable-workflow/server/issues/325) tracks this +runtime parity gap. The runnable example uses sequential children and committed +start handles; it does not establish deferred-group compatibility on Native. + ## Evolve workflow code with version markers Give each long-lived code change a stable ID and keep the old branch while histories can still reach it. `getVersion()` records the maximum supported version for a new run and returns that same decision after redelivery, worker restart, or cold replay. Including `-1` in the first range lets histories that passed this point before the marker existed stay on the legacy branch: @@ -460,6 +560,7 @@ The context provides these replay-aware commands: | `saga()` | Reverse-order compensation through ordinary activity history. | | `sleep()` | A durable timer that survives worker downtime. | | `childWorkflow()` | A separately identified durable execution. | +| `startChildWorkflow()` | A started child whose identity the parent reads before awaiting its result. | | `all()`, `parallel()` | An ordered, fail-fast join over deferred activities, child workflows, and timers. | | `sideEffect()` | A small nondeterministic value recorded once in history. | | `getVersion()`, `patched()`, `deprecatePatch()` | A durable workflow-code evolution decision. | diff --git a/examples/child-workflow-client.php b/examples/child-workflow-client.php new file mode 100644 index 00000000..f004c02d --- /dev/null +++ b/examples/child-workflow-client.php @@ -0,0 +1,35 @@ +startWorkflow( + 'quickstart.php.child-examples', + 'php-child-example-'.bin2hex(random_bytes(12)), + quickstartEnvironment('DURABLE_WORKFLOW_TASK_QUEUE'), + ['PHP', $mode, $reject], +); +$result = $parent->result(timeoutSeconds: 90, pollIntervalSeconds: 1); +$deadline = microtime(true) + 15; +do { + $child = $client->describeWorkflow($result['identity']['workflow_id']); + if (in_array($child->status, ['completed', 'failed', 'cancelled', 'terminated'], true)) { + break; + } + usleep(200_000); +} while (microtime(true) < $deadline); + +echo json_encode(['parent_result' => $result, 'child_status' => $child->status], JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT).PHP_EOL; diff --git a/examples/child-workflow-worker.php b/examples/child-workflow-worker.php new file mode 100644 index 00000000..4bd95c45 --- /dev/null +++ b/examples/child-workflow-worker.php @@ -0,0 +1,89 @@ +childWorkflow('quickstart.php.child-greeting', [$name, false]); + + $policy = match ($mode) { + 'await', 'abandon' => ParentClosePolicy::Abandon, + 'cancel' => ParentClosePolicy::RequestCancel, + 'terminate' => ParentClosePolicy::Terminate, + default => throw new InvalidArgumentException('Unknown child example mode.'), + }; + $child = $context->startChildWorkflow('quickstart.php.child-greeting', [$name, $reject], [ + 'parent_close_policy' => $policy, + ]); + $identity = $child->identity(); + $snapshot = ['call_id' => $identity->callId, 'workflow_id' => $identity->workflowId, 'initial_run_id' => $identity->initialRunId]; + $context->activity('quickstart.php.record-child', [$snapshot]); + $result = ['sequential' => $sequential, 'identity' => $snapshot, 'parent_close_policy' => $policy->value]; + if ($mode !== 'await') { + // Closing with an open child applies the selected parent-close policy. + return $result; + } + + try { + $result['child_result'] = $child->await(); + } catch (ChildWorkflowFailed $failure) { + $result['child_failure'] = ['type' => $failure->failureType, 'message' => $failure->getMessage()]; + } + + return $result; + } +} + +final class ChildGreetingWorkflow +{ + #[Workflow('quickstart.php.child-greeting')] + public function greet(WorkflowContext $context, string $name, bool $reject): array + { + $context->sleep(3); + if ($reject) { + throw new ChildExampleRejected('The child declined this greeting.'); + } + + return ['greeting' => 'Hello, '.$name, 'accepted' => true]; + } +} + +final class ChildExampleActivities +{ + #[Activity('quickstart.php.record-child')] + public function record(ActivityContext $context, array $identity): array + { + echo json_encode(['child_started' => $identity], JSON_THROW_ON_ERROR).PHP_EOL; + + return $identity; + } +} + +$client = new Client( + quickstartEnvironment('DURABLE_WORKFLOW_RUNTIME_URL'), + namespace: quickstartEnvironment('DURABLE_WORKFLOW_NAMESPACE'), + workerToken: quickstartEnvironment('DURABLE_WORKFLOW_WORKER_TOKEN'), +); + +Worker::create($client, quickstartEnvironment('DURABLE_WORKFLOW_TASK_QUEUE')) + ->register(ChildWorkflowExamples::class, ChildGreetingWorkflow::class, ChildExampleActivities::class) + ->run(); diff --git a/scripts/ci/run-replay-regression-fixture.php b/scripts/ci/run-replay-regression-fixture.php index 60b7ca8e..242f2e72 100644 --- a/scripts/ci/run-replay-regression-fixture.php +++ b/scripts/ci/run-replay-regression-fixture.php @@ -735,6 +735,15 @@ private static function workflow(string $workflowType): callable return ['child' => $result]; }, + 'golden.child-start-handle' => static function ( + WorkflowContext $context, + mixed $workflowType, + ): array { + $child = $context->startChildWorkflow((string) $workflowType, ['golden-input']); + $identity = $child->identity(); + + return ['identity' => [$identity->callId, $identity->workflowId, $identity->initialRunId], 'child' => $child->await()]; + }, 'golden.side-effect' => static function (WorkflowContext $context, mixed $value): array { $result = $context->sideEffect(static fn (): mixed => $value); diff --git a/scripts/probe-deferred-child-group.php b/scripts/probe-deferred-child-group.php new file mode 100644 index 00000000..1a503519 --- /dev/null +++ b/scripts/probe-deferred-child-group.php @@ -0,0 +1,50 @@ +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('probe.child-group.parent', 'parent-'.$suffix, $queue); +$workerId = $queue.'-worker'; +$registration = $worker->registerWorker($workerId, $queue, ['probe.child-group.parent'], []); +try { + $task = $worker->pollWorkflowTask($workerId, $queue, 1); + if (!is_array($task)) { + throw new LogicException('Probe did not receive its initial parent claim.'); + } + $workflow = static fn (WorkflowContext $context): array => $context->all([ + $context->deferChildWorkflow('probe.child-group.child', ['child input']), + ]); + $commands = (new Replayer($worker->payloadCodec()))->replay($workflow, $task['history_events'] ?? [], [], $queue)->commands; + try { + $response = $worker->completeWorkflowTask($task['task_id'], $task['lease_owner'], $task['workflow_task_attempt'], $commands); + $outcome = ['accepted' => true, 'receipt' => $response]; + } catch (ServerException $error) { + $outcome = ['accepted' => false, 'status' => $error->status, 'reason' => $error->reason]; + } + $page = $control->workflowHistory($parent->workflowId, (string) $parent->selectedRunId, 200); + echo json_encode([ + 'observed_at' => gmdate(DATE_ATOM), 'namespace' => $namespace, + 'workflow_id' => $parent->workflowId, 'run_id' => $parent->selectedRunId, + 'commands' => $commands, 'outcome' => $outcome, + 'history' => array_column($page['events'] ?? $page['history_events'] ?? [], 'event_type'), + ], JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT).PHP_EOL; +} finally { + $worker->deregisterWorkerRegistration($workerId, $registration['registration_token'] ?? null); +} diff --git a/scripts/qualify-child-start-compatibility.php b/scripts/qualify-child-start-compatibility.php new file mode 100644 index 00000000..6c8d115e --- /dev/null +++ b/scripts/qualify-child-start-compatibility.php @@ -0,0 +1,112 @@ + str_contains($exception->getMessage(), 'child_workflow_start_turn_unavailable'), + 'exception_type' => $exception::class, + ], JSON_THROW_ON_ERROR)); + $worker->requestShutdown(); + } + }, + ); + $worker->registerWorkflow('compat.parent', static function (WorkflowContext $context, string $value) use ($mode): mixed { + if ($mode === 'gate') { + return $context->startChildWorkflow('compat.child', [$value])->await(); + } + + return $context->childWorkflow('compat.child', [$value]); + }); + $worker->registerWorkflow('compat.child', static fn (WorkflowContext $context, string $value): array => [$value, PHP_INT_MAX, null]); + $worker->run(1); + exit(0); +} + +function check(bool $condition, string $message): void +{ + if (!$condition) { + throw new ChildStartCompatibilityFailed($message); + } + echo "ok: {$message}\n"; +} + +$startedAt = gmdate(DATE_ATOM); +$admin = new Client($url, controlToken: 'claimed-task-admin'); +$capability = $admin->clusterInfo()->capabilities['child_workflow_start_turn'] ?? null; +check($capability !== true, 'published Server does not advertise the child start turn'); +$results = []; +foreach (['gate', 'legacy'] as $mode) { + $suffix = bin2hex(random_bytes(6)); + $namespace = 'php-child-compat-'.$suffix; + $queue = 'child-compat-'.$suffix; + $admin->createNamespace($namespace); + $client = new Client($url, namespace: $namespace, controlToken: 'claimed-task-client'); + $parent = $client->startWorkflow('compat.parent', 'parent-'.$suffix, $queue, ['typed PHP result']); + $receipt = sys_get_temp_dir().'/child-compat-'.$suffix.'.json'; + $process = proc_open([PHP_BINARY, $script, $autoload, $url, '--worker', $namespace, $queue, $mode, $receipt], [1 => STDOUT, 2 => STDERR], $pipes); + check(is_resource($process), 'managed worker process starts'); + try { + $deadline = microtime(true) + 45; + do { + usleep(100_000); + $execution = $client->describeWorkflow($parent->workflowId); + if (($mode === 'gate' && is_file($receipt)) || ($mode === 'legacy' && $execution->status === 'completed')) { + break; + } + } while (microtime(true) < $deadline); + if ($mode === 'gate') { + check(is_file($receipt), 'the worker reports the unsupported start turn'); + $refusal = json_decode((string) file_get_contents($receipt), true, flags: JSON_THROW_ON_ERROR); + check($refusal['unsupported'] === true, 'the refusal names child_workflow_start_turn_unavailable'); + $workerDeadline = microtime(true) + 5; + while (proc_get_status($process)['running'] && microtime(true) < $workerDeadline) { + usleep(100_000); + } + } else { + check($execution->status === 'completed', 'legacy childWorkflow completes on the published Server'); + check($parent->result(5) === ['typed PHP result', PHP_INT_MAX, null], 'legacy childWorkflow preserves its typed result'); + } + $history = $client->workflowHistory($parent->workflowId, (string) $parent->selectedRunId, 200); + $events = $history['events'] ?? $history['history_events'] ?? []; + $types = array_column($events, 'event_type'); + $childStarts = count(array_filter($types, static fn (string $type): bool => $type === 'ChildWorkflowScheduled')); + check($childStarts === ($mode === 'gate' ? 0 : 1), $mode === 'gate' ? 'unsupported start committed no child' : 'legacy call committed exactly one child'); + $results[$mode] = ['child_starts' => $childStarts, 'history' => $types]; + } finally { + if (proc_get_status($process)['running']) { + proc_terminate($process); + } + proc_close($process); + if (is_file($receipt)) { + unlink($receipt); + } + } +} +echo json_encode(['started_at' => $startedAt, 'finished_at' => gmdate(DATE_ATOM), 'php' => PHP_VERSION, 'cases' => $results], JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT)."\n"; + diff --git a/scripts/qualify-child-start-handle.php b/scripts/qualify-child-start-handle.php new file mode 100644 index 00000000..ca81625a --- /dev/null +++ b/scripts/qualify-child-start-handle.php @@ -0,0 +1,101 @@ +registerWorkflow('qualification.child-handle.parent', static function (WorkflowContext $context, string $account) use ($queue): array { + $child = $context->startChildWorkflow('qualification.child-handle.child', [$account], [ + 'queue' => $queue, 'parent_close_policy' => ParentClosePolicy::Abandon, + ]); + $identity = $child->identity(); + // Durable work while the child keeps running, before awaiting it. + $context->sleep(1); + + return ['identity' => [$identity->callId, $identity->workflowId, $identity->initialRunId], 'result' => $child->await()]; + }); + $worker->registerWorkflow('qualification.child-handle.child', static function (WorkflowContext $context, string $account): string { + $context->sleep(3); + + return 'reconciled '.$account; + }); + try { + $worker->run(1); + } catch (Throwable $error) { + fwrite(STDERR, 'worker stopped: '.$error::class.': '.$error->getMessage()."\n"); + exit(1); + } + exit(0); +} + +function check(bool $condition, string $message): void +{ + if (!$condition) { + fwrite(STDERR, "FAIL: {$message}\n"); + exit(1); + } + echo "ok: {$message}\n"; +} + +$admin = new Client($url, controlToken: 'claimed-task-admin'); +check(($admin->clusterInfo()->capabilities['child_workflow_start_turn'] ?? null) === true, + 'cluster info advertises child_workflow_start_turn'); +$suffix = bin2hex(random_bytes(6)); +$namespace = 'php-child-handle-'.$suffix; +$queue = 'child-handle-'.$suffix; +$admin->createNamespace($namespace); +$client = new Client($url, namespace: $namespace, controlToken: 'claimed-task-client'); +$process = proc_open([PHP_BINARY, $script, $autoload, $url, '--worker', $namespace, $queue], [1 => STDOUT, 2 => STDERR], $pipes); + +try { + $parent = $client->startWorkflow('qualification.child-handle.parent', 'parent-'.$suffix, $queue, ['acct-1']); + $deadline = microtime(true) + 90; + $execution = null; + while (microtime(true) < $deadline) { + usleep(500_000); + $execution = $client->describeWorkflow($parent->workflowId); + if ($execution->status !== 'running' && $execution->status !== 'waiting' && $execution->status !== 'pending') { + break; + } + } + $history = $client->workflowHistory($parent->workflowId, (string) $parent->selectedRunId, 200); + $events = $history['events'] ?? $history['history_events'] ?? []; + if ($execution?->status !== 'completed') { + foreach ($events as $event) { + fwrite(STDERR, ($event['recorded_at'] ?? '').' '.$event['event_type']."\n"); + } + } + check($execution?->status === 'completed', 'the parent completes (status: '.($execution?->status ?? 'none').')'); + + $types = array_column($events, 'event_type'); + $scheduled = array_values(array_filter($events, static fn (array $event): bool => $event['event_type'] === 'ChildWorkflowScheduled')); + check(count($scheduled) === 1, 'exactly one child start was committed'); + $recorded = [$scheduled[0]['payload']['child_call_id'], $scheduled[0]['payload']['child_workflow_instance_id'], $scheduled[0]['payload']['child_workflow_run_id']]; + $workedAt = array_search('TimerScheduled', $types, true); + $resolvedAt = array_search('ChildRunCompleted', $types, true); + check($workedAt !== false && $resolvedAt !== false && $workedAt < $resolvedAt, + 'the parent did durable work between the committed start and the child result'); + $result = $parent->result(5); + check($result === ['identity' => $recorded, 'result' => 'reconciled acct-1'], + 'the parent returns the committed identity and the awaited child result'); + echo json_encode(['namespace' => $namespace, 'identity' => $recorded, 'history' => $types], JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT)."\n"; +} finally { + proc_terminate($process); + proc_close($process); +} diff --git a/scripts/qualify-child-start-lifecycle.php b/scripts/qualify-child-start-lifecycle.php new file mode 100644 index 00000000..b8674d89 --- /dev/null +++ b/scripts/qualify-child-start-lifecycle.php @@ -0,0 +1,365 @@ +http = new Psr18Transport(); + } + + 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)); + } + + private function observe(string $uri, ?array $body, ?array $response): ?array + { + if (preg_match('~/workflow-tasks/([^/]+)/complete$~', $uri, $match) === 1 + && ($body['commands'][0]['await'] ?? null) === 'started') { + file_put_contents($this->receipt, json_encode([ + 'task_id' => rawurldecode($match[1]), 'body' => $body, 'response' => $response, + ], JSON_THROW_ON_ERROR)); + ($this->started)(); + } + + return $response; + } +} + +function lifecycleCheck(bool $condition, string $message): void +{ + if (!$condition) { + throw new ChildLifecycleFailure($message); + } + echo 'ok: '.$message.PHP_EOL; +} + +/** @return list */ +function lifecycleValue(): array +{ + return ['typed child ✓', PHP_INT_MAX, null, true, -0.0]; +} + +if (($argv[3] ?? null) === '--worker') { + [$namespace, $queue, $workerId, $stopAfterStart, $receipt, $cooperative] = array_slice($argv, 4); + $worker = null; + $transport = new ChildLifecycleTransport($receipt, static function () use (&$worker, $stopAfterStart): void { + if ($stopAfterStart === 'yes') { + $worker->requestShutdown(); + } + }); + $client = new Client($url, namespace: $namespace, workerToken: 'claimed-task-worker', transport: $transport); + $worker = new Worker($client, $queue, workerId: $workerId, enableCooperativeCancellation: $cooperative === 'yes'); + $worker->registerWorkflow('qualification.lifecycle.parent', static function ( + WorkflowContext $context, string $mode, string $childQueue, array $options, + ): array { + $child = $context->startChildWorkflow('qualification.lifecycle.child', [$mode, 0], ['queue' => $childQueue, ...$options]); + $identity = $child->identity(); + $snapshot = [$identity->callId, $identity->workflowId, $identity->initialRunId]; + if ($mode === 'close') { + return ['identity' => $snapshot]; + } + $context->sideEffect(static fn (): array => $snapshot); + try { + return ['identity' => $snapshot, 'result' => $child->await()]; + } catch (ChildWorkflowFailed $failure) { + return ['identity' => $snapshot, 'failure' => [ + 'class' => $failure::class, 'workflow_type' => $failure->workflowType, + 'failure_type' => $failure->failureType, 'message' => $failure->getMessage(), + ]]; + } + }); + $worker->registerWorkflow('qualification.lifecycle.child', static function ( + WorkflowContext $context, string $mode, int $stage, + ): array { + if ($mode === 'failure' || $mode === 'retry') { + throw new QualificationChildRejected('Qualification child rejected.'); + } + if ($mode === 'continue' && $stage === 0) { + $context->continueAsNew([$mode, 1]); + } + if ($mode === 'cancel') { + $context->sleep(60); + } + + return lifecycleValue(); + }); + $worker->run(1); + exit(0); +} + +/** @return list> */ +function lifecycleHistory(Client $client, string $workflow, string $run): array +{ + $page = $client->workflowHistory($workflow, $run, 200); + lifecycleCheck(empty($page['next_page_token']), 'qualification history fits one complete page'); + + return $page['events'] ?? $page['history_events'] ?? []; +} + +function lifecycleWait(callable $condition, string $description, int $seconds = 35): mixed +{ + $deadline = microtime(true) + $seconds; + do { + $result = $condition(); + if ($result) { + return $result; + } + usleep(100_000); + } while (microtime(true) < $deadline); + + throw new ChildLifecycleFailure('Timed out: '.$description); +} + +/** @return resource */ +function lifecycleWorker(string $namespace, string $queue, string $workerId, string $receipt, bool $once = false, bool $cooperative = false) +{ + global $script, $autoload, $url; + $process = proc_open([ + PHP_BINARY, $script, $autoload, $url, '--worker', $namespace, $queue, $workerId, + $once ? 'yes' : 'no', $receipt, $cooperative ? 'yes' : 'no', + ], [1 => STDOUT, 2 => STDERR], $pipes); + if (!is_resource($process)) { + throw new ChildLifecycleFailure('Could not start a managed worker.'); + } + + return $process; +} + +/** @param resource $process */ +function lifecycleStop($process): void +{ + if (proc_get_status($process)['running']) { + proc_terminate($process); + $deadline = microtime(true) + 5; + while (proc_get_status($process)['running'] && microtime(true) < $deadline) { + usleep(100_000); + } + if (proc_get_status($process)['running']) { + proc_terminate($process, 9); + } + } + proc_close($process); +} + +/** @return array */ +function lifecycleDuplicate(Client $client, array $start, string $queue): array +{ + $body = $start['body']; + try { + $repeat = $client->completeWorkflowTask($start['task_id'], $body['lease_owner'], $body['workflow_task_attempt'], $body['commands']); + lifecycleCheck(($repeat['recorded'] ?? null) === false && ($repeat['reason'] ?? null) === 'already_completed', 'duplicate start completion is an explicit no-op'); + $duplicateReceipt = 'already_completed'; + } catch (ServerException $error) { + lifecycleCheck($error->status === 409 && $error->reason === 'task_not_leased', 'duplicate completion from the stopped worker is fenced'); + $duplicateReceipt = '409/task_not_leased'; + } + + $workerId = $queue.'-redelivery'; + $registration = $client->registerWorker($workerId, $queue, ['qualification.lifecycle.parent'], []); + try { + $requestId = $queue.'-same-poll'; + $first = $client->pollWorkflowTaskResponse($workerId, $queue, 1, $requestId); + $second = $client->pollWorkflowTaskResponse($workerId, $queue, 1, $requestId); + $task = $first['task'] ?? null; + lifecycleCheck(is_array($task), 'a fresh parent turn is available before child execution'); + foreach (['task_id', 'lease_owner', 'workflow_task_attempt'] as $field) { + lifecycleCheck($task[$field] === ($second['task'][$field] ?? null), 'duplicate parent poll preserves '.$field); + } + } finally { + $released = $client->deregisterWorkerRegistration($workerId, $registration['registration_token'] ?? null); + } + lifecycleCheck($released['recovered_workflow_task_count'] === 1, 'the delivery-control worker releases its one claim for replacement'); + try { + $client->completeWorkflowTask($task['task_id'], $task['lease_owner'], $task['workflow_task_attempt'], [[ + 'type' => 'complete_workflow', 'result' => $client->payloadCodec()->envelope('stale worker'), + ]]); + throw new ChildLifecycleFailure('A released claim completed the parent.'); + } catch (ServerException $error) { + lifecycleCheck($error->status === 409 && $error->reason === 'task_not_leased', 'the released parent attempt is fenced'); + } + + return ['duplicate_completion' => $duplicateReceipt, 'duplicate_delivery' => 'same_claim', 'released_attempt' => 'task_not_leased']; +} + +$startedAt = gmdate(DATE_ATOM); +$admin = new Client($url, controlToken: 'claimed-task-admin'); +lifecycleCheck(($admin->clusterInfo()->capabilities['child_workflow_start_turn'] ?? null) === true, 'runtime advertises the committed-start turn'); +$results = []; +$cases = [ + ['result', ['parent_close_policy' => 'abandon']], + ['failure', []], + ['continue', []], + ['retry', ['retry_policy' => ['max_attempts' => 2, 'backoff_seconds' => [0]]]], +]; +foreach (['abandon', 'request_cancel', 'terminate', 'request_cancellation'] as $policy) { + $cases[] = ['close', ['parent_close_policy' => $policy]]; +} +foreach (['abandon', 'try_cancel', 'wait_cancellation_completed'] as $policy) { + $cases[] = ['cancel', ['parent_close_policy' => 'abandon', 'cancellation_policy' => $policy]]; +} +if (($argv[3] ?? null) === '--policies') { + $cases = array_filter($cases, static fn (array $case): bool => in_array($case[0], ['close', 'cancel'], true)); +} + +foreach ($cases as [$mode, $options]) { + $suffix = bin2hex(random_bytes(6)); + $namespace = 'php-child-lifecycle-'.$suffix; + $parentQueue = 'parent-'.$suffix; + $childQueue = 'child-'.$suffix; + $receipt = sys_get_temp_dir().'/child-lifecycle-'.$suffix.'.json'; + $admin->createNamespace($namespace); + $client = new Client($url, namespace: $namespace, controlToken: 'claimed-task-client'); + $workerClient = new Client($url, namespace: $namespace, workerToken: 'claimed-task-worker'); + $cooperative = $mode === 'cancel' || ($options['parent_close_policy'] ?? null) === 'request_cancellation'; + $parent = $client->startWorkflow('qualification.lifecycle.parent', 'parent-'.$suffix, $parentQueue, [$mode, $childQueue, $options]); + $processes = []; + $identity = null; + try { + $first = lifecycleWorker($namespace, $parentQueue, $parentQueue.'-original', $receipt, once: true, cooperative: $cooperative); + $processes[] = $first; + lifecycleWait(static fn (): bool => is_file($receipt) && !proc_get_status($first)['running'], 'original worker commits the start and exits'); + $start = json_decode((string) file_get_contents($receipt), true, flags: JSON_THROW_ON_ERROR); + $before = lifecycleHistory($client, $parent->workflowId, (string) $parent->selectedRunId); + $scheduled = array_values(array_filter($before, static fn (array $event): bool => $event['event_type'] === 'ChildWorkflowScheduled')); + lifecycleCheck(count($scheduled) === 1, 'exactly one child was started before replacement'); + $payload = $scheduled[0]['payload']; + $identity = [$payload['child_call_id'], $payload['child_workflow_instance_id'], $payload['child_workflow_run_id']]; + $controls = $mode === 'result' ? lifecycleDuplicate($workerClient, $start, $parentQueue) : []; + + $other = 'php-child-isolation-'.$suffix; + $admin->createNamespace($other); + try { + (new Client($url, namespace: $other, controlToken: 'claimed-task-client'))->describeWorkflow($identity[1]); + throw new ChildLifecycleFailure('A child was visible in another namespace.'); + } catch (ServerException $error) { + lifecycleCheck($error->status === 404 && $error->reason === 'instance_not_found', 'the child identity remains namespace-scoped ('.$error->status.'/'.$error->reason.')'); + } + + $processes[] = lifecycleWorker($namespace, $parentQueue, $parentQueue.'-replacement', $receipt, cooperative: $cooperative); + if ($mode !== 'close') { + lifecycleWait(static function () use ($client, $parent): bool { + $page = $client->workflowHistory($parent->workflowId, (string) $parent->selectedRunId, 200); + + return in_array('SideEffectRecorded', array_column($page['events'] ?? $page['history_events'] ?? [], 'event_type'), true); + }, 'replacement recovers identity and records its snapshot'); + } + + if ($mode === 'cancel') { + $request = $parent->requestCancellation('Qualification parent cancellation.', 20); + $repeat = $parent->requestCancellation('Repeated request must not reset the budget.', 30); + lifecycleCheck($repeat['duplicate'] === true, 'a repeated cancellation is recognized'); + foreach (['request_id', 'cleanup_deadline_at'] as $field) { + lifecycleCheck($request['cancellation_request'][$field] === $repeat['cancellation_request'][$field], 'repeated cancellation preserves '.$field); + } + } + if ($mode !== 'close') { + $processes[] = lifecycleWorker($namespace, $childQueue, $childQueue.'-worker', $receipt, cooperative: $cooperative); + } + $execution = lifecycleWait(static function () use ($client, $parent) { + $value = $client->describeWorkflow($parent->workflowId, (string) $parent->selectedRunId); + + return in_array($value->status, ['completed', 'cancelled', 'terminated', 'failed'], true) ? $value : null; + }, 'parent reaches its expected terminal state'); + lifecycleCheck($execution->status === ($mode === 'cancel' ? 'cancelled' : 'completed'), 'parent terminal state for '.$mode); + if ($mode === 'close' && in_array($options['parent_close_policy'], ['abandon', 'request_cancellation'], true)) { + $processes[] = lifecycleWorker($namespace, $childQueue, $childQueue.'-worker', $receipt, cooperative: $cooperative); + } + $history = lifecycleHistory($client, $parent->workflowId, (string) $parent->selectedRunId); + $types = array_column($history, 'event_type'); + lifecycleCheck(count(array_filter($types, static fn (string $type): bool => $type === 'ChildWorkflowScheduled')) === 1, 'replacement and duplicate delivery did not start another child'); + if ($mode !== 'cancel') { + $result = $parent->result(5); + lifecycleCheck($result['identity'] === $identity, 'initial child identity is unchanged at parent completion'); + if (in_array($mode, ['result', 'continue'], true)) { + lifecycleCheck(serialize($result['result']) === serialize(lifecycleValue()), 'the awaited value preserves its PHP types and negative zero'); + } + if ($mode === 'failure' || $mode === 'retry') { + lifecycleCheck($result['failure'] === [ + 'class' => ChildWorkflowFailed::class, 'workflow_type' => 'qualification.lifecycle.child', + 'failure_type' => 'ChildRunFailed', 'message' => 'Qualification child rejected.', + ], 'the parent receives the typed child failure'); + } + } + $starts = array_values(array_filter($history, static fn (array $event): bool => $event['event_type'] === 'ChildRunStarted')); + if ($mode === 'continue' || $mode === 'retry') { + lifecycleCheck(count($starts) === 2 && $starts[1]['payload']['child_workflow_run_id'] !== $identity[2], 'the child chain has a distinct second run'); + lifecycleCheck($starts[1]['payload']['child_call_id'] === $identity[0], 'the second run belongs to the same child call'); + } + if ($mode === 'close' || $mode === 'cancel') { + $expected = $mode === 'close' ? match ($options['parent_close_policy']) { + 'abandon' => 'completed', 'terminate' => 'terminated', default => 'cancelled', + } : ($options['cancellation_policy'] === 'abandon' ? 'open' : 'cancelled'); + $child = lifecycleWait(static function () use ($client, $identity, $expected) { + $value = $client->describeWorkflow($identity[1]); + + return ($expected === 'open' ? in_array($value->status, ['pending', 'running', 'waiting'], true) : $value->status === $expected) ? $value : null; + }, 'child lifecycle policy '.$expected); + lifecycleCheck($expected === 'open' || $child->status === $expected, 'child policy result '.$child->status); + if (($mode === 'close' && $options['parent_close_policy'] === 'request_cancellation') + || ($mode === 'cancel' && $options['cancellation_policy'] !== 'abandon')) { + $childHistory = lifecycleHistory($client, $identity[1], $identity[2]); + $parentRequests = array_values(array_filter($history, static fn (array $event): bool => in_array($event['event_type'], ['CooperativeCancellationRequested', 'ParentCloseCancellationRequested'], true))); + $childRequests = array_values(array_filter($childHistory, static fn (array $event): bool => $event['event_type'] === 'CooperativeCancellationRequested')); + lifecycleCheck(count($parentRequests) === 1 && count($childRequests) === 1, 'one canonical cancellation root reaches the child'); + $root = $parentRequests[0]['payload']['cancellation']; + $descendant = $childRequests[0]['payload']['cancellation']; + foreach (['root_request_id', 'cleanup_deadline_at'] as $field) { + lifecycleCheck($root[$field] === $descendant[$field], 'child cancellation preserves '.$field); + } + $controls['cancellation_deadline'] = 'preserved'; + } + } + $results[] = ['mode' => $mode, 'options' => $options, 'identity' => $identity, 'history' => $types, 'controls' => $controls]; + } catch (Throwable $error) { + fwrite(STDERR, json_encode([ + 'mode' => $mode, 'options' => $options, 'namespace' => $namespace, + 'parent_id' => $parent->workflowId, 'identity' => $identity ?? null, + ], JSON_THROW_ON_ERROR).PHP_EOL); + throw $error; + } finally { + foreach (array_reverse($processes) as $process) { + lifecycleStop($process); + } + @unlink($receipt); + } +} +echo json_encode(['started_at' => $startedAt, 'finished_at' => gmdate(DATE_ATOM), 'scenarios' => $results], JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT).PHP_EOL; diff --git a/src/Worker.php b/src/Worker.php index 8cade7c9..e40f067c 100644 --- a/src/Worker.php +++ b/src/Worker.php @@ -108,6 +108,7 @@ final class Worker /** @var array|null */ private ?array $workflowMemoCapability = null; private bool $serviceOperationsSupported = false; + private bool $childStartTurnSupported = false; private bool $queryTaskAuthoritySupported = false; private ?QueryTaskAuthority $queryTaskAuthority = null; private bool $activityLeaseRenewalSupported = false; @@ -807,6 +808,8 @@ public function tick(int $pollTimeoutSeconds = 1): bool $supportedCommands = is_array($capabilities) ? ($capabilities['supported_workflow_task_commands'] ?? null) : null; $this->serviceOperationsSupported = is_array($supportedCommands) && in_array('start_service_operation', $supportedCommands, true); + $this->childStartTurnSupported = is_array($capabilities) + && ($capabilities['child_workflow_start_turn'] ?? null) === true; $this->heartbeatIfDue(); if ($this->crossKindPollWakeReceived($workflowPoll)) { $pollTimeoutSeconds = 0; @@ -1459,6 +1462,7 @@ private function executeWorkflowTask(array $task): void } $this->assertWorkflowMemoUpdatesAvailable($commands); $this->assertServiceOperationsAvailable($commands); + $this->assertChildStartTurnAvailable($commands); $this->assertCancellationPoliciesAvailable($commands); $stickyClaim = $this->stickyCacheClaim($task); $this->retryStorageAdmission('workflow_complete', fn (): array => $this->client->completeWorkflowTask( @@ -2412,6 +2416,25 @@ private function assertServiceOperationsAvailable(array $commands): void } } + /** + * A runtime without the capability ignores the unknown await field and + * would start the child without the parent turn startChildWorkflow() waits + * for, so the claim is abandoned before anything is submitted. + * + * @param list> $commands + */ + private function assertChildStartTurnAvailable(array $commands): void + { + foreach ($commands as $command) { + if (($command['type'] ?? null) === 'start_child_workflow' && ($command['await'] ?? null) === 'started' + && !$this->childStartTurnSupported) { + throw new WorkflowClaimAborted( + 'child_workflow_start_turn_unavailable: startChildWorkflow() needs a runtime that advertises child_workflow_start_turn.', + ); + } + } + } + /** @param list> $commands */ private function assertCancellationPoliciesAvailable(array $commands): void { diff --git a/src/Worker/ChildWorkflowHandle.php b/src/Worker/ChildWorkflowHandle.php new file mode 100644 index 00000000..74a18490 --- /dev/null +++ b/src/Worker/ChildWorkflowHandle.php @@ -0,0 +1,53 @@ +identity; + } + + /** Wait for the child's result in this workflow execution. */ + public function await(): mixed + { + return $this->context->awaitChildWorkflow($this); + } + + /** @internal */ + public function belongsTo(WorkflowContext $context): bool + { + return $this->context === $context; + } + + /** @internal */ + public function assertOwner(WorkflowContext $context): void + { + if (!$this->belongsTo($context)) { + throw new LogicException('A child workflow handle can only be awaited by the workflow execution that started the child.'); + } + } +} diff --git a/src/Worker/ChildWorkflowIdentity.php b/src/Worker/ChildWorkflowIdentity.php new file mode 100644 index 00000000..96be67f6 --- /dev/null +++ b/src/Worker/ChildWorkflowIdentity.php @@ -0,0 +1,27 @@ +type === 'await_child_workflow') { + // A started child's await consumes no sequence: it reads the + // original call's resolution, through continuation and retries. + $awaited = $stepsBySequence[$suspended->attributes['sequence']] ?? null; + if (($awaited['shape'] ?? null) !== 'child_workflow') { + throw new NonDeterministicWorkflow('An awaited child workflow has no recorded child call.', + (int) $suspended->attributes['sequence'], reason: 'child_workflow_handle_mismatch'); + } + if ($awaited['resolved'] !== true) { + return $this->result($commands, $context); + } + $this->advanceCancellationClock($context, $history, $this->blockingResolutionOrders($awaited)); + $suspended = $awaited['failure'] instanceof Throwable + ? $execution->throw($awaited['failure']) + : $execution->resume($awaited['value']); + continue; + } + $step = $steps[$stepCursor] ?? null; if ($step !== null) { if ($strictSequences && $step['sequence'] !== $nextSequence) { @@ -748,6 +766,15 @@ public function replay( } ++$stepCursor; $nextSequence = max($nextSequence, $step['sequence'] + 1); + if ($suspended->type === 'start_child_workflow' && ($suspended->attributes['await'] ?? null) === 'started') { + $identity = $this->stepChildIdentity($step); + if ($identity === null) { + throw new NonDeterministicWorkflow('A committed child start lacks its child call, instance or run identity.', + $step['sequence'], reason: 'child_start_identity_missing'); + } + $suspended = $execution->resume(new ChildWorkflowHandle($context, $step['sequence'], $identity)); + continue; + } if ($suspended->type === 'open_condition_wait') { $this->assertConditionWaitCompatible($step, $suspended); $nextStep = $steps[$stepCursor] ?? null; @@ -1183,6 +1210,29 @@ private function isRootCancellationBoundary( return true; } + /** @param array $payload */ + private function childIdentity(array $payload): ?ChildWorkflowIdentity + { + $call = $payload['child_call_id'] ?? null; + $instance = $payload['child_workflow_instance_id'] ?? $payload['workflow_instance_id'] ?? null; + $run = $payload['child_workflow_run_id'] ?? $payload['workflow_run_id'] ?? null; + foreach ([$call, $instance, $run] as $value) { + if (!is_string($value) || trim($value) === '') { + return null; + } + } + + return new ChildWorkflowIdentity($call, $instance, $run); + } + + /** @param array $step */ + private function stepChildIdentity(array $step): ?ChildWorkflowIdentity + { + $identity = $step['child_identity'] ?? null; + + return $identity instanceof ChildWorkflowIdentity ? $identity : null; + } + /** @param array $step */ private function stepOccurrence(array $step): ?string { @@ -1376,6 +1426,7 @@ private function recordedSteps(array $history, bool $preserveConditionReopens, C $timerCleanup = []; $conditionStepsByWaitId = []; $conditionOpenedAt = []; + $childIdentities = []; $conditionOpenedOrder = []; $conditionOccurrences = []; $versionMarkerSequences = []; @@ -1634,6 +1685,11 @@ private function recordedSteps(array $history, bool $preserveConditionReopens, C $this->payloadDetail($payload, 'child_workflow'), parallelPath: $this->parallelPath($payload, $sequence), ); + // Preserve a missing initial identity too. A later continued + // or retried run cannot establish which run started the child. + if (!array_key_exists($key, $childIdentities)) { + $childIdentities[$key] = $this->childIdentity($payload); + } } elseif ($type === 'ChildRunCompleted') { if (($steps[$key]['resolved'] ?? false) === true) { continue; @@ -1875,6 +1931,11 @@ private function recordedSteps(array $history, bool $preserveConditionReopens, C $steps[$key]['occurrence_id'] = $conditionOccurrences[$key] ?? null; } } + foreach ($childIdentities as $key => $identity) { + if (($steps[$key]['shape'] ?? null) === 'child_workflow' && $identity !== null) { + $steps[$key]['child_identity'] = $identity; + } + } ksort($steps, SORT_NUMERIC); $steps = $this->bindConditionReopens(array_values($steps)); diff --git a/src/Worker/WorkflowCommand.php b/src/Worker/WorkflowCommand.php index 8bb1afe1..485e19d0 100644 --- a/src/Worker/WorkflowCommand.php +++ b/src/Worker/WorkflowCommand.php @@ -223,6 +223,9 @@ public function conditionSatisfied(): bool */ public static function childWorkflow(string $workflowType, array $arguments, array $options = []): self { + if (array_key_exists('await', $options)) { + throw new InvalidArgumentException('Child workflow await is chosen by the API: use startChildWorkflow() for a committed start handle.'); + } $policies = self::canonicalChildWorkflowPolicies($options); unset($options['parent_close_policy'], $options['cancellation_policy']); foreach (self::CHILD_WORKFLOW_TIMEOUT_FIELDS as $field) { @@ -237,6 +240,15 @@ public static function childWorkflow(string $workflowType, array $arguments, arr ])); } + /** + * @internal Waits on a started child's recorded resolution. The replayer + * resolves it from history; it is never sent to the runtime. + */ + public static function awaitChildWorkflow(int $sequence): self + { + return new self('await_child_workflow', 'child_workflow', ['sequence' => $sequence]); + } + /** * Accept typed policies and their portable string values without changing omitted defaults. * diff --git a/src/Worker/WorkflowContext.php b/src/Worker/WorkflowContext.php index 2568e302..c709a70d 100644 --- a/src/Worker/WorkflowContext.php +++ b/src/Worker/WorkflowContext.php @@ -451,6 +451,49 @@ public function childWorkflow(string $workflowType, array $arguments = [], array return $this->suspend($operation->command); } + /** + * Start a child workflow and return a handle once its start is committed. + * + * The current turn ends after the start command. The runtime commits + * ChildWorkflowScheduled and ChildRunStarted and gives the parent a fresh + * turn, so the handle's identity comes from canonical history while the + * child keeps running. Call await() on the handle for the child's result. + * Requires a runtime that advertises child_workflow_start_turn; the worker + * refuses the start before submitting it otherwise. Not available inside + * all()/parallel() groups. + * + * @param list $arguments + * @param array $options the same options as childWorkflow() + */ + public function startChildWorkflow(string $workflowType, array $arguments = [], array $options = []): ChildWorkflowHandle + { + $this->assertActiveFiber(); + if ($this->isCapturing()) { + throw new LogicException('startChildWorkflow() cannot run inside an all() or parallel() group. Use deferChildWorkflow() there.'); + } + $command = $this->withCancellationScope( + WorkflowCommand::childWorkflow($workflowType, $arguments, $options)->withAttributes(['await' => 'started']), + ); + $handle = $this->suspend($command); + if (!$handle instanceof ChildWorkflowHandle) { + throw new LogicException('A started child workflow must resume with its committed handle.'); + } + + return $handle; + } + + /** @internal Called by ChildWorkflowHandle::await(). */ + public function awaitChildWorkflow(ChildWorkflowHandle $handle): mixed + { + $this->assertActiveFiber(); + $handle->assertOwner($this); + if ($this->isCapturing()) { + throw new LogicException('A child workflow handle cannot be awaited inside an all() or parallel() group.'); + } + + return $this->suspend(WorkflowCommand::awaitChildWorkflow($handle->sequence)); + } + /** * Prepare an activity without scheduling it until an all/parallel barrier is reached. * diff --git a/tests/ChildWorkflowHandleTest.php b/tests/ChildWorkflowHandleTest.php new file mode 100644 index 00000000..7cb0c855 --- /dev/null +++ b/tests/ChildWorkflowHandleTest.php @@ -0,0 +1,270 @@ +codec = new AvroPayloadCodec(); + } + + public function testTheStartEndsTheTurnWithAStartedAwaitAndNoOtherCommand(): void + { + $commands = $this->replay([]); + + self::assertCount(1, $commands); + self::assertSame('start_child_workflow', $commands[0]['type']); + self::assertSame('billing.reconcile', $commands[0]['workflow_type']); + self::assertSame('started', $commands[0]['await']); + self::assertSame('abandon', $commands[0]['parent_close_policy']); + } + + public function testTheFreshTurnReturnsTheCommittedIdentityWhileTheChildRuns(): void + { + $commands = $this->replay($this->started()); + + self::assertCount(1, $commands); + self::assertSame('schedule_activity', $commands[0]['type']); + self::assertSame(['call-1', 'child-instance', 'child-run-1'], $this->codec->decodeEnvelope($commands[0]['arguments'])); + } + + public function testAwaitingAnOpenChildEndsTheTurnWithoutACommand(): void + { + self::assertSame([], $this->replay([...$this->started(), ...$this->notified()])); + } + + public function testAwaitReturnsTheChildResultWithoutANewSequence(): void + { + $commands = $this->replay([...$this->started(), ...$this->notified(), $this->resolved('ChildRunCompleted', ['result' => $this->codec->envelope('reconciled')])]); + + self::assertSame('complete_workflow', $commands[0]['type']); + self::assertSame('reconciled', $this->codec->decodeEnvelope($commands[0]['result'])); + } + + public function testAwaitThrowsTheChildFailure(): void + { + $workflow = static function (WorkflowContext $context): string { + $child = $context->startChildWorkflow('billing.reconcile', ['acct-1']); + try { + return (string) $child->await(); + } catch (ChildWorkflowFailed $failure) { + return 'failed: '.$failure->getMessage(); + } + }; + $commands = (new Replayer($this->codec))->replay($workflow, [ + ...$this->started(), + $this->resolved('ChildRunFailed', ['message' => 'Ledger mismatch.']), + ], [], 'php-workers')->commands; + + self::assertSame('failed: Ledger mismatch.', $this->codec->decodeEnvelope($commands[0]['result'])); + } + + public function testTheInitialRunStaysFixedThroughChildContinueAsNew(): void + { + $workflow = static function (WorkflowContext $context): array { + $child = $context->startChildWorkflow('billing.reconcile', ['acct-1']); + $result = $child->await(); + + return [$child->identity()->callId, $child->identity()->workflowId, $child->identity()->initialRunId, $result]; + }; + $continued = ['event_type' => 'ChildRunStarted', 'payload' => [ + 'sequence' => 1, 'child_call_id' => 'call-1', 'child_workflow_instance_id' => 'child-instance', 'child_workflow_run_id' => 'child-run-2', + ]]; + $commands = (new Replayer($this->codec))->replay($workflow, [ + ...$this->started(), $continued, + $this->resolved('ChildRunCompleted', ['child_workflow_run_id' => 'child-run-2', 'result' => $this->codec->envelope('done')]), + ], [], 'php-workers')->commands; + + self::assertSame(['call-1', 'child-instance', 'child-run-1', 'done'], $this->codec->decodeEnvelope($commands[0]['result'])); + } + + public function testColdReplayRebuildsTheSameHandleWithoutAnotherStart(): void + { + $history = [...$this->started(), ...$this->notified()]; + + self::assertSame($this->replay($history), $this->replay($history)); + self::assertNotContains('start_child_workflow', array_column($this->replay($history), 'type')); + } + + public function testACommittedStartWithoutIdentityIsRefused(): void + { + $history = $this->started(); + unset($history[0]['payload']['child_call_id'], $history[1]['payload']['child_call_id']); + + try { + $this->replay($history); + self::fail('A start without its committed identity must be refused.'); + } catch (NonDeterministicWorkflow $exception) { + self::assertSame('child_start_identity_missing', $exception->reason); + } + } + + public function testALaterRunCannotSupplyTheMissingInitialIdentity(): void + { + $history = $this->started(); + unset($history[0]['payload']['child_call_id'], $history[1]['payload']['child_call_id']); + $history[] = ['event_type' => 'ChildRunStarted', 'payload' => [ + 'sequence' => 1, 'child_call_id' => 'call-1', 'child_workflow_instance_id' => 'child-instance', + 'child_workflow_run_id' => 'child-run-2', 'child_workflow_type' => 'billing.reconcile', + ]]; + + try { + $this->replay($history); + self::fail('A later run must not become the initial child identity.'); + } catch (NonDeterministicWorkflow $exception) { + self::assertSame('child_start_identity_missing', $exception->reason); + } + } + + public function testTheExistingChildCallKeepsReturningTheResult(): void + { + $workflow = static fn (WorkflowContext $context): mixed => $context->childWorkflow('billing.reconcile', ['acct-1']); + $commands = (new Replayer($this->codec))->replay($workflow, [], [], 'php-workers')->commands; + + self::assertSame('start_child_workflow', $commands[0]['type']); + self::assertArrayNotHasKey('await', $commands[0]); + } + + public function testAnAwaitOptionIsNotAcceptedOnTheResultCall(): void + { + $workflow = static fn (WorkflowContext $context): mixed => $context->childWorkflow('billing.reconcile', [], ['await' => 'started']); + + $this->expectException(InvalidArgumentException::class); + $this->expectExceptionMessage('use startChildWorkflow()'); + + (new Replayer($this->codec))->replay($workflow, [], [], 'php-workers'); + } + + public function testAStartedChildCannotJoinAGroup(): void + { + $workflow = static fn (WorkflowContext $context): array => $context->all([ + static fn (): ChildWorkflowHandle => $context->startChildWorkflow('billing.reconcile'), + ]); + + $this->expectException(LogicException::class); + $this->expectExceptionMessage('deferChildWorkflow()'); + + (new Replayer($this->codec))->replay($workflow, [], [], 'php-workers'); + } + + /** + * A runtime without child_workflow_start_turn ignores the await field and + * gives no parent turn, so the worker refuses before submitting the start. + */ + public function testTheWorkerSubmitsAStartedChildOnlyWhenTheRuntimeAdvertisesTheTurn(): void + { + foreach ([false, true] as $supported) { + $commands = null; + $failure = null; + $transport = new FakeTransport(handler: static function ( + string $method, string $uri, array $headers, ?array $body, + ) use ($supported, &$commands, &$failure): array { + if (str_ends_with($uri, '/workflow-tasks/poll')) { + return ['server_capabilities' => ['child_workflow_start_turn' => $supported], + 'poll_status' => 'leased', 'task' => ['task_id' => 'parent-task', 'workflow_task_attempt' => 1, + 'lease_owner' => 'worker-1', 'workflow_id' => 'parent', 'run_id' => 'parent-run', + 'workflow_type' => 'billing.parent', 'payload_codec' => 'avro', 'history_events' => []]]; + } + if (str_ends_with($uri, '/parent-task/heartbeat')) { + return ['task_id' => 'parent-task', 'workflow_task_attempt' => 1, 'lease_owner' => 'worker-1', 'renewed' => true]; + } + if (str_ends_with($uri, '/parent-task/complete')) { + $commands = $body['commands']; + + return ['completed' => true]; + } + if (str_ends_with($uri, '/parent-task/fail')) { + $failure = $body['failure']['message'] ?? null; + + return ['recorded' => true]; + } + if (str_ends_with($uri, '/activity-tasks/poll')) { + return ['task' => null, 'poll_status' => 'stopped', 'reason' => 'worker_stopped']; + } + self::fail("Unexpected request {$method} {$uri}"); + }); + $worker = new Worker(new Client('https://runtime.test', transport: $transport), 'parents', workerId: 'worker-1'); + $worker->registerWorkflow('billing.parent', static fn (WorkflowContext $context): mixed => + $context->startChildWorkflow('billing.reconcile', ['acct-1'])->await()); + + self::assertTrue($worker->tick(0)); + if ($supported) { + self::assertNull($failure); + self::assertSame('started', $commands[0]['await'] ?? null); + } else { + self::assertNull($commands, 'nothing is submitted without the advertised turn'); + self::assertStringContainsString('child_workflow_start_turn_unavailable', (string) $failure); + } + } + } + + /** @param list> $history + * @return list> + */ + private function replay(array $history): array + { + $workflow = static function (WorkflowContext $context): mixed { + $child = $context->startChildWorkflow('billing.reconcile', ['acct-1'], [ + 'parent_close_policy' => ParentClosePolicy::Abandon, + ]); + $identity = $child->identity(); + $context->activity('notify.child-started', [$identity->callId, $identity->workflowId, $identity->initialRunId]); + + return $child->await(); + }; + + return (new Replayer($this->codec))->replay($workflow, $history, [], 'php-workers')->commands; + } + + /** @return list> */ + private function started(): array + { + $identity = ['sequence' => 1, 'child_call_id' => 'call-1', 'workflow_link_id' => 'call-1', + 'child_workflow_instance_id' => 'child-instance', 'child_workflow_run_id' => 'child-run-1', 'child_workflow_type' => 'billing.reconcile']; + + return [ + ['event_type' => 'ChildWorkflowScheduled', 'payload' => $identity], + ['event_type' => 'ChildRunStarted', 'payload' => $identity], + ]; + } + + /** @return list> */ + private function notified(): array + { + return [ + ['event_type' => 'ActivityScheduled', 'payload' => ['sequence' => 2, 'activity_type' => 'notify.child-started', 'activity_execution_id' => 'act-2']], + ['event_type' => 'ActivityCompleted', 'payload' => ['sequence' => 2, 'activity_type' => 'notify.child-started', 'activity_execution_id' => 'act-2', 'result' => $this->codec->envelope(null)]], + ]; + } + + /** @param array $payload + * @return array + */ + private function resolved(string $type, array $payload): array + { + return ['event_type' => $type, 'payload' => [ + 'sequence' => 1, 'child_call_id' => 'call-1', 'child_workflow_instance_id' => 'child-instance', + 'child_workflow_run_id' => 'child-run-1', 'child_workflow_type' => 'billing.reconcile', ...$payload, + ]]; + } +} diff --git a/tests/Support/ReplayRegressionFixture.php b/tests/Support/ReplayRegressionFixture.php index f1619fa3..38785c02 100644 --- a/tests/Support/ReplayRegressionFixture.php +++ b/tests/Support/ReplayRegressionFixture.php @@ -432,6 +432,15 @@ private static function workflow(string $workflowType): callable return ['child' => $result]; }, + 'golden.child-start-handle' => static function ( + WorkflowContext $context, + mixed $workflowType, + ): array { + $child = $context->startChildWorkflow((string) $workflowType, ['golden-input']); + $identity = $child->identity(); + + return ['identity' => [$identity->callId, $identity->workflowId, $identity->initialRunId], 'child' => $child->await()]; + }, 'golden.side-effect' => static function (WorkflowContext $context, mixed $value): array { $result = $context->sideEffect(static fn (): mixed => $value); diff --git a/tests/fixtures/replay-regressions/child-start-handle-identity-and-result.json b/tests/fixtures/replay-regressions/child-start-handle-identity-and-result.json new file mode 100644 index 00000000..8ea3f1ca --- /dev/null +++ b/tests/fixtures/replay-regressions/child-start-handle-identity-and-result.json @@ -0,0 +1,67 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "php-child-start-handle-identity-and-result", + "protocol_version": "1.20", + "bindings": ["php"], + "workflow": { + "type": "golden.child-start-handle", + "input": ["billing.reconcile"] + }, + "history": [ + { + "event_type": "ChildWorkflowScheduled", + "payload": { + "sequence": 1, + "workflow_link_id": "call-1", + "child_call_id": "call-1", + "child_workflow_instance_id": "child-instance", + "child_workflow_run_id": "child-run-1", + "child_workflow_type": "billing.reconcile" + } + }, + { + "event_type": "ChildRunStarted", + "payload": { + "sequence": 1, + "workflow_link_id": "call-1", + "child_call_id": "call-1", + "child_workflow_instance_id": "child-instance", + "child_workflow_run_id": "child-run-1", + "child_workflow_type": "billing.reconcile" + } + }, + { + "event_type": "ChildRunStarted", + "payload": { + "sequence": 1, + "workflow_link_id": "link-2", + "child_call_id": "call-1", + "child_workflow_instance_id": "child-instance", + "child_workflow_run_id": "child-run-2", + "child_workflow_type": "billing.reconcile" + } + }, + { + "event_type": "ChildRunCompleted", + "payload": { + "sequence": 1, + "child_call_id": "call-1", + "child_workflow_instance_id": "child-instance", + "child_workflow_run_id": "child-run-2", + "child_workflow_type": "billing.reconcile", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoUcmVjb25jaWxlZA==" + } + } + } + ], + "expected": { + "type": "complete_workflow", + "result": { + "identity": ["call-1", "child-instance", "child-run-1"], + "child": "reconciled" + } + } +} diff --git a/tests/fixtures/replay-regressions/child-start-missing-initial-identity.json b/tests/fixtures/replay-regressions/child-start-missing-initial-identity.json new file mode 100644 index 00000000..c241ffbf --- /dev/null +++ b/tests/fixtures/replay-regressions/child-start-missing-initial-identity.json @@ -0,0 +1,31 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "php-child-start-missing-initial-identity", + "protocol_version": "1.20", + "bindings": ["php"], + "workflow": {"type": "golden.child-start-handle", "input": ["billing.reconcile"]}, + "history": [ + {"event_type": "ChildWorkflowScheduled", "payload": { + "sequence": 1, "workflow_link_id": "call-1", + "child_workflow_instance_id": "child-instance", "child_workflow_run_id": "child-run-1", + "child_workflow_type": "billing.reconcile" + }}, + {"event_type": "ChildRunStarted", "payload": { + "sequence": 1, "workflow_link_id": "call-1", + "child_workflow_instance_id": "child-instance", "child_workflow_run_id": "child-run-1", + "child_workflow_type": "billing.reconcile" + }}, + {"event_type": "ChildRunStarted", "payload": { + "sequence": 1, "workflow_link_id": "link-2", "child_call_id": "call-1", + "child_workflow_instance_id": "child-instance", "child_workflow_run_id": "child-run-2", + "child_workflow_type": "billing.reconcile" + }}, + {"event_type": "ChildRunCompleted", "payload": { + "sequence": 1, "child_call_id": "call-1", + "child_workflow_instance_id": "child-instance", "child_workflow_run_id": "child-run-2", + "child_workflow_type": "billing.reconcile" + }} + ], + "expected": {"command_sequence": [{"type": "replay_error", "reason": "child_start_identity_missing", "sequence": 1}]} +}