diff --git a/CHANGELOG.md b/CHANGELOG.md index 642288c7d..be73f2fdc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ ## Unreleased +- Experimental embedded child-start milestone: `Workflow::startChild()` returns + a versioned `StartedChildWorkflow` receipt after native child admission, and + `$child->await()` observes that same child later. Cold replay reuses the + original admission; a queued request alone cannot supply a receipt. The + existing `child()` still waits for completion. This adds a + `ChildWorkflowAwaited` history event and an optional + `child_completion_mode` on `ChildWorkflowScheduled`, with no database + migration. Worker-protocol execution refuses the new calls. State-bound + business bindings and retained compensation remain separate #1069 work. + - Require patched Laravel 12.72 or 13.41 minimums for the current Composer security advisories. Keep existing application audits enabled and update application lockfiles before upgrading Workflow (#1133). diff --git a/docs/business-state-machines.md b/docs/business-state-machines.md index 6bb9b8410..8334153cc 100644 --- a/docs/business-state-machines.md +++ b/docs/business-state-machines.md @@ -177,7 +177,26 @@ The next nested slice must preserve the same four methods and add: - Verified child compensation before earlier parent compensation. Technical child completion, accepted signals and timeouts are insufficient evidence. - Acyclic bounded binding graphs, resource limits and explicit adoption ownership. -Existing `child()` can be called from a step and durably await a child result. That does not yet implement the separate state-bound start milestone. This slice deliberately adds no alternate child scheduler or compensation engine. +Existing `child()` can be called from a step and durably await a child result. The separate native admission primitive below supplies a confirmed start milestone; it does not itself implement the state-bound binding protocol. Both paths use the existing child scheduler and history. + +## Confirmed native child admission + +Embedded V2 workflows can separate native child admission from waiting for its result: + +```php +$child = self::startChild(ChildMachineWorkflow::class, $input); +self::activity(RecordChildAdmission::class, $child->toArray()); +// Continue the parent, then observe this exact child when its result is needed. +$result = $child->await(); +``` + +The child class and activity above are application implementations. `startChild()` accepts the same optional `ChildWorkflowOptions` as `child()`. The child may still be pending execution when the receipt returns: its native instance, run, lineage and task have been admitted, rather than merely queued as a request to create them. A cold duplicate resumes with the same admission. The later await creates neither another child nor another compensation owner, and works whether the child completes before or after the await. + +`StartedChildWorkflow::toArray()` is the version-1 serialized receipt. It binds namespace, original parent instance/run, owning start sequence, child instance, initial child run, child-call ID and workflow type. `fromArray()` validates its shape; the runtime separately checks it against that parent's paired schedule/start history before awaiting. Modified, foreign-parent and missing-admission receipts refuse. Awaiting is scoped to the original parent execution. Query replay uses committed parent resolution history, not a live terminal row as an unrecorded result. + +Native retries and continue-as-new remain attached only through the recorded descendant chain rooted at that receipt. A later current run in the same instance does not authorize substitution. Await cancellation records both its own workflow step and the original child operation, so cold replay preserves the selected child cancellation policy. + +This primitive is experimental and embedded-only. Worker-protocol execution refuses these calls. It preserves ordinary `child()` terminal-wait behavior and parent-close/cancellation policies. `RecordChildAdmission` still owns application authorization, expected state/revision checks and a protected idempotent business commit; a receipt does not implement those boundaries. The scope/slot/generation binding, policy pinning, adoption ownership, business-outcome verification and retained recovery in the next section remain required before claiming nested business-machine support. ## Follow-up implementation sequence diff --git a/src/V2/Enums/HistoryEventType.php b/src/V2/Enums/HistoryEventType.php index dbf2141aa..2cbac6feb 100644 --- a/src/V2/Enums/HistoryEventType.php +++ b/src/V2/Enums/HistoryEventType.php @@ -11,6 +11,7 @@ enum HistoryEventType: string case WorkflowStarted = 'WorkflowStarted'; case WorkflowContinuedAsNew = 'WorkflowContinuedAsNew'; case ChildWorkflowScheduled = 'ChildWorkflowScheduled'; + case ChildWorkflowAwaited = 'ChildWorkflowAwaited'; case ChildCancellationRequested = 'ChildCancellationRequested'; case ChildCancellationResolved = 'ChildCancellationResolved'; case ChildRunStarted = 'ChildRunStarted'; diff --git a/src/V2/StartedChildWorkflow.php b/src/V2/StartedChildWorkflow.php new file mode 100644 index 000000000..74405c7a5 --- /dev/null +++ b/src/V2/StartedChildWorkflow.php @@ -0,0 +1,77 @@ + + */ + public function toArray(): array + { + return [ + 'version' => 1, + ...get_object_vars($this), + ]; + } + + /** + * @param array $value + */ + public static function fromArray(array $value): self + { + $strings = ['namespace', 'parentInstanceId', 'parentRunId', 'childInstanceId', + 'childRunId', 'childCallId', 'workflowType']; + if (($value['version'] ?? null) !== 1 || ! is_int($value['startSequence'] ?? null) + || count($value) !== count($strings) + 2) { + throw new InvalidArgumentException('Unsupported or malformed child admission.'); + } + foreach ($strings as $key) { + if (! is_string($value[$key] ?? null)) { + throw new InvalidArgumentException('Child admission identifiers must be strings.'); + } + } + return new self( + $value['namespace'], + $value['parentInstanceId'], + $value['parentRunId'], + $value['startSequence'], + $value['childInstanceId'], + $value['childRunId'], + $value['childCallId'], + $value['workflowType'] + ); + } +} diff --git a/src/V2/Support/AwaitStartedChildWorkflowCall.php b/src/V2/Support/AwaitStartedChildWorkflowCall.php new file mode 100644 index 000000000..523d061c6 --- /dev/null +++ b/src/V2/Support/AwaitStartedChildWorkflowCall.php @@ -0,0 +1,16 @@ +historyEvents->first(static fn (WorkflowHistoryEvent $event): bool => + $event->event_type === HistoryEventType::ChildRunStarted + && ($event->payload['sequence'] ?? null) === $sequence); + if ($scheduled === null && $started === null) { + return null; + } + if ($scheduled === null || $started === null + || ($scheduled->payload['child_completion_mode'] ?? null) !== 'started') { + throw self::mismatch($sequence, 'A confirmed start needs its original schedule and native admission.'); + } + foreach (['child_workflow_instance_id', 'child_workflow_run_id', 'child_call_id', + 'child_workflow_type'] as $key) { + if (! is_string($scheduled->payload[$key] ?? null) || $scheduled->payload[$key] === '' + || ($started->payload[$key] ?? null) !== $scheduled->payload[$key]) { + throw self::mismatch($sequence, 'The native child admission differs from its owning schedule.'); + } + } + if ($scheduled->workflow_run_id !== $run->id || $started->workflow_run_id !== $run->id) { + throw self::mismatch($sequence, 'Child admission history belongs to another parent.'); + } + return new StartedChildWorkflow( + (string) $run->namespace, + $run->workflow_instance_id, + $run->id, + $sequence, + $scheduled->payload['child_workflow_instance_id'], + $scheduled->payload['child_workflow_run_id'], + $scheduled->payload['child_call_id'], + $scheduled->payload['child_workflow_type'] + ); + } + + public static function assertReceipt(WorkflowRun $run, StartedChildWorkflow $child, int $sequence): void + { + $recorded = self::receipt($run, $child->startSequence); + if ($recorded === null || $recorded->toArray() !== $child->toArray() + || $child->startSequence >= $sequence) { + throw self::mismatch($sequence, 'The child receipt is not an earlier admission of this parent execution.'); + } + } + + public static function cancellationDelivery( + WorkflowRun $run, + StartedChildWorkflow $child, + int $sequence, + ): ?WorkflowHistoryEvent { + $event = CooperativeCancellationDelivery::recorded($run); + if ($event === null || ($event->payload['sequence'] ?? null) !== $sequence) { + return null; + } + $invalid = CooperativeCancellationDelivery::validate( + $run, + (string) $run->cancellation_request_command_id, + $sequence, + 'started_child_await', + operationSequence: $child->startSequence + ); + if ($invalid !== null) { + throw self::mismatch($sequence, $invalid); + } + return $event; + } + + public static function awaitedEvent( + WorkflowRun $run, + StartedChildWorkflow $child, + int $sequence, + ): ?WorkflowHistoryEvent { + $event = $run->historyEvents->first(static fn (WorkflowHistoryEvent $event): bool => + $event->event_type === HistoryEventType::ChildWorkflowAwaited + && ($event->payload['sequence'] ?? null) === $sequence); + if ($event !== null) { + foreach (self::awaitPayload($child, $sequence) as $key => $value) { + if (($event->payload[$key] ?? null) !== $value) { + throw self::mismatch($sequence, 'The await step belongs to a different child admission.'); + } + } + } + return $event; + } + + /** + * @return array + */ + public static function awaitPayload(StartedChildWorkflow $child, int $sequence): array + { + return [ + 'sequence' => $sequence, + 'receipt' => $child->toArray(), + 'child_workflow_instance_id' => $child->childInstanceId, + 'child_workflow_run_id' => $child->childRunId, + 'child_call_id' => $child->childCallId, + 'child_workflow_type' => $child->workflowType, + ]; + } + + /** + * Resolve only the recorded chain, never mutable current-run order. + */ + public static function target( + WorkflowRun $parent, + StartedChildWorkflow $child, + ?WorkflowHistoryEvent $resolution, + int $sequence, + ): ?WorkflowRun { + $starts = $parent->historyEvents->filter(static fn (WorkflowHistoryEvent $event): bool => + $event->event_type === HistoryEventType::ChildRunStarted + && ($event->payload['sequence'] ?? null) === $child->startSequence)->sortBy('sequence'); + $previous = null; + $seen = []; + foreach ($starts as $event) { + $id = self::recordedRunId($parent, $child, $event, $seen, $sequence); + /** @var WorkflowRun|null $target */ + $target = ConfiguredV2Models::query('run_model', WorkflowRun::class) + ->where('namespace', $child->namespace) + ->where('workflow_instance_id', $child->childInstanceId) + ->where('workflow_type', $child->workflowType) + ->find($id); + if ($target === null) { + // A complete parent outcome can outlive the initial child's + // detail row. It cannot invent a missing descendant chain. + if ($id === $child->childRunId && $starts->count() === 1 && $resolution !== null) { + self::assertResolution($child, $resolution, null, $sequence); + return null; + } + throw self::mismatch($sequence, 'The admitted child lineage is unavailable.'); + } + $target->setRelation('historyEvents', $target->historyEvents()->whereIn('event_type', [ + HistoryEventType::WorkflowStarted, HistoryEventType::WorkflowContinuedAsNew, + HistoryEventType::WorkflowCompleted, HistoryEventType::WorkflowFailed, + HistoryEventType::WorkflowCancelled, HistoryEventType::WorkflowTerminated, + ])->get()); + if ($previous !== null) { + self::assertSuccessor($parent, $child, $previous, $target, $event, $sequence); + } + $seen[$id] = true; + $previous = $target; + } + if ($previous === null || $previous->closed_reason === 'continued') { + throw self::mismatch($sequence, 'The admitted child has no complete recorded continuation chain.'); + } + self::assertResolution($child, $resolution, $previous, $sequence); + return $previous; + } + + public static function assertResolution( + StartedChildWorkflow $child, + ?WorkflowHistoryEvent $resolution, + ?WorkflowRun $target, + int $sequence, + ): void { + if ($target !== null && ($target->namespace !== $child->namespace + || $target->workflow_instance_id !== $child->childInstanceId + || $target->workflow_type !== $child->workflowType)) { + throw self::mismatch($sequence, 'The child target escaped its admitted namespace or instance.'); + } + if ($resolution !== null && (($resolution->payload['child_workflow_instance_id'] ?? null) + !== $child->childInstanceId || ($resolution->payload['child_call_id'] ?? null) !== $child->childCallId + || ($resolution->payload['child_workflow_run_id'] ?? null) !== ($target?->id ?? $child->childRunId) + || ($resolution->payload['child_workflow_type'] ?? null) !== $child->workflowType)) { + throw self::mismatch($sequence, 'The terminal outcome belongs to a different child admission.'); + } + } + + private static function recordedRunId( + WorkflowRun $parent, + StartedChildWorkflow $child, + WorkflowHistoryEvent $event, + array $seen, + int $sequence, + ): string { + $id = $event->payload['child_workflow_run_id'] ?? null; + if (! is_string($id) || isset($seen[$id]) + || $event->workflow_run_id !== $parent->id + || ($event->payload['child_call_id'] ?? null) !== $child->childCallId + || ($event->payload['child_workflow_instance_id'] ?? null) !== $child->childInstanceId + || ($event->payload['child_workflow_type'] ?? null) !== $child->workflowType) { + throw self::mismatch($sequence, 'The child admission chain is not linear or has substituted identity.'); + } + if ($seen === [] && $id !== $child->childRunId) { + throw self::mismatch($sequence, 'The child chain does not begin at the admitted run.'); + } + return $id; + } + + private static function assertSuccessor( + WorkflowRun $parent, + StartedChildWorkflow $child, + WorkflowRun $predecessor, + WorkflowRun $successor, + WorkflowHistoryEvent $parentStart, + int $sequence, + ): void { + $starts = $successor->historyEvents->filter(static fn (WorkflowHistoryEvent $event): bool => + $event->event_type === HistoryEventType::WorkflowStarted); + $started = $starts->first(); + $payload = $started?->payload ?? []; + if ($starts->count() !== 1 || ! is_string($payload['workflow_link_id'] ?? null) + || $payload['workflow_link_id'] === '' + || ! is_string($parentStart->payload['workflow_link_id'] ?? null) + || $parentStart->payload['workflow_link_id'] === '' + || ($payload['workflow_instance_id'] ?? null) !== $child->childInstanceId + || ($payload['workflow_type'] ?? null) !== $child->workflowType + || ($payload['workflow_run_id'] ?? null) !== $successor->id + || ($payload['parent_workflow_instance_id'] ?? null) !== $parent->workflow_instance_id + || ($payload['parent_workflow_run_id'] ?? null) !== $parent->id + || ($payload['parent_sequence'] ?? null) !== $child->startSequence + || ($payload['child_call_id'] ?? null) !== $child->childCallId) { + throw self::mismatch($sequence, 'The successor start is not bound to the admitted child call.'); + } + $retry = ($payload['retry_of_child_workflow_run_id'] ?? null) === $predecessor->id + && $parentStart->payload['workflow_link_id'] === $payload['workflow_link_id'] + && ($payload['continued_from_run_id'] ?? null) === null + && ($payload['recovery_kind'] ?? null) !== 'redrive' + && ($parentStart->payload['retry_of_child_workflow_run_id'] ?? null) === $predecessor->id + && $predecessor->historyEvents->contains(static fn (WorkflowHistoryEvent $event): bool => + $event->event_type === HistoryEventType::WorkflowFailed); + $continued = ($payload['continued_from_run_id'] ?? null) === $predecessor->id + && ($payload['retry_of_child_workflow_run_id'] ?? null) === null + && ($payload['recovery_kind'] ?? null) !== 'redrive' + && $predecessor->historyEvents->contains(static fn (WorkflowHistoryEvent $event): bool => + $event->event_type === HistoryEventType::WorkflowContinuedAsNew + && ($event->payload['continued_to_run_id'] ?? null) === $successor->id + && ($event->payload['workflow_link_id'] ?? null) === ($payload['workflow_link_id'] ?? null)); + if (! $retry && ! $continued) { + throw self::mismatch($sequence, 'The successor lacks the exact typed retry or continuation edge.'); + } + } + + private static function mismatch(int $sequence, string $detail): HistoryEventShapeMismatchException + { + return new HistoryEventShapeMismatchException( + $sequence, + 'confirmed child admission', + [HistoryEventType::ChildRunStarted->value], + $detail + ); + } +} diff --git a/src/V2/Support/CooperativeCancellationDelivery.php b/src/V2/Support/CooperativeCancellationDelivery.php index 1f898ec95..7c6bb5909 100644 --- a/src/V2/Support/CooperativeCancellationDelivery.php +++ b/src/V2/Support/CooperativeCancellationDelivery.php @@ -10,6 +10,7 @@ use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowTask; +use Workflow\V2\StartedChildWorkflow; /** * Canonical cancellation delivery shared by embedded and portable execution. @@ -24,6 +25,7 @@ final class CooperativeCancellationDelivery 'condition' => WorkflowStepHistory::CONDITION_WAIT, 'signal' => WorkflowStepHistory::SIGNAL_WAIT, 'child' => WorkflowStepHistory::CHILD_WORKFLOW, + 'started_child_await' => WorkflowStepHistory::CHILD_WORKFLOW_AWAIT, 'service_operation' => WorkflowStepHistory::SERVICE_OPERATION, ]; @@ -94,6 +96,11 @@ public static function validate( return $membership; } + $boundary = self::validateStartedChildBoundary($run, $request, $sequence, $callKind, $operationSequence); + if ($boundary !== null) { + return $boundary; + } + $existing = self::recorded($run); if ($existing instanceof WorkflowHistoryEvent) { return ($existing->workflow_command_id === $requestId @@ -150,7 +157,7 @@ public static function validateBoundarySyntax( || ($limit > 0 && $operationSequenceSpan > $limit) || (! isset(self::CALL_SHAPES[$callKind]) && ! in_array($callKind, ['parallel', 'selection_handle'], true)) || ($callKind !== 'parallel' && $sequenceSpan !== 1) - || ($callKind === 'selection_handle') !== ($operationSequence !== null) + || in_array($callKind, ['selection_handle', 'started_child_await'], true) !== ($operationSequence !== null) || ($callKind !== 'selection_handle' && $operationSequenceSpan !== 1)) ? 'invalid_cancellation_delivery' : null; } @@ -176,6 +183,36 @@ public static function validateCallBoundary( return 'cancellation_delivery_sequence_mismatch'; } + if ($callKind === 'started_child_await') { + if ($operationSequence === null || $operationSequence < 1 || $operationSequence >= $sequence) { + return 'cancellation_delivery_sequence_mismatch'; + } + try { + WorkflowStepHistory::assertCompatible($run, $sequence, WorkflowStepHistory::CHILD_WORKFLOW_AWAIT); + WorkflowStepHistory::assertTypedHistoryRecorded( + $run, + $sequence, + WorkflowStepHistory::CHILD_WORKFLOW_AWAIT + ); + $event = $run->historyEvents->first(static fn (WorkflowHistoryEvent $event): bool => + $event->event_type === HistoryEventType::ChildWorkflowAwaited + && ($event->payload['sequence'] ?? null) === $sequence); + if (! is_array($event?->payload['receipt'] ?? null)) { + return 'cancellation_delivery_shape_mismatch'; + } + $receipt = StartedChildWorkflow::fromArray($event->payload['receipt']); + ChildStartHistory::assertReceipt($run, $receipt, $sequence); + ChildStartHistory::awaitedEvent($run, $receipt, $sequence); + if ($receipt->startSequence !== $operationSequence) { + return 'cancellation_delivery_sequence_mismatch'; + } + } catch (\InvalidArgumentException|HistoryEventShapeMismatchException $error) { + return 'cancellation_delivery_shape_mismatch'; + } + return self::resolvedBeforeRequest($run, $operationSequence, $request) + ? 'cancellation_delivery_not_eligible' : null; + } + if ($callKind === 'selection_handle') { if ($sequence !== $nextSequence || $operationSequence === null || $operationSequence < 1 || $operationSequence >= $nextSequence @@ -307,6 +344,7 @@ public static function callKindAt(WorkflowRun $run, int $sequence): ?string HistoryEventType::ConditionWaitOpened => 'condition', HistoryEventType::SignalWaitOpened => 'signal', HistoryEventType::ChildWorkflowScheduled => 'child', + HistoryEventType::ChildWorkflowAwaited => 'started_child_await', HistoryEventType::ServiceOperationScheduled, HistoryEventType::ServiceCallStarted => 'service_operation', HistoryEventType::TimerScheduled => match ($event->payload['timer_kind'] ?? null) { @@ -324,6 +362,25 @@ public static function callKindAt(WorkflowRun $run, int $sequence): ?string return null; } + private static function validateStartedChildBoundary( + WorkflowRun $run, + WorkflowHistoryEvent $request, + int $sequence, + string $callKind, + ?int $operationSequence, + ): ?string { + // Syntax already restricts this call to one await and one child start. + return $callKind === 'started_child_await' + ? self::validateCallBoundary( + $run, + $request, + $sequence, + $callKind, + operationSequence: $operationSequence + ) + : null; + } + /** * Root delivery actors have no frozen scoped inventory. Recorded scoped * operations must use scope preparation and delivery, including a group diff --git a/src/V2/Support/HistoryEventPayloadContract.php b/src/V2/Support/HistoryEventPayloadContract.php index c33b856ae..236e3aa7a 100644 --- a/src/V2/Support/HistoryEventPayloadContract.php +++ b/src/V2/Support/HistoryEventPayloadContract.php @@ -641,6 +641,7 @@ final class HistoryEventPayloadContract ], 'ChildWorkflowScheduled' => [ 'sequence', + 'child_completion_mode', 'workflow_link_id', 'child_call_id', 'child_workflow_instance_id', @@ -662,6 +663,8 @@ final class HistoryEventPayloadContract 'reused_from_run_id', 'reused_recorded_at', ], + 'ChildWorkflowAwaited' => ['sequence', 'receipt', 'child_workflow_instance_id', + 'child_workflow_run_id', 'child_call_id', 'child_workflow_type'], 'ChildRunStarted' => [ 'sequence', 'workflow_link_id', diff --git a/src/V2/Support/HistoryTimeline.php b/src/V2/Support/HistoryTimeline.php index e2dff29be..897e64ce5 100644 --- a/src/V2/Support/HistoryTimeline.php +++ b/src/V2/Support/HistoryTimeline.php @@ -429,6 +429,7 @@ private static function kindFor(HistoryEventType $eventType): string HistoryEventType::UpdateApplied, HistoryEventType::UpdateCompleted => 'update', HistoryEventType::ChildWorkflowScheduled, + HistoryEventType::ChildWorkflowAwaited, HistoryEventType::ChildRunStarted, HistoryEventType::ChildRunCompleted, HistoryEventType::ChildRunFailed, @@ -530,6 +531,7 @@ private static function summaryFor( ?? (string) (self::intValue($payload['member_index'] ?? null) ?? 0), ), HistoryEventType::ChildWorkflowScheduled => sprintf('Scheduled child workflow %s.', $childLabel), + HistoryEventType::ChildWorkflowAwaited => sprintf('Awaited admitted child workflow %s.', $childLabel), HistoryEventType::ChildRunStarted => sprintf('Child workflow %s started.', $childLabel), HistoryEventType::ChildRunCompleted => sprintf('Child workflow %s completed.', $childLabel), HistoryEventType::ChildRunFailed => $message === null @@ -1286,6 +1288,7 @@ private static function sourceKindFor(WorkflowHistoryEvent $event): string HistoryEventType::SignalWaitOpened, HistoryEventType::SignalApplied => 'signal_wait', HistoryEventType::ChildWorkflowScheduled, + HistoryEventType::ChildWorkflowAwaited, HistoryEventType::ChildRunStarted, HistoryEventType::ChildRunCompleted, HistoryEventType::ChildRunFailed, diff --git a/src/V2/Support/QueryStateReplayer.php b/src/V2/Support/QueryStateReplayer.php index 9d89c857b..08a8bdd7a 100644 --- a/src/V2/Support/QueryStateReplayer.php +++ b/src/V2/Support/QueryStateReplayer.php @@ -523,12 +523,80 @@ public function replayState(WorkflowRun $run): ReplayState continue; } + if ($current instanceof StartChildWorkflowCall) { + $historySequence = $this->historySequenceForReplayPosition($historySequencesByPosition, $sequence); + $this->applyRecordedUpdates($run, $workflow, $historySequence); + WorkflowStepHistory::assertCompatible($run, $historySequence, WorkflowStepHistory::CHILD_WORKFLOW, [ + 'child_workflow_type' => $current->child->workflow, + 'child_completion_mode' => 'started', + ]); + $receipt = ChildStartHistory::receipt($run, $historySequence); + if ($receipt === null) { + $this->syncWorkflowCursor($workflow, $sequence); + return new ReplayState($workflow, $sequence, $current); + } + $this->syncWorkflowCursor($workflow, $sequence + 1); + $current = $workflowExecution->send($receipt, RedriveReplayClock::recordedAt( + ChildRunHistory::scheduledEventForSequence($run, $historySequence) + )); + ++$sequence; + continue; + } + + if ($current instanceof AwaitStartedChildWorkflowCall) { + $historySequence = $this->historySequenceForReplayPosition($historySequencesByPosition, $sequence); + $this->applyRecordedUpdates($run, $workflow, $historySequence); + WorkflowStepHistory::assertCompatible( + $run, + $historySequence, + WorkflowStepHistory::CHILD_WORKFLOW_AWAIT + ); + $child = $current->child; + ChildStartHistory::assertReceipt($run, $child, $historySequence); + $awaited = ChildStartHistory::awaitedEvent($run, $child, $historySequence); + $cancelled = ChildStartHistory::cancellationDelivery($run, $child, $historySequence); + if ($cancelled !== null) { + $context = CooperativeCancellationDelivery::context($run); + if ($context === null) { + throw new \LogicException('The child-await cancellation has no canonical context.'); + } + $this->syncWorkflowCursor($workflow, $sequence + 1); + $current = $workflowExecution->throw( + \Workflow\V2\Exceptions\WorkflowCancellationRequestedException::forContext($context), + $cancelled->recorded_at + ); + ++$sequence; + continue; + } + $resolution = ChildRunHistory::resolutionEventForSequence($run, $child->startSequence); + $target = ChildStartHistory::target($run, $child, $resolution, $historySequence); + // Live terminal rows do not authorize query replay before the + // parent's typed resolution commits. + if ($awaited === null || $resolution === null) { + $this->syncWorkflowCursor($workflow, $sequence); + return new ReplayState($workflow, $sequence, $current); + } + $this->syncWorkflowCursor($workflow, $sequence + 1); + $current = $resolution->event_type === HistoryEventType::ChildRunCompleted + ? $workflowExecution->send( + ChildRunHistory::outputForResolution($resolution, $target), + RedriveReplayClock::recordedAt($resolution) + ) + : $workflowExecution->throw( + ChildRunHistory::exceptionForResolution($resolution, $target), + RedriveReplayClock::recordedAt($resolution) + ); + ++$sequence; + continue; + } + if ($current instanceof ChildWorkflowCall) { $historySequence = $this->historySequenceForReplayPosition($historySequencesByPosition, $sequence); $this->applyRecordedUpdates($run, $workflow, $historySequence); WorkflowStepHistory::assertCompatible($run, $historySequence, WorkflowStepHistory::CHILD_WORKFLOW, [ 'child_workflow_type' => $current->workflow, + 'child_completion_mode' => 'result', ]); $resolutionEvent = ChildRunHistory::resolutionEventForSequence($run, $historySequence); @@ -657,6 +725,7 @@ public function replayState(WorkflowRun $run): ReplayState 'activity_type' => $call->activity, ] : ($call instanceof ChildWorkflowCall ? [ 'child_workflow_type' => $call->workflow, + 'child_completion_mode' => 'result', ] : []), ); @@ -2022,6 +2091,7 @@ private function hasWorkflowCommandSequence(?string $type): bool HistoryEventType::SignalWaitOpened->value, HistoryEventType::SignalApplied->value, HistoryEventType::ChildWorkflowScheduled->value, + HistoryEventType::ChildWorkflowAwaited->value, HistoryEventType::ChildRunStarted->value, HistoryEventType::ChildRunCompleted->value, HistoryEventType::ChildRunFailed->value, diff --git a/src/V2/Support/StartChildWorkflowCall.php b/src/V2/Support/StartChildWorkflowCall.php new file mode 100644 index 000000000..50fa65faa --- /dev/null +++ b/src/V2/Support/StartChildWorkflowCall.php @@ -0,0 +1,15 @@ +task, $state->frame ), + $current instanceof StartChildWorkflowCall => $this->executeChildStart( + $state->run, + $state->task, + $state->frame + ), + $current instanceof AwaitStartedChildWorkflowCall => $this->executeStartedChildAwait( + $state->run, + $state->task, + $state->frame + ), $current instanceof ChildWorkflowCall => $this->executeChild($state->run, $state->task, $state->frame), $current instanceof AllCall => $this->executeAll($state->run, $state->task, $state->frame), $current instanceof ContinueAsNewCall => $this->executeContinueAsNew( @@ -1592,6 +1602,130 @@ private function resumeServiceOperation( return WorkflowTaskStep::ContinueExecution; } + private function executeChildStart( + WorkflowRun $run, + WorkflowTask $task, + WorkflowContinuation $frame, + ): WorkflowTask|WorkflowTaskStep|null { + /** @var StartChildWorkflowCall $call */ + $call = $frame->current; + $sequence = $frame->sequence; + $this->syncWorkflowCursorForCurrent($frame->workflow, $run, $call, $sequence); + if (! $this->applyRecordedUpdates($run, $frame->workflow, $sequence, $task)) { + return $this->restartAfterPendingUpdateFailure($run, $task); + } + if (! $this->ensureStepHistoryCompatible($run, $task, $sequence, WorkflowStepHistory::CHILD_WORKFLOW, [ + 'child_workflow_type' => $call->child->workflow, + 'child_completion_mode' => 'started', + ])) { + return null; + } + try { + $receipt = ChildStartHistory::receipt($run, $sequence); + $admitted = ChildRunHistory::scheduledEventForSequence($run, $sequence); + $delivery = $this->deliverCancellationAtCall($run, $task, $sequence, 'child', $admitted, $frame->execution); + if ($delivery === CancellationDeliveryState::Waiting) { + return $this->waitForNextResumeSource($run, $task); + } + if ($delivery === CancellationDeliveryState::Delivered) { + return $this->resumeCancellation($run, $task, $frame); + } + if ($receipt === null) { + $childTask = $this->scheduleChildWorkflowOrFailRun( + $run, + $task, + $sequence, + $call->child, + parkRun: false, + startMilestone: true + ); + if ($childTask === null) { + return null; + } + $run->load('historyEvents'); + $receipt = ChildStartHistory::receipt($run, $sequence); + if ($receipt === null) { + throw new LogicException('Native child creation produced no confirmed admission.'); + } + TaskDispatcher::dispatch($childTask); + } + $this->syncWorkflowCursor($frame->workflow, $sequence + 1); + $frame->current = $frame->execution->send($receipt, RedriveReplayClock::recordedAt( + ChildRunHistory::scheduledEventForSequence($run, $sequence) + )); + ++$frame->sequence; + return WorkflowTaskStep::ContinueExecution; + } catch (Throwable $error) { + $this->failRun($run, $task, $error, 'workflow_run', $run->id); + return null; + } + } + + private function executeStartedChildAwait( + WorkflowRun $run, + WorkflowTask $task, + WorkflowContinuation $frame, + ): WorkflowTask|WorkflowTaskStep|null { + /** @var AwaitStartedChildWorkflowCall $call */ + $call = $frame->current; + $sequence = $frame->sequence; + $child = $call->child; + $this->syncWorkflowCursorForCurrent($frame->workflow, $run, $call, $sequence); + if (! $this->applyRecordedUpdates($run, $frame->workflow, $sequence, $task)) { + return $this->restartAfterPendingUpdateFailure($run, $task); + } + if (! $this->ensureStepHistoryCompatible($run, $task, $sequence, WorkflowStepHistory::CHILD_WORKFLOW_AWAIT)) { + return null; + } + try { + ChildStartHistory::assertReceipt($run, $child, $sequence); + $awaited = ChildStartHistory::awaitedEvent($run, $child, $sequence); + if (ChildStartHistory::cancellationDelivery($run, $child, $sequence) !== null) { + return $this->resumeCancellation($run, $task, $frame); + } + $resolution = ChildRunHistory::resolutionEventForSequence($run, $child->startSequence); + $target = ChildStartHistory::target($run, $child, $resolution, $sequence); + if ($awaited === null) { + WorkflowHistoryEvent::record( + $run, + HistoryEventType::ChildWorkflowAwaited, + ChildStartHistory::awaitPayload($child, $sequence), + $task + ); + $run->load('historyEvents'); + } + $delivery = $this->deliverCancellationAtCall( + $run, + $task, + $sequence, + 'started_child_await', + $resolution, + $frame->execution, + operationSequence: $child->startSequence + ); + if ($delivery === CancellationDeliveryState::Waiting) { + return $this->waitForNextResumeSource($run, $task); + } + if ($delivery === CancellationDeliveryState::Delivered) { + return $this->resumeCancellation($run, $task, $frame); + } + if ($resolution === null && $target !== null) { + $status = ChildRunHistory::resolvedStatus(null, $target); + if ($status !== null && $status->isTerminal()) { + $resolution = $this->recordChildResolution($run, $task, $child->startSequence, $target); + } + } + if ($resolution !== null) { + return $this->resumeChildResolution($run, $task, $frame, $resolution, $target); + } + $this->syncWorkflowCursor($frame->workflow, $sequence + 1); + return $this->waitForNextResumeSource($run, $task); + } catch (Throwable $error) { + $this->failRun($run, $task, $error, 'workflow_run', $run->id); + return null; + } + } + private function executeChild( WorkflowRun $run, WorkflowTask $task, @@ -1609,6 +1743,7 @@ private function executeChild( } if (! $this->ensureStepHistoryCompatible($run, $task, $sequence, WorkflowStepHistory::CHILD_WORKFLOW, [ 'child_workflow_type' => $current->workflow, + 'child_completion_mode' => 'result', ])) { return null; } @@ -1780,6 +1915,7 @@ private function parallelMemberIdentity(mixed $call): array ], $call instanceof ChildWorkflowCall => [ 'child_workflow_type' => $call->workflow, + 'child_completion_mode' => 'result', ], default => [], }; @@ -2752,6 +2888,7 @@ private function scheduleChildWorkflow( ChildWorkflowCall $childWorkflowCall, ?array $parallelMetadata = null, bool $parkRun = true, + bool $startMilestone = false, ): WorkflowTask { StructuralLimits::guardPendingChildren($run); $this->logApproachingLimit(StructuralLimits::warnApproachingPendingChildren($run), $run); @@ -2882,6 +3019,9 @@ private function scheduleChildWorkflow( WorkflowHistoryEvent::record($run, HistoryEventType::ChildWorkflowScheduled, array_merge([ 'sequence' => $sequence, + ...($startMilestone ? [ + 'child_completion_mode' => 'started', + ] : []), 'workflow_link_id' => $link->id, 'child_call_id' => $childCallId, 'child_workflow_instance_id' => $childInstance->id, @@ -2977,6 +3117,7 @@ private function scheduleChildWorkflowOrFailRun( ChildWorkflowCall $childWorkflowCall, ?array $parallelMetadata = null, bool $parkRun = true, + bool $startMilestone = false, ): ?WorkflowTask { try { return $this->scheduleChildWorkflow( @@ -2985,7 +3126,8 @@ private function scheduleChildWorkflowOrFailRun( $sequence, $childWorkflowCall, $parallelMetadata, - $parkRun + $parkRun, + $startMilestone ); } catch (Throwable $throwable) { $this->failRun($run, $task, $throwable, 'workflow_run', $run->id); diff --git a/src/V2/Support/WorkflowFiberRunner.php b/src/V2/Support/WorkflowFiberRunner.php index 1ea52a870..355ba4e41 100644 --- a/src/V2/Support/WorkflowFiberRunner.php +++ b/src/V2/Support/WorkflowFiberRunner.php @@ -312,6 +312,12 @@ private function nextObservableStep(): WorkflowStep $current = $this->execution->current(); + if ($current instanceof StartChildWorkflowCall || $current instanceof AwaitStartedChildWorkflowCall) { + throw new UnsupportedWorkflowYieldException( + 'Confirmed child admission currently requires embedded native execution.' + ); + } + if (! $current instanceof YieldedCommand) { throw new UnsupportedWorkflowYieldException(sprintf( 'Worker protocol runner received an unsupported workflow suspension of type %s.', @@ -807,6 +813,17 @@ private function nextObservableStep(): WorkflowStep if ($current instanceof ChildWorkflowCall) { $historySequence = $this->historySequenceForCurrentPosition(); + if ($historySequence !== null) { + WorkflowStepHistory::assertCompatible( + $this->workflow->run, + $historySequence, + WorkflowStepHistory::CHILD_WORKFLOW, + [ + 'child_workflow_type' => $current->workflow, + 'child_completion_mode' => 'result', + ] + ); + } $recorded = $historySequence === null ? null : ($this->recordedChildOutcomes[$historySequence] ?? null); diff --git a/src/V2/Support/WorkflowStepHistory.php b/src/V2/Support/WorkflowStepHistory.php index 9a5525f29..de8b10944 100644 --- a/src/V2/Support/WorkflowStepHistory.php +++ b/src/V2/Support/WorkflowStepHistory.php @@ -22,6 +22,8 @@ final class WorkflowStepHistory public const CHILD_WORKFLOW = 'child workflow'; + public const CHILD_WORKFLOW_AWAIT = 'await admitted child workflow'; + public const CANCELLATION_SCOPE = 'cancellation scope'; public const CONDITION_WAIT = 'condition wait'; @@ -60,6 +62,7 @@ final class WorkflowStepHistory HistoryEventType::ActivityCancelled, HistoryEventType::ActivityTimedOut, HistoryEventType::ChildWorkflowScheduled, + HistoryEventType::ChildWorkflowAwaited, HistoryEventType::ChildRunStarted, HistoryEventType::ChildRunCompleted, HistoryEventType::ChildRunFailed, @@ -137,6 +140,19 @@ public static function assertCompatible( throw new HistoryEventShapeMismatchException($sequence, $expectedShape, $conflictingEventTypes); } + if ($expectedShape === self::CHILD_WORKFLOW && array_key_exists('child_completion_mode', $expectedDetails)) { + $scheduled = ChildRunHistory::scheduledEventForSequence($run, $sequence); + if ($scheduled !== null && ($scheduled->payload['child_completion_mode'] ?? 'result') + !== ($expectedDetails['child_completion_mode'] ?? 'result')) { + throw new HistoryEventShapeMismatchException( + $sequence, + $expectedShape, + [HistoryEventType::ChildWorkflowScheduled->value], + 'Starting a child and waiting for its result are different durable steps.' + ); + } + } + $detailMismatch = self::detailMismatchForSequence($run, $sequence, $expectedShape, $expectedDetails); if ($detailMismatch !== null) { @@ -441,6 +457,7 @@ private static function eventMatchesShape(WorkflowHistoryEvent $event, string $e HistoryEventType::ActivityCancelled, HistoryEventType::ActivityTimedOut, ], true) && self::isLocalActivityEvent($event), + self::CHILD_WORKFLOW_AWAIT => $event->event_type === HistoryEventType::ChildWorkflowAwaited, self::CHILD_WORKFLOW => in_array($event->event_type, [ HistoryEventType::ChildWorkflowScheduled, HistoryEventType::ChildRunStarted, diff --git a/src/V2/Workflow.php b/src/V2/Workflow.php index 304c98448..3a83e4ded 100644 --- a/src/V2/Workflow.php +++ b/src/V2/Workflow.php @@ -474,6 +474,17 @@ public static function executeChildWorkflow(string $workflow, mixed ...$argument return child($workflow, ...$arguments); } + /** + * Confirm native child admission, then await its result separately if needed. + * + * This milestone is available in embedded execution. A queued request is + * insufficient: the returned receipt identifies the durable child run. + */ + public static function startChild(string $workflow, mixed ...$arguments): StartedChildWorkflow + { + return startChild($workflow, ...$arguments); + } + /** * Run a callable as an auto-generated child workflow. * diff --git a/src/V2/functions.php b/src/V2/functions.php index 55c13f278..ad959a2e3 100644 --- a/src/V2/functions.php +++ b/src/V2/functions.php @@ -26,6 +26,7 @@ use Workflow\V2\Support\ServiceOperationOptions; use Workflow\V2\Support\SideEffectCall; use Workflow\V2\Support\SignalCall; +use Workflow\V2\Support\StartChildWorkflowCall; use Workflow\V2\Support\TimerCall; use Workflow\V2\Support\UpsertMemoCall; use Workflow\V2\Support\UpsertSearchAttributesCall; @@ -117,6 +118,20 @@ function child(string $workflow, mixed ...$arguments): mixed } } +if (! function_exists(__NAMESPACE__ . '\\startChild')) { + /** Confirm an embedded native child admission without waiting for completion. */ + function startChild(string $workflow, mixed ...$arguments): StartedChildWorkflow + { + $options = null; + if (($arguments[0] ?? null) instanceof ChildWorkflowOptions) { + $options = array_shift($arguments); + } + return WorkflowFiberContext::suspend(new StartChildWorkflowCall( + new ChildWorkflowCall($workflow, $arguments, $options) + )); + } +} + if (! function_exists(__NAMESPACE__ . '\\async')) { function async(callable $callback): mixed { diff --git a/tests/Fixtures/V2/ReplayRegression/child-start-cannot-become-result-wait.json b/tests/Fixtures/V2/ReplayRegression/child-start-cannot-become-result-wait.json new file mode 100644 index 000000000..99478f59c --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/child-start-cannot-become-result-wait.json @@ -0,0 +1,46 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "child-start-cannot-become-result-wait", + "protocol_version": "1.20", + "bindings": [ + "php" + ], + "consumers": [ + "workflow-fiber-runner", + "query-state-replayer", + "workflow-executor" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestParentChildWorkflow", + "arguments": [ + "Taylor" + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-11-13T12:00:00Z" + }, + { + "sequence": 2, + "event_type": "ChildWorkflowScheduled", + "payload": { + "sequence": 1, + "child_workflow_type": "Tests\\Fixtures\\V2\\TestChildGreetingWorkflow", + "child_workflow_run_id": "01j00000000000000000000061", + "child_workflow_instance_id": "corpus-admitted-child", + "child_call_id": "01j00000000000000000000062", + "child_completion_mode": "started" + }, + "recorded_at": "2026-11-13T12:00:01Z" + } + ], + "expected_failure": { + "type": "workflow_history_shape_mismatch", + "exception": "Workflow\\V2\\Exceptions\\HistoryEventShapeMismatchException" + } +} diff --git a/tests/Unit/V2/CancellationContextTest.php b/tests/Unit/V2/CancellationContextTest.php index fd0e7b353..09ead3f83 100644 --- a/tests/Unit/V2/CancellationContextTest.php +++ b/tests/Unit/V2/CancellationContextTest.php @@ -23,10 +23,8 @@ public function testSnapshotAndDeadlineRemainImmutableWhenTheCallerChangesItsCop $snapshot['reason'] = 'changed'; $snapshot['requester']['id'] = 'changed'; $snapshot['lineage'][0]['request_id'] = 'changed'; - $context->deadline() - ->addHour(); - $context->requestedAt() - ->addDay(); + $this->assertSame('2026-10-01T01:00:30.000000Z', $context->deadline()->addHour()->toISOString()); + $this->assertSame('2026-10-02T00:00:00.000000Z', $context->requestedAt()->addDay()->toISOString()); $this->assertSame('maintenance', $context->reason); $this->assertSame('operator-1', $context->requester['id']); diff --git a/tests/Unit/V2/HostServiceQuerySqliteWriteRetryTest.php b/tests/Unit/V2/HostServiceQuerySqliteWriteRetryTest.php index 123b3e989..7d1c75189 100644 --- a/tests/Unit/V2/HostServiceQuerySqliteWriteRetryTest.php +++ b/tests/Unit/V2/HostServiceQuerySqliteWriteRetryTest.php @@ -4,6 +4,7 @@ namespace Tests\Unit\V2; +use Illuminate\Database\Events\TransactionRolledBack; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; use PHPUnit\Framework\Attributes\DataProvider; @@ -66,12 +67,11 @@ public function testShortWritesWaitOnlyAfterAnOwnedRootRollback(string $phase, b 'id' => $callId, ]); fwrite($pair[1], "locked\n"); - // Hold the lock briefly. The owned root path retries for about - // 60 ms (3 attempts, 20 ms then 40 ms apart), so a short fixed - // hold keeps real contention without depending on how quickly - // a busy runner reschedules this process (#1096). - usleep(5_000); + if (fgets($pair[1]) !== "release\n") { + throw new RuntimeException('The original root did not roll back before writer release.'); + } DB::commit(); + fwrite($pair[1], "committed\n"); fclose($pair[1]); exit(0); } catch (Throwable $failure) { @@ -86,6 +86,35 @@ public function testShortWritesWaitOnlyAfterAnOwnedRootRollback(string $phase, b $connection = DB::connection(); $connection->statement('PRAGMA busy_timeout=0'); $entered = false; + $released = false; + $releaseLevels = []; + $connectionName = $connection->getName(); + $dispatcher = $connection->getEventDispatcher(); + $this->assertNotNull($dispatcher); + $dispatcher->listen( + TransactionRolledBack::class, + static function (TransactionRolledBack $event) use ( + &$entered, + &$released, + &$releaseLevels, + $connectionName, + $pair + ): void { + if (! $entered || $released || $event->connection->getName() !== $connectionName + || $event->connection->transactionLevel() !== 0) { + return; + } + // Release only after the original root rollback. A committed + // acknowledgement removes the fixed-sleep scheduling race + // without changing the runtime's attempt or deadline limits. + $released = true; + $releaseLevels[] = $event->connection->transactionLevel(); + fwrite($pair[0], "release\n"); + if (fgets($pair[0]) !== "committed\n") { + throw new RuntimeException('The competing writer did not confirm its commit.'); + } + } + ); DB::listen(static function ($event) use (&$entered, $pair): void { if ($entered || ! str_starts_with($event->sql, 'select * from "workflow_service_calls"')) { return; @@ -119,6 +148,8 @@ public function testShortWritesWaitOnlyAfterAnOwnedRootRollback(string $phase, b $this->assertNotNull($result); $this->assertSame(0, $connection->transactionLevel()); } + $this->assertTrue($released); + $this->assertSame([0], $releaseLevels); pcntl_waitpid($pid, $status); $this->assertTrue(pcntl_wifexited($status)); $this->assertSame(0, pcntl_wexitstatus($status)); diff --git a/tests/Unit/V2/ScopedCancellationContextTest.php b/tests/Unit/V2/ScopedCancellationContextTest.php index a8248a626..7f7239bc2 100644 --- a/tests/Unit/V2/ScopedCancellationContextTest.php +++ b/tests/Unit/V2/ScopedCancellationContextTest.php @@ -69,10 +69,8 @@ public function testStricterAcceptedScopeBudgetDoesNotRewriteOriginalMetadataOrG $this->assertSame('api', $child->rootContext->source); $copy = $child->toArray(); $copy['lineage'][1]['scope_id'] = 'changed'; - $child->deadline() - ->addHour(); - $child->rootDeadline() - ->addHour(); + $this->assertSame('2026-10-03T01:00:20.000001Z', $child->deadline()->addHour()->toISOString()); + $this->assertSame('2026-10-03T01:00:30.123456Z', $child->rootDeadline()->addHour()->toISOString()); $this->assertSame('scope-a', $child->lineage[1]['scope_id']); $this->assertSame('2026-10-03T00:00:20.000001Z', $child->deadline()->toISOString()); $this->expectException(InvalidArgumentException::class); diff --git a/tests/Unit/V2/StartedChildWorkflowTest.php b/tests/Unit/V2/StartedChildWorkflowTest.php new file mode 100644 index 000000000..7eb2a45e6 --- /dev/null +++ b/tests/Unit/V2/StartedChildWorkflowTest.php @@ -0,0 +1,608 @@ + 'redis', + 'workflows.v2.task_dispatch_mode' => 'queue', + 'workflows.v2.namespaces' => ['client-a'], + ]); + } + + public function testConfirmedStartPrecedesChildExecutionAndSurvivesAColdDuplicateResume(): void + { + [$parent, $run] = $this->startParent(); + $receipt = $this->queryAdmission($run); + $this->assertSame($run->id, $receipt['parentRunId']); + $child = WorkflowRun::query()->findOrFail($receipt['childRunId']); + $this->assertSame(RunStatus::Pending, $child->status); + $this->assertSame($child->workflow_instance_id, $receipt['childInstanceId']); + $this->assertSame('client-a', $receipt['namespace']); + $this->assertSame(1, WorkflowLink::query()->where('parent_workflow_run_id', $run->id)->count()); + + // The acknowledgement can be lost while admission has committed. + // A completely new task/fiber must reuse the original start. + $this->resume($run); + $this->assertSame($receipt, $this->queryAdmission($run->fresh())); + $this->assertSame(2, WorkflowInstance::query()->count()); + $this->assertSame( + 1, + $run->historyEvents() + ->where('event_type', HistoryEventType::ChildWorkflowScheduled)->count() + ); + $this->assertSame( + 0, + $run->historyEvents() + ->where('event_type', HistoryEventType::ChildWorkflowAwaited)->count() + ); + $this->assertFalse($parent->completed()); + } + + public function testLaterAwaitUsesTheSameChildWithoutAnotherAdmission(): void + { + [$parent, $run] = $this->startParent(); + $receipt = $this->queryAdmission($run); + $parent->signal('proceed'); + $this->drainParent($run); + $this->assertSame( + 1, + $run->historyEvents() + ->where('event_type', HistoryEventType::ChildWorkflowAwaited)->count() + ); + $this->assertFalse($parent->completed()); + $child = WorkflowRun::query()->findOrFail($receipt['childRunId']); + $this->runTaskFor($child); + $this->drainParent($run); + $parent = WorkflowStub::loadRun($run->id); + $this->assertTrue($parent->completed()); + $this->assertSame([ + 'admission' => $receipt, + 'result' => 'child-result', + ], $parent->output()); + $this->assertSame(2, WorkflowInstance::query()->count()); + $this->assertSame(1, WorkflowLink::query()->where('parent_workflow_run_id', $run->id)->count()); + $this->assertSame($receipt, $this->queryAdmission($run->fresh())); + } + + public function testChildMayCompleteBeforeTheLaterAwaitWithoutChangingAdmissionIdentity(): void + { + [$parent, $run] = $this->startParent(); + $receipt = $this->queryAdmission($run); + $this->runTaskFor(WorkflowRun::query()->findOrFail($receipt['childRunId'])); + $this->drainParent($run); + $this->assertSame($receipt, $this->queryAdmission($run->fresh())); + $this->assertFalse($parent->completed()); + $parent->signal('proceed'); + $this->drainParent($run); + $this->assertSame('child-result', WorkflowStub::loadRun($run->id)->output()['result']); + $this->assertSame(2, WorkflowInstance::query()->count()); + } + + public static function substitutedTargets(): iterable + { + yield 'unrecorded current continuation' => [false]; + yield 'foreign terminal resolution in the same instance' => [true]; + } + + #[DataProvider('substitutedTargets')] + public function testLaterSameInstanceRunsCannotReplaceTheAdmittedTarget(bool $resolution): void + { + [$parent, $run] = $this->startParent(); + $receipt = $this->queryAdmission($run); + $parent->signal('proceed'); + $this->drainParent($run); + $initial = WorkflowRun::query()->findOrFail($receipt['childRunId']); + $later = $initial->replicate(); + $later->run_number = 2; + $later->status = RunStatus::Completed; + $later->closed_at = now(); + $later->closed_reason = 'completed'; + $later->save(); + WorkflowHistoryEvent::record($later, HistoryEventType::WorkflowStarted, [ + 'workflow_type' => $later->workflow_type, + 'workflow_instance_id' => $later->workflow_instance_id, + 'workflow_run_id' => $later->id, + ]); + WorkflowHistoryEvent::record($later, HistoryEventType::WorkflowCompleted, []); + $initial->instance() + ->update([ + 'current_run_id' => $later->id, + 'run_count' => 2, + ]); + if ($resolution) { + WorkflowHistoryEvent::record($run, HistoryEventType::ChildRunCompleted, [ + 'sequence' => $receipt['startSequence'], + 'child_workflow_instance_id' => $receipt['childInstanceId'], + 'child_workflow_run_id' => $later->id, + 'child_call_id' => $receipt['childCallId'], + 'child_workflow_type' => $receipt['workflowType'], + ]); + } else { + $initial->update([ + 'status' => RunStatus::Completed, + 'closed_reason' => 'continued', + ]); + } + $this->assertReadAndExecutionRefuseWithoutHistoryWrites($run); + } + + public static function childSuccessions(): iterable + { + yield 'native continue as new' => ['continue', false]; + yield 'native retry' => ['retry', false]; + yield 'substituted continuation edge' => ['continue', true]; + yield 'substituted retry edge' => ['retry', true]; + } + + #[DataProvider('childSuccessions')] + public function testOnlyRecordedNativeSuccessionsExtendTheReceipt(string $mode, bool $substitute): void + { + [$parent, $run] = $this->startParent(childMode: $mode); + $receipt = $this->queryAdmission($run); + if ($mode === 'retry') { + // The embedded options do not author retry policy yet. Exercise + // the real native retry producer with its recorded policy input. + $scheduled = $run->historyEvents() + ->where('event_type', HistoryEventType::ChildWorkflowScheduled)->sole(); + $scheduled->update([ + 'payload' => array_merge($scheduled->payload, [ + 'retry_policy' => [ + 'snapshot_version' => 1, + 'max_attempts' => 2, + 'backoff_seconds' => [], + 'non_retryable_error_types' => [], + ], + ]), + ]); + } + $this->runTaskFor(WorkflowRun::query()->findOrFail($receipt['childRunId'])); + $successor = WorkflowRun::query()->where('workflow_instance_id', $receipt['childInstanceId']) + ->where('run_number', 2) + ->sole(); + $this->assertSame(RunStatus::Pending, $successor->status); + $this->assertSame($receipt, $this->queryAdmission($run->fresh())); + $parent->signal('proceed'); + $this->drainParent($run); + $this->assertSame(RunStatus::Waiting, $run->fresh()->status); + if ($substitute) { + $started = $successor->historyEvents() + ->where('event_type', HistoryEventType::WorkflowStarted)->sole(); + $payload = $started->payload; + $payload[$mode === 'retry' ? 'retry_of_child_workflow_run_id' : 'continued_from_run_id'] = 'unrelated'; + $started->update([ + 'payload' => $payload, + ]); + $this->assertReadAndExecutionRefuseWithoutHistoryWrites($run); + return; + } + $this->runTaskFor($successor); + $this->drainParent($run); + $this->assertSame(RunStatus::Completed, $run->fresh()->status); + $this->assertSame('child-result', WorkflowStub::loadRun($run->id)->output()['result']); + $this->assertSame($receipt, $this->queryAdmission($run->fresh())); + } + + public static function cancellationPolicies(): iterable + { + foreach (CancellationPolicy::cases() as $policy) { + yield $policy->value => [$policy]; + } + } + + #[DataProvider('cancellationPolicies')] + public function testStartedAwaitCancellationHasACanonicalColdBoundary(CancellationPolicy $policy): void + { + Carbon::setTestNow('2026-11-13T12:00:00Z'); + try { + [$parent, $run, $child] = $this->startCancellationParent($policy); + $root = $parent->requestCancellation('maintenance', 30) + ->cancellationContext(); + $this->drainParent($run); + if ($policy === CancellationPolicy::WaitCancellationCompleted) { + $this->assertSame(RunStatus::Waiting, $run->fresh()->status); + $this->assertNull(CooperativeCancellationDelivery::recorded($run->fresh(['historyEvents']))); + $this->runTaskFor($child); + Carbon::setTestNow(now()->addSeconds(2)); + $timer = WorkflowTask::query()->where('workflow_run_id', $child->id) + ->where('task_type', TaskType::Timer)->where('status', TaskStatus::Ready)->firstOrFail(); + app() + ->call([new RunTimerTask($timer->id), 'handle']); + $this->drainParent($child); + $this->assertSame(RunStatus::Cancelled, $child->fresh()->status); + $this->drainParent($run); + } + $fresh = $run->fresh(['historyEvents']); + $this->assertSame(RunStatus::Waiting, $fresh->status); + $event = CooperativeCancellationDelivery::recorded($fresh); + $this->assertNotNull($event); + $this->assertSame('started_child_await', $event->payload['call_kind']); + $this->assertSame(2, $event->payload['sequence']); + $this->assertSame(1, $event->payload['operation_sequence']); + $this->assertNull(CooperativeCancellationDelivery::validate( + $fresh, + $root->requestId, + 2, + 'started_child_await', + operationSequence: 1 + )); + $before = $fresh->historyEvents() + ->get() + ->toArray(); + $this->assertSame('waiting', (new QueryStateReplayer())->query($fresh, 'phase')); + $this->assertSame($before, $fresh->historyEvents()->get()->toArray()); + $this->resume($run); + $this->assertSame($before, $run->historyEvents()->get()->toArray()); + Carbon::setTestNow(now()->addSecond()); + $timer = WorkflowTask::query()->where('workflow_run_id', $run->id) + ->where('task_type', TaskType::Timer)->where('status', TaskStatus::Ready)->firstOrFail(); + app() + ->call([new RunTimerTask($timer->id), 'handle']); + $this->drainParent($run); + $this->assertSame(RunStatus::Cancelled, $run->fresh()->status); + $this->assertSame('cleaned', (new QueryStateReplayer())->query($run->fresh(), 'phase')); + $this->assertSame( + $policy === CancellationPolicy::Abandon, + $child->fresh() +->cancellation_request_command_id === null + ); + if ($policy !== CancellationPolicy::WaitCancellationCompleted) { + $this->assertSame(RunStatus::Waiting, $child->fresh()->status); + } + } finally { + Carbon::setTestNow(); + } + } + + public function testCancellationCannotReplaceTheOriginalOperationWithAnotherAdmittedChild(): void + { + [$parent, $run] = $this->startCancellationParent(CancellationPolicy::Abandon, secondChild: true); + $root = $parent->requestCancellation('maintenance', 30) + ->cancellationContext(); + $this->drainParent($run); + $event = CooperativeCancellationDelivery::recorded($run->fresh(['historyEvents'])); + $this->assertNotNull($event); + $event->update([ + 'payload' => array_merge($event->payload, [ + 'operation_sequence' => 2, + ]), + ]); + $fresh = $run->fresh(['historyEvents']); + $this->assertNotNull(CooperativeCancellationDelivery::validate( + $fresh, + $root->requestId, + 3, + 'started_child_await', + operationSequence: 2 + )); + $before = $fresh->historyEvents() + ->get() + ->toArray(); + try { + (new QueryStateReplayer())->query($fresh, 'phase'); + $this->fail('A later valid child admission cannot replace the awaited receipt.'); + } catch (HistoryEventShapeMismatchException) { + $this->assertSame($before, $fresh->historyEvents()->get()->toArray()); + } + $this->resume($run); + $this->assertSame($before, $run->historyEvents()->get()->toArray()); + $this->assertSame(RunStatus::Waiting, $run->fresh()->status); + $blocked = WorkflowTask::query()->where('workflow_run_id', $run->id)->where( + 'status', + TaskStatus::Failed + )->sole(); + $this->assertSame('history_shape_mismatch', $blocked->payload['replay_blocked_reason']); + } + + public static function receiptSubstitutions(): iterable + { + foreach (['namespace', 'parentInstanceId', 'parentRunId', 'childInstanceId', 'childRunId', + 'childCallId', 'workflowType', 'startSequence'] as $field) { + yield $field => [$field]; + } + } + + #[DataProvider('receiptSubstitutions')] + public function testForeignOrStaleReceiptsCannotAuthorizeAnAwait(string $field): void + { + [, $run] = $this->startParent($field); + $stub = WorkflowStub::loadRun($run->id); + $stub->signal('proceed'); + $this->drainParent($run); + $this->assertFalse($run->fresh()->status->isTerminal()); + $this->assertSame( + 0, + $run->historyEvents() + ->where('event_type', HistoryEventType::ChildWorkflowAwaited)->count() + ); + $this->assertSame(2, WorkflowInstance::query()->count()); + $this->assertSame(0, $run->historyEvents()->where('event_type', HistoryEventType::WorkflowFailed)->count()); + $blocked = WorkflowTask::query()->where('workflow_run_id', $run->id)->where( + 'status', + TaskStatus::Failed + )->sole(); + $this->assertTrue($blocked->payload['replay_blocked']); + $this->assertSame('history_shape_mismatch', $blocked->payload['replay_blocked_reason']); + } + + public function testReadOnlyReplayRefusesAnIncompleteAdmissionWithoutInventingAChild(): void + { + [, $run] = $this->startParent(); + $run->historyEvents() + ->where('event_type', HistoryEventType::ChildRunStarted)->delete(); + $before = $run->historyEvents() + ->get() + ->toArray(); + try { + $this->queryAdmission($run->fresh()); + $this->fail('A scheduled request alone must not become a start receipt.'); + } catch (HistoryEventShapeMismatchException) { + $this->assertSame($before, $run->historyEvents()->get()->toArray()); + $this->assertSame(2, WorkflowInstance::query()->count()); + } + } + + public function testOrdinaryChildStillWaitsForItsTerminalResult(): void + { + $parent = WorkflowStub::make(OrdinaryMilestoneParent::class, 'ordinary-parent', 'client-a'); + $parent->start(); + $run = WorkflowRun::query()->findOrFail($parent->runId()); + $this->runTaskFor($run); + $this->assertFalse($parent->completed()); + $child = WorkflowRun::query()->where('id', '!=', $run->id)->sole(); + $this->runTaskFor($child); + $this->drainParent($run); + $this->assertSame('child-result', WorkflowStub::loadRun($run->id)->output()); + $this->assertArrayNotHasKey('child_completion_mode', $run->historyEvents()->where( + 'event_type', + HistoryEventType::ChildWorkflowScheduled + )->sole() +->payload); + } + + public function testPortableWorkerCannotMistakeAQueuedCommandForNativeAdmission(): void + { + try { + WorkflowFiberRunner::forClass( + MilestoneParent::class, + 'portable-parent', + 'portable-run', + [''], + 'avro' + )->step(); + $this->fail('Portable admission is not implemented by the embedded milestone.'); + } catch (UnsupportedWorkflowYieldException $error) { + $this->assertStringContainsString('embedded native execution', $error->getMessage()); + $this->assertSame(0, WorkflowInstance::query()->count()); + } + } + + public function testAStartedMilestoneCannotBeReplayedAsAnOrdinaryTerminalChildCall(): void + { + [, $run] = $this->startParent(); + $before = $run->historyEvents() + ->get() + ->toArray(); + try { + WorkflowStepHistory::assertCompatible( + $run->fresh(['historyEvents']), + 1, + WorkflowStepHistory::CHILD_WORKFLOW, + [ + 'child_completion_mode' => 'result', + ] + ); + $this->fail('Changing child call completion semantics must require a compatible definition.'); + } catch (HistoryEventShapeMismatchException) { + $this->assertSame($before, $run->historyEvents()->get()->toArray()); + } + } + + private function startCancellationParent(CancellationPolicy $policy, bool $secondChild = false): array + { + $parent = WorkflowStub::make(CancellationMilestoneParent::class, 'cancelling-parent', 'client-a'); + $parent->start($policy->value, $secondChild); + $run = WorkflowRun::query()->findOrFail($parent->runId()); + $this->runTaskFor($run); + $first = $run->historyEvents() + ->where('event_type', HistoryEventType::ChildRunStarted)->firstOrFail(); + $child = WorkflowRun::query()->findOrFail($first->payload['child_workflow_run_id']); + $this->runTaskFor($child); + return [$parent, $run, $child]; + } + + private function assertReadAndExecutionRefuseWithoutHistoryWrites(WorkflowRun $run): void + { + $before = $run->historyEvents() + ->get() + ->toArray(); + try { + $this->queryAdmission($run->fresh()); + $this->fail('A receipt must refuse an unrecorded or substituted descendant.'); + } catch (HistoryEventShapeMismatchException) { + $this->assertSame($before, $run->historyEvents()->get()->toArray()); + } + $this->resume($run); + $this->assertSame($before, $run->historyEvents()->get()->toArray()); + $this->assertSame(RunStatus::Waiting, $run->fresh()->status); + $failed = WorkflowTask::query()->where('workflow_run_id', $run->id)->where( + 'status', + TaskStatus::Failed + )->sole(); + $this->assertSame('history_shape_mismatch', $failed->payload['replay_blocked_reason']); + } + + private function startParent(string $mutation = '', string $childMode = ''): array + { + $stub = WorkflowStub::make(MilestoneParent::class, 'milestone-parent', 'client-a'); + $stub->start($mutation, $childMode); + $run = WorkflowRun::query()->findOrFail($stub->runId()); + $this->runTaskFor($run); + return [$stub, $run]; + } + + private function queryAdmission(WorkflowRun $run): array + { + return (new QueryStateReplayer())->query($run->fresh(), 'admitted'); + } + + private function runTaskFor(WorkflowRun $run): void + { + $task = WorkflowTask::query()->where('workflow_run_id', $run->id) + ->where('task_type', TaskType::Workflow)->where('status', TaskStatus::Ready)->firstOrFail(); + app() + ->call([new RunWorkflowTask($task->id), 'handle']); + } + + private function resume(WorkflowRun $run): void + { + $task = WorkflowTask::query()->create([ + 'workflow_run_id' => $run->id, + 'namespace' => $run->namespace, + 'task_type' => TaskType::Workflow, + 'status' => TaskStatus::Ready, + 'payload' => [], + 'available_at' => now(), + 'connection' => $run->connection, + 'queue' => $run->queue, + 'compatibility' => $run->compatibility, + ]); + app() + ->call([new RunWorkflowTask($task->id), 'handle']); + } + + private function drainParent(WorkflowRun $run): void + { + for ($i = 0; $i < 8; ++$i) { + if (! WorkflowTask::query()->where('workflow_run_id', $run->id)->where('task_type', TaskType::Workflow) + ->where('status', TaskStatus::Ready)->exists()) { + return; + } + $this->runTaskFor($run); + } + $this->fail('The finite parent task drain did not settle.'); + } +} + +#[Signal('proceed')] +final class MilestoneParent extends Workflow +{ + public array $admission = []; + + public function handle(string $mutation, string $childMode = ''): array + { + $child = self::startChild(MilestoneChild::class, $childMode); + $this->admission = $child->toArray(); + self::awaitSignal('proceed'); + if ($mutation !== '') { + $value = $child->toArray(); + $value[$mutation] = $mutation === 'startSequence' ? 99 : 'foreign-value'; + $child = StartedChildWorkflow::fromArray($value); + } + return [ + 'admission' => $this->admission, + 'result' => $child->await(), + ]; + } + + #[QueryMethod] + public function admitted(): array + { + return $this->admission; + } +} + +final class MilestoneChild extends Workflow +{ + public function handle(string $mode = '', int $generation = 0): string + { + if ($mode === 'continue' && $generation === 0) { + self::continueAsNew($mode, 1); + } + if ($mode === 'retry' && $this->run->run_number === 1) { + throw new RuntimeException('first child attempt fails'); + } + return 'child-result'; + } +} + +final class OrdinaryMilestoneParent extends Workflow +{ + public function handle(): mixed + { + return self::child(MilestoneChild::class); + } +} + +final class CancellationMilestoneParent extends Workflow +{ + public string $state = 'waiting'; + + public function handle(string $policy, bool $secondChild): void + { + try { + $child = self::startChild( + TestChildPolicyCleanupWorkflow::class, + new ChildWorkflowOptions(cancellationPolicy: CancellationPolicy::from($policy)) + ); + if ($secondChild) { + self::startChild(MilestoneChild::class); + } + $child->await(); + } catch (WorkflowCancellationRequestedException) { + \Workflow\V2\cancellationShield(function (): void { + \Workflow\V2\timer(1); + \Workflow\V2\upsertMemo([ + 'parent_cleanup' => 'complete', + ]); + $this->state = 'cleaned'; + }); + } + } + + #[QueryMethod] + public function phase(): string + { + return $this->state; + } +}