diff --git a/CHANGELOG.md b/CHANGELOG.md index d979fcca..c8f8101d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,22 @@ ## Unreleased +- Require a trusted host authorizer for workflow service operations. Refuse + worker-supplied principal overrides, foreign caller namespaces and service + call IDs or idempotency keys owned by another run, instance or operation. + Preserve explicit embedded null namespaces instead of using ambient config. +- Service-operation hosts must bind `WorkflowServiceOperationAuthorizer` before + upgrading. The default refuses service operations; explicitly trusted embedded + hosts can opt into `RunServiceOperationAuthorizer` for same-namespace calls + with an unprivileged workflow principal. Authenticated servers supply their + own current principal and authorize requested target scope. +- Recheck current service-call replay policy without reserving fresh admission + capacity, including cached checkpoint receipts. Custom policies must provide + the pure `authorizeReplay` hook. +- Serialize keyed service-call creation with a unique canonical scope digest. + Reauthorize the actual locked row after a concurrent insert or uniqueness + conflict. Apply the nullable schema expansion and drain older writers before + relying on concurrent deduplication. - Bound `record_local_activity` timing fields (`duration_ms`, heartbeat `elapsed_ms`, `backoff_seconds`) and refuse a report whose walk back from completion leaves the storable timestamp range (#930). diff --git a/docs/architecture/workflow-service-calls-architecture.md b/docs/architecture/workflow-service-calls-architecture.md index 9c94198b..0355612e 100644 --- a/docs/architecture/workflow-service-calls-architecture.md +++ b/docs/architecture/workflow-service-calls-architecture.md @@ -352,6 +352,86 @@ of the five reasons, and a Cancelled call MUST carry `ServiceCallFailureReason::Cancellation` even though its terminal status is Cancelled rather than Failed. +## Workflow operation authority + +Workflow service operations require a host binding for +`Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer`. The default binding +refuses them with `service_operation_authority_invalid`. This applies to embedded +execution and external worker completions, including retained-claim prefixes. +Upgrade the host binding before accepting service-operation work. + +An authorizer receives the locked caller run and normalized command intent. +It must validate the requested target scope against trusted host/run context, +then return a `ServiceCallPrincipal` derived from that context. It must not +reconstruct authority from the command's principal fields or metadata. It is an +authorization check, so it must not reserve admission budgets or perform +external effects; a transaction retry or preflight may invoke it again. + +An embedded host that explicitly trusts its own workflow code to invoke +services within each run's namespace can opt into the provided workload policy: + +```php +$app->bind( + \Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer::class, + \Workflow\V2\Support\RunServiceOperationAuthorizer::class, +); +``` + +That policy permits only the run's namespace. Its principal is +`workflow:`, method `workflow`, with no roles, tenant or claims. It does +not satisfy a catalog policy that requires privileged roles. A server with +authenticated workers instead binds an authorizer that checks its current +authentication/namespace policy and returns the authenticated worker principal. +Authorized cross-namespace routing remains possible through that trusted host +binding; a worker's requested `namespace` alone grants no permission. + +Policies used for existing workflow-call replay must provide +`authorizeReplay(ServiceBoundaryRequest): ServiceBoundaryDecision`; the optional +`ServiceBoundaryReplayPolicy` interface declares this contract. The default +policy checks current authorization and circuit state without reserving rate +or concurrency capacity. Existing compatible host methods work without adopting +the interface. A custom policy without this pure check fails closed with +`service_operation_replay_authority_unavailable`; implement the hook as part of +the host upgrade. Cached checkpoint receipts require the same current check, +including after catalog policy changes or replacement. + +Caller namespace and workflow run/instance identity always come from the locked +run. A mismatched non-null `caller_namespace` or non-null `principal_*` override +is refused before applying sibling commands. Unknown fields do not grant +authority. Explicit null run namespaces stay null even when the process has a +configured default namespace. + +An explicit `service_call_id` or an existing idempotency-key match must bind to +the same caller namespace, run and instance, the authorized target namespace, +and the exact endpoint/service/operation names. Existing rows are checked under +locks before replay or dispatch. A missing or foreign explicit call, or a +foreign key match, refuses the worker turn with `service_call_not_owned` and +leaves the lease and durable rows unchanged. New default idempotency keys are +checked by the same rule. Server adapters retain their current-policy replay +checks before handing the call to Core. + +The run, service call and catalog models must use the completion's database +connection. A separate connection is refused before sibling writes with +`service_operation_storage_connection_mismatch`; separate transactions cannot +provide the atomic refusal contract. + +Keyed calls use a unique SHA-256 digest of the exact target namespace (including +null), catalog names and opaque key. The full scope is still compared after +lookup; a digest or database collation match never grants authority. Creation +uses a savepoint, then locks and reauthorizes the canonical row if another +transaction wins. Explicit call IDs obey the same keyed-call invariant. + +Apply the nullable `idempotency_scope_hash` expansion before starting upgraded +writers. It does not backfill existing calls: an existing candidate must pass +current caller, catalog and policy checks before it can claim the key. Conflicting +legacy rows are refused rather than reassigned. Stop or drain older service-call +writers before relying on concurrent deduplication, since they do not populate +the digest. Retaining the column and index supports code rollback without +discarding canonical claims. The migration uses a short metadata-lock timeout +and online index creation on PostgreSQL and MySQL; retry a refused migration +after removing the lock contention. No production migration has been qualified +by the source tests alone. + ## Observability surface The observability surface for a cross-namespace service call MUST be diff --git a/src/Providers/WorkflowServiceProvider.php b/src/Providers/WorkflowServiceProvider.php index d20f4a74..9efa531f 100644 --- a/src/Providers/WorkflowServiceProvider.php +++ b/src/Providers/WorkflowServiceProvider.php @@ -43,6 +43,7 @@ use Workflow\V2\Contracts\ServiceBoundaryPolicy; use Workflow\V2\Contracts\ServiceControlPlane; use Workflow\V2\Contracts\WorkflowControlPlane; +use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowLink; @@ -63,6 +64,7 @@ use Workflow\V2\Support\DefaultServiceControlPlane; use Workflow\V2\Support\DefaultWorkflowControlPlane; use Workflow\V2\Support\DefaultWorkflowTaskBridge; +use Workflow\V2\Support\DenyWorkflowServiceOperationAuthorizer; use Workflow\V2\Support\ExecutionTracing; use Workflow\V2\Support\HistoryProjectionMaintenanceFallback; use Workflow\V2\Support\InMemoryTaskFairnessState; @@ -132,6 +134,10 @@ static function ($app): HistoryProjectionMaintenanceRole { ); $this->app->singletonIf(ServiceControlPlane::class, DefaultServiceControlPlane::class); + $this->app->singletonIf( + WorkflowServiceOperationAuthorizer::class, + DenyWorkflowServiceOperationAuthorizer::class + ); $this->app->singletonIf(SchedulerRole::class, DefaultSchedulerRole::class); diff --git a/src/V2/Contracts/ServiceBoundaryReplayPolicy.php b/src/V2/Contracts/ServiceBoundaryReplayPolicy.php new file mode 100644 index 00000000..2d273e43 --- /dev/null +++ b/src/V2/Contracts/ServiceBoundaryReplayPolicy.php @@ -0,0 +1,14 @@ + $command + */ + public function authorize(WorkflowRun $run, array $command): ServiceCallPrincipal; +} diff --git a/src/V2/Exceptions/ServiceOperationAuthorizationException.php b/src/V2/Exceptions/ServiceOperationAuthorizationException.php new file mode 100644 index 00000000..63b6c846 --- /dev/null +++ b/src/V2/Exceptions/ServiceOperationAuthorizationException.php @@ -0,0 +1,15 @@ +checkAuthorization($request) + ?? $this->checkCircuitBreak($request) + ?? ServiceBoundaryDecision::allow(policyName: self::POLICY_NAME); + } + /** * @param array $rules */ diff --git a/src/V2/Support/DefaultServiceControlPlane.php b/src/V2/Support/DefaultServiceControlPlane.php index 33cfdd7e..0315ff63 100644 --- a/src/V2/Support/DefaultServiceControlPlane.php +++ b/src/V2/Support/DefaultServiceControlPlane.php @@ -17,6 +17,7 @@ use Workflow\V2\Enums\ServiceCallOperationMode; use Workflow\V2\Enums\ServiceCallOutcome; use Workflow\V2\Enums\ServiceCallStatus; +use Workflow\V2\Exceptions\ServiceOperationAuthorizationException; use Workflow\V2\Models\WorkflowService; use Workflow\V2\Models\WorkflowServiceCall; use Workflow\V2\Models\WorkflowServiceEndpoint; @@ -60,12 +61,30 @@ public function execute( CodecRegistry::canonicalize($this->stringOption($options, 'payload_codec')); $explicitCallId = $this->stringFrom($options['service_call_id'] ?? null); + $key = $this->stringOption($options, 'idempotency_key'); + $connection = ConfiguredV2Models::query('service_call_model', WorkflowServiceCall::class) + ->getModel() + ->getConnection(); + if (($key !== null || $explicitCallId !== null) && $connection->transactionLevel() === 0) { + return $connection->transaction(fn (): array => $this->execute( + $endpointName, + $serviceName, + $operationName, + $options, + )); + } $idempotent = $explicitCallId === null ? $this->idempotentCall($namespace, $endpointKey, $serviceKey, $operationKey, $options) : null; if ($idempotent !== null) { + $this->authorizeWorkflowCall($idempotent, $namespace, $endpointKey, $serviceKey, $operationKey, $options); + $idempotent = ServiceCallIdempotency::save( + $idempotent, + [$namespace, $endpointKey, $serviceKey, $operationKey, $key] + ); + $this->authorizeWorkflowCall($idempotent, $namespace, $endpointKey, $serviceKey, $operationKey, $options); return $this->serialize($idempotent) + [ 'accepted' => $this->acceptedShape($idempotent), 'idempotent_replay' => true, @@ -74,10 +93,24 @@ public function execute( } $call = $this->existingCall($explicitCallId, $namespace); + if ($call instanceof WorkflowServiceCall) { + $this->authorizeWorkflowCall($call, $namespace, $endpointKey, $serviceKey, $operationKey, $options); + } if ($call instanceof WorkflowServiceCall && $namespace === null) { $namespace = $this->stringFrom($call->target_namespace) ?? $this->stringFrom($call->namespace); } + if ($call instanceof WorkflowServiceCall && is_string($call->idempotency_key) + && $call->idempotency_key !== '') { + if ($key !== null && $key !== $call->idempotency_key) { + throw new ServiceOperationAuthorizationException('service_call_idempotency_scope_conflict'); + } + $scope = [$namespace, $endpointKey, $serviceKey, $operationKey, $call->idempotency_key]; + $canonical = ServiceCallIdempotency::find($scope) ?? $call; + $this->authorizeWorkflowCall($canonical, $namespace, $endpointKey, $serviceKey, $operationKey, $options); + $call = ServiceCallIdempotency::save($canonical, $scope); + $this->authorizeWorkflowCall($call, $namespace, $endpointKey, $serviceKey, $operationKey, $options); + } if ($explicitCallId !== null && ! $call instanceof WorkflowServiceCall) { return [ @@ -121,9 +154,10 @@ public function execute( $call ??= $this->newCall(); $principal = $this->principal($options); - $callerNamespace = $this->namespaceOption($options, 'caller_namespace') - ?? ($call instanceof WorkflowServiceCall ? $this->stringFrom($call->caller_namespace) : null) - ?? $this->configuredNamespace(); + $callerNamespace = array_key_exists('caller_namespace', $options) + ? $this->namespaceOption($options, 'caller_namespace') + : (($call instanceof WorkflowServiceCall ? $this->stringFrom($call->caller_namespace) : null) + ?? $this->configuredNamespace()); if ($call->accepted_at === null) { $this->stampPending( @@ -136,7 +170,23 @@ public function execute( $principal, $options, ); - $call->save(); + if (is_string($call->idempotency_key) && $call->idempotency_key !== '') { + $candidate = $call; + $call = ServiceCallIdempotency::save( + $call, + [$namespace, $endpointKey, $serviceKey, $operationKey, $call->idempotency_key] + ); + if (! $call->is($candidate)) { + $this->authorizeWorkflowCall($call, $namespace, $endpointKey, $serviceKey, $operationKey, $options); + return $this->serialize($call) + [ + 'accepted' => $this->acceptedShape($call), + 'idempotent_replay' => true, + 'reason' => null, + ]; + } + } else { + $call->save(); + } } $endpoint = $this->endpoint($namespace, $endpointKey); @@ -467,9 +517,13 @@ private function emptyCallShape(): array */ private function targetNamespace(array $options): ?string { - return $this->namespaceOption($options, 'target_namespace') - ?? $this->namespaceOption($options, 'namespace') - ?? $this->configuredNamespace(); + foreach (['target_namespace', 'namespace'] as $field) { + if (array_key_exists($field, $options)) { + return $this->namespaceOption($options, $field); + } + } + + return $this->configuredNamespace(); } /** @@ -543,7 +597,8 @@ private function existingCall(mixed $serviceCallId, ?string $namespace = null): $query->where('namespace', $namespace); } - return $query->first(); + return $query->lockForUpdate() + ->first(); } /** @@ -585,17 +640,35 @@ private function idempotentCall( return null; } - $model = ConfiguredV2Models::resolve('service_call_model', WorkflowServiceCall::class); + return ServiceCallIdempotency::find([$namespace, $endpointName, $serviceName, $operationName, $key]); + } - return $model::query() - ->where('target_namespace', $namespace) - ->where('endpoint_name', $endpointName) - ->where('service_name', $serviceName) - ->where('operation_name', $operationName) - ->where('idempotency_key', $key) - ->oldest('created_at') - ->oldest('id') - ->first(); + /** + * @param array $options + */ + private function authorizeWorkflowCall( + WorkflowServiceCall $call, + ?string $namespace, + string $endpoint, + string $service, + string $operation, + array $options, + ): void { + $run = $this->stringOption($options, 'caller_workflow_run_id'); + $instance = $this->stringOption($options, 'caller_workflow_instance_id'); + if ($run === null && $instance === null) { + return; + } + WorkflowServiceOperationAuthority::authorizeExistingCall($call, $this->principal($options), [ + 'namespace' => $namespace, + 'caller_namespace' => array_key_exists('caller_namespace', $options) + ? $this->namespaceOption($options, 'caller_namespace') : $this->configuredNamespace(), + 'caller_workflow_instance_id' => $instance, + 'caller_workflow_run_id' => $run, + 'endpoint_name' => $endpoint, + 'service_name' => $service, + 'operation_name' => $operation, + ], $this->boundaryPolicy()); } /** diff --git a/src/V2/Support/DefaultWorkflowTaskBridge.php b/src/V2/Support/DefaultWorkflowTaskBridge.php index 126788ee..469de996 100644 --- a/src/V2/Support/DefaultWorkflowTaskBridge.php +++ b/src/V2/Support/DefaultWorkflowTaskBridge.php @@ -40,6 +40,7 @@ use Workflow\V2\Exceptions\ActivityTimeoutException; use Workflow\V2\Exceptions\HistoryEventShapeMismatchException; use Workflow\V2\Exceptions\HistoryTransactionLimitExceededException; +use Workflow\V2\Exceptions\ServiceOperationAuthorizationException; use Workflow\V2\Models\ActivityAttempt; use Workflow\V2\Models\ActivityExecution; use Workflow\V2\Models\WorkflowChildCall; @@ -1366,6 +1367,17 @@ public function complete(string $taskId, array $commands): array $sequence = WorkflowStepHistory::nextDurableCommandSequence($run); $createdTaskIds = []; + $serviceError = $this->serviceOperationAuthorityError($run, $parsed['non_terminal'], $sequence); + if ($serviceError !== null) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => $run->id, + 'run_status' => $run->status->value, + 'created_task_ids' => [], + 'reason' => $serviceError, + ]; + } $scopeError = $this->validateCancellationScopeMembership($run, $parsed['non_terminal'], $sequence); if ($scopeError !== null) { return [ @@ -1460,6 +1472,15 @@ public function complete(string $taskId, array $commands): array 'reason' => null, ]; }); + } catch (ServiceOperationAuthorizationException $refusal) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => null, + 'run_status' => null, + 'created_task_ids' => [], + 'reason' => $refusal->getMessage(), + ]; } catch (HistoryTransactionLimitExceededException $refusal) { return [ 'completed' => false, @@ -1652,6 +1673,10 @@ private function checkpointCommands( || ($run->run_deadline_at !== null && now()->gte($run->run_deadline_at))) { return $refused('run_deadline_expired'); } + $serviceError = $this->serviceOperationAuthorityError($run, $parsed['non_terminal'], $startSequence); + if ($serviceError !== null) { + return $refused($serviceError); + } $payload = is_array($task->payload) ? $task->payload : []; $receiptKey = $scopePrefix ? 'portable_scope_checkpoint' : ($group ? 'portable_local_group_checkpoint' : 'portable_local_checkpoint'); @@ -1814,6 +1839,8 @@ private function checkpointCommands( 'reason' => null, ]; }, 5); + } catch (ServiceOperationAuthorizationException $refusal) { + return $refused($refusal->getMessage()); } catch (HistoryTransactionLimitExceededException $refusal) { return $refused(self::meterRefusalReason($refusal)); } @@ -3606,17 +3633,13 @@ private function applyStartServiceOperation( $payloadCodec = is_string($command['payload_codec'] ?? null) && $command['payload_codec'] !== '' ? $command['payload_codec'] : ($run->payload_codec ?? CodecRegistry::defaultCodec()); + $options = $this->serviceOperationControlPlaneOptions($run, $command, $sequence, $payloadCodec); $namespace = is_string($run->namespace) ? $run->namespace : null; $requestPayload = isset($command['request_payload']) && is_string($command['request_payload']) ? ExternalPayloads::externalizeForNamespace($command['request_payload'], $payloadCodec, $namespace) : null; $surface = $this->serviceControlPlane() - ->execute( - $command['endpoint_name'], - $command['service_name'], - $command['operation_name'], - $this->serviceOperationControlPlaneOptions($run, $command, $sequence, $payloadCodec), - ); + ->execute($command['endpoint_name'], $command['service_name'], $command['operation_name'], $options); $eventType = self::serviceOperationEventTypeForSurface($surface); $event = WorkflowHistoryEvent::record($run, $eventType, $this->serviceOperationEventPayload( @@ -3664,6 +3687,9 @@ private function serviceOperationControlPlaneOptions( int $sequence, string $payloadCodec, ): array { + if (! isset($command['service_call_id'])) { + $command['idempotency_key'] ??= $this->defaultServiceOperationIdempotencyKey($run, $sequence); + } $metadata = is_array($command['metadata'] ?? null) ? $command['metadata'] : []; $metadata = [ 'caller_sdk_language' => self::metadataString($metadata, 'caller_sdk_language') ?? 'workflow-php', @@ -3672,20 +3698,17 @@ private function serviceOperationControlPlaneOptions( 'workflow_sequence' => $sequence, ] + $metadata; - return array_filter([ - 'namespace' => self::nonEmptyString($command['namespace'] ?? null) ?? $run->namespace, + return WorkflowServiceOperationAuthority::options($run, $command) + array_filter([ 'arguments' => $this->serviceOperationRequestArguments($command, $payloadCodec), 'payload_blob' => self::nonEmptyString($command['request_payload'] ?? null), 'payload_codec' => $payloadCodec, 'service_call_id' => self::nonEmptyString($command['service_call_id'] ?? null), - 'idempotency_key' => self::nonEmptyString($command['idempotency_key'] ?? null) - ?? $this->defaultServiceOperationIdempotencyKey($run, $sequence), + 'idempotency_key' => self::nonEmptyString($command['idempotency_key'] ?? null), 'mode_override' => self::nonEmptyString($command['mode_override'] ?? null), 'wait_for' => self::nonEmptyString($command['wait_for'] ?? null), 'wait_timeout_seconds' => is_int($command['wait_timeout_seconds'] ?? null) ? $command['wait_timeout_seconds'] : null, - 'caller_namespace' => self::nonEmptyString($command['caller_namespace'] ?? null) ?? $run->namespace, 'caller_workflow_instance_id' => $run->workflow_instance_id, 'caller_workflow_run_id' => $run->id, 'target_workflow_instance_id' => self::nonEmptyString($command['target_workflow_instance_id'] ?? null), @@ -3701,14 +3724,32 @@ private function serviceOperationControlPlaneOptions( 'duplicate_start_policy' => self::nonEmptyString($command['duplicate_start_policy'] ?? null), 'metadata' => $metadata, 'request_payload_reference' => self::nonEmptyString($command['request_payload_reference'] ?? null), - 'principal_subject' => self::nonEmptyString($command['principal_subject'] ?? null), - 'principal_method' => self::nonEmptyString($command['principal_method'] ?? null), - 'principal_roles' => is_array($command['principal_roles'] ?? null) ? $command['principal_roles'] : null, - 'principal_tenant' => self::nonEmptyString($command['principal_tenant'] ?? null), - 'principal_claims' => is_array($command['principal_claims'] ?? null) ? $command['principal_claims'] : null, ], static fn (mixed $value): bool => $value !== null); } + /** + * @param list $commands + */ + private function serviceOperationAuthorityError(WorkflowRun $run, array $commands, int $sequence): ?string + { + try { + foreach ($commands as $command) { + if ($command['type'] === 'start_service_operation') { + if (! isset($command['service_call_id'])) { + $command['idempotency_key'] ??= $this->defaultServiceOperationIdempotencyKey($run, $sequence); + } + WorkflowServiceOperationAuthority::options($run, $command); + } + if ($command['type'] !== 'cancel_selection_operation') { + ++$sequence; + } + } + } catch (ServiceOperationAuthorizationException $exception) { + return $exception->getMessage(); + } + return null; + } + /** * @param array $command */ diff --git a/src/V2/Support/DenyWorkflowServiceOperationAuthorizer.php b/src/V2/Support/DenyWorkflowServiceOperationAuthorizer.php new file mode 100644 index 00000000..b6943820 --- /dev/null +++ b/src/V2/Support/DenyWorkflowServiceOperationAuthorizer.php @@ -0,0 +1,18 @@ +namespace; + if ($target !== $run->namespace) { + throw new ServiceOperationAuthorizationException(); + } + + return new ServiceCallPrincipal(subject: 'workflow:' . $run->id, method: 'workflow'); + } +} diff --git a/src/V2/Support/ServiceCallIdempotency.php b/src/V2/Support/ServiceCallIdempotency.php new file mode 100644 index 00000000..3235eeac --- /dev/null +++ b/src/V2/Support/ServiceCallIdempotency.php @@ -0,0 +1,94 @@ +where('idempotency_scope_hash', self::digest($scope)); + $call = ($lock ? $query->lockForUpdate() : $query) + ->first(); + if ($call !== null) { + self::assertScope($call, $scope); + return $call; + } + [$namespace, $endpoint, $service, $operation, $key] = $scope; + $legacy = ConfiguredV2Models::query('service_call_model', WorkflowServiceCall::class) + ->whereNull('idempotency_scope_hash') + ->where('target_namespace', $namespace) + ->where('endpoint_name', $endpoint) + ->where('service_name', $service) + ->where('operation_name', $operation) + ->where('idempotency_key', $key) + ->oldest('created_at') + ->oldest('id'); + $call = ($lock ? $legacy->lockForUpdate() : $legacy) + ->first(); + if ($call !== null) { + self::assertScope($call, $scope); + } + return $call; + } + + /** + * Claim only after caller/catalog authorization. The savepoint keeps a + * PostgreSQL uniqueness loser usable for locked refetch in the outer turn. + * @param array{?string, string, string, string, string} $scope + */ + public static function save(WorkflowServiceCall $call, array $scope): WorkflowServiceCall + { + self::assertScope($call, $scope); + if ($call->exists && $call->idempotency_scope_hash === self::digest($scope)) { + return $call; + } + $connection = $call->getConnection(); + try { + $connection->transaction(static function () use ($call, $scope): void { + $call->idempotency_scope_hash = self::digest($scope); + $call->save(); + }); + return $call; + } catch (UniqueConstraintViolationException $failure) { + $query = ConfiguredV2Models::query('service_call_model', WorkflowServiceCall::class) + ->where('idempotency_scope_hash', self::digest($scope))->lockForUpdate(); + $winner = $query->first(); + if ($winner === null) { + throw $failure; + } + self::assertScope($winner, $scope); + return $winner; + } + } + + /** + * @param array{?string, string, string, string, string} $scope + */ + private static function assertScope(WorkflowServiceCall $call, array $scope): void + { + $actual = [$call->target_namespace, $call->endpoint_name, $call->service_name, + $call->operation_name, $call->idempotency_key]; + if ($call->namespace !== $scope[0] || $actual !== $scope) { + throw new ServiceOperationAuthorizationException('service_call_idempotency_scope_conflict'); + } + } +} diff --git a/src/V2/Support/ServiceCallPrincipal.php b/src/V2/Support/ServiceCallPrincipal.php index 4e2ef5e6..9987006e 100644 --- a/src/V2/Support/ServiceCallPrincipal.php +++ b/src/V2/Support/ServiceCallPrincipal.php @@ -11,8 +11,8 @@ * the audit row so accepted *and* rejected calls share the same * audit shape. This object is intentionally narrow — only the fields * that survive into the durable audit row — so it can be reconstructed - * cleanly from server-side `Principal`, from worker-supplied identity - * headers, or from a synthetic "system" caller. + * cleanly from a server-authenticated `Principal` or an explicitly + * authorized embedded workload. Worker command fields are not identity. * * Privacy boundary: this DTO carries identity fields (subject, method, * roles, tenant, optional opaque claims). Payload material — request diff --git a/src/V2/Support/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php index 5a5ab588..24e42312 100644 --- a/src/V2/Support/WorkflowExecutor.php +++ b/src/V2/Support/WorkflowExecutor.php @@ -5969,14 +5969,16 @@ private function recordServiceOperationEvent( : CodecRegistry::defaultCodec()); $namespace = is_string($run->namespace) ? $run->namespace : null; $serializedRequest = Serializer::serializeWithCodec($payloadCodec, $call->requestPayload); + $options = $this->serviceOperationControlPlaneOptions( + $run, + $sequence, + $call, + $payloadCodec, + $serializedRequest + ); $storedRequest = ExternalPayloads::externalizeForNamespace($serializedRequest, $payloadCodec, $namespace); $surface = $this->serviceControlPlane() - ->execute( - $call->endpointName, - $call->serviceName, - $call->operationName, - $this->serviceOperationControlPlaneOptions($run, $sequence, $call, $payloadCodec, $serializedRequest), - ); + ->execute($call->endpointName, $call->serviceName, $call->operationName, $options); $eventType = self::serviceOperationEventTypeForSurface($surface); $event = WorkflowHistoryEvent::record( @@ -6001,6 +6003,9 @@ private function serviceOperationControlPlaneOptions( string $serializedRequest, ): array { $options = $call->options?->toCommandOptions() ?? []; + if (! isset($options['service_call_id'])) { + $options['idempotency_key'] ??= $this->defaultServiceOperationIdempotencyKey($run, $sequence); + } $metadata = is_array($options['metadata'] ?? null) ? $options['metadata'] : []; $metadata = [ 'caller_sdk_language' => 'workflow-php', @@ -6009,20 +6014,21 @@ private function serviceOperationControlPlaneOptions( 'workflow_sequence' => $sequence, ] + $metadata; - return array_filter([ - 'namespace' => self::nonEmptyString($options['namespace'] ?? null) ?? $run->namespace, + return WorkflowServiceOperationAuthority::options($run, $options + [ + 'endpoint_name' => $call->endpointName, + 'service_name' => $call->serviceName, + 'operation_name' => $call->operationName, + ]) + array_filter([ 'arguments' => $call->requestPayload, 'payload_blob' => $serializedRequest, 'payload_codec' => $payloadCodec, 'service_call_id' => self::nonEmptyString($options['service_call_id'] ?? null), - 'idempotency_key' => self::nonEmptyString($options['idempotency_key'] ?? null) - ?? $this->defaultServiceOperationIdempotencyKey($run, $sequence), + 'idempotency_key' => self::nonEmptyString($options['idempotency_key'] ?? null), 'mode_override' => self::nonEmptyString($options['mode_override'] ?? null), 'wait_for' => self::nonEmptyString($options['wait_for'] ?? null), 'wait_timeout_seconds' => is_int($options['wait_timeout_seconds'] ?? null) ? $options['wait_timeout_seconds'] : null, - 'caller_namespace' => self::nonEmptyString($options['caller_namespace'] ?? null) ?? $run->namespace, 'caller_workflow_instance_id' => $run->workflow_instance_id, 'caller_workflow_run_id' => $run->id, 'target_workflow_instance_id' => self::nonEmptyString($options['target_workflow_instance_id'] ?? null), @@ -6038,11 +6044,6 @@ private function serviceOperationControlPlaneOptions( 'duplicate_start_policy' => self::nonEmptyString($options['duplicate_start_policy'] ?? null), 'metadata' => $metadata, 'request_payload_reference' => self::nonEmptyString($options['request_payload_reference'] ?? null), - 'principal_subject' => self::nonEmptyString($options['principal_subject'] ?? null), - 'principal_method' => self::nonEmptyString($options['principal_method'] ?? null), - 'principal_roles' => is_array($options['principal_roles'] ?? null) ? $options['principal_roles'] : null, - 'principal_tenant' => self::nonEmptyString($options['principal_tenant'] ?? null), - 'principal_claims' => is_array($options['principal_claims'] ?? null) ? $options['principal_claims'] : null, ], static fn (mixed $value): bool => $value !== null); } diff --git a/src/V2/Support/WorkflowServiceOperationAuthority.php b/src/V2/Support/WorkflowServiceOperationAuthority.php new file mode 100644 index 00000000..96c6b752 --- /dev/null +++ b/src/V2/Support/WorkflowServiceOperationAuthority.php @@ -0,0 +1,221 @@ + $command + * @return array + */ + public static function options(WorkflowRun $run, array $command): array + { + // Caller, call and catalog locks must share the completion transaction. + // A separately configured model connection cannot provide atomic refusal. + $connection = $run->getConnection(); + if ($connection !== DB::connection()) { + throw new ServiceOperationAuthorizationException('service_operation_storage_connection_mismatch'); + } + foreach ([ + ['service_call_model', WorkflowServiceCall::class], + ['service_endpoint_model', WorkflowServiceEndpoint::class], + ['service_model', WorkflowService::class], + ['service_operation_model', WorkflowServiceOperation::class], + ] as [$role, $model]) { + if (ConfiguredV2Models::query($role, $model)->getModel()->getConnection() !== $connection) { + throw new ServiceOperationAuthorizationException('service_operation_storage_connection_mismatch'); + } + } + foreach (['principal', 'principal_subject', 'principal_method', 'principal_roles', + 'principal_tenant', 'principal_claims'] as $field) { + if (array_key_exists($field, $command)) { + throw new ServiceOperationAuthorizationException(); + } + } + if (isset($command['caller_namespace']) && $command['caller_namespace'] !== $run->namespace) { + throw new ServiceOperationAuthorizationException(); + } + + $principal = app(WorkflowServiceOperationAuthorizer::class)->authorize($run, $command); + $namespace = $command['namespace'] ?? $run->namespace; + self::checkReference($run, $command, $namespace, $principal); + + // Preserve explicit null namespaces: null is the embedded namespace, + // not permission to use ambient configuration or any existing call. + return [ + 'namespace' => $namespace, + 'caller_namespace' => $run->namespace, + 'caller_workflow_instance_id' => $run->workflow_instance_id, + 'caller_workflow_run_id' => $run->id, + 'principal' => $principal, + ]; + } + + /** + * Repeated by the control plane on the actual row selected for an effect. + * @param array $scope + */ + public static function authorizeExistingCall( + WorkflowServiceCall $call, + ServiceCallPrincipal $principal, + array $scope, + ?ServiceBoundaryPolicy $policy = null, + ): void { + foreach ($scope + [ + 'target_namespace' => $scope['namespace'], + ] as $field => $expected) { + if ($call->getAttribute($field) !== $expected) { + throw new ServiceOperationAuthorizationException('service_call_not_owned'); + } + } + self::checkCatalog($call, $scope['namespace'], $principal, $policy); + } + + /** + * @param array $command + */ + private static function checkReference( + WorkflowRun $run, + array $command, + ?string $namespace, + ServiceCallPrincipal $principal, + ): void { + $callId = $command['service_call_id'] ?? null; + $key = $command['idempotency_key'] ?? null; + if ($callId === null && $key === null) { + return; + } + + if ($callId !== null) { + $call = ConfiguredV2Models::query('service_call_model', WorkflowServiceCall::class) + ->whereKey($callId) + ->lockForUpdate() + ->first(); + } else { + $call = ServiceCallIdempotency::find([$namespace, strtolower(trim($command['endpoint_name'])), + strtolower(trim($command['service_name'])), strtolower(trim($command['operation_name'])), $key]); + } + if ($call === null && $callId === null) { + return; + } + if ($call === null) { + throw new ServiceOperationAuthorizationException('service_call_not_owned'); + } + self::authorizeExistingCall($call, $principal, [ + 'namespace' => $namespace, + 'caller_namespace' => $run->namespace, + 'caller_workflow_instance_id' => $run->workflow_instance_id, + 'caller_workflow_run_id' => $run->id, + 'endpoint_name' => strtolower(trim($command['endpoint_name'])), + 'service_name' => strtolower(trim($command['service_name'])), + 'operation_name' => strtolower(trim($command['operation_name'])), + ]); + } + + private static function checkCatalog( + WorkflowServiceCall $call, + ?string $namespace, + ServiceCallPrincipal $principal, + ?ServiceBoundaryPolicy $policy = null, + ): void { + $catalog = []; + $resolutionFailure = $call->status === ServiceCallStatus::Failed->value + && $call->outcome === ServiceCallOutcome::RejectedNotFound + && ($call->outcome_metadata['failure_reason'] ?? null) === 'resolution_failure'; + foreach ([ + [ + 'service_endpoint_model', WorkflowServiceEndpoint::class, $call->workflow_service_endpoint_id, + [ + 'endpoint_name' => $call->endpoint_name, + ]], + [ + 'service_model', WorkflowService::class, $call->workflow_service_id, + [ + 'workflow_service_endpoint_id' => $call->workflow_service_endpoint_id, + 'service_name' => $call->service_name, + ]], + [ + 'service_operation_model', WorkflowServiceOperation::class, $call->workflow_service_operation_id, + [ + 'workflow_service_endpoint_id' => $call->workflow_service_endpoint_id, + 'workflow_service_id' => $call->workflow_service_id, + 'operation_name' => $call->operation_name, + ]], + ] as [$role, $model, $id, $attributes]) { + // A retained resolution failure may have no catalog component. + // Known components still have to be this call's scoped objects. + if ($id === 'unresolved' && $resolutionFailure) { + continue; + } + $row = ConfiguredV2Models::query($role, $model)->where('namespace', $namespace) + ->lockForUpdate() + ->find($id); + if ($row === null) { + throw new ServiceOperationAuthorizationException('service_call_not_owned'); + } + foreach ([ + 'id' => $id, + 'namespace' => $namespace, + ] + $attributes as $field => $expected) { + if ($row->getAttribute($field) !== $expected) { + throw new ServiceOperationAuthorizationException('service_call_not_owned'); + } + } + $catalog[$role] = $row; + } + // A retained resolution failure has no operation to invoke. Its known + // components and caller scope were still checked above. + if (count($catalog) !== 3) { + return; + } + $policy ??= app(ServiceBoundaryPolicy::class); + if (! is_callable([$policy, 'authorizeReplay'])) { + throw new ServiceOperationAuthorizationException('service_operation_replay_authority_unavailable'); + } + $operation = $catalog['service_operation_model']; + $decision = $policy->authorizeReplay(new ServiceBoundaryRequest( + principal: $principal, + callerNamespace: $call->caller_namespace, + targetNamespace: $namespace ?? '', + endpointName: $call->endpoint_name, + serviceName: $call->service_name, + operationName: $call->operation_name, + operationMode: ServiceCallOperationMode::tryFromCatalog($call->operation_mode) + ?? ServiceCallOperationMode::Async, + resolvedBindingKind: $call->resolved_binding_kind, + resolvedTargetReference: $call->resolved_target_reference, + callerWorkflowInstanceId: $call->caller_workflow_instance_id, + callerWorkflowRunId: $call->caller_workflow_run_id, + linkedWorkflowInstanceId: $call->linked_workflow_instance_id, + linkedWorkflowRunId: $call->linked_workflow_run_id, + linkedWorkflowUpdateId: $call->linked_workflow_update_id, + idempotencyKey: $call->idempotency_key, + endpointBoundaryPolicy: $catalog['service_endpoint_model']->boundary_policy ?? [], + serviceBoundaryPolicy: $catalog['service_model']->boundary_policy ?? [], + operationBoundaryPolicy: $operation->boundary_policy ?? [], + )); + if (! $decision instanceof ServiceBoundaryDecision || $decision->isDenied()) { + throw new ServiceOperationAuthorizationException( + $decision instanceof ServiceBoundaryDecision + ? ($decision->reason ?? 'service_operation_authority_invalid') + : 'service_operation_replay_authority_unavailable', + ); + } + } +} diff --git a/src/migrations/2026_10_25_000100_add_service_call_idempotency_scope.php b/src/migrations/2026_10_25_000100_add_service_call_idempotency_scope.php new file mode 100644 index 00000000..b19401ec --- /dev/null +++ b/src/migrations/2026_10_25_000100_add_service_call_idempotency_scope.php @@ -0,0 +1,96 @@ +getDriverName(); + $previous = null; + if (in_array($driver, ['mysql', 'mariadb'], true)) { + $previous = (int) ((array) $connection->selectOne( + 'SELECT @@SESSION.lock_wait_timeout AS seconds' + ))['seconds']; + $connection->statement('SET SESSION lock_wait_timeout = 1'); + } elseif ($driver === 'pgsql') { + $previous = ((array) $connection->selectOne('SHOW lock_timeout'))['lock_timeout']; + $connection->selectOne("SELECT set_config('lock_timeout', '1s', false)"); + } + try { + $table = $connection->getQueryGrammar() + ->wrapTable(self::TABLE); + if (! Schema::hasColumn(self::TABLE, 'idempotency_scope_hash')) { + if (in_array($driver, ['mysql', 'mariadb'], true)) { + $connection->statement( + "ALTER TABLE {$table} ADD COLUMN idempotency_scope_hash CHAR(64) NULL, ALGORITHM=INSTANT" + ); + } else { + Schema::table(self::TABLE, static function (Blueprint $table): void { + $table->char('idempotency_scope_hash', 64) + ->nullable(); + }); + } + } + $physical = $connection->getTablePrefix() . self::TABLE; + $index = $connection->getQueryGrammar() + ->wrap(self::INDEX); + if ($driver === 'pgsql') { + $existing = $connection->selectOne( + 'SELECT ix.indisvalid FROM pg_index ix JOIN pg_class i ON i.oid = ix.indexrelid ' + . 'JOIN pg_class t ON t.oid = ix.indrelid JOIN pg_namespace n ON n.oid = t.relnamespace ' + . 'WHERE n.nspname = current_schema() AND t.relname = ? AND i.relname = ?', + [$physical, self::INDEX], + ); + if ($existing !== null && ! in_array($existing->indisvalid, [true, 1, '1', 't'], true)) { + $connection->statement("DROP INDEX CONCURRENTLY {$index}"); + $existing = null; + } + if ($existing === null) { + $connection->statement( + "CREATE UNIQUE INDEX CONCURRENTLY {$index} ON {$table} (idempotency_scope_hash)" + ); + } + } elseif (in_array($driver, ['mysql', 'mariadb'], true)) { + $existing = $connection->selectOne( + 'SELECT INDEX_NAME FROM information_schema.statistics WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME = ?', + [$physical, self::INDEX], + ); + if ($existing === null) { + $connection->statement( + "ALTER TABLE {$table} ADD UNIQUE INDEX {$index} (idempotency_scope_hash), ALGORITHM=INPLACE, LOCK=NONE" + ); + } + } else { + $indexes = $connection->select("PRAGMA index_list({$table})"); + if (! in_array(self::INDEX, array_column($indexes, 'name'), true)) { + Schema::table(self::TABLE, static function (Blueprint $table): void { + $table->unique('idempotency_scope_hash', self::INDEX); + }); + } + } + } finally { + if (in_array($driver, ['mysql', 'mariadb'], true)) { + $connection->statement('SET SESSION lock_wait_timeout = ' . $previous); + } elseif ($driver === 'pgsql') { + $connection->selectOne("SELECT set_config('lock_timeout', ?, false)", [$previous]); + } + } + } + + public function down(): void + { + // Keep the nullable expansion and canonical claims on rollback. Older + // readers can ignore them; dropping the guard would race live writers. + } +}; diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 05ed98a9..8d9df9de 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -333,6 +333,12 @@ public function testFixturesExecuteThroughDeclaredReplayConsumers(): void continue; } + if ($fixture['id'] === 'workflow-service-operation-host-authority') { + $this->clearWorkflowState(); + $this->assertNexusWorkerCannotSupplyAuthority($fixture); + continue; + } + if ($fixture['id'] === 'nexus-waiting-caller-terminal-delivery') { $this->clearWorkflowState(); $this->assertWaitingNexusCallerReceivesResult( diff --git a/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-host-authority.json b/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-host-authority.json new file mode 100644 index 00000000..4f01b6bb --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-host-authority.json @@ -0,0 +1,48 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "workflow-service-operation-host-authority", + "protocol_version": "1.19", + "bindings": ["php"], + "consumers": ["workflow-fiber-runner", "workflow-executor"], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestNexusWaitingCallerWorkflow", + "arguments": ["world"], + "payload_codec": "avro" + }, + "history": [ + {"sequence": 1, "event_type": "WorkflowStarted", "payload": {}}, + { + "sequence": 2, + "event_type": "ServiceCallStarted", + "payload": { + "sequence": 1, + "service_call_id": "authority-recorded-call", + "endpoint_name": "php-service", + "service_name": "greeter", + "operation_name": "greet", + "status": "started", + "operation_mode": "sync", + "wait_for": "completed", + "service_call": {"service_call_id": "authority-recorded-call", "status": "started"} + } + } + ], + "authority": { + "commands": [ + {"type": "start_timer", "delay_seconds": 60}, + { + "type": "start_service_operation", + "endpoint_name": "billing", + "service_name": "invoices", + "operation_name": "create", + "namespace": "foreign", + "caller_namespace": "foreign", + "principal_subject": "administrator", + "principal_roles": ["admin"] + } + ], + "reason": "service_operation_authority_invalid" + }, + "expected": {"completed": false, "result": null, "commands": []} +} diff --git a/tests/Support/NexusCallerDeliveryScenario.php b/tests/Support/NexusCallerDeliveryScenario.php index 5aea820a..390fcd3f 100644 --- a/tests/Support/NexusCallerDeliveryScenario.php +++ b/tests/Support/NexusCallerDeliveryScenario.php @@ -10,20 +10,26 @@ use Workflow\Serializers\Serializer; use Workflow\V2\Contracts\ServiceControlPlane; use Workflow\V2\Contracts\WorkflowControlPlane; +use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; use Workflow\V2\Enums\TaskStatus; +use Workflow\V2\Exceptions\ServiceOperationAuthorizationException; use Workflow\V2\Jobs\RunWorkflowTask; +use Workflow\V2\Models\WorkflowCommand; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowInstance; use Workflow\V2\Models\WorkflowRun; +use Workflow\V2\Models\WorkflowRunSummary; use Workflow\V2\Models\WorkflowService; use Workflow\V2\Models\WorkflowServiceCall; use Workflow\V2\Models\WorkflowServiceEndpoint; use Workflow\V2\Models\WorkflowServiceOperation; use Workflow\V2\Models\WorkflowTask; +use Workflow\V2\Models\WorkflowTimer; use Workflow\V2\Support\DefaultServiceControlPlane; +use Workflow\V2\Support\ServiceCallPrincipal; use Workflow\V2\Support\WorkflowFiberRunner; use Workflow\V2\TaskWatchdog; use Workflow\V2\Workflow; @@ -31,12 +37,51 @@ trait NexusCallerDeliveryScenario { + /** + * @param array $fixture + */ + protected function assertNexusWorkerCannotSupplyAuthority(array $fixture): void + { + $previousDispatchMode = config('workflows.v2.task_dispatch_mode'); + config([ + 'workflows.v2.task_dispatch_mode' => 'poll', + ]); + try { + $workflow = WorkflowStub::make($fixture['workflow']['type'], 'authority-corpus-caller', 'caller'); + $this->assertTrue($workflow->attemptStart(...$fixture['workflow']['arguments'])->accepted()); + $task = WorkflowTask::query()->where('workflow_run_id', $workflow->runId())->sole(); + $bridge = $this->app->make(WorkflowTaskBridge::class); + $this->assertTrue($bridge->claimStatus($task->id, 'authority-corpus-worker')['claimed']); + $models = [WorkflowInstance::class, WorkflowRun::class, WorkflowTask::class, WorkflowTimer::class, + WorkflowCommand::class, WorkflowHistoryEvent::class, WorkflowRunSummary::class, WorkflowServiceCall::class]; + $snapshot = static function () use ($models): array { + $rows = []; + foreach ($models as $model) { + $rows[$model] = $model::query()->orderBy('id')->get()->map( + static fn ($row): array => $row->getRawOriginal(), + )->all(); + } + return $rows; + }; + $before = $snapshot(); + $result = $bridge->complete($task->id, $fixture['authority']['commands']); + $this->assertFalse($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $this->assertSame($fixture['authority']['reason'], $result['reason']); + $this->assertSame($before, $snapshot()); + } finally { + config([ + 'workflows.v2.task_dispatch_mode' => $previousDispatchMode, + ]); + } + } + /** * @param array $fixture */ protected function assertNexusCallerSdkAttribution(array $fixture): void { $previous = $this->app->make(ServiceControlPlane::class); + $previousAuthority = $this->installNexusServiceAuthority(); $previousDispatchMode = config('workflows.v2.task_dispatch_mode'); config([ 'workflows.v2.task_dispatch_mode' => 'poll', @@ -96,6 +141,9 @@ protected function assertNexusCallerSdkAttribution(array $fixture): void 'workflows.v2.task_dispatch_mode' => $previousDispatchMode, ]); $this->app->instance(ServiceControlPlane::class, $previous); + if ($previousAuthority !== null) { + $this->app->instance(WorkflowServiceOperationAuthorizer::class, $previousAuthority); + } } } @@ -113,6 +161,7 @@ protected function assertWaitingNexusCallerReceivesResult( ): void { $previousDispatchMode = config('workflows.v2.task_dispatch_mode'); $previousControlPlane = $this->app->make(ServiceControlPlane::class); + $previousAuthority = $this->installNexusServiceAuthority(); $previousNow = Carbon::getTestNow(); config([ 'workflows.v2.task_dispatch_mode' => 'poll', @@ -243,6 +292,35 @@ protected function assertWaitingNexusCallerReceivesResult( 'workflows.v2.task_dispatch_mode' => $previousDispatchMode, ]); $this->app->instance(ServiceControlPlane::class, $previousControlPlane); + if ($previousAuthority !== null) { + $this->app->instance(WorkflowServiceOperationAuthorizer::class, $previousAuthority); + } + } + } + + private function installNexusServiceAuthority(): ?object + { + // The retained consumer also runs against the pre-authorizer base. + if (! interface_exists(WorkflowServiceOperationAuthorizer::class)) { + return null; } + $previous = $this->app->make(WorkflowServiceOperationAuthorizer::class); + $this->app->instance( + WorkflowServiceOperationAuthorizer::class, + new class() implements WorkflowServiceOperationAuthorizer { + public function authorize(WorkflowRun $run, array $command): ServiceCallPrincipal + { + $target = $command['namespace'] ?? $run->namespace; + $operation = [$command['endpoint_name'], $command['service_name'], $command['operation_name']]; + $allowed = ($target === 'caller' && $operation === ['billing', 'invoices', 'create']) + || ($target === 'service' && $operation === ['php-service', 'greeter', 'greet']); + if ($run->namespace !== 'caller' || ! $allowed) { + throw new ServiceOperationAuthorizationException(); + } + return new ServiceCallPrincipal(subject: 'workflow:' . $run->id, method: 'workflow'); + } + } + ); + return $previous; } } diff --git a/tests/Unit/V2/DefaultWorkflowTaskBridgeServiceOperationTest.php b/tests/Unit/V2/DefaultWorkflowTaskBridgeServiceOperationTest.php index 7dc2e92b..d5173165 100644 --- a/tests/Unit/V2/DefaultWorkflowTaskBridgeServiceOperationTest.php +++ b/tests/Unit/V2/DefaultWorkflowTaskBridgeServiceOperationTest.php @@ -13,6 +13,7 @@ use Workflow\Serializers\Serializer; use Workflow\V2\Attributes\Type; use Workflow\V2\Contracts\ServiceControlPlane; +use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; @@ -23,6 +24,8 @@ use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowRunSummary; use Workflow\V2\Models\WorkflowTask; +use Workflow\V2\Support\RunServiceOperationAuthorizer; +use Workflow\V2\Support\ServiceCallPrincipal; use Workflow\V2\Support\WorkerCompatibilityFleet; use Workflow\V2\Workflow; use Workflow\V2\WorkflowStub; @@ -38,6 +41,7 @@ final class DefaultWorkflowTaskBridgeServiceOperationTest extends TestCase protected function setUp(): void { parent::setUp(); + $this->app->instance(WorkflowServiceOperationAuthorizer::class, new RunServiceOperationAuthorizer()); Carbon::setTestNow('2026-10-08T02:00:00Z'); config([ 'workflows.v2.task_dispatch_mode' => 'poll', @@ -304,9 +308,8 @@ public function testServiceOptionsPreserveTypedValuesAndAuthoritativeCallerMetad ]; $payload = Serializer::serialize($arguments); $options = [ - 'namespace' => 'billing', - 'caller_namespace' => 'finance', - 'service_call_id' => 'requested-service-call', + 'namespace' => 'caller', + 'caller_namespace' => 'caller', 'idempotency_key' => 'invoice-request-42', 'mode_override' => ' SYNC ', 'wait_for' => ' COMPLETED ', @@ -327,13 +330,6 @@ public function testServiceOptionsPreserveTypedValuesAndAuthoritativeCallerMetad ], 'duplicate_start_policy' => 'return_existing_active', 'request_payload_reference' => 'invoice-request-reference', - 'principal_subject' => 'user:billing-clerk', - 'principal_method' => 'token', - 'principal_roles' => ['invoice:create', 'invoice:read'], - 'principal_tenant' => 'customer-a', - 'principal_claims' => [ - 'approved' => false, - ], 'metadata' => [ 'caller_sdk_language' => 'sdk-python', 'caller_workflow_instance_id' => 'forged-owner', @@ -368,7 +364,13 @@ public function testServiceOptionsPreserveTypedValuesAndAuthoritativeCallerMetad 'service_sdk_language' => 'rust', 'trace_id' => 'trace-42', ]; + $this->assertInstanceOf(ServiceCallPrincipal::class, $forwarded['principal']); + $forwarded['principal'] = $forwarded['principal']->toAuditArray(); $expected += [ + 'principal' => (new ServiceCallPrincipal( + subject: 'workflow:' . $this->run->id, + method: 'workflow' + ))->toAuditArray(), 'arguments' => $arguments, 'payload_blob' => $payload, 'payload_codec' => 'avro', diff --git a/tests/Unit/V2/EmbeddedContinuationTest.php b/tests/Unit/V2/EmbeddedContinuationTest.php index e84f667f..0d5144c5 100644 --- a/tests/Unit/V2/EmbeddedContinuationTest.php +++ b/tests/Unit/V2/EmbeddedContinuationTest.php @@ -35,7 +35,9 @@ use Workflow\V2\Attributes\Type; use Workflow\V2\Contracts\ServiceControlPlane; use Workflow\V2\Contracts\WorkflowControlPlane; +use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer; use Workflow\V2\Enums\HistoryEventType; +use Workflow\V2\Exceptions\ServiceOperationAuthorizationException; use Workflow\V2\Exceptions\WorkflowEntryArgumentsMismatchException; use Workflow\V2\Jobs\RunWorkflowTask; use Workflow\V2\Models\WorkflowHistoryEvent; @@ -49,6 +51,7 @@ use Workflow\V2\StartOptions; use Workflow\V2\Support\DefaultServiceControlPlane; use Workflow\V2\Support\ExecutionTracing; +use Workflow\V2\Support\ServiceCallPrincipal; use Workflow\V2\Support\ServiceOperationOptions; use Workflow\V2\Support\ServiceOperationResult; use Workflow\V2\Support\TaskRepairPolicy; @@ -1321,6 +1324,20 @@ private function workOnce(): void private function registerNexusService(int &$starts): WorkflowRun { + $this->app->instance( + WorkflowServiceOperationAuthorizer::class, + new class() implements WorkflowServiceOperationAuthorizer { + public function authorize(WorkflowRun $run, array $command): ServiceCallPrincipal + { + if ($run->namespace !== 'continuation-test' || ($command['namespace'] ?? null) !== 'service' + || [$command['endpoint_name'], $command['service_name'], $command['operation_name']] + !== ['php-service', 'greeter', 'greet']) { + throw new ServiceOperationAuthorizationException(); + } + return new ServiceCallPrincipal(subject: 'workflow:' . $run->id, method: 'workflow'); + } + } + ); $instance = WorkflowInstance::query()->create([ 'namespace' => 'service', 'workflow_type' => 'nexus.service', diff --git a/tests/Unit/V2/LinkedServiceCallCloserTest.php b/tests/Unit/V2/LinkedServiceCallCloserTest.php index 173906cb..efd4cef1 100644 --- a/tests/Unit/V2/LinkedServiceCallCloserTest.php +++ b/tests/Unit/V2/LinkedServiceCallCloserTest.php @@ -15,6 +15,7 @@ use Workflow\V2\Contracts\ExternalPayloadStoragePolicy; use Workflow\V2\Contracts\ServiceControlPlane; use Workflow\V2\Contracts\WorkflowControlPlane; +use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; @@ -22,6 +23,7 @@ use Workflow\V2\Enums\ServiceCallOutcome; use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Enums\TaskType; +use Workflow\V2\Exceptions\ServiceOperationAuthorizationException; use Workflow\V2\Jobs\RunWorkflowTask; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowInstance; @@ -36,6 +38,7 @@ use Workflow\V2\Support\ExternalPayloadStorage; use Workflow\V2\Support\LinkedServiceCallCloser; use Workflow\V2\Support\ServiceCallDeadline; +use Workflow\V2\Support\ServiceCallPrincipal; use Workflow\V2\Support\ServiceCallView; use Workflow\V2\Support\ServiceCallWorkflowDelivery; use Workflow\V2\Support\ServiceOperationOptions; @@ -456,6 +459,7 @@ public function testEmbeddedCallerWaitsAndThenReplaysTheActualServiceResult(): v 'workflow_run_id' => $service->id, ]); $this->registerWorkflowService(); + $this->registerTrustedServiceOperationHost(); $this->app->instance(ServiceControlPlane::class, new DefaultServiceControlPlane($control)); $workflow = WorkflowStub::make(NexusWaitingCallerWorkflow::class, 'waiting-caller', 'caller'); $this->assertTrue($workflow->attemptStart()->accepted()); @@ -997,6 +1001,25 @@ private function registerWorkflowService(): void ]); } + private function registerTrustedServiceOperationHost(): void + { + $this->app->instance( + WorkflowServiceOperationAuthorizer::class, + new class() implements WorkflowServiceOperationAuthorizer { + public function authorize(WorkflowRun $run, array $command): ServiceCallPrincipal + { + if ($run->namespace !== 'caller' || ($command['namespace'] ?? null) !== 'service' + || [$command['endpoint_name'], $command['service_name'], $command['operation_name']] + !== ['php-service', 'greeter', 'greet']) { + throw new ServiceOperationAuthorizationException(); + } + + return new ServiceCallPrincipal(subject: 'workflow:' . $run->id, method: 'workflow'); + } + } + ); + } + private function createRun(string $namespace = 'service'): WorkflowRun { $instance = WorkflowInstance::query()->create([ diff --git a/tests/Unit/V2/NexusCancellationDeliveryTest.php b/tests/Unit/V2/NexusCancellationDeliveryTest.php index e5e5aa11..794f3a22 100644 --- a/tests/Unit/V2/NexusCancellationDeliveryTest.php +++ b/tests/Unit/V2/NexusCancellationDeliveryTest.php @@ -11,20 +11,24 @@ use Tests\Fixtures\V2\TestNexusWaitingCallerWorkflow; use Tests\TestCase; use Workflow\Serializers\Serializer; +use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Enums\TaskType; +use Workflow\V2\Exceptions\ServiceOperationAuthorizationException; use Workflow\V2\Exceptions\WorkflowCancellationRequestedException; use Workflow\V2\Jobs\RunTimerTask; use Workflow\V2\Jobs\RunWorkflowTask; +use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowService; use Workflow\V2\Models\WorkflowServiceCall; use Workflow\V2\Models\WorkflowServiceEndpoint; use Workflow\V2\Models\WorkflowServiceOperation; use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Support\CooperativeCancellationDelivery; +use Workflow\V2\Support\ServiceCallPrincipal; use Workflow\V2\Support\ServiceOperationCancellation; use Workflow\V2\Support\WorkflowFiberRunner; use Workflow\V2\WorkflowStub; @@ -40,6 +44,21 @@ protected function setUp(): void 'workflows.v2.compatibility.current' => 'build-a', 'workflows.v2.compatibility.supported' => ['build-a'], ]); + $this->app->instance( + WorkflowServiceOperationAuthorizer::class, + new class() implements WorkflowServiceOperationAuthorizer { + public function authorize(WorkflowRun $run, array $command): ServiceCallPrincipal + { + if ($run->namespace !== 'caller' || ($command['namespace'] ?? null) !== 'service' + || [$command['endpoint_name'], $command['service_name'], $command['operation_name']] + !== ['php-service', 'greeter', 'greet']) { + throw new ServiceOperationAuthorizationException(); + } + + return new ServiceCallPrincipal(subject: 'workflow:' . $run->id, method: 'workflow'); + } + } + ); Carbon::setTestNow('2026-10-12T08:00:00Z'); } diff --git a/tests/Unit/V2/WorkflowExecutorServiceOperationTest.php b/tests/Unit/V2/WorkflowExecutorServiceOperationTest.php index 843b1f78..f4b03591 100644 --- a/tests/Unit/V2/WorkflowExecutorServiceOperationTest.php +++ b/tests/Unit/V2/WorkflowExecutorServiceOperationTest.php @@ -15,6 +15,7 @@ use Workflow\Serializers\Serializer; use Workflow\V2\Attributes\Type; use Workflow\V2\Contracts\ServiceControlPlane; +use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Exceptions\RestoredWorkflowException; @@ -26,7 +27,9 @@ use Workflow\V2\Models\WorkflowRunSummary; use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Support\DefaultWorkflowTaskBridge; +use Workflow\V2\Support\RunServiceOperationAuthorizer; use Workflow\V2\Support\RunTimelineProjector; +use Workflow\V2\Support\ServiceCallPrincipal; use Workflow\V2\Support\ServiceOperationOptions; use Workflow\V2\Support\ServiceOperationResult; use Workflow\V2\Support\WorkerCompatibilityFleet; @@ -39,6 +42,7 @@ final class WorkflowExecutorServiceOperationTest extends TestCase protected function setUp(): void { parent::setUp(); + $this->app->instance(WorkflowServiceOperationAuthorizer::class, new RunServiceOperationAuthorizer()); Carbon::setTestNow('2026-10-08 17:10:00'); Queue::fake(); config() @@ -107,6 +111,10 @@ public function testPublicExecutionRecordsTheBoundaryOutcomeAndReplaysWithoutCal $this->assertCount(1, $service->executions); $request = Serializer::serializeWithCodec('avro', ExecutorServiceOperationWorkflow::REQUEST); $expectedOptions = [ + 'principal' => (new ServiceCallPrincipal( + subject: 'workflow:' . $run->id, + method: 'workflow' + ))->toAuditArray(), 'namespace' => 'service-caller', 'arguments' => ExecutorServiceOperationWorkflow::REQUEST, 'payload_blob' => $request, @@ -128,6 +136,8 @@ public function testPublicExecutionRecordsTheBoundaryOutcomeAndReplaysWithoutCal 'wait_for' => 'accepted', ]; } + $this->assertInstanceOf(ServiceCallPrincipal::class, $service->executions[0][3]['principal']); + $service->executions[0][3]['principal'] = $service->executions[0][3]['principal']->toAuditArray(); $this->assertSame( $this->ordered(['billing', 'invoices', 'create', $expectedOptions]), $this->ordered($service->executions[0]) diff --git a/tests/Unit/V2/WorkflowServiceOperationAuthorityTest.php b/tests/Unit/V2/WorkflowServiceOperationAuthorityTest.php new file mode 100644 index 00000000..4542e789 --- /dev/null +++ b/tests/Unit/V2/WorkflowServiceOperationAuthorityTest.php @@ -0,0 +1,971 @@ +app->instance(WorkflowServiceOperationAuthorizer::class, new RunServiceOperationAuthorizer()); + Carbon::setTestNow('2026-10-25T21:00:00Z'); + config([ + 'workflows.v2.task_dispatch_mode' => 'poll', + 'workflows.v2.compatibility.current' => 'authority-test', + 'workflows.v2.compatibility.supported' => ['authority-test'], + 'workflows.v2.namespace' => null, + ]); + WorkerCompatibilityFleet::clear(); + Queue::fake(); + $workflow = WorkflowStub::make(ServiceAuthorityWorkflow::class, 'authority-owner', 'caller'); + $this->assertTrue($workflow->attemptStart()->accepted()); + $this->run = WorkflowRun::query()->findOrFail($workflow->runId()); + $this->task = WorkflowTask::query()->where('workflow_run_id', $this->run->id)->sole(); + $this->bridge = $this->app->make(WorkflowTaskBridge::class); + $this->assertTrue($this->bridge->claimStatus($this->task->id, 'authority-worker')['claimed']); + } + + protected function tearDown(): void + { + Carbon::setTestNow(); + WorkerCompatibilityFleet::clear(); + parent::tearDown(); + } + + /** + * @param array $options + */ + #[DataProvider('forgedAuthority')] + public function testWorkerAuthorityOverridesRefuseTheWholeCompletion(array $options): void + { + $before = $this->records(); + $result = $this->bridge->complete($this->task->id, [ + [ + 'type' => 'start_timer', + 'delay_seconds' => 60, + ], + $this->command() + $options, + ]); + + $this->assertFalse($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $this->assertSame('service_operation_authority_invalid', $result['reason']); + $this->assertSame($before, $this->records()); + } + + /** + * @return iterable}> + */ + public static function forgedAuthority(): iterable + { + yield 'foreign namespace' => [[ + 'namespace' => 'foreign', + ]]; + yield 'foreign caller' => [[ + 'caller_namespace' => 'foreign', + ]]; + yield 'principal subject' => [[ + 'principal_subject' => 'administrator', + ]]; + yield 'principal method' => [[ + 'principal_method' => 'system', + ]]; + yield 'principal roles' => [[ + 'principal_roles' => ['admin'], + ]]; + yield 'principal tenant' => [[ + 'principal_tenant' => 'foreign', + ]]; + yield 'principal claims' => [[ + 'principal_claims' => [ + 'administrator' => true, + ], + ]]; + } + + #[DataProvider('foreignCalls')] + public function testForeignCallReferencesRefuseBeforeAnyCommandWrites( + string $status, + string $mismatch, + bool $idempotency, + ): void { + $call = $this->serviceCall($status); + $call->forceFill(match ($mismatch) { + 'run' => [ + 'caller_workflow_run_id' => 'another-run', + ], + 'instance' => [ + 'caller_workflow_instance_id' => 'another-owner', + ], + 'caller' => [ + 'caller_namespace' => 'another-caller', + ], + 'catalog' => [ + 'operation_name' => 'another-operation', + ], + 'namespace' => [ + 'namespace' => 'foreign', + 'target_namespace' => 'foreign', + ], + 'endpoint id' => [ + 'workflow_service_endpoint_id' => 'foreign-endpoint', + ], + 'service id' => [ + 'workflow_service_id' => 'foreign-service', + ], + 'operation id' => [ + 'workflow_service_operation_id' => 'foreign-operation', + ], + })->save(); + $options = $idempotency ? [ + 'idempotency_key' => $call->idempotency_key, + ] + : [ + 'service_call_id' => $call->id, + ]; + $before = $this->records(); + $result = $this->bridge->complete($this->task->id, [ + [ + 'type' => 'start_timer', + 'delay_seconds' => 60, + ], + $this->command() + $options, + ]); + + $this->assertFalse($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $this->assertSame('service_call_not_owned', $result['reason']); + $this->assertSame($before, $this->records()); + } + + /** + * @return iterable + */ + public static function foreignCalls(): iterable + { + foreach (['pending', 'started', 'completed'] as $status) { + foreach ([ + 'run', + 'instance', + 'caller', + 'catalog', + 'namespace', + 'endpoint id', + 'service id', + 'operation id', + ] as $mismatch) { + yield $status . ' ' . $mismatch => [$status, $mismatch, false]; + } + foreach (['run', 'instance', 'caller'] as $mismatch) { + yield $status . ' idempotency ' . $mismatch => [$status, $mismatch, true]; + } + } + } + + #[DataProvider('ownedCalls')] + public function testOwnedCallReplayRemainsUsable(string $status, bool $idempotency): void + { + $call = $this->serviceCall($status); + $result = $this->bridge->complete($this->task->id, [ + $this->command() + ($idempotency + ? [ + 'idempotency_key' => $call->idempotency_key, + ] : [ + 'service_call_id' => $call->id, + ])]); + $this->assertTrue($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $this->assertSame(1, WorkflowServiceCall::query()->count()); + $this->assertSame($this->run->id, $call->fresh()->caller_workflow_run_id); + } + + /** + * @return iterable + */ + public static function ownedCalls(): iterable + { + foreach (['started', 'completed'] as $status) { + yield $status . ' by id' => [$status, false]; + yield $status . ' by key' => [$status, true]; + } + } + + public function testNewCallsUseTheRunWorkloadPrincipal(): void + { + $result = $this->bridge->complete($this->task->id, [$this->command()]); + $this->assertTrue($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $call = WorkflowServiceCall::query()->sole(); + $this->assertSame('caller', $call->namespace); + $this->assertSame('caller', $call->caller_namespace); + $this->assertSame('workflow:' . $this->run->id, $call->caller_principal_subject); + $this->assertSame('workflow', $call->caller_principal_method); + $this->assertNull($call->caller_principal_roles); + $this->assertNull($call->caller_principal_tenant); + $this->assertNull($call->caller_principal_claims); + } + + public function testMissingTrustedContextRefusesBeforeAnyCommandWrites(): void + { + $this->app->forgetInstance(WorkflowServiceOperationAuthorizer::class); + $this->assertInstanceOf( + DenyWorkflowServiceOperationAuthorizer::class, + $this->app->make(WorkflowServiceOperationAuthorizer::class) + ); + $before = $this->records(); + $result = $this->bridge->complete($this->task->id, [ + [ + 'type' => 'start_timer', + 'delay_seconds' => 60, + ], + $this->command(), + ]); + $this->assertFalse($result['completed']); + $this->assertSame('service_operation_authority_invalid', $result['reason']); + $this->assertSame($before, $this->records()); + } + + public function testNullRunNamespaceDoesNotUseAmbientConfiguration(): void + { + $this->run->instance->forceFill([ + 'namespace' => null, + ])->save(); + $this->run->forceFill([ + 'namespace' => null, + ])->save(); + $this->task->forceFill([ + 'namespace' => null, + ])->save(); + config([ + 'workflows.v2.namespace' => 'ambient-namespace', + ]); + + $result = $this->bridge->complete($this->task->id, [$this->command()]); + $this->assertTrue($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $call = WorkflowServiceCall::query()->sole(); + $this->assertNull($call->namespace); + $this->assertNull($call->target_namespace); + $this->assertNull($call->caller_namespace); + $this->assertSame($this->run->id, $call->caller_workflow_run_id); + } + + public function testDefaultIdempotencyKeyCannotReplayAnotherRunsCall(): void + { + $call = $this->serviceCall('completed'); + $call->forceFill([ + 'caller_workflow_run_id' => 'another-run', + 'idempotency_key' => 'workflow-service-operation:' . $this->run->workflow_instance_id . ':' . $this->run->id . ':2', + ])->save(); + $before = $this->records(); + $result = $this->bridge->complete($this->task->id, [ + [ + 'type' => 'start_timer', + 'delay_seconds' => 60, + ], + $this->command(), + ]); + $this->assertFalse($result['completed']); + $this->assertSame('service_call_not_owned', $result['reason']); + $this->assertSame($before, $this->records()); + } + + public function testRetainedClaimPrefixRefusesForeignCallBeforeWritingItsSibling(): void + { + $call = $this->serviceCall('completed'); + $call->forceFill([ + 'caller_workflow_instance_id' => 'another-instance', + ])->save(); + $before = $this->records(); + $result = $this->app->make(\Workflow\V2\Contracts\PreparedLocalActivityTaskBridge::class) + ->checkpointLocalActivityPrefix( + $this->task->id, + 'authority-worker', + $this->task->fresh() +->attempt_count, + 'authority-checkpoint', + 1, + [ + [ + 'type' => 'start_timer', + 'delay_seconds' => 60, + ], + $this->command() + [ + 'service_call_id' => $call->id, + ], + ], + '1.20' + ); + $this->assertFalse($result['checkpointed'], json_encode($result, JSON_THROW_ON_ERROR)); + $this->assertSame('service_call_not_owned', $result['reason']); + $this->assertSame($before, $this->records()); + } + + public function testDuplicatePrefixStillRequiresCurrentTrustedAuthority(): void + { + $call = $this->serviceCall('completed'); + $bridge = $this->app->make(\Workflow\V2\Contracts\PreparedLocalActivityTaskBridge::class); + $commands = [ + $this->command() + [ + 'service_call_id' => $call->id, + ]]; + $attempt = $this->task->fresh() +->attempt_count; + $first = $bridge->checkpointLocalActivityPrefix( + $this->task->id, + 'authority-worker', + $attempt, + 'authority-checkpoint', + 1, + $commands, + '1.20' + ); + $this->assertTrue($first['checkpointed'], json_encode($first, JSON_THROW_ON_ERROR)); + $before = $this->records(); + $this->app->instance(WorkflowServiceOperationAuthorizer::class, new DenyWorkflowServiceOperationAuthorizer()); + $duplicate = $bridge->checkpointLocalActivityPrefix( + $this->task->id, + 'authority-worker', + $attempt, + 'authority-checkpoint', + 1, + $commands, + '1.20' + ); + $this->assertFalse($duplicate['checkpointed'], json_encode($duplicate, JSON_THROW_ON_ERROR)); + $this->assertSame('service_operation_authority_invalid', $duplicate['reason']); + $this->assertSame($before, $this->records()); + } + + public function testOwnedResolutionFailureCanBeReplayed(): void + { + $first = $this->bridge->complete($this->task->id, [$this->command()]); + $this->assertTrue($first['completed']); + $call = WorkflowServiceCall::query()->sole(); + $before = $call->getRawOriginal(); + $this->assertSame('failed', $call->status); + $task = WorkflowTask::query()->whereKey($first['created_task_ids'][0])->sole(); + $this->assertTrue($this->bridge->claimStatus($task->id, 'authority-worker')['claimed']); + $replay = $this->bridge->complete($task->id, [ + $this->command() + [ + 'service_call_id' => $call->id, + ]]); + $this->assertTrue($replay['completed'], json_encode($replay, JSON_THROW_ON_ERROR)); + $this->assertSame(1, WorkflowServiceCall::query()->count()); + $this->assertSame($before, $call->fresh()->getRawOriginal()); + } + + #[DataProvider('duplicateAuthorityChanges')] + public function testDuplicatePrefixChecksCurrentCallAndCatalogAuthority(string $change, ?string $reason): void + { + $call = $this->serviceCall('completed'); + $bridge = $this->app->make(\Workflow\V2\Contracts\PreparedLocalActivityTaskBridge::class); + $commands = [ + $this->command() + [ + 'service_call_id' => $call->id, + ]]; + $attempt = $this->task->fresh() +->attempt_count; + $first = $bridge->checkpointLocalActivityPrefix( + $this->task->id, + 'authority-worker', + $attempt, + 'authority-checkpoint', + 1, + $commands, + '1.20' + ); + $this->assertTrue($first['checkpointed'], json_encode($first, JSON_THROW_ON_ERROR)); + match ($change) { + 'run' => $call->forceFill([ + 'caller_workflow_run_id' => 'another-run', + ])->save(), + 'instance' => $call->forceFill([ + 'caller_workflow_instance_id' => 'another-instance', + ])->save(), + 'caller' => $call->forceFill([ + 'caller_namespace' => 'another-namespace', + ])->save(), + 'target' => $call->forceFill([ + 'target_namespace' => 'another-namespace', + ])->save(), + 'deleted' => $call->operation->delete(), + 'replacement' => $this->replaceOperation($call), + 'endpoint policy' => $call->endpoint->forceFill([ + 'boundary_policy' => [ + 'required_roles' => ['admin'], + ], + ])->save(), + 'service policy' => $call->service->forceFill([ + 'boundary_policy' => [ + 'required_roles' => ['admin'], + ], + ])->save(), + 'operation policy' => $call->operation->forceFill([ + 'boundary_policy' => [ + 'required_roles' => ['admin'], + ], + ])->save(), + default => null, + }; + $before = $this->records(); + $duplicate = $bridge->checkpointLocalActivityPrefix( + $this->task->id, + 'authority-worker', + $attempt, + 'authority-checkpoint', + 1, + $commands, + '1.20' + ); + $this->assertSame($reason === null, $duplicate['checkpointed'], json_encode($duplicate, JSON_THROW_ON_ERROR)); + $this->assertSame($reason, $duplicate['reason']); + $this->assertSame($before, $this->records()); + } + + /** + * @return iterable + */ + public static function duplicateAuthorityChanges(): iterable + { + foreach (['run', 'instance', 'caller', 'target', 'deleted', 'replacement'] as $change) { + yield $change => [$change, 'service_call_not_owned']; + } + foreach (['endpoint policy', 'service policy', 'operation policy'] as $change) { + yield $change => [$change, 'missing_required_role']; + } + yield 'unchanged owned receipt' => ['none', null]; + } + + #[DataProvider('admissionCaps')] + public function testOwnedReplayDoesNotReserveFreshAdmissionCapacity(array $rules): void + { + $call = $this->serviceCall('completed'); + $policy = new DefaultServiceBoundaryPolicy($rules); + $this->app->instance(ServiceBoundaryPolicy::class, $policy); + $request = new ServiceBoundaryRequest( + principal: new ServiceCallPrincipal(subject: 'workflow:' . $this->run->id, method: 'workflow'), + callerNamespace: 'caller', + targetNamespace: 'caller', + endpointName: 'billing', + serviceName: 'invoices', + operationName: 'create', + operationMode: ServiceCallOperationMode::Async, + ); + $this->assertTrue($policy->evaluate($request)->isAllowed()); + $this->assertTrue($policy->evaluate($request)->isDenied()); + $result = $this->bridge->complete($this->task->id, [ + $this->command() + [ + 'service_call_id' => $call->id, + ]]); + $this->assertTrue($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $this->assertSame(1, WorkflowServiceCall::query()->count()); + } + + /** + * @return iterable}> + */ + public static function admissionCaps(): iterable + { + yield 'rate' => [[ + 'rate_limit' => [ + 'requests_per_minute' => 1, + ], + ]]; + yield 'concurrency' => [[ + 'concurrency' => [ + 'max_in_flight' => 1, + ], + ]]; + } + + public function testLegacyPolicyWithoutReplayAuthorizationFailsClosed(): void + { + $call = $this->serviceCall('completed'); + $this->app->instance(ServiceBoundaryPolicy::class, new class() implements ServiceBoundaryPolicy { + public function evaluate(ServiceBoundaryRequest $request): ServiceBoundaryDecision + { + throw new \LogicException('Replay must not consume a new admission.'); + } + }); + $before = $this->records(); + $result = $this->bridge->complete($this->task->id, [ + $this->command() + [ + 'service_call_id' => $call->id, + ]]); + $this->assertFalse($result['completed']); + $this->assertSame('service_operation_replay_authority_unavailable', $result['reason']); + $this->assertSame($before, $this->records()); + } + + public function testLegacyPolicyWithPureReplayMethodCanRefuseTheCall(): void + { + $call = $this->serviceCall('completed'); + $this->app->instance(ServiceBoundaryPolicy::class, new class() implements ServiceBoundaryPolicy { + public function evaluate(ServiceBoundaryRequest $request): ServiceBoundaryDecision + { + throw new \LogicException('Replay must not consume a new admission.'); + } + + public function authorizeReplay(ServiceBoundaryRequest $request): ServiceBoundaryDecision + { + return ServiceBoundaryDecision::denyAuthorization( + reason: 'custom_scope_denied', + message: 'Current host replay policy refuses the call.', + ); + } + }); + $before = $this->records(); + $result = $this->bridge->complete($this->task->id, [ + $this->command() + [ + 'service_call_id' => $call->id, + ]]); + $this->assertFalse($result['completed']); + $this->assertSame('custom_scope_denied', $result['reason']); + $this->assertSame($before, $this->records()); + } + + #[DataProvider('concurrentInsertions')] + public function testConcurrentKeyInsertionIsReauthorizedAtTheEffectingLookup( + string $pauseAt, + bool $foreign, + ): void { + $driver = DB::connection()->getDriverName(); + if (! in_array($driver, ['pgsql', 'mysql'], true)) { + $this->markTestSkipped('This interleaving requires independent row-locking transactions.'); + } + if (! function_exists('pcntl_fork') || ! function_exists('posix_kill')) { + $this->markTestSkipped('Process control is required for the transaction interleaving.'); + } + $call = $this->serviceCall(ServiceCallStatus::Completed->value); + if ($foreign) { + $call->caller_workflow_run_id = 'foreign-run'; + } + $attributes = $call->getAttributes(); + $call->delete(); + $scope = ['caller', 'billing', 'invoices', 'create', 'shared-idempotency-key']; + DB::purge(); + $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); + $this->assertIsArray($sockets); + $pid = pcntl_fork(); + $this->assertNotSame(-1, $pid); + if ($pid === 0) { + fclose($sockets[0]); + stream_set_timeout($sockets[1], 15); + $ok = false; + try { + DB::reconnect(); + DB::statement($driver === 'pgsql' + ? 'SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL READ COMMITTED' + : 'SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED'); + $paused = false; + $pause = static function () use (&$paused, $sockets): void { + if ($paused) { + return; + } + $paused = true; + fwrite($sockets[1], "absent\n"); + if (fgets($sockets[1]) !== "committed\n") { + throw new RuntimeException('Concurrent insertion was not committed.'); + } + }; + if ($pauseAt === 'lookup') { + $lookups = 0; + DB::listen(static function (QueryExecuted $query) use ($pause, &$lookups): void { + if (str_starts_with(strtolower($query->sql), 'select') + && str_contains($query->sql, 'workflow_service_calls') + && str_contains($query->sql, 'idempotency_key')) { + // The first lookup is the whole-turn preflight; + // the second is the last authority check before + // the control plane selects its effecting row. + if (++$lookups === 2) { + $pause(); + } + } + }); + } else { + WorkflowServiceCall::creating(static function () use ($pause): void { + $pause(); + }); + } + $result = $this->bridge->complete($this->task->id, [ + [ + 'type' => 'start_timer', + 'delay_seconds' => 60, + ], + $this->command() + [ + 'idempotency_key' => 'shared-idempotency-key', + ], + ]); + fwrite($sockets[1], json_encode([ + 'result' => $result, + ], JSON_THROW_ON_ERROR) . "\n"); + $ok = true; + } catch (Throwable $error) { + fwrite($sockets[1], json_encode([ + 'error' => $error::class . ': ' . $error->getMessage(), + ], JSON_THROW_ON_ERROR) . "\n"); + } finally { + DB::disconnect(); + fclose($sockets[1]); + } + exit($ok ? 0 : 1); + } + fclose($sockets[1]); + stream_set_timeout($sockets[0], 20); + try { + $this->assertSame( + "absent\n", + fgets($sockets[0]), + 'Actor must observe the absent key before the competing commit.' + ); + DB::reconnect(); + DB::transaction(static function () use ($attributes, $pauseAt, $scope): void { + $insert = $attributes; + // The lookup case also covers an old writer without a digest. + if ($pauseAt === 'insert') { + $insert['idempotency_scope_hash'] = ServiceCallIdempotency::digest($scope); + } + DB::table('workflow_service_calls')->insert($insert); + }); + $before = $this->records(); + fwrite($sockets[0], "committed\n"); + $message = json_decode((string) fgets($sockets[0]), true, 512, JSON_THROW_ON_ERROR); + $this->assertArrayNotHasKey('error', $message, json_encode($message, JSON_THROW_ON_ERROR)); + $this->assertSame(! $foreign, $message['result']['completed'], json_encode($message, JSON_THROW_ON_ERROR)); + if ($foreign) { + $this->assertSame('service_call_not_owned', $message['result']['reason']); + $this->assertSame($before, $this->records()); + } else { + $this->assertSame(1, WorkflowServiceCall::query()->count()); + $this->assertSame($attributes['id'], WorkflowServiceCall::query()->sole()->id); + $this->assertSame( + ServiceCallIdempotency::digest($scope), + WorkflowServiceCall::query()->sole()->idempotency_scope_hash + ); + } + pcntl_waitpid($pid, $status); + $pid = 0; + $this->assertSame(0, pcntl_wexitstatus($status)); + } finally { + if ($pid > 0) { + posix_kill($pid, SIGKILL); + pcntl_waitpid($pid, $status); + } + fclose($sockets[0]); + DB::purge(); + DB::reconnect(); + } + } + + /** + * @return iterable + */ + public static function concurrentInsertions(): iterable + { + yield 'legacy foreign insert after absent preflight' => ['lookup', true]; + yield 'legacy same-owner insert after absent preflight' => ['lookup', false]; + yield 'canonical foreign insert wins creation' => ['insert', true]; + yield 'canonical same-owner insert wins creation' => ['insert', false]; + } + + #[DataProvider('canonicalScopes')] + public function testCanonicalKeyAdoptsLegacyRowsAndSurvivesMigrationRetry(?string $namespace, string $key): void + { + $call = $this->serviceCall(ServiceCallStatus::Completed->value); + $call->forceFill([ + 'namespace' => $namespace, + 'target_namespace' => $namespace, + 'idempotency_key' => $key, + ])->save(); + $scope = [$namespace, 'billing', 'invoices', 'create', $key]; + $this->assertNull($call->idempotency_scope_hash); + $this->assertSame($call->id, ServiceCallIdempotency::find($scope)->id); + DB::transaction(static fn () => ServiceCallIdempotency::save($call, $scope)); + $before = $call->fresh() + ->getRawOriginal(); + $migration = require __DIR__ . '/../../../src/migrations/2026_10_25_000100_add_service_call_idempotency_scope.php'; + $migration->up(); + $migration->down(); + $migration->up(); + $this->assertSame($before, $call->fresh()->getRawOriginal()); + $this->assertSame($call->id, ServiceCallIdempotency::find($scope)->id); + $this->assertSame(1, WorkflowServiceCall::query()->count()); + $other = $scope; + $other[0] = $namespace === null ? '' : null; + $this->assertNotSame(ServiceCallIdempotency::digest($scope), ServiceCallIdempotency::digest($other)); + $this->assertNull(ServiceCallIdempotency::find($other)); + } + + /** + * @return iterable + */ + public static function canonicalScopes(): iterable + { + yield 'null namespace' => [null, 'same-key']; + yield 'empty namespace' => ['', 'same-key']; + yield 'maximum ASCII key' => ['caller', str_repeat('a', 191)]; + yield 'maximum Unicode key' => ['caller', str_repeat('界', 191)]; + } + + public function testSqliteUniqueInsertLoserRefetchesTheCanonicalCallInAUsableTransaction(): void + { + if (DB::connection()->getDriverName() !== 'sqlite') { + $this->markTestSkipped('SQLite has database-wide writer admission; row-lock races are covered separately.'); + } + if (! function_exists('pcntl_fork') || ! function_exists('posix_kill')) { + $this->markTestSkipped('Process control is required.'); + } + $call = $this->serviceCall(ServiceCallStatus::Completed->value); + $scope = ['caller', 'billing', 'invoices', 'create', $call->idempotency_key]; + $candidate = $call->replicate(); + $candidate->id = (string) Str::ulid(); + $call->delete(); + DB::purge(); + $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); + $this->assertIsArray($sockets); + $pid = pcntl_fork(); + $this->assertNotSame(-1, $pid); + if ($pid === 0) { + fclose($sockets[0]); + stream_set_timeout($sockets[1], 15); + try { + DB::reconnect(); + WorkflowServiceCall::creating(static function () use ($sockets): void { + fwrite($sockets[1], "insert-ready\n"); + if (fgets($sockets[1]) !== "committed\n") { + throw new RuntimeException('Canonical writer was not committed.'); + } + }); + $winner = DB::transaction(static fn () => ServiceCallIdempotency::save($candidate, $scope)); + fwrite($sockets[1], json_encode([ + 'id' => $winner->id, + 'level' => DB::transactionLevel(), + ], JSON_THROW_ON_ERROR) . "\n"); + DB::disconnect(); + exit(0); + } catch (Throwable $error) { + fwrite($sockets[1], json_encode([ + 'error' => $error::class . ': ' . $error->getMessage(), + ], JSON_THROW_ON_ERROR) . "\n"); + exit(1); + } + } + fclose($sockets[1]); + stream_set_timeout($sockets[0], 20); + try { + $this->assertSame("insert-ready\n", fgets($sockets[0])); + DB::reconnect(); + $call->exists = false; + DB::transaction(static fn () => ServiceCallIdempotency::save($call, $scope)); + $before = $this->records(); + fwrite($sockets[0], "committed\n"); + $message = json_decode((string) fgets($sockets[0]), true, 512, JSON_THROW_ON_ERROR); + $this->assertSame([ + 'id' => $call->id, + 'level' => 0, + ], $message); + $this->assertSame($before, $this->records()); + pcntl_waitpid($pid, $status); + $pid = 0; + $this->assertSame(0, pcntl_wexitstatus($status)); + } finally { + if ($pid > 0) { + posix_kill($pid, SIGKILL); + pcntl_waitpid($pid, $status); + } + fclose($sockets[0]); + DB::purge(); + DB::reconnect(); + } + // A canonical row still needs caller authorization after locked refetch. + $call->caller_workflow_run_id = 'foreign-run'; + $call->save(); + $before = $this->records(); + try { + DB::transaction(function () use ($candidate, $scope): void { + $winner = ServiceCallIdempotency::save($candidate, $scope); + WorkflowServiceOperationAuthority::authorizeExistingCall( + $winner, + new ServiceCallPrincipal('workflow:' . $this->run->id, 'workflow'), + [ + 'namespace' => 'caller', + 'caller_namespace' => 'caller', + 'caller_workflow_run_id' => $this->run->id, + 'caller_workflow_instance_id' => $this->run->workflow_instance_id, + 'endpoint_name' => 'billing', + 'service_name' => 'invoices', + 'operation_name' => 'create', + ] + ); + }); + $this->fail('Foreign canonical winner must be refused.'); + } catch (ServiceOperationAuthorizationException $error) { + $this->assertSame('service_call_not_owned', $error->getMessage()); + } + $this->assertSame($before, $this->records()); + } + + public function testSeparateCallConnectionRefusesBeforeSiblingWrites(): void + { + config([ + 'database.connections.authority_other' => config('database.connections.' . DB::getDefaultConnection()), + 'workflows.v2.service_call_model' => SeparateAuthorityServiceCall::class, + ]); + $before = $this->records(); + $result = $this->bridge->complete($this->task->id, [ + [ + 'type' => 'start_timer', + 'delay_seconds' => 60, + ], + $this->command(), + ]); + $this->assertFalse($result['completed']); + $this->assertSame('service_operation_storage_connection_mismatch', $result['reason']); + $this->assertSame($before, $this->records()); + } + + private function replaceOperation(WorkflowServiceCall $call): void + { + $replacement = $call->operation->replicate(); + $call->operation->delete(); + $replacement->save(); + $this->assertNotSame($call->workflow_service_operation_id, $replacement->id); + } + + /** + * @return array + */ + private function command(): array + { + return [ + 'type' => 'start_service_operation', + 'endpoint_name' => 'billing', + 'service_name' => 'invoices', + 'operation_name' => 'create', + ]; + } + + private function serviceCall(string $status): WorkflowServiceCall + { + $endpoint = WorkflowServiceEndpoint::query()->create([ + 'namespace' => 'caller', + 'endpoint_name' => 'billing', + ]); + $service = WorkflowService::query()->create([ + 'namespace' => 'caller', + 'workflow_service_endpoint_id' => $endpoint->id, + 'service_name' => 'invoices', + ]); + $operation = WorkflowServiceOperation::query()->create([ + 'namespace' => 'caller', + 'workflow_service_endpoint_id' => $endpoint->id, + 'workflow_service_id' => $service->id, + 'operation_name' => 'create', + 'operation_mode' => 'async', + 'handler_binding_kind' => ServiceCallBindingKind::WorkflowRun->value, + 'handler_target_reference' => 'service-authority-workflow', + ]); + return WorkflowServiceCall::query()->create([ + 'namespace' => 'caller', + 'target_namespace' => 'caller', + 'caller_namespace' => 'caller', + 'caller_workflow_instance_id' => $this->run->workflow_instance_id, + 'caller_workflow_run_id' => $this->run->id, + 'workflow_service_endpoint_id' => $endpoint->id, + 'workflow_service_id' => $service->id, + 'workflow_service_operation_id' => $operation->id, + 'endpoint_name' => 'billing', + 'service_name' => 'invoices', + 'operation_name' => 'create', + 'operation_mode' => 'async', + 'status' => $status, + 'outcome' => $status === 'completed' ? ServiceCallOutcome::Completed->value + : ServiceCallOutcome::Accepted->value, + 'resolved_binding_kind' => ServiceCallBindingKind::WorkflowRun->value, + 'idempotency_key' => 'shared-idempotency-key', + 'accepted_at' => $status === ServiceCallStatus::Pending->value ? null : now(), + 'started_at' => $status === ServiceCallStatus::Started->value ? now() : null, + 'completed_at' => $status === ServiceCallStatus::Completed->value ? now() : null, + ]); + } + + /** + * @return array>> + */ + private function records(): array + { + $records = []; + foreach ([WorkflowInstance::class, WorkflowRun::class, WorkflowTask::class, WorkflowCommand::class, + WorkflowHistoryEvent::class, WorkflowRunSummary::class, WorkflowServiceCall::class, + WorkflowServiceEndpoint::class, WorkflowService::class, WorkflowServiceOperation::class, + WorkflowTimer::class] as $model) { + $records[$model] = $model::query()->orderBy('id')->get()->map( + static fn (Model $row): array => $row->getRawOriginal(), + )->all(); + } + return $records; + } +} + +#[Type('service-authority-workflow')] +final class ServiceAuthorityWorkflow extends Workflow +{ + public function handle(): string + { + return 'done'; + } +} + +final class SeparateAuthorityServiceCall extends WorkflowServiceCall +{ + public function getConnectionName(): ?string + { + return 'authority_other'; + } +}