diff --git a/docs/business-state-machines.md b/docs/business-state-machines.md index 8334153cc..731a864dd 100644 --- a/docs/business-state-machines.md +++ b/docs/business-state-machines.md @@ -2,7 +2,7 @@ This is the first implementation slice of the composable business state-machine proposal. It provides an optional embedded V2 transition protocol with a four-method step contract, recorded assessments, versioned profiles and active-run native Saga integration. Existing workflows and `Workflow\States\StateMachine` are unchanged. Symfony Workflow is not a dependency. -**Experimental.** The `Workflow\V2\StateMachine` API is under review and may change in a minor release until the nested slices land; it carries no backward-compatibility promise yet. This slice does not implement state-bound child start/adoption, retained compensation scopes, a business-state database schema or a persistence adapter for a specific application. It does not complete that proposal. +**Experimental.** The `Workflow\V2\StateMachine` API is under review and may change in a minor release until the nested slices land; it carries no backward-compatibility promise yet. The complete state-bound binding runtime, retained compensation adapter, business-state database schema and application persistence integration remain unfinished. This does not complete the proposal. ## Responsibilities and flow @@ -231,3 +231,15 @@ Tests exercise refusal/unknown outcomes, source and scope substitution, changed | 21–22 | Runnable V2 regression example, existing APIs unchanged | Full integration example, persistence crash tests and upstream qualification | This table reports scope, not completion of the issue's unchecked acceptance criteria. + +## Native child reservation boundary + +The experimental `Binding\ChildReservationStore` provides the native reservation needed before state-bound child admission. Call `reserve()` and `close()` inside recorded application activities, outside caller-owned SQL transactions. Each method owns its transaction so a returned closure is committed and host authorization runs before native locks. The supplied `BindingAuthorization` must validate the authenticated principal, original application/business owner, generation and requested operation before protected storage is read. Identity values and a serialized reservation are not credentials. + +A reservation pins the original parent instance/run, business execution, slot/generation, policies, required outcome, portable input fingerprint, child definition fingerprint and child options. Repeating the exact reservation returns the stored descriptor. Reusing that lookup key with changed content refuses. New reservations require the active original parent; retrieving an existing reservation does not follow a newer current run or state entry. + +Within the original parent workflow, `ChildReservation::start(...$arguments)` uses the normal confirmed native child-start command. Admission locks the reserved child instance before payload storage or child creation, checks the actual executing parent and pinned definition/input/options, and commits the original child receipt with the run and paired parent history. Replay compares the original reservation and reuses its original admission. It does not start another child. + +`close()` locks that same instance. If admission committed first, it returns the exact `ChildBinding`. Otherwise it permanently closes the unused reservation and returns `NeverAdmittedChild` with the stored closure identity. Repeating close after a lost acknowledgement returns the same result. Late native work and ordinary start/signal-with-start cannot reopen that reservation. A missing record, malformed state or uncertain run pointer refuses rather than proving absence. Closed unused instances must remain retained; the reservation migration refuses rollback while any reservation exists. + +This primitive does not resolve an existing `dw.child-scope` participant whose child scope is missing. The retained-scope adapter must independently reread authoritative closure evidence or wait for the exact admitted child's verified compensation. A deserialized `NeverAdmittedChild`, technical cancellation or lookup absence is insufficient. That adapter, protected business cancellation/commit integration, real database race qualification and the two-integration example remain required before the full binding contract is delivered. Forward ownership transfer through continue-as-new or redrive is not supported for these reserved children. diff --git a/src/V2/StateMachine/Binding/BindingAuthorization.php b/src/V2/StateMachine/Binding/BindingAuthorization.php new file mode 100644 index 000000000..4d5574f13 --- /dev/null +++ b/src/V2/StateMachine/Binding/BindingAuthorization.php @@ -0,0 +1,14 @@ + 512) { + throw new InvalidArgumentException('Binding identity fields must contain between 1 and 512 bytes.'); + } + } + foreach ([$parentExecution->policyFingerprint, $childPolicyFingerprint, $inputFingerprint] as $fingerprint) { + if (preg_match('/\A[0-9a-f]{64}\z/D', $fingerprint) !== 1) { + throw new InvalidArgumentException('Binding policy and input fingerprints must be SHA-256 values.'); + } + } + } + + /** + * Exact original generation lookup; never substitute a latest/current binding. + */ + public function key(): string + { + return hash('sha256', json_encode([ + 'version' => 1, + 'scope' => $this->parentExecution->scope, + 'subject' => $this->parentExecution->subject, + 'executionId' => $this->parentExecution->executionId, + 'namespace' => $this->namespace, + 'parentInstanceId' => $this->parentInstanceId, + 'parentRunId' => $this->parentRunId, + 'slot' => $this->slot, + 'generation' => $this->generation, + ], JSON_THROW_ON_ERROR)); + } + + /** + * The immutable descriptor must also match when an existing key is reused. + */ + public function fingerprint(): string + { + return hash('sha256', json_encode($this->toArray(), JSON_THROW_ON_ERROR)); + } + + /** + * @return array + */ + public function toArray(): array + { + return [ + 'version' => 1, + 'parentExecution' => [ + 'scope' => $this->parentExecution->scope, + 'subject' => $this->parentExecution->subject, + 'executionId' => $this->parentExecution->executionId, + 'policyFingerprint' => $this->parentExecution->policyFingerprint, + 'sourceState' => $this->parentExecution->sourceState, + 'sourceRevision' => $this->parentExecution->sourceRevision, + ], + 'namespace' => $this->namespace, + 'parentInstanceId' => $this->parentInstanceId, + 'parentRunId' => $this->parentRunId, + 'slot' => $this->slot, + 'generation' => $this->generation, + 'hostingState' => $this->hostingState, + 'childWorkflowType' => $this->childWorkflowType, + 'childPolicyFingerprint' => $this->childPolicyFingerprint, + 'requiredOutcome' => $this->requiredOutcome, + 'inputFingerprint' => $this->inputFingerprint, + 'childOptions' => $this->childOptions->toSnapshot(), + ]; + } + + /** + * @param array $payload + */ + public static function fromArray(array $payload): self + { + $strings = ['namespace', 'parentInstanceId', 'parentRunId', 'slot', 'generation', + 'hostingState', 'childWorkflowType', 'childPolicyFingerprint', 'requiredOutcome', 'inputFingerprint']; + self::strings($payload, $strings, ['version', 'parentExecution', 'childOptions']); + if ($payload['version'] !== 1 || ! is_array($payload['parentExecution']) || ! is_array( + $payload['childOptions'] + )) { + throw new InvalidArgumentException('Unsupported or malformed binding identity.'); + } + $parent = $payload['parentExecution']; + self::strings( + $parent, + ['scope', 'subject', 'executionId', 'policyFingerprint', 'sourceState', 'sourceRevision'] + ); + $options = $payload['childOptions']; + self::strings($options, ['parent_close_policy', 'cancellation_policy'], ['connection', 'queue']); + foreach (['connection', 'queue'] as $key) { + if ($options[$key] !== null && ! is_string($options[$key])) { + throw new InvalidArgumentException('Binding child routing must be a string or null.'); + } + } + return new self( + new ExecutionReference( + $parent['scope'], + $parent['subject'], + $parent['executionId'], + $parent['policyFingerprint'], + $parent['sourceState'], + $parent['sourceRevision'] + ), + $payload['namespace'], + $payload['parentInstanceId'], + $payload['parentRunId'], + $payload['slot'], + $payload['generation'], + $payload['hostingState'], + $payload['childWorkflowType'], + $payload['childPolicyFingerprint'], + $payload['requiredOutcome'], + $payload['inputFingerprint'], + new ChildWorkflowOptions( + ParentClosePolicy::from($options['parent_close_policy']), + $options['connection'], + $options['queue'], + CancellationPolicy::from($options['cancellation_policy']), + ), + ); + } + + /** + * @param array $payload + * @param list $strings + * @param list $other + */ + private static function strings(array $payload, array $strings, array $other = []): void + { + $expected = [...$strings, ...$other]; + if (count($payload) !== count($expected) || array_diff($expected, array_keys($payload)) !== []) { + throw new InvalidArgumentException('Unexpected or missing binding identity field.'); + } + foreach ($strings as $key) { + if (! is_string($payload[$key])) { + throw new InvalidArgumentException('Binding identity fields must be strings.'); + } + } + } +} diff --git a/src/V2/StateMachine/Binding/ChildBinding.php b/src/V2/StateMachine/Binding/ChildBinding.php new file mode 100644 index 000000000..fda3ea6c4 --- /dev/null +++ b/src/V2/StateMachine/Binding/ChildBinding.php @@ -0,0 +1,59 @@ +namespace !== $admission->namespace + || $identity->parentInstanceId !== $admission->parentInstanceId + || $identity->parentRunId !== $admission->parentRunId + || $identity->childWorkflowType !== $admission->workflowType) { + throw new InvalidArgumentException('The child admission belongs to a different binding owner or type.'); + } + } + + public function matches(BindingIdentity $expected): bool + { + return $this->identity->toArray() === $expected->toArray(); + } + + /** + * @return array + */ + public function toArray(): array + { + return [ + 'version' => 1, + 'identity' => $this->identity->toArray(), + 'admission' => $this->admission->toArray(), + ]; + } + + /** + * @param array $payload + */ + public static function fromArray(array $payload): self + { + if (count($payload) !== 3 || ($payload['version'] ?? null) !== 1 + || ! is_array($payload['identity'] ?? null) || ! is_array($payload['admission'] ?? null)) { + throw new InvalidArgumentException('Unsupported or malformed child binding.'); + } + return new self(BindingIdentity::fromArray($payload['identity']), StartedChildWorkflow::fromArray( + $payload['admission'] + )); + } +} diff --git a/src/V2/StateMachine/Binding/ChildReservation.php b/src/V2/StateMachine/Binding/ChildReservation.php new file mode 100644 index 000000000..fcb93ea82 --- /dev/null +++ b/src/V2/StateMachine/Binding/ChildReservation.php @@ -0,0 +1,90 @@ + 512 + || trim($authorizationVersion) === '' || strlen($authorizationVersion) > 128 + || preg_match('/\Asha256:[0-9a-f]{64}\z/D', $definitionFingerprint) !== 1) { + throw new InvalidArgumentException('Invalid child reservation definition or authorization version.'); + } + PortableDescriptor::encode($this->toArray()); + } + + public function childInstanceId(): string + { + return 'dw-bound-child-' . $this->identity->key(); + } + + /** + * Start through the original parent's native command/history boundary. + */ + public function start(mixed ...$arguments): StartedChildWorkflow + { + return WorkflowFiberContext::suspend(new StartChildWorkflowCall(new ChildWorkflowCall( + $this->workflowClass, + $arguments, + $this->identity->childOptions, + $this, + ))); + } + + /** + * @param array $arguments + */ + public static function inputFingerprint(array $arguments): string + { + return hash('sha256', PortableDescriptor::encode($arguments)); + } + + /** + * @return array + */ + public function toArray(): array + { + return [ + 'version' => 1, + 'identity' => $this->identity->toArray(), + 'workflowClass' => $this->workflowClass, + 'definitionFingerprint' => $this->definitionFingerprint, + 'authorizationVersion' => $this->authorizationVersion, + ]; + } + + /** + * @param array $value + */ + public static function fromArray(array $value): self + { + if (count($value) !== 5 || ($value['version'] ?? null) !== 1 + || ! is_array($value['identity'] ?? null) + || ! is_string($value['workflowClass'] ?? null) + || ! is_string($value['definitionFingerprint'] ?? null) + || ! is_string($value['authorizationVersion'] ?? null)) { + throw new InvalidArgumentException('Unsupported or malformed child reservation.'); + } + return new self( + BindingIdentity::fromArray($value['identity']), + $value['workflowClass'], + $value['definitionFingerprint'], + $value['authorizationVersion'], + ); + } +} diff --git a/src/V2/StateMachine/Binding/ChildReservationStore.php b/src/V2/StateMachine/Binding/ChildReservationStore.php new file mode 100644 index 000000000..d0ce15a43 --- /dev/null +++ b/src/V2/StateMachine/Binding/ChildReservationStore.php @@ -0,0 +1,315 @@ +authorize($identity, $principal, 'reserve'); + if (! is_a($workflowClass, \Workflow\V2\Workflow::class, true) + || TypeRegistry::for($workflowClass) !== $identity->childWorkflowType) { + throw new LogicException('The binding child type does not match its definition.'); + } + $fingerprint = WorkflowDefinition::fingerprint($workflowClass); + if ($fingerprint === null) { + throw new LogicException('The child definition has no immutable fingerprint.'); + } + $candidate = new ChildReservation($identity, $workflowClass, $fingerprint, $version); + $query = self::instances(); + $connection = $query->getModel() + ->getConnection(); + if (! $connection->getSchemaBuilder()->hasColumn($query->getModel()->getTable(), 'native_child_reservation')) { + throw new LogicException('Native child reservations require the reservation migration.'); + } + + return SqliteTransaction::run(static function () use ($candidate, $identity): ChildReservation { + // The parent row serializes first creation of this deterministic child. + $parent = self::instances()->lockForUpdate()->find($identity->parentInstanceId); + $child = self::instances()->lockForUpdate()->find($candidate->childInstanceId()); + if ($child !== null) { + self::assertInstance($child, $candidate); + $state = self::read($child); + $stored = ChildReservation::fromArray($state['reservation']); + self::matching($child, $stored); + if ($stored->identity->toArray() !== $identity->toArray() + || $stored->workflowClass !== $candidate->workflowClass + || $stored->definitionFingerprint !== $candidate->definitionFingerprint) { + throw new LogicException('An existing binding generation has a different immutable descriptor.'); + } + return $stored; + } + $parentRun = ConfiguredV2Models::query('run_model', WorkflowRun::class) + ->whereKey($identity->parentRunId) + ->first(); + if ($parent === null || $parent->namespace !== $identity->namespace + || $parent->current_run_id !== $identity->parentRunId + || $parentRun === null || $parentRun->workflow_instance_id !== $parent->id + || $parentRun->namespace !== $identity->namespace + || $parentRun->status->isTerminal() || $parentRun->cancellation_requested_at !== null) { + throw new LogicException('A new binding requires its active original parent run.'); + } + /** @var WorkflowInstance $child */ + $child = self::instances()->create([ + 'id' => $candidate->childInstanceId(), + 'workflow_class' => $candidate->workflowClass, + 'workflow_type' => $identity->childWorkflowType, + 'namespace' => $identity->namespace, + 'reserved_at' => now(), + 'run_count' => 0, + ]); + self::write($child, [ + 'version' => 1, + 'reservation' => $candidate->toArray(), + 'status' => 'reserved', + 'admission' => null, + 'closureId' => null, + ]); + return $candidate; + }, admission: static function () use ($identity): void { + self::acquireWriter($identity->parentInstanceId); + }, connection: $connection); + } + + /** + * Linearize against native admission. A lost acknowledgement returns the + * same closed receipt or the exact admitted child, never lookup absence. + */ + public function close(ChildReservation $reservation, string $principal): ChildBinding|NeverAdmittedChild + { + $this->authorize($reservation->identity, $principal, 'close'); + return SqliteTransaction::run(static function () use ($reservation): ChildBinding|NeverAdmittedChild { + $child = self::instances()->lockForUpdate()->findOrFail($reservation->childInstanceId()); + $state = self::matching($child, $reservation); + if ($state['status'] === 'admitted') { + return ChildBinding::fromArray($state['admission']); + } + if ((int) $child->run_count !== 0 || $child->current_run_id !== null || $child->runs()->exists()) { + throw new LogicException('An uncertain child cannot be certified never admitted.'); + } + if ($state['status'] === 'reserved') { + $state['status'] = 'closed'; + $state['closureId'] = (string) Str::ulid(); + self::write($child, $state); + } + return new NeverAdmittedChild($reservation, $state['closureId']); + }, admission: static function () use ($reservation): void { + self::acquireWriter($reservation->childInstanceId()); + }, connection: self::instances()->getModel()->getConnection()); + } + + /** + * @internal Native scheduler only, inside its existing root task transaction. + */ + public static function lockForAdmission(WorkflowRun $parent, ChildWorkflowCall $call): WorkflowInstance + { + $reservation = $call->reservation; + if ($reservation === null || $parent->getConnection()->transactionLevel() === 0 + || $parent->getConnection() !== self::instances()->getModel()->getConnection()) { + throw new LogicException('A reserved child needs native transactional admission.'); + } + $identity = $reservation->identity; + // Check the actual executing parent before looking up the protected child. + if ($parent->namespace !== $identity->namespace || $parent->workflow_instance_id !== $identity->parentInstanceId + || $parent->id !== $identity->parentRunId || $call->workflow !== $reservation->workflowClass + || ChildReservation::inputFingerprint($call->arguments) !== $identity->inputFingerprint + || $call->options?->toSnapshot() !== $identity->childOptions->toSnapshot()) { + throw new LogicException('The native child command differs from its reserved owner, definition or input.'); + } + $child = self::instances()->lockForUpdate()->findOrFail($reservation->childInstanceId()); + $state = self::matching($child, $reservation); + if ($state['status'] !== 'reserved' || (int) $child->run_count !== 0 || $child->current_run_id !== null + || $child->runs() + ->exists() || $parent->status->isTerminal() + || $parent->cancellation_requested_at !== null) { + throw new LogicException('The native child reservation is closed, already admitted or no longer eligible.'); + } + // Persisted data does not choose a class to autoload. First bind the + // supplied descriptor to the host-authorized reservation under its lock. + if (TypeRegistry::for($call->workflow) !== $identity->childWorkflowType + || WorkflowDefinition::fingerprint($call->workflow) !== $reservation->definitionFingerprint) { + throw new LogicException('The native child definition differs from its reserved source.'); + } + return $child; + } + + /** + * @internal Store together with the child's native run, command and paired parent history. + */ + public static function recordAdmission( + WorkflowInstance $child, + ChildReservation $reservation, + StartedChildWorkflow $receipt + ): void { + if ($child->getConnection()->transactionLevel() === 0 + || $child->current_run_id !== $receipt->childRunId || (int) $child->run_count !== 1) { + throw new LogicException('A native admission receipt must be committed with its original child run.'); + } + $state = self::matching($child, $reservation); + if ($state['status'] !== 'reserved' || $receipt->childInstanceId !== $child->id) { + throw new LogicException('The reservation cannot record this native admission.'); + } + $binding = new ChildBinding($reservation->identity, $receipt); + $state['status'] = 'admitted'; + $state['admission'] = $binding->toArray(); + self::write($child, $state); + } + + /** + * @internal Ordinary start/redrive/signal-with-start must never bypass the native fence. + */ + public static function assertOrdinaryStartAllowed(WorkflowInstance $instance): void + { + if ($instance->getRawOriginal('native_child_reservation') !== null) { + throw new LogicException('A reserved native child can only start through its original parent command.'); + } + } + + /** + * @internal Symmetric replay check: adding or removing a reservation changes the command. + */ + public static function assertRecordedCommand(?array $recorded, ?ChildReservation $reservation): void + { + $descriptor = $recorded === null ? null : ChildReservation::fromArray($recorded)->toArray(); + if ($descriptor !== $reservation?->toArray()) { + throw new LogicException('The recorded native child reservation differs from this command.'); + } + } + + private function authorize(BindingIdentity $identity, string $principal, string $operation): string + { + // A returned closure must already be committed, and host authorization + // must not execute beneath a caller's SQL locks. + $connection = self::instances()->getModel()->getConnection(); + if ($connection->transactionLevel() !== 0) { + throw new LogicException('Binding operations require their own recorded activity transaction boundary.'); + } + $version = $this->authorization->assertAllowed($identity, $principal, $operation); + if ($connection->transactionLevel() !== 0 + || self::instances()->getModel()->getConnection()->transactionLevel() !== 0) { + throw new LogicException('Binding authorization must leave the native transaction boundary closed.'); + } + if (trim($version) === '' || strlen($version) > 128) { + throw new LogicException('Binding authorization must identify its policy version.'); + } + return $version; + } + + /** + * @return Builder + */ + private static function instances(): Builder + { + return ConfiguredV2Models::query('instance_model', WorkflowInstance::class); + } + + private static function acquireWriter(string $id): void + { + $query = self::instances(); + $query->whereKey($id) + ->toBase() + ->update([ + $query->getModel() + ->getKeyName() => $id, + ]); + } + + /** + * @return array + */ + private static function matching(WorkflowInstance $child, ChildReservation $reservation): array + { + self::assertInstance($child, $reservation); + $state = self::read($child); + if ($state['reservation'] !== $reservation->toArray()) { + throw new LogicException('The stored native child reservation does not match the original descriptor.'); + } + return $state; + } + + private static function assertInstance(WorkflowInstance $child, ChildReservation $reservation): void + { + if ($child->id !== $reservation->childInstanceId() + || $child->namespace !== $reservation->identity->namespace + || $child->workflow_class !== $reservation->workflowClass + || $child->workflow_type !== $reservation->identity->childWorkflowType) { + throw new LogicException('The stored native child reservation does not match the original descriptor.'); + } + } + + /** + * @return array + */ + private static function read(WorkflowInstance $child): array + { + $raw = $child->getRawOriginal('native_child_reservation'); + if (! is_string($raw) || strlen($raw) > 32_768) { + throw new LogicException('The native child has no valid reservation record.'); + } + $state = json_decode($raw, true, 32, JSON_THROW_ON_ERROR); + if (! is_array($state) || count($state) !== 5 || ($state['version'] ?? null) !== 1 + || ! is_array($state['reservation'] ?? null) + || ! array_key_exists('admission', $state) || ! array_key_exists('closureId', $state) + || ! in_array($state['status'] ?? null, ['reserved', 'admitted', 'closed'], true)) { + throw new LogicException('Malformed native child reservation state.'); + } + $reservation = ChildReservation::fromArray($state['reservation']); + // SQL JSON types may reorder object keys. Reconstruct the closed value + // before comparing its immutable descriptor on every backend. + $state['reservation'] = $reservation->toArray(); + if ($state['status'] === 'admitted') { + if (! is_array($state['admission']) || $state['closureId'] !== null) { + throw new LogicException('Malformed native child admission state.'); + } + $binding = ChildBinding::fromArray($state['admission']); + if (! $binding->matches($reservation->identity) + || $binding->admission->childInstanceId !== $reservation->childInstanceId()) { + throw new LogicException('Stored child admission belongs to another reservation.'); + } + $state['admission'] = $binding->toArray(); + } elseif ($state['admission'] !== null || ($state['status'] === 'reserved' ? $state['closureId'] !== null + : ! is_string($state['closureId']) || $state['closureId'] === '')) { + throw new LogicException('Malformed native child closure state.'); + } + return $state; + } + + /** + * @param array $state + */ + private static function write(WorkflowInstance $child, array $state): void + { + $encoded = json_encode($state, JSON_THROW_ON_ERROR); + self::instances()->whereKey($child->id)->toBase()->update([ + 'native_child_reservation' => $encoded, + ]); + $child->setRawAttributes([ + ...$child->getAttributes(), + 'native_child_reservation' => $encoded, + ], true); + } +} diff --git a/src/V2/StateMachine/Binding/NeverAdmittedChild.php b/src/V2/StateMachine/Binding/NeverAdmittedChild.php new file mode 100644 index 000000000..1ea475b25 --- /dev/null +++ b/src/V2/StateMachine/Binding/NeverAdmittedChild.php @@ -0,0 +1,18 @@ +getRawOriginal('native_child_reservation') !== null) { + return null; + } + $nextRunNumber = ((int) WorkflowRun::query() ->where('workflow_instance_id', $failedChildRun->workflow_instance_id) ->max('run_number')) + 1; diff --git a/src/V2/Support/ChildStartHistory.php b/src/V2/Support/ChildStartHistory.php index a8d6d4e96..6937b8068 100644 --- a/src/V2/Support/ChildStartHistory.php +++ b/src/V2/Support/ChildStartHistory.php @@ -36,6 +36,9 @@ public static function receipt(WorkflowRun $run, int $sequence): ?StartedChildWo if ($scheduled->workflow_run_id !== $run->id || $started->workflow_run_id !== $run->id) { throw self::mismatch($sequence, 'Child admission history belongs to another parent.'); } + if (($scheduled->payload['child_reservation'] ?? null) !== ($started->payload['child_reservation'] ?? null)) { + throw self::mismatch($sequence, 'The native child reservation differs between its paired history events.'); + } return new StartedChildWorkflow( (string) $run->namespace, $run->workflow_instance_id, diff --git a/src/V2/Support/ChildWorkflowCall.php b/src/V2/Support/ChildWorkflowCall.php index dff6c3eec..3d8930dca 100644 --- a/src/V2/Support/ChildWorkflowCall.php +++ b/src/V2/Support/ChildWorkflowCall.php @@ -5,6 +5,7 @@ namespace Workflow\V2\Support; use Workflow\V2\Contracts\YieldedCommand; +use Workflow\V2\StateMachine\Binding\ChildReservation; final class ChildWorkflowCall implements YieldedCommand { @@ -15,6 +16,7 @@ public function __construct( public readonly string $workflow, public readonly array $arguments, public readonly ?ChildWorkflowOptions $options = null, + public readonly ?ChildReservation $reservation = null, ) { } } diff --git a/src/V2/Support/DefaultWorkflowControlPlane.php b/src/V2/Support/DefaultWorkflowControlPlane.php index 448229297..1157499c8 100644 --- a/src/V2/Support/DefaultWorkflowControlPlane.php +++ b/src/V2/Support/DefaultWorkflowControlPlane.php @@ -87,6 +87,12 @@ public function redrive(string $instanceId, string $failedRunId, array $options return; } + if ($instance->getRawOriginal('native_child_reservation') !== null) { + $result['reason'] = 'native_child_reservation_requires_original_run'; + $result['status'] = 409; + return; + } + /** @var WorkflowRun|null $source */ $source = $this->runQuery() ->where('workflow_instance_id', $instanceId) @@ -1252,6 +1258,8 @@ private function resolveOrCreateInstance( ->lockForUpdate() ->findOrFail($instanceId); + \Workflow\V2\StateMachine\Binding\ChildReservationStore::assertOrdinaryStartAllowed($instance); + if ($instance->namespace !== $namespace) { throw new LogicException(sprintf( 'Workflow instance [%s] cannot be reused with a different namespace.', diff --git a/src/V2/Support/DefaultWorkflowTaskBridge.php b/src/V2/Support/DefaultWorkflowTaskBridge.php index db9964fdc..6a8126458 100644 --- a/src/V2/Support/DefaultWorkflowTaskBridge.php +++ b/src/V2/Support/DefaultWorkflowTaskBridge.php @@ -5361,6 +5361,7 @@ private function applyContinueAsNew( ->findOrFail($run->workflow_instance_id); /** @var WorkflowRun $continuedRun */ + \Workflow\V2\StateMachine\Binding\ChildReservationStore::assertOrdinaryStartAllowed($instance); $continuedRun = WorkflowRun::query()->create([ 'workflow_instance_id' => $run->workflow_instance_id, 'run_number' => $run->run_number + 1, diff --git a/src/V2/Support/EmbeddedV2HistoryImport.php b/src/V2/Support/EmbeddedV2HistoryImport.php index 859a298cf..66f0ba84c 100644 --- a/src/V2/Support/EmbeddedV2HistoryImport.php +++ b/src/V2/Support/EmbeddedV2HistoryImport.php @@ -357,6 +357,8 @@ private static function writeBundle( throw new InvalidArgumentException('The stored workflow instance no longer matches the target namespace.'); } + \Workflow\V2\StateMachine\Binding\ChildReservationStore::assertOrdinaryStartAllowed($instance); + /** @var WorkflowRun $run */ $run = WorkflowRun::query()->create([ 'id' => $runId, diff --git a/src/V2/Support/HistoryEventPayloadContract.php b/src/V2/Support/HistoryEventPayloadContract.php index fbbffb49b..dee750104 100644 --- a/src/V2/Support/HistoryEventPayloadContract.php +++ b/src/V2/Support/HistoryEventPayloadContract.php @@ -643,6 +643,7 @@ final class HistoryEventPayloadContract 'ChildWorkflowScheduled' => [ 'sequence', 'child_completion_mode', + 'child_reservation', 'workflow_link_id', 'child_call_id', 'child_workflow_instance_id', @@ -668,6 +669,7 @@ final class HistoryEventPayloadContract 'child_workflow_run_id', 'child_call_id', 'child_workflow_type'], 'ChildRunStarted' => [ 'sequence', + 'child_reservation', 'workflow_link_id', 'child_call_id', 'child_workflow_instance_id', @@ -1174,5 +1176,17 @@ public static function assertKnownPayloadKeys(HistoryEventType $eventType, array implode(', ', $unknown), )); } + + if (array_key_exists('child_reservation', $payload)) { + if (! is_array($payload['child_reservation'])) { + throw new InvalidArgumentException('A native child reservation must be its versioned descriptor.'); + } + $reservation = \Workflow\V2\StateMachine\Binding\ChildReservation::fromArray($payload['child_reservation']); + if (($payload['child_workflow_instance_id'] ?? null) !== $reservation->childInstanceId() + || ($payload['child_workflow_type'] ?? null) !== $reservation->identity->childWorkflowType + || ($payload['child_workflow_class'] ?? null) !== $reservation->workflowClass) { + throw new InvalidArgumentException('Native child history does not match its reserved definition.'); + } + } } } diff --git a/src/V2/Support/QueryStateReplayer.php b/src/V2/Support/QueryStateReplayer.php index 08a8bdd7a..a7f9aa015 100644 --- a/src/V2/Support/QueryStateReplayer.php +++ b/src/V2/Support/QueryStateReplayer.php @@ -535,6 +535,13 @@ public function replayState(WorkflowRun $run): ReplayState $this->syncWorkflowCursor($workflow, $sequence); return new ReplayState($workflow, $sequence, $current); } + \Workflow\V2\StateMachine\Binding\ChildReservationStore::assertRecordedCommand( + ChildRunHistory::scheduledEventForSequence( + $run, + $historySequence + )?->payload['child_reservation'] ?? null, + $current->child->reservation, + ); $this->syncWorkflowCursor($workflow, $sequence + 1); $current = $workflowExecution->send($receipt, RedriveReplayClock::recordedAt( ChildRunHistory::scheduledEventForSequence($run, $historySequence) diff --git a/src/V2/Support/StandaloneActivityStartService.php b/src/V2/Support/StandaloneActivityStartService.php index 7ee7ef5b7..98e17a804 100644 --- a/src/V2/Support/StandaloneActivityStartService.php +++ b/src/V2/Support/StandaloneActivityStartService.php @@ -286,6 +286,8 @@ private function resolveOrCreateInstance( ->lockForUpdate() ->findOrFail($providedActivityId); + \Workflow\V2\StateMachine\Binding\ChildReservationStore::assertOrdinaryStartAllowed($instance); + if ($instance->workflow_type !== StandaloneActivityHostType::WORKFLOW_TYPE) { throw new InvalidArgumentException(sprintf( 'Identifier [%s] is already reserved for workflow type [%s] and cannot be reused for a standalone activity.', diff --git a/src/V2/Support/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php index 28120811b..9d79f48e6 100644 --- a/src/V2/Support/WorkflowExecutor.php +++ b/src/V2/Support/WorkflowExecutor.php @@ -55,6 +55,8 @@ use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Models\WorkflowTimer; use Workflow\V2\Models\WorkflowUpdate; +use Workflow\V2\StartedChildWorkflow; +use Workflow\V2\StateMachine\Binding\ChildReservationStore; use Workflow\V2\Workflow; use Workflow\WorkflowMetadata; @@ -1623,6 +1625,12 @@ private function executeChildStart( try { $receipt = ChildStartHistory::receipt($run, $sequence); $admitted = ChildRunHistory::scheduledEventForSequence($run, $sequence); + if ($admitted !== null) { + ChildReservationStore::assertRecordedCommand( + $admitted->payload['child_reservation'] ?? null, + $call->child->reservation, + ); + } $delivery = $this->deliverCancellationAtCall($run, $task, $sequence, 'child', $admitted, $frame->execution); if ($delivery === CancellationDeliveryState::Waiting) { return $this->waitForNextResumeSource($run, $task); @@ -2890,6 +2898,8 @@ private function scheduleChildWorkflow( bool $parkRun = true, bool $startMilestone = false, ): WorkflowTask { + $reservedChild = $childWorkflowCall->reservation === null ? null + : ChildReservationStore::lockForAdmission($run, $childWorkflowCall); StructuralLimits::guardPendingChildren($run); $this->logApproachingLimit(StructuralLimits::warnApproachingPendingChildren($run), $run); @@ -2925,7 +2935,7 @@ private function scheduleChildWorkflow( StructuralLimits::guardPayloadSize($storedChildArguments); /** @var WorkflowInstance $childInstance */ - $childInstance = WorkflowInstance::query()->create([ + $childInstance = $reservedChild ?? WorkflowInstance::query()->create([ 'workflow_class' => $childWorkflowCall->workflow, 'workflow_type' => $workflowType, 'namespace' => $run->namespace, @@ -2933,6 +2943,12 @@ private function scheduleChildWorkflow( 'started_at' => $now, 'run_count' => 1, ]); + if ($reservedChild !== null) { + $childInstance->forceFill([ + 'started_at' => $now, + 'run_count' => 1, + ])->save(); + } /** @var WorkflowRun $childRun */ $childRun = WorkflowRun::query()->create([ @@ -2948,8 +2964,10 @@ private function scheduleChildWorkflow( 'compatibility' => $run->compatibility ?? WorkerCompatibility::current(), 'payload_codec' => $childCodec, 'arguments' => $storedChildArguments, - 'connection' => RoutingResolver::workflowConnection($childWorkflowCall->workflow, $metadata), - 'queue' => RoutingResolver::workflowQueue($childWorkflowCall->workflow, $metadata), + 'connection' => ($reservedChild !== null ? $childWorkflowCall->options?->connection : null) + ?? RoutingResolver::workflowConnection($childWorkflowCall->workflow, $metadata), + 'queue' => ($reservedChild !== null ? $childWorkflowCall->options?->queue : null) + ?? RoutingResolver::workflowQueue($childWorkflowCall->workflow, $metadata), 'started_at' => $now, 'last_progress_at' => $now, 'last_history_sequence' => 0, @@ -3019,6 +3037,9 @@ private function scheduleChildWorkflow( WorkflowHistoryEvent::record($run, HistoryEventType::ChildWorkflowScheduled, array_merge([ 'sequence' => $sequence, + ...($childWorkflowCall->reservation === null ? [] : [ + 'child_reservation' => $childWorkflowCall->reservation->toArray(), + ]), ...($startMilestone ? [ 'child_completion_mode' => 'started', ] : []), @@ -3034,6 +3055,9 @@ private function scheduleChildWorkflow( WorkflowHistoryEvent::record($run, HistoryEventType::ChildRunStarted, array_merge([ 'sequence' => $sequence, + ...($childWorkflowCall->reservation === null ? [] : [ + 'child_reservation' => $childWorkflowCall->reservation->toArray(), + ]), 'workflow_link_id' => $link->id, 'child_call_id' => $childCallId, 'child_workflow_instance_id' => $childInstance->id, @@ -3107,6 +3131,22 @@ private function scheduleChildWorkflow( ]) ); + if ($childWorkflowCall->reservation !== null) { + ChildReservationStore::recordAdmission( + $childInstance, + $childWorkflowCall->reservation, + new StartedChildWorkflow( + (string) $run->namespace, + $run->workflow_instance_id, + $run->id, + $sequence, + $childInstance->id, + $childRun->id, + $childCallId, + $workflowType, + ) + ); + } return $childTask; } @@ -3120,6 +3160,22 @@ private function scheduleChildWorkflowOrFailRun( bool $startMilestone = false, ): ?WorkflowTask { try { + if ($childWorkflowCall->reservation !== null) { + if (! $startMilestone || $run->getConnection()->transactionLevel() === 0) { + throw new LogicException('Reserved children require the native start milestone transaction.'); + } + // Roll back every child write on refusal before failing the parent. + return $run->getConnection() + ->transaction(fn (): WorkflowTask => $this->scheduleChildWorkflow( + $run, + $task, + $sequence, + $childWorkflowCall, + $parallelMetadata, + $parkRun, + $startMilestone, + )); + } return $this->scheduleChildWorkflow( $run, $task, @@ -3858,6 +3914,8 @@ private function continueAsNew( ->lockForUpdate() ->findOrFail($run->workflow_instance_id); + ChildReservationStore::assertOrdinaryStartAllowed($instance); + if ($instance->workflow_class !== $workflowClass) { $instance->forceFill([ 'workflow_class' => $workflowClass, diff --git a/src/V2/WorkflowStub.php b/src/V2/WorkflowStub.php index 54a2a3790..c1bc88583 100644 --- a/src/V2/WorkflowStub.php +++ b/src/V2/WorkflowStub.php @@ -1777,6 +1777,7 @@ private function admitStart( ->findOrFail($this->instance->id); $recoveryIntent = null; + \Workflow\V2\StateMachine\Binding\ChildReservationStore::assertOrdinaryStartAllowed($instance); if ($recoveryAdmission !== null) { $workflowClass = TypeRegistry::resolveWorkflowClass( $instance->workflow_class, @@ -3627,6 +3628,7 @@ private function admitSignalWithStart( ->findOrFail($this->instance->id); $commandContext = $this->signalWithStartCommandContext($intakeGroupId); + \Workflow\V2\StateMachine\Binding\ChildReservationStore::assertOrdinaryStartAllowed($instance); // A keyed retry returns the original intake instead of recording a // second start/signal pair under a new intake group. $retainedReceipt = $this->retainedCommandForRequest( @@ -6104,6 +6106,8 @@ private static function reserveCallerSuppliedInstance( /** @var WorkflowInstance $instance */ $instance = $instanceQuery->findOrFail($instanceId); + \Workflow\V2\StateMachine\Binding\ChildReservationStore::assertOrdinaryStartAllowed($instance); + if ($instance->workflow_type !== $workflowType) { throw new LogicException(sprintf( 'Workflow instance [%s] is reserved for durable type [%s] and cannot be reused for [%s].', diff --git a/src/migrations/2026_11_14_000100_add_native_child_reservation_to_workflow_instances.php b/src/migrations/2026_11_14_000100_add_native_child_reservation_to_workflow_instances.php new file mode 100644 index 000000000..e33ea0eba --- /dev/null +++ b/src/migrations/2026_11_14_000100_add_native_child_reservation_to_workflow_instances.php @@ -0,0 +1,35 @@ +withSchemaLockBudget(static function (): void { + Schema::table('workflow_instances', static function (Blueprint $table): void { + $table->json('native_child_reservation') + ->nullable(); + }); + }); + } + } + + public function down(): void + { + if (\Illuminate\Support\Facades\DB::table('workflow_instances')->whereNotNull( + 'native_child_reservation' + )->exists()) { + throw new LogicException('Native child reservation fences cannot be discarded by a schema rollback.'); + } + Schema::table('workflow_instances', static function (Blueprint $table): void { + $table->dropColumn('native_child_reservation'); + }); + } +}; diff --git a/tests/Feature/V2/V2ChildReservationConcurrencyTest.php b/tests/Feature/V2/V2ChildReservationConcurrencyTest.php new file mode 100644 index 000000000..04b295929 --- /dev/null +++ b/tests/Feature/V2/V2ChildReservationConcurrencyTest.php @@ -0,0 +1,332 @@ +getDriverName(), ['mysql', 'mariadb', 'pgsql'], true) + || ! function_exists('pcntl_fork')) { + $this->markTestSkipped('Separate processes and SQL row-lock observation are required.'); + } + self::stopWorkers(); + Queue::fake(); + } + + public static function winners(): array + { + return [ + 'closure wins' => ['close'], + 'admission wins' => ['admit'], + ]; + } + + #[DataProvider('winners')] + public function testNativeAdmissionAndClosureSerializeOnTheReservedInstance(string $winner): void + { + [$parent, $task, $reservation] = $this->reserve(); + $results = $this->race($parent, $task, $reservation, $winner); + if ($winner === 'close') { + $this->assertSame('closed', $results[0]['kind']); + $this->assertSame('refused', $results[1]['kind']); + $this->assertSame( + 0, + WorkflowRun::query()->where('workflow_instance_id', $reservation->childInstanceId())->count() + ); + $closed = $this->store() + ->close($reservation, 'owner'); + $this->assertInstanceOf(NeverAdmittedChild::class, $closed); + $this->assertSame($results[0]['closureId'], $closed->closureId); + } else { + $this->assertSame('admitted', $results[0]['kind']); + $this->assertSame('admitted', $results[1]['kind']); + $this->assertSame($results[0]['binding'], $results[1]['binding']); + $this->assertSame( + 1, + WorkflowRun::query()->where('workflow_instance_id', $reservation->childInstanceId())->count() + ); + $this->assertSame(1, $parent->historyEvents()->where('event_type', 'ChildRunStarted')->count()); + } + } + + private function store(): ChildReservationStore + { + return new ChildReservationStore(new class() implements BindingAuthorization { + public function assertAllowed(BindingIdentity $identity, string $principal, string $operation): string + { + if ($principal !== 'owner' || $identity->namespace !== 'race-tenant' + || DB::connection()->transactionLevel() !== 0) { + throw new LogicException('Race fixture authorization refused.'); + } + return 'race-policy-v1'; + } + }); + } + + private function reserve(): array + { + $type = TypeRegistry::for(TestGreetingWorkflow::class); + $instance = WorkflowInstance::query()->create([ + 'workflow_class' => TestGreetingWorkflow::class, + 'workflow_type' => $type, + 'namespace' => 'race-tenant', + 'reserved_at' => now(), + 'run_count' => 1, + ]); + $run = WorkflowRun::query()->create([ + 'workflow_instance_id' => $instance->id, + 'run_number' => 1, + 'workflow_class' => TestGreetingWorkflow::class, + 'workflow_type' => $type, + 'namespace' => 'race-tenant', + 'status' => RunStatus::Running, + 'started_at' => now(), + 'last_history_sequence' => 0, + ]); + $instance->forceFill([ + 'current_run_id' => $run->id, + ])->save(); + $identity = new BindingIdentity( + new ExecutionReference('race-app', 'subject', 'execution', str_repeat('a', 64), 'ready', '1'), + 'race-tenant', + $instance->id, + $run->id, + 'child', + 'entry-1', + 'working', + $type, + str_repeat('b', 64), + 'delivered', + ChildReservation::inputFingerprint(['Taylor']), + ); + $reservation = $this->store() + ->reserve($identity, TestGreetingWorkflow::class, 'owner'); + $task = WorkflowTask::query()->create([ + 'workflow_run_id' => $run->id, + 'namespace' => $run->namespace, + 'task_type' => TaskType::Workflow, + 'status' => TaskStatus::Ready, + 'available_at' => now(), + 'payload' => [], + ]); + return [$run, $task, $reservation]; + } + + private function operate( + string $operation, + WorkflowRun $parent, + WorkflowTask $task, + ChildReservation $reservation + ): array { + if ($operation === 'admit') { + try { + DB::transaction(static function () use ($parent, $task, $reservation): void { + (new ReflectionMethod(WorkflowExecutor::class, 'scheduleChildWorkflow'))->invoke( + new WorkflowExecutor(), + $parent, + $task, + 1, + new ChildWorkflowCall($reservation->workflowClass, [ + 'Taylor', + ], $reservation->identity->childOptions, $reservation), + null, + false, + true, + ); + }); + } catch (LogicException $error) { + if (! str_contains($error->getMessage(), 'closed, already admitted')) { + throw $error; + } + return [ + 'kind' => 'refused', + ]; + } + } + $outcome = $this->store() + ->close($reservation, 'owner'); + return $outcome instanceof ChildBinding + ? [ + 'kind' => 'admitted', + 'binding' => $outcome->toArray(), + ] + : [ + 'kind' => 'closed', + 'closureId' => $outcome->closureId, + ]; + } + + /** + * @return list> + */ + private function race(WorkflowRun $parent, WorkflowTask $task, ChildReservation $reservation, string $winner): array + { + $root = sys_get_temp_dir() . '/dw-child-reservation-race-' . bin2hex(random_bytes(8)); + mkdir($root, 0700); + $driver = DB::connection()->getDriverName(); + DB::purge(); + $children = []; + try { + for ($index = 0; $index < 2; ++$index) { + $pid = pcntl_fork(); + if ($pid === -1) { + throw new RuntimeException('Cannot fork the child reservation race.'); + } + if ($pid === 0) { + try { + DB::reconnect(); + $backend = $driver === 'pgsql' + ? DB::selectOne('SELECT pg_backend_pid() AS id')->id + : DB::selectOne('SELECT CONNECTION_ID() AS id')->id; + if ($driver === 'pgsql') { + DB::statement("SET lock_timeout = '15s'"); + } else { + DB::statement('SET SESSION innodb_lock_wait_timeout = 15'); + } + file_put_contents($root . '/backend-' . $index, (string) $backend); + if ($index === 0) { + $held = false; + DB::listen(static function (QueryExecuted $query) use ($root, $winner, &$held): void { + if ($held || ! str_starts_with(strtolower($query->sql), 'update') + || ! str_contains($query->sql, 'native_child_reservation')) { + return; + } + foreach ($query->bindings as $binding) { + $state = is_string($binding) ? json_decode($binding, true) : null; + if (is_array( + $state + ) && ($state['status'] ?? null) === ($winner === 'close' ? 'closed' : 'admitted')) { + if ($query->connection->transactionLevel() === 0) { + throw new RuntimeException( + 'The winner did not retain its native transaction.' + ); + } + $held = true; + touch($root . '/held'); + self::waitFile($root . '/release'); + } + } + }); + } else { + self::waitFile($root . '/held'); + } + $result = $this->operate( + $index === 0 ? $winner : ($winner === 'close' ? 'admit' : 'close'), + $parent, + $task, + $reservation + ); + file_put_contents($root . '/' . $index . '.json', json_encode([ + 'result' => $result, + ], JSON_THROW_ON_ERROR)); + DB::disconnect(); + exit(0); + } catch (Throwable $error) { + file_put_contents($root . '/' . $index . '.json', json_encode([ + 'error' => $error::class . ': ' . $error->getMessage(), + ], JSON_THROW_ON_ERROR)); + DB::disconnect(); + exit(1); + } + } + $children[] = $pid; + } + self::waitFile($root . '/held'); + self::waitFile($root . '/backend-1'); + $backend = (int) file_get_contents($root . '/backend-1'); + DB::reconnect(); + $deadline = microtime(true) + 5; + $waiting = false; + while (! $waiting && microtime(true) < $deadline) { + $rows = $driver === 'pgsql' + ? DB::select( + 'SELECT pid AS id, query AS statement, wait_event_type FROM pg_stat_activity WHERE pid = ?', + [$backend] + ) + : DB::select('SHOW FULL PROCESSLIST'); + foreach ($rows as $row) { + $id = (int) ($driver === 'pgsql' ? $row->id : $row->Id); + $sql = strtolower((string) ($driver === 'pgsql' ? $row->statement : $row->Info)); + $waiting = $id === $backend && str_starts_with($sql, 'select') + && str_contains($sql, 'workflow_instances') && str_contains($sql, 'for update') + && ($driver !== 'pgsql' || $row->wait_event_type === 'Lock'); + if ($waiting) { + break; + } + } + if (! $waiting) { + usleep(10_000); + } + } + $this->assertTrue($waiting, 'The competing native reservation row-lock query was not observed.'); + touch($root . '/release'); + $results = []; + foreach ($children as $index => $pid) { + pcntl_waitpid($pid, $status); + $payload = json_decode( + file_get_contents($root . '/' . $index . '.json'), + true, + flags: JSON_THROW_ON_ERROR + ); + $this->assertSame(0, pcntl_wexitstatus($status), $payload['error'] ?? 'Reservation contender failed.'); + $results[] = $payload['result']; + } + return $results; + } finally { + touch($root . '/release'); + foreach ($children as $pid) { + if (pcntl_waitpid($pid, $status, WNOHANG) === 0) { + posix_kill($pid, SIGKILL); + pcntl_waitpid($pid, $status); + } + } + foreach (glob($root . '/*') as $file) { + unlink($file); + } + rmdir($root); + DB::reconnect(); + } + } + + private static function waitFile(string $path): void + { + $deadline = microtime(true) + 10; + while (! is_file($path)) { + if (microtime(true) > $deadline) { + throw new RuntimeException('The native reservation race barrier timed out.'); + } + usleep(10_000); + } + } +} diff --git a/tests/Feature/V2/V2ChildReservationTest.php b/tests/Feature/V2/V2ChildReservationTest.php new file mode 100644 index 000000000..19512c32a --- /dev/null +++ b/tests/Feature/V2/V2ChildReservationTest.php @@ -0,0 +1,483 @@ +parent(); + $reservation = $this->reserve($parent); + $parent->forceFill([ + 'status' => RunStatus::Completed, + ])->save(); + $again = $this->reserve($parent); + $this->assertSame($reservation->toArray(), $again->toArray()); + $this->assertSame(2, WorkflowInstance::query()->count()); + $this->assertSame(1, WorkflowRun::query()->count()); + } + + public function testCloseBeforeAdmissionPersistsTheSameFenceAndRefusesLateNativeWork(): void + { + $parent = $this->parent(); + $reservation = $this->reserve($parent); + $closed = $this->store() + ->close($reservation, 'trusted-owner'); + $this->assertInstanceOf(NeverAdmittedChild::class, $closed); + $again = $this->store() + ->close(ChildReservation::fromArray($reservation->toArray()), 'trusted-owner'); + $this->assertInstanceOf(NeverAdmittedChild::class, $again); + $this->assertSame($closed->closureId, $again->closureId); + try { + DB::transaction(static fn () => ChildReservationStore::lockForAdmission($parent, new ChildWorkflowCall( + $reservation->workflowClass, + ['Taylor'], + $reservation->identity->childOptions, + $reservation, + ))); + $this->fail('The closed reservation admitted late work.'); + } catch (LogicException $error) { + $this->assertStringContainsString('closed', $error->getMessage()); + } + $this->assertSame( + 0, + WorkflowRun::query()->where('workflow_instance_id', $reservation->childInstanceId())->count() + ); + $this->assertSame(0, WorkflowHistoryEvent::query()->count()); + } + + public function testAdmissionBeforeCloseReturnsExactNativeReceiptAndHonorsPinnedRouting(): void + { + $parent = $this->parent(); + $reservation = $this->reserve($parent); + $task = WorkflowTask::query()->create([ + 'workflow_run_id' => $parent->id, + 'namespace' => $parent->namespace, + 'task_type' => TaskType::Workflow, + 'status' => TaskStatus::Ready, + 'available_at' => now(), + 'payload' => [], + ]); + DB::transaction(static function () use ($parent, $reservation, $task): void { + (new ReflectionMethod(WorkflowExecutor::class, 'scheduleChildWorkflow'))->invoke( + new WorkflowExecutor(), + $parent, + $task, + 1, + new ChildWorkflowCall($reservation->workflowClass, [ + 'Taylor', + ], $reservation->identity->childOptions, $reservation), + null, + false, + true, + ); + }); + $binding = $this->store() + ->close($reservation, 'trusted-owner'); + $this->assertInstanceOf(ChildBinding::class, $binding); + $this->assertSame($reservation->childInstanceId(), $binding->admission->childInstanceId); + $parent->load('historyEvents'); + $this->assertSame(ChildStartHistory::receipt($parent, 1)->toArray(), $binding->admission->toArray()); + $this->assertSame($binding->toArray(), $this->store()->close($reservation, 'trusted-owner')->toArray()); + $childRun = WorkflowRun::query()->findOrFail($binding->admission->childRunId); + $this->assertSame('database', $childRun->connection); + $this->assertSame('bound-child', $childRun->queue); + $this->assertSame( + 1, + WorkflowRun::query()->where('workflow_instance_id', $reservation->childInstanceId())->count() + ); + $this->assertSame(1, $parent->historyEvents->where('event_type', HistoryEventType::ChildRunStarted)->count()); + } + + public function testForgedReservationCannotSelectAnAutoloadedDefinition(): void + { + $parent = $this->parent(); + $reservation = $this->reserve($parent); + $untrustedClass = 'Tests\\Fixtures\\V2\\UntrustedReservationDefinition'; + $forged = new ChildReservation( + $reservation->identity, + $untrustedClass, + $reservation->definitionFingerprint, + $reservation->authorizationVersion, + ); + $task = WorkflowTask::query()->create([ + 'workflow_run_id' => $parent->id, + 'namespace' => $parent->namespace, + 'task_type' => TaskType::Workflow, + 'status' => TaskStatus::Ready, + 'available_at' => now(), + 'payload' => [], + ]); + $autoloaded = []; + $loader = static function (string $class) use (&$autoloaded): void { + $autoloaded[] = $class; + }; + spl_autoload_register($loader); + $failure = null; + try { + DB::transaction(static function () use ($parent, $task, $forged): void { + (new ReflectionMethod(WorkflowExecutor::class, 'scheduleChildWorkflow'))->invoke( + new WorkflowExecutor(), + $parent, + $task, + 1, + new ChildWorkflowCall($forged->workflowClass, ['Taylor'], $forged->identity->childOptions, $forged), + null, + false, + true, + ); + }); + } catch (\Throwable $error) { + $failure = $error; + } finally { + spl_autoload_unregister($loader); + } + $this->assertInstanceOf(LogicException::class, $failure); + $this->assertStringContainsString('stored native child reservation does not match', $failure->getMessage()); + $this->assertNotContains($untrustedClass, $autoloaded); + $this->assertSame( + 0, + WorkflowRun::query()->where('workflow_instance_id', $reservation->childInstanceId())->count() + ); + } + + public function testForeignInstanceOwnershipIsRefusedBeforeDecodingItsReservation(): void + { + $parent = $this->parent(); + $reservation = $this->reserve($parent); + WorkflowInstance::query()->whereKey($reservation->childInstanceId())->toBase()->update([ + 'namespace' => 'foreign-tenant', + 'native_child_reservation' => json_encode([ + 'protected' => 'not-a-reservation', + ], JSON_THROW_ON_ERROR), + ]); + foreach (['reserve', 'close'] as $operation) { + try { + if ($operation === 'reserve') { + $this->reserve($parent); + } else { + $this->store() + ->close($reservation, 'trusted-owner'); + } + $this->fail('A foreign native instance reached reservation decoding.'); + } catch (LogicException $error) { + $this->assertStringContainsString( + 'stored native child reservation does not match', + $error->getMessage() + ); + } + } + } + + public function testChangedImmutableDescriptorCannotReuseItsOriginalLookupKey(): void + { + $parent = $this->parent(); + $reservation = $this->reserve($parent); + $changed = $reservation->identity->toArray(); + $changed['requiredOutcome'] = 'other-outcome'; + $this->expectException(LogicException::class); + $this->expectExceptionMessage('different immutable descriptor'); + $this->store() + ->reserve(BindingIdentity::fromArray($changed), TestGreetingWorkflow::class, 'trusted-owner'); + } + + public function testAuthorizationRefusalOccursBeforeAnyProtectedDatabaseRead(): void + { + $parent = $this->parent(); + $reservation = $this->reserve($parent); + $queries = []; + DB::listen(static function ($query) use (&$queries): void { + $queries[] = $query->sql; + }); + try { + $this->store(false) + ->close($reservation, 'untrusted'); + $this->fail('An unauthorized caller reached binding storage.'); + } catch (LogicException $error) { + $this->assertSame('binding_access_refused', $error->getMessage()); + } + $this->assertSame([], $queries); + } + + public function testCallerTransactionsCannotExposeUncommittedClosureOrRunHostAuthorizationUnderLocks(): void + { + $parent = $this->parent(); + $reservation = $this->reserve($parent); + $authorization = new class() implements BindingAuthorization { + public int $calls = 0; + + public function assertAllowed(BindingIdentity $identity, string $principal, string $operation): string + { + ++$this->calls; + return 'policy-v1'; + } + }; + $store = new ChildReservationStore($authorization); + foreach (['reserve', 'close'] as $operation) { + try { + DB::transaction(static function () use ($operation, $store, $reservation): void { + if ($operation === 'reserve') { + $store->reserve($reservation->identity, $reservation->workflowClass, 'owner'); + } else { + $store->close($reservation, 'owner'); + } + }); + $this->fail('A binding outcome escaped an uncommitted caller transaction.'); + } catch (LogicException $error) { + $this->assertStringContainsString('own recorded activity transaction boundary', $error->getMessage()); + } + } + $this->assertSame(0, $authorization->calls); + $this->assertInstanceOf(NeverAdmittedChild::class, $store->close($reservation, 'owner')); + } + + public function testAuthorizationCannotLeaveAnUncommittedNativeTransaction(): void + { + $reservation = $this->reserve($this->parent()); + $authorization = new class() implements BindingAuthorization { + public function assertAllowed(BindingIdentity $identity, string $principal, string $operation): string + { + DB::beginTransaction(); + return 'policy-v1'; + } + }; + try { + (new ChildReservationStore($authorization))->close($reservation, 'owner'); + $this->fail('Authorization left an uncommitted native closure boundary.'); + } catch (LogicException $error) { + $this->assertStringContainsString('transaction boundary closed', $error->getMessage()); + } finally { + if (DB::transactionLevel() !== 0) { + DB::rollBack(); + } + } + $this->assertInstanceOf(NeverAdmittedChild::class, $this->store()->close($reservation, 'trusted-owner')); + } + + public function testOrdinaryStartCannotBypassAClosedNativeReservation(): void + { + $reservation = $this->reserve($this->parent()); + $this->store() + ->close($reservation, 'trusted-owner'); + $this->expectException(LogicException::class); + $this->expectExceptionMessage('original parent command'); + WorkflowStub::load($reservation->childInstanceId(), 'tenant-a')->attemptStart('Taylor'); + } + + public function testSignalWithStartCannotBypassAClosedNativeReservation(): void + { + $reservation = $this->reserve($this->parent()); + $this->store() + ->close($reservation, 'trusted-owner'); + $this->expectException(LogicException::class); + $this->expectExceptionMessage('original parent command'); + WorkflowStub::load($reservation->childInstanceId(), 'tenant-a')->attemptSignalWithStart( + 'finish', + [], + ['Taylor'] + ); + } + + public function testOrdinaryMakeCannotReplaceTheReservedDefinition(): void + { + $reservation = $this->reserve($this->parent()); + $this->expectException(LogicException::class); + $this->expectExceptionMessage('original parent command'); + WorkflowStub::make($reservation->workflowClass, $reservation->childInstanceId(), 'tenant-a'); + } + + public function testReplayCannotAddOrRemoveAReservation(): void + { + $reservation = $this->reserve($this->parent()); + ChildReservationStore::assertRecordedCommand($reservation->toArray(), $reservation); + ChildReservationStore::assertRecordedCommand(array_reverse($reservation->toArray(), true), $reservation); + foreach ([[$reservation->toArray(), null], [null, $reservation]] as [$recorded, $current]) { + try { + ChildReservationStore::assertRecordedCommand($recorded, $current); + $this->fail('A changed reservation replayed as the same native command.'); + } catch (LogicException $error) { + $this->assertStringContainsString('differs', $error->getMessage()); + } + } + } + + public function testChangedInputAndForeignParentCannotReachTheReservedChild(): void + { + $parent = $this->parent(); + $reservation = $this->reserve($parent); + $foreign = $this->parent(); + foreach ([[$parent, ['changed']], [$foreign, ['Taylor']]] as [$executing, $arguments]) { + try { + DB::transaction( + static fn () => ChildReservationStore::lockForAdmission($executing, new ChildWorkflowCall( + $reservation->workflowClass, + $arguments, + $reservation->identity->childOptions, + $reservation, + )) + ); + $this->fail('A changed native command reached the reservation.'); + } catch (LogicException $error) { + $this->assertStringContainsString('reserved owner, definition or input', $error->getMessage()); + } + } + $this->assertSame( + 0, + WorkflowRun::query()->where('workflow_instance_id', $reservation->childInstanceId())->count() + ); + } + + public function testAnUncertainNativePointerCannotBecomeNeverAdmitted(): void + { + $reservation = $this->reserve($this->parent()); + WorkflowInstance::query()->whereKey($reservation->childInstanceId())->update([ + 'run_count' => 1, + ]); + $this->expectException(LogicException::class); + $this->expectExceptionMessage('uncertain child'); + $this->store() + ->close($reservation, 'trusted-owner'); + } + + public function testAClosedReservationWithAnUnexpectedRunCannotCertifyAbsence(): void + { + $parent = $this->parent(); + $reservation = $this->reserve($parent); + $this->store() + ->close($reservation, 'trusted-owner'); + WorkflowRun::query()->create([ + 'workflow_instance_id' => $reservation->childInstanceId(), + 'run_number' => 1, + 'workflow_class' => $reservation->workflowClass, + 'workflow_type' => $reservation->identity->childWorkflowType, + 'namespace' => $reservation->identity->namespace, + 'status' => RunStatus::Pending, + 'started_at' => now(), + 'last_history_sequence' => 0, + ]); + $this->expectException(LogicException::class); + $this->expectExceptionMessage('uncertain child'); + $this->store() + ->close($reservation, 'trusted-owner'); + } + + public function testAReservationTombstoneSurvivesItsParentRunClosing(): void + { + $parent = $this->parent(); + $reservation = $this->reserve($parent); + $closed = $this->store() + ->close($reservation, 'trusted-owner'); + $parent->forceFill([ + 'status' => RunStatus::Cancelled, + ])->save(); + $this->assertSame($closed->closureId, $this->store()->close($reservation, 'trusted-owner')->closureId); + $this->assertSame($reservation->toArray(), $this->reserve($parent)->toArray()); + } + + public function testCancellationPreventsCreatingANewReservation(): void + { + $parent = $this->parent(); + $parent->forceFill([ + 'cancellation_requested_at' => now(), + ])->save(); + $this->expectException(LogicException::class); + $this->expectExceptionMessage('active original parent'); + $this->reserve($parent); + } + + private function store(bool $allowed = true): ChildReservationStore + { + return new ChildReservationStore(new class($allowed) implements BindingAuthorization { + public function __construct( + private readonly bool $allowed + ) { + } + + public function assertAllowed(BindingIdentity $identity, string $principal, string $operation): string + { + if (! $this->allowed || $principal !== 'trusted-owner' || $identity->namespace !== 'tenant-a') { + throw new LogicException('binding_access_refused'); + } + return 'host-policy-v1'; + } + }); + } + + private function parent(): WorkflowRun + { + $instance = WorkflowInstance::query()->create([ + 'workflow_class' => TestGreetingWorkflow::class, + 'workflow_type' => TypeRegistry::for(TestGreetingWorkflow::class), + 'namespace' => 'tenant-a', + 'run_count' => 1, + 'reserved_at' => now(), + ]); + $run = WorkflowRun::query()->create([ + 'workflow_instance_id' => $instance->id, + 'run_number' => 1, + 'workflow_class' => TestGreetingWorkflow::class, + 'workflow_type' => $instance->workflow_type, + 'namespace' => 'tenant-a', + 'status' => RunStatus::Running, + 'last_history_sequence' => 0, + 'started_at' => now(), + ]); + $instance->forceFill([ + 'current_run_id' => $run->id, + ])->save(); + return $run; + } + + private function identity(WorkflowRun $parent): BindingIdentity + { + return new BindingIdentity( + new ExecutionReference('app', 'business-1', 'execution-1', str_repeat('a', 64), 'ready', 'revision-1'), + 'tenant-a', + $parent->workflow_instance_id, + $parent->id, + 'shipment', + 'entry-1', + 'delivering', + TypeRegistry::for(TestGreetingWorkflow::class), + str_repeat('b', 64), + 'delivered', + ChildReservation::inputFingerprint(['Taylor']), + new ChildWorkflowOptions(connection: 'database', queue: 'bound-child'), + ); + } + + private function reserve(WorkflowRun $parent): ChildReservation + { + return $this->store() + ->reserve($this->identity($parent), TestGreetingWorkflow::class, 'trusted-owner'); + } +} diff --git a/tests/Fixtures/V2/ReplayRegression/reserved-child-admission-cannot-become-ordinary-result-wait.json b/tests/Fixtures/V2/ReplayRegression/reserved-child-admission-cannot-become-ordinary-result-wait.json new file mode 100644 index 000000000..4925eb5de --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/reserved-child-admission-cannot-become-ordinary-result-wait.json @@ -0,0 +1,130 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "reserved-child-admission-cannot-become-ordinary-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": "dw-bound-child-207b51bd65b5d7a568d4cb9964f605e5683238f45475b3e83334390d465b86d6", + "child_call_id": "01j00000000000000000000062", + "child_completion_mode": "started", + "child_workflow_class": "Tests\\Fixtures\\V2\\TestChildGreetingWorkflow", + "child_reservation": { + "version": 1, + "identity": { + "version": 1, + "parentExecution": { + "scope": "fixture-app", + "subject": "fixture-subject", + "executionId": "execution-1", + "policyFingerprint": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "sourceState": "ready", + "sourceRevision": "rev-1" + }, + "namespace": "fixture-tenant", + "parentInstanceId": "original-parent", + "parentRunId": "original-parent-run", + "slot": "child", + "generation": "entry-1", + "hostingState": "working", + "childWorkflowType": "Tests\\Fixtures\\V2\\TestChildGreetingWorkflow", + "childPolicyFingerprint": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "requiredOutcome": "delivered", + "inputFingerprint": "aa1c10b6ce2d97c28c6ccc8289d751c9eae388196aca97c2fdfbeabeb6ee07c1", + "childOptions": { + "parent_close_policy": "abandon", + "connection": null, + "queue": null, + "cancellation_policy": "abandon" + } + }, + "workflowClass": "Tests\\Fixtures\\V2\\TestChildGreetingWorkflow", + "definitionFingerprint": "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", + "authorizationVersion": "fixture-policy-v1" + } + }, + "recorded_at": "2026-11-13T12:00:01Z" + }, + { + "sequence": 3, + "event_type": "ChildRunStarted", + "payload": { + "sequence": 1, + "child_workflow_type": "Tests\\Fixtures\\V2\\TestChildGreetingWorkflow", + "child_workflow_run_id": "01j00000000000000000000061", + "child_workflow_instance_id": "dw-bound-child-207b51bd65b5d7a568d4cb9964f605e5683238f45475b3e83334390d465b86d6", + "child_call_id": "01j00000000000000000000062", + "child_workflow_class": "Tests\\Fixtures\\V2\\TestChildGreetingWorkflow", + "child_reservation": { + "version": 1, + "identity": { + "version": 1, + "parentExecution": { + "scope": "fixture-app", + "subject": "fixture-subject", + "executionId": "execution-1", + "policyFingerprint": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "sourceState": "ready", + "sourceRevision": "rev-1" + }, + "namespace": "fixture-tenant", + "parentInstanceId": "original-parent", + "parentRunId": "original-parent-run", + "slot": "child", + "generation": "entry-1", + "hostingState": "working", + "childWorkflowType": "Tests\\Fixtures\\V2\\TestChildGreetingWorkflow", + "childPolicyFingerprint": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "requiredOutcome": "delivered", + "inputFingerprint": "aa1c10b6ce2d97c28c6ccc8289d751c9eae388196aca97c2fdfbeabeb6ee07c1", + "childOptions": { + "parent_close_policy": "abandon", + "connection": null, + "queue": null, + "cancellation_policy": "abandon" + } + }, + "workflowClass": "Tests\\Fixtures\\V2\\TestChildGreetingWorkflow", + "definitionFingerprint": "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", + "authorizationVersion": "fixture-policy-v1" + } + }, + "recorded_at": "2026-11-13T12:00:01Z" + } + ], + "expected_failure": { + "type": "workflow_history_shape_mismatch", + "exception": "Workflow\\V2\\Exceptions\\HistoryEventShapeMismatchException" + }, + "reservation_replay": { + "current": null, + "expected": "refused" + } +} diff --git a/tests/Unit/V2/ReplayRegressionCorpusTest.php b/tests/Unit/V2/ReplayRegressionCorpusTest.php index af16aabca..9f696906d 100644 --- a/tests/Unit/V2/ReplayRegressionCorpusTest.php +++ b/tests/Unit/V2/ReplayRegressionCorpusTest.php @@ -83,6 +83,20 @@ protected function tearDown(): void public function testFixtureExecutesThroughColdReplayRunner(array $fixture): void { $this->assertHistoryPayloadContract($fixture); + if (isset($fixture['reservation_replay'])) { + $scheduled = array_values(array_filter($fixture['history'], static fn (array $event): bool => + $event['event_type'] === 'ChildWorkflowScheduled'))[0]; + try { + \Workflow\V2\StateMachine\Binding\ChildReservationStore::assertRecordedCommand( + $scheduled['payload']['child_reservation'], + $fixture['reservation_replay']['current'], + ); + $this->fail('A recorded reservation was silently removed during cold replay.'); + } catch (\LogicException $error) { + $this->assertSame('refused', $fixture['reservation_replay']['expected']); + $this->assertStringContainsString('recorded native child reservation differs', $error->getMessage()); + } + } if (isset($fixture['redrive_plan'])) { $this->assertRetriedActivityRedrivePlan($fixture); } diff --git a/tests/Unit/V2/ReservedChildWorkflowTest.php b/tests/Unit/V2/ReservedChildWorkflowTest.php new file mode 100644 index 000000000..340376cb7 --- /dev/null +++ b/tests/Unit/V2/ReservedChildWorkflowTest.php @@ -0,0 +1,206 @@ + 'redis', + 'workflows.v2.task_dispatch_mode' => 'queue', + 'workflows.v2.namespaces' => ['tenant-a'], + ]); + } + + public function testRecordedReservationAndColdNativeReplayKeepOneChildAndOriginalAdmission(): void + { + [$stub, $parent, $reservation] = $this->reserveFromRecordedActivity(); + $this->runNext($parent, TaskType::Workflow); + $original = (new QueryStateReplayer())->query($parent->fresh(), 'admitted'); + $this->assertSame($reservation->childInstanceId(), $original['childInstanceId']); + $this->assertSame(2, WorkflowInstance::query()->count()); + $this->assertSame(2, WorkflowRun::query()->count()); + $this->assertSame(1, ActivityExecution::query()->count()); + $this->assertSame(1, ActivityAttempt::query()->count()); + + // A fresh task/executor/fiber replays the lost workflow acknowledgement. + WorkflowTask::query()->create([ + 'workflow_run_id' => $parent->id, + 'namespace' => $parent->namespace, + 'task_type' => TaskType::Workflow, + 'status' => TaskStatus::Ready, + 'available_at' => now(), + 'payload' => [], + 'connection' => $parent->connection, + 'queue' => $parent->queue, + 'compatibility' => $parent->compatibility, + ]); + $this->runNext($parent, TaskType::Workflow); + $this->assertSame($original, (new QueryStateReplayer())->query($parent->fresh(), 'admitted')); + $this->assertSame(2, WorkflowRun::query()->count()); + $this->assertSame(1, ActivityAttempt::query()->count()); + $binding = (new ChildReservationStore(new ReservationFixtureAuthorization()))->close( + $reservation, + 'fixture-owner' + ); + $this->assertInstanceOf(ChildBinding::class, $binding); + $this->assertSame($original, $binding->admission->toArray()); + + $child = WorkflowRun::query()->findOrFail($original['childRunId']); + $this->runNext($child, TaskType::Workflow); + $this->assertSame(RunStatus::Completed, $child->fresh()->status); + $stub->signal('finish'); + $this->runNext($parent, TaskType::Workflow); + $this->assertSame(RunStatus::Completed, $parent->fresh()->status); + $this->assertSame('Hello, Taylor!', $stub->refresh()->output()); + $this->assertSame(2, WorkflowRun::query()->count()); + } + + public function testCloseAfterReservationActivityFencesTheDelayedParentTask(): void + { + [, $parent, $reservation] = $this->reserveFromRecordedActivity(); + $store = new ChildReservationStore(new ReservationFixtureAuthorization()); + $closed = $store->close($reservation, 'fixture-owner'); + $this->assertInstanceOf(NeverAdmittedChild::class, $closed); + $this->runNext($parent, TaskType::Workflow); + $this->assertSame(RunStatus::Failed, $parent->fresh()->status); + $this->assertSame( + 0, + WorkflowRun::query()->where('workflow_instance_id', $reservation->childInstanceId())->count() + ); + $this->assertSame($closed->closureId, $store->close($reservation, 'fixture-owner')->closureId); + $this->assertSame(1, ActivityAttempt::query()->count()); + } + + private function reserveFromRecordedActivity(): array + { + $stub = WorkflowStub::make(ReservedParentFixture::class, 'reserved-parent-fixture', 'tenant-a'); + $stub->start(); + $parent = WorkflowRun::query()->findOrFail($stub->runId()); + $this->runNext($parent, TaskType::Workflow); + $this->runNext($parent, TaskType::Activity); + $reservation = (new QueryStateReplayer())->query($parent->fresh(), 'reserved'); + return [$stub, $parent, ChildReservation::fromArray($reservation)]; + } + + private function runNext(WorkflowRun $run, TaskType $type): void + { + $task = WorkflowTask::query()->where('workflow_run_id', $run->id)->where('task_type', $type) + ->where('status', TaskStatus::Ready)->firstOrFail(); + $job = $type === TaskType::Activity ? new RunActivityTask($task->id) : new RunWorkflowTask($task->id); + app() + ->call([$job, 'handle']); + } +} + +final class ReservationFixtureAuthorization implements BindingAuthorization +{ + public function assertAllowed(BindingIdentity $identity, string $principal, string $operation): string + { + if ($principal !== 'fixture-owner' || $identity->namespace !== 'tenant-a' + || $identity->parentExecution->scope !== 'fixture-app' + || $identity->parentExecution->subject !== 'fixture-subject') { + throw new LogicException('fixture_authorization_refused'); + } + return 'fixture-policy-v1'; + } +} + +final class ReserveChildFixtureActivity extends Activity +{ + public function handle(): array + { + $identity = new BindingIdentity( + new ExecutionReference('fixture-app', 'fixture-subject', 'execution-1', str_repeat( + 'a', + 64 + ), 'ready', 'rev-1'), + 'tenant-a', + $this->workflowId(), + $this->runId(), + 'child', + 'entry-1', + 'working', + TypeRegistry::for(ReservedChildFixture::class), + str_repeat('b', 64), + 'delivered', + ChildReservation::inputFingerprint(['Taylor']), + ); + return (new ChildReservationStore(new ReservationFixtureAuthorization()))->reserve( + $identity, + ReservedChildFixture::class, + 'fixture-owner', + )->toArray(); + } +} + +#[Signal('finish')] +final class ReservedParentFixture extends Workflow +{ + public array $reservation = []; + + public array $admission = []; + + public function handle(): mixed + { + $this->reservation = self::activity(ReserveChildFixtureActivity::class); + $child = ChildReservation::fromArray($this->reservation)->start('Taylor'); + $this->admission = $child->toArray(); + self::awaitSignal('finish'); + return $child->await(); + } + + #[QueryMethod] + public function admitted(): array + { + return $this->admission; + } + + #[QueryMethod] + public function reserved(): array + { + return $this->reservation; + } +} + +final class ReservedChildFixture extends Workflow +{ + public function handle(string $name): string + { + return 'Hello, ' . $name . '!'; + } +} diff --git a/tests/Unit/V2/StateMachine/ChildBindingTest.php b/tests/Unit/V2/StateMachine/ChildBindingTest.php new file mode 100644 index 000000000..33273d257 --- /dev/null +++ b/tests/Unit/V2/StateMachine/ChildBindingTest.php @@ -0,0 +1,178 @@ +identity(); + $this->assertSame('922278dd098dc77d71e6f3a84ef4d2d7995d46fb561c1d851083cce7855e9253', $identity->key()); + $this->assertSame('66ed5bc95b201363bd3e27f739837e5539a787f7aa635b8701eb3fd35ea51c96', $identity->fingerprint()); + } + + public function testStoredBindingRoundTripsWithoutAWorkflowOrMutableStateLookup(): void + { + $binding = new ChildBinding($this->identity(), $this->admission()); + $stored = json_encode($binding->toArray(), JSON_THROW_ON_ERROR); + $restored = ChildBinding::fromArray(json_decode($stored, true, flags: JSON_THROW_ON_ERROR)); + $this->assertSame($binding->toArray(), $restored->toArray()); + $this->assertTrue($restored->matches($this->identity())); + $this->assertSame('parent-run-1', $restored->admission->parentRunId); + $this->assertSame('child-run-1', $restored->admission->childRunId); + $this->assertSame('entry-1', $restored->identity->generation); + } + + public static function descriptorChanges(): iterable + { + foreach (['hostingState', 'requiredOutcome', 'childWorkflowType'] as $key) { + yield $key => [$key, 'changed']; + } + foreach (['childPolicyFingerprint', 'inputFingerprint'] as $key) { + yield $key => [$key, str_repeat('d', 64)]; + } + } + + #[DataProvider('descriptorChanges')] + public function testAReusedKeyCannotMakeAChangedDescriptorMatch(string $key, string $value): void + { + $original = $this->identity(); + $payload = $original->toArray(); + $payload[$key] = $value; + $changed = BindingIdentity::fromArray($payload); + $this->assertSame($original->key(), $changed->key()); + $this->assertNotSame($original->fingerprint(), $changed->fingerprint()); + $this->assertFalse((new ChildBinding($original, $this->admission()))->matches($changed)); + } + + public static function originalIdentityChanges(): iterable + { + foreach (['namespace', 'parentInstanceId', 'parentRunId', 'slot', 'generation'] as $key) { + yield $key => [$key, false]; + } + foreach (['scope', 'subject', 'executionId'] as $key) { + yield $key => [$key, true]; + } + } + + #[DataProvider('originalIdentityChanges')] + public function testOriginalOwnerSlotAndGenerationEachHaveTheirOwnLookupKey(string $key, bool $parent): void + { + $original = $this->identity(); + $payload = $original->toArray(); + if ($parent) { + $payload['parentExecution'][$key] = 'different'; + } else { + $payload[$key] = 'different'; + } + $this->assertNotSame($original->key(), BindingIdentity::fromArray($payload)->key()); + } + + public static function receiptSubstitutions(): iterable + { + foreach (['namespace', 'parentInstanceId', 'parentRunId', 'workflowType'] as $key) { + yield $key => [$key]; + } + } + + #[DataProvider('receiptSubstitutions')] + public function testAReceiptFromAnotherOwnerOrTypeCannotBeBound(string $key): void + { + $receipt = $this->admission() + ->toArray(); + $receipt[$key] = 'substituted'; + $this->expectException(InvalidArgumentException::class); + new ChildBinding($this->identity(), StartedChildWorkflow::fromArray($receipt)); + } + + public static function invalidPersistedIdentities(): iterable + { + yield 'unknown version' => [static function (array &$p): void { + $p['version'] = 2; + }]; + yield 'extra field' => [static function (array &$p): void { + $p['latest'] = true; + }]; + yield 'missing generation' => [static function (array &$p): void { + unset($p['generation']); + }]; + yield 'untyped generation' => [static function (array &$p): void { + $p['generation'] = 1; + }]; + yield 'empty generation' => [static function (array &$p): void { + $p['generation'] = ''; + }]; + yield 'changed parent schema' => [static function (array &$p): void { + $p['parentExecution']['extra'] = ''; + }]; + yield 'missing policy' => [static function (array &$p): void { + $p['childPolicyFingerprint'] = ''; + }]; + yield 'missing outcome' => [static function (array &$p): void { + $p['requiredOutcome'] = ''; + }]; + yield 'unbound input' => [static function (array &$p): void { + $p['inputFingerprint'] = 'latest'; + }]; + yield 'extra options' => [static function (array &$p): void { + $p['childOptions']['extra'] = null; + }]; + yield 'wrong routing type' => [static function (array &$p): void { + $p['childOptions']['queue'] = false; + }]; + } + + #[DataProvider('invalidPersistedIdentities')] + public function testClosedVersionedIdentityRefusesAmbiguousStoredData(callable $mutate): void + { + $payload = $this->identity() + ->toArray(); + $mutate($payload); + $this->expectException(InvalidArgumentException::class); + BindingIdentity::fromArray($payload); + } + + private function identity(): BindingIdentity + { + return new BindingIdentity( + new ExecutionReference('tenant-a', 'order-1', 'transition-1', str_repeat( + 'a', + 64 + ), 'Received', 'revision-7'), + 'client-a', + 'parent-1', + 'parent-run-1', + 'payment', + 'entry-1', + 'Processing', + 'example.child.v1', + str_repeat('b', 64), + 'Paid', + str_repeat('c', 64), + ); + } + + private function admission(): StartedChildWorkflow + { + return new StartedChildWorkflow( + 'client-a', + 'parent-1', + 'parent-run-1', + 7, + 'child-1', + 'child-run-1', + 'child-call-1', + 'example.child.v1' + ); + } +}