From ed7cf4bf9504f8ff06a39fb27f66725ece4140cd Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 19:42:23 +0000 Subject: [PATCH 1/6] Add an embedded child admission milestone and separate await --- CHANGELOG.md | 10 + docs/business-state-machines.md | 19 +- src/V2/Enums/HistoryEventType.php | 1 + src/V2/StartedChildWorkflow.php | 77 +++++ .../Support/AwaitStartedChildWorkflowCall.php | 16 + src/V2/Support/ChildStartHistory.php | 119 +++++++ .../Support/HistoryEventPayloadContract.php | 3 + src/V2/Support/HistoryTimeline.php | 3 + src/V2/Support/QueryStateReplayer.php | 56 ++++ src/V2/Support/StartChildWorkflowCall.php | 15 + src/V2/Support/WorkflowExecutor.php | 141 +++++++- src/V2/Support/WorkflowFiberRunner.php | 17 + src/V2/Support/WorkflowStepHistory.php | 17 + src/V2/Workflow.php | 11 + src/V2/functions.php | 15 + tests/Unit/V2/StartedChildWorkflowTest.php | 307 ++++++++++++++++++ 16 files changed, 825 insertions(+), 2 deletions(-) create mode 100644 src/V2/StartedChildWorkflow.php create mode 100644 src/V2/Support/AwaitStartedChildWorkflowCall.php create mode 100644 src/V2/Support/ChildStartHistory.php create mode 100644 src/V2/Support/StartChildWorkflowCall.php create mode 100644 tests/Unit/V2/StartedChildWorkflowTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 8706e54be..7af7931a1 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. + - Failed-run redrive accepts a verified linear activity retry chain. A retried completed prefix is reused and only the failed boundary step runs again. Substituted attempt/task identities refuse with diff --git a/docs/business-state-machines.md b/docs/business-state-machines.md index 6bb9b8410..43c415876 100644 --- a/docs/business-state-machines.md +++ b/docs/business-state-machines.md @@ -177,7 +177,24 @@ 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 portable 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. + +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 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, + ]; + } + + 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)) { + throw self::mismatch($sequence, 'The terminal outcome belongs to a different child admission.'); + } + } + + 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/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..d3f05e9fa 100644 --- a/src/V2/Support/QueryStateReplayer.php +++ b/src/V2/Support/QueryStateReplayer.php @@ -523,12 +523,67 @@ 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); + $resolution = ChildRunHistory::resolutionEventForSequence($run, $child->startSequence); + // 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); + } + $target = ChildRunHistory::childRunForSequence($run, $child->startSequence); + ChildStartHistory::assertResolution($child, $resolution, $target, $historySequence); + $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 +712,7 @@ public function replayState(WorkflowRun $run): ReplayState 'activity_type' => $call->activity, ] : ($call instanceof ChildWorkflowCall ? [ 'child_workflow_type' => $call->workflow, + 'child_completion_mode' => 'result', ] : []), ); 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,127 @@ 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); + $resolution = ChildRunHistory::resolutionEventForSequence($run, $child->startSequence); + $target = ChildRunHistory::childRunForSequence($run, $child->startSequence); + ChildStartHistory::assertResolution($child, $resolution, $target, $sequence); + $delivery = $this->deliverCancellationAtCall( + $run, + $task, + $sequence, + 'child', + $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 ($awaited === null) { + WorkflowHistoryEvent::record( + $run, + HistoryEventType::ChildWorkflowAwaited, + ChildStartHistory::awaitPayload($child, $sequence), + $task + ); + } + 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 +1740,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 +1912,7 @@ private function parallelMemberIdentity(mixed $call): array ], $call instanceof ChildWorkflowCall => [ 'child_workflow_type' => $call->workflow, + 'child_completion_mode' => 'result', ], default => [], }; @@ -2748,6 +2881,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); @@ -2878,6 +3012,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, @@ -2973,6 +3110,7 @@ private function scheduleChildWorkflowOrFailRun( ChildWorkflowCall $childWorkflowCall, ?array $parallelMetadata = null, bool $parkRun = true, + bool $startMilestone = false, ): ?WorkflowTask { try { return $this->scheduleChildWorkflow( @@ -2981,7 +3119,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/Unit/V2/StartedChildWorkflowTest.php b/tests/Unit/V2/StartedChildWorkflowTest.php new file mode 100644 index 000000000..1c32bec9b --- /dev/null +++ b/tests/Unit/V2/StartedChildWorkflowTest.php @@ -0,0 +1,307 @@ + '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 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 startParent(string $mutation = ''): array + { + $stub = WorkflowStub::make(MilestoneParent::class, 'milestone-parent', 'client-a'); + $stub->start($mutation); + $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): array + { + $child = self::startChild(MilestoneChild::class); + $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 + { + return 'child-result'; + } +} + +final class OrdinaryMilestoneParent extends Workflow +{ + public function handle(): mixed + { + return self::child(MilestoneChild::class); + } +} From d57c56b8d7c82adb5ca695f0b128f7ab9402862e Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 20:30:42 +0000 Subject: [PATCH 2/6] fix: bind started-child awaits to native lineage and cancellation --- docs/business-state-machines.md | 4 +- src/V2/Support/ChildStartHistory.php | 129 ++++++- .../CooperativeCancellationDelivery.php | 50 ++- src/V2/Support/QueryStateReplayer.php | 18 +- src/V2/Support/WorkflowExecutor.php | 25 +- ...child-start-cannot-become-result-wait.json | 46 +++ tests/Unit/V2/StartedChildWorkflowTest.php | 318 +++++++++++++++++- 7 files changed, 565 insertions(+), 25 deletions(-) create mode 100644 tests/Fixtures/V2/ReplayRegression/child-start-cannot-become-result-wait.json diff --git a/docs/business-state-machines.md b/docs/business-state-machines.md index 43c415876..8334153cc 100644 --- a/docs/business-state-machines.md +++ b/docs/business-state-machines.md @@ -192,7 +192,9 @@ $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 portable 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. +`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. diff --git a/src/V2/Support/ChildStartHistory.php b/src/V2/Support/ChildStartHistory.php index 636542207..219345586 100644 --- a/src/V2/Support/ChildStartHistory.php +++ b/src/V2/Support/ChildStartHistory.php @@ -57,6 +57,28 @@ public static function assertReceipt(WorkflowRun $run, StartedChildWorkflow $chi } } + 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, @@ -90,6 +112,65 @@ public static function awaitPayload(StartedChildWorkflow $child, int $sequence): ]; } + /** + * 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 = $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 ($previous === null && $id !== $child->childRunId) { + throw self::mismatch($sequence, 'The child chain does not begin at the admitted run.'); + } + /** @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, @@ -102,11 +183,57 @@ public static function assertResolution( 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)) { + !== $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 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( diff --git a/src/V2/Support/CooperativeCancellationDelivery.php b/src/V2/Support/CooperativeCancellationDelivery.php index 1f898ec95..31cf49d64 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,21 @@ public static function validate( return $membership; } + if ($callKind === 'started_child_await') { + $boundary = self::validateCallBoundary( + $run, + $request, + $sequence, + $callKind, + $sequenceSpan, + $operationSequence, + $operationSequenceSpan + ); + if ($boundary !== null) { + return $boundary; + } + } + $existing = self::recorded($run); if ($existing instanceof WorkflowHistoryEvent) { return ($existing->workflow_command_id === $requestId @@ -150,7 +167,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 +193,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 +354,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) { diff --git a/src/V2/Support/QueryStateReplayer.php b/src/V2/Support/QueryStateReplayer.php index d3f05e9fa..08a8bdd7a 100644 --- a/src/V2/Support/QueryStateReplayer.php +++ b/src/V2/Support/QueryStateReplayer.php @@ -554,15 +554,28 @@ public function replayState(WorkflowRun $run): ReplayState $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); } - $target = ChildRunHistory::childRunForSequence($run, $child->startSequence); - ChildStartHistory::assertResolution($child, $resolution, $target, $historySequence); $this->syncWorkflowCursor($workflow, $sequence + 1); $current = $resolution->event_type === HistoryEventType::ChildRunCompleted ? $workflowExecution->send( @@ -2078,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/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php index 4c3adafa0..cf1fb14a2 100644 --- a/src/V2/Support/WorkflowExecutor.php +++ b/src/V2/Support/WorkflowExecutor.php @@ -1680,14 +1680,25 @@ private function executeStartedChildAwait( 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 = ChildRunHistory::childRunForSequence($run, $child->startSequence); - ChildStartHistory::assertResolution($child, $resolution, $target, $sequence); + $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, - 'child', + 'started_child_await', $resolution, $frame->execution, operationSequence: $child->startSequence @@ -1698,14 +1709,6 @@ private function executeStartedChildAwait( if ($delivery === CancellationDeliveryState::Delivered) { return $this->resumeCancellation($run, $task, $frame); } - if ($awaited === null) { - WorkflowHistoryEvent::record( - $run, - HistoryEventType::ChildWorkflowAwaited, - ChildStartHistory::awaitPayload($child, $sequence), - $task - ); - } if ($resolution === null && $target !== null) { $status = ChildRunHistory::resolvedStatus(null, $target); if ($status !== null && $status->isTerminal()) { 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/StartedChildWorkflowTest.php b/tests/Unit/V2/StartedChildWorkflowTest.php index 1c32bec9b..aa2a9be0e 100644 --- a/tests/Unit/V2/StartedChildWorkflowTest.php +++ b/tests/Unit/V2/StartedChildWorkflowTest.php @@ -4,23 +4,32 @@ namespace Tests\Unit\V2; +use Illuminate\Support\Carbon; use Illuminate\Support\Facades\Queue; use PHPUnit\Framework\Attributes\DataProvider; +use RuntimeException; +use Tests\Fixtures\V2\TestChildPolicyCleanupWorkflow; use Tests\TestCase; use Workflow\QueryMethod; use Workflow\V2\Attributes\Signal; +use Workflow\V2\Enums\CancellationPolicy; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Enums\TaskType; use Workflow\V2\Exceptions\HistoryEventShapeMismatchException; use Workflow\V2\Exceptions\UnsupportedWorkflowYieldException; +use Workflow\V2\Exceptions\WorkflowCancellationRequestedException; +use Workflow\V2\Jobs\RunTimerTask; use Workflow\V2\Jobs\RunWorkflowTask; +use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowInstance; use Workflow\V2\Models\WorkflowLink; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowTask; use Workflow\V2\StartedChildWorkflow; +use Workflow\V2\Support\ChildWorkflowOptions; +use Workflow\V2\Support\CooperativeCancellationDelivery; use Workflow\V2\Support\QueryStateReplayer; use Workflow\V2\Support\WorkflowFiberRunner; use Workflow\V2\Support\WorkflowStepHistory; @@ -59,11 +68,13 @@ public function testConfirmedStartPrecedesChildExecutionAndSurvivesAColdDuplicat $this->assertSame(2, WorkflowInstance::query()->count()); $this->assertSame( 1, - $run->historyEvents()->where('event_type', HistoryEventType::ChildWorkflowScheduled)->count() + $run->historyEvents() + ->where('event_type', HistoryEventType::ChildWorkflowScheduled)->count() ); $this->assertSame( 0, - $run->historyEvents()->where('event_type', HistoryEventType::ChildWorkflowAwaited)->count() + $run->historyEvents() + ->where('event_type', HistoryEventType::ChildWorkflowAwaited)->count() ); $this->assertFalse($parent->completed()); } @@ -76,7 +87,8 @@ public function testLaterAwaitUsesTheSameChildWithoutAnotherAdmission(): void $this->drainParent($run); $this->assertSame( 1, - $run->historyEvents()->where('event_type', HistoryEventType::ChildWorkflowAwaited)->count() + $run->historyEvents() + ->where('event_type', HistoryEventType::ChildWorkflowAwaited)->count() ); $this->assertFalse($parent->completed()); $child = WorkflowRun::query()->findOrFail($receipt['childRunId']); @@ -107,6 +119,220 @@ public function testChildMayCompleteBeforeTheLaterAwaitWithoutChangingAdmissionI $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', @@ -125,7 +351,8 @@ public function testForeignOrStaleReceiptsCannotAuthorizeAnAwait(string $field): $this->assertFalse($run->fresh()->status->isTerminal()); $this->assertSame( 0, - $run->historyEvents()->where('event_type', HistoryEventType::ChildWorkflowAwaited)->count() + $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()); @@ -210,10 +437,44 @@ public function testAStartedMilestoneCannotBeReplayedAsAnOrdinaryTerminalChildCa } } - private function startParent(string $mutation = ''): array + 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); + $stub->start($mutation, $childMode); $run = WorkflowRun::query()->findOrFail($stub->runId()); $this->runTaskFor($run); return [$stub, $run]; @@ -267,9 +528,9 @@ final class MilestoneParent extends Workflow { public array $admission = []; - public function handle(string $mutation): array + public function handle(string $mutation, string $childMode = ''): array { - $child = self::startChild(MilestoneChild::class); + $child = self::startChild(MilestoneChild::class, $childMode); $this->admission = $child->toArray(); self::awaitSignal('proceed'); if ($mutation !== '') { @@ -292,8 +553,14 @@ public function admitted(): array final class MilestoneChild extends Workflow { - public function handle(): string + 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'; } } @@ -305,3 +572,36 @@ 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; + } +} From 09ec4ffe7e8af4925cce3b8fba2ab84dfef14596 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 20:36:57 +0000 Subject: [PATCH 3/6] style: preserve the canonical child cancellation assertion layout --- tests/Unit/V2/StartedChildWorkflowTest.php | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/Unit/V2/StartedChildWorkflowTest.php b/tests/Unit/V2/StartedChildWorkflowTest.php index aa2a9be0e..7eb2a45e6 100644 --- a/tests/Unit/V2/StartedChildWorkflowTest.php +++ b/tests/Unit/V2/StartedChildWorkflowTest.php @@ -283,7 +283,8 @@ public function testStartedAwaitCancellationHasACanonicalColdBoundary(Cancellati $this->assertSame('cleaned', (new QueryStateReplayer())->query($run->fresh(), 'phase')); $this->assertSame( $policy === CancellationPolicy::Abandon, - $child->fresh()->cancellation_request_command_id === null + $child->fresh() +->cancellation_request_command_id === null ); if ($policy !== CancellationPolicy::WaitCancellationCompleted) { $this->assertSame(RunStatus::Waiting, $child->fresh()->status); From 95c0c34f51884d4ee23324dafb4eb12a0d320fec Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 21:00:55 +0000 Subject: [PATCH 4/6] refactor: separate child receipt and cancellation boundary checks --- src/V2/Support/ChildStartHistory.php | 33 +++++++++++------ .../CooperativeCancellationDelivery.php | 35 ++++++++++++------- 2 files changed, 44 insertions(+), 24 deletions(-) diff --git a/src/V2/Support/ChildStartHistory.php b/src/V2/Support/ChildStartHistory.php index 219345586..a8d6d4e96 100644 --- a/src/V2/Support/ChildStartHistory.php +++ b/src/V2/Support/ChildStartHistory.php @@ -127,17 +127,7 @@ public static function target( $previous = null; $seen = []; foreach ($starts as $event) { - $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 ($previous === null && $id !== $child->childRunId) { - throw self::mismatch($sequence, 'The child chain does not begin at the admitted run.'); - } + $id = self::recordedRunId($parent, $child, $event, $seen, $sequence); /** @var WorkflowRun|null $target */ $target = ConfiguredV2Models::query('run_model', WorkflowRun::class) ->where('namespace', $child->namespace) @@ -190,6 +180,27 @@ public static function assertResolution( } } + 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, diff --git a/src/V2/Support/CooperativeCancellationDelivery.php b/src/V2/Support/CooperativeCancellationDelivery.php index 31cf49d64..7c6bb5909 100644 --- a/src/V2/Support/CooperativeCancellationDelivery.php +++ b/src/V2/Support/CooperativeCancellationDelivery.php @@ -96,19 +96,9 @@ public static function validate( return $membership; } - if ($callKind === 'started_child_await') { - $boundary = self::validateCallBoundary( - $run, - $request, - $sequence, - $callKind, - $sequenceSpan, - $operationSequence, - $operationSequenceSpan - ); - if ($boundary !== null) { - return $boundary; - } + $boundary = self::validateStartedChildBoundary($run, $request, $sequence, $callKind, $operationSequence); + if ($boundary !== null) { + return $boundary; } $existing = self::recorded($run); @@ -372,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 From fab709b6523e9a1d721dd045dbb8745154b44cb9 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 23:38:33 +0000 Subject: [PATCH 5/6] Synchronize SQLite contention fixture with root rollback --- .../HostServiceQuerySqliteWriteRetryTest.php | 41 ++++++++++++++++--- 1 file changed, 36 insertions(+), 5 deletions(-) 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)); From 1a17bd4b3e1c5eb0f7f11c5bf81a617f004e454a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 00:27:05 +0000 Subject: [PATCH 6/6] Assert immutable cancellation date results on PHP 8.5 --- tests/Unit/V2/CancellationContextTest.php | 6 ++---- tests/Unit/V2/ScopedCancellationContextTest.php | 6 ++---- 2 files changed, 4 insertions(+), 8 deletions(-) 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/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);