diff --git a/docs/portal/build/failures-retries.md b/docs/portal/build/failures-retries.md index 1e90ea22..e6801a7e 100644 --- a/docs/portal/build/failures-retries.md +++ b/docs/portal/build/failures-retries.md @@ -18,6 +18,7 @@ next: | `ServerException` | The runtime rejected or failed an operation. | Inspect status, reason, and details. | | `ActivityFailed` | A recorded activity reached a failed terminal attempt. | Catch in workflow code only when compensation is deterministic. | | `NonDeterministicWorkflow` | Code no longer matches committed history. | Restore compatible code; do not blindly retry. | +| `WorkflowEntryArgumentsMismatch` | Recorded start input cannot bind to the entry method, before its body runs. The workflow task is refused and the run remains recoverable. | Restore a compatible entry method, inspect the task and use run-scoped repair. | | `WorkflowFailed` | The execution closed as failed. | Surface the recorded failure to the caller. | | `WorkflowTimedOut` | A Server deadline closed the run. | Start a new workflow only if business policy permits. | | `WorkflowCancelled` | The run closed as Cancelled, after terminal cancellation or successful cooperative cleanup. | Inspect the cancellation lifecycle and cleanup outcome. | diff --git a/docs/portal/operate/deployment.md b/docs/portal/operate/deployment.md index c56da46b..de91228e 100644 --- a/docs/portal/operate/deployment.md +++ b/docs/portal/operate/deployment.md @@ -50,6 +50,21 @@ Do not use the SDK repository's `main` branch as a production Composer source. I Start new workers before draining the old pool. Keep old and new builds registered only when both can replay the histories routed to that queue. If a change is not replay-compatible, use a new task queue or a worker build-routing plan rather than hoping restarts choose the right code. +## Preserve workflow entry arguments + +Recorded start input must still bind to the deployed workflow entry method. +Adding a required parameter or changing its type can prevent the method from +starting, before a version branch inside it can run. The worker reports +`WorkflowEntryArgumentsMismatch` as a nonterminal workflow-task replay refusal, +with reason `workflow_entry_arguments_mismatch`. It leaves the original run, +input, history and deadlines intact. + +Restore a compatible entry method and use the runtime's run-scoped repair +operation to resume the same run. Inspect the refused task before repairing it; +repeated incompatible deployments do not fix its input. A compatible optional +parameter is allowed. A `TypeError` thrown after valid entry binding remains an +application failure, rather than being relabelled as a deployment mismatch. + ## Readiness and shutdown Runtime readiness, worker admission, and process liveness are different signals: diff --git a/regression-corpus-policy.json b/regression-corpus-policy.json index fed65cce..8f64a3ef 100644 --- a/regression-corpus-policy.json +++ b/regression-corpus-policy.json @@ -38,6 +38,12 @@ { "glob": "src/Worker/Replayer.php" }, + { + "glob": "src/Worker/WorkflowEntryArguments.php" + }, + { + "glob": "src/Exception/WorkflowEntryArgumentsMismatch.php" + }, { "glob": "src/Worker/WorkflowCommand.php" }, diff --git a/scripts/ci/run-replay-regression-fixture.php b/scripts/ci/run-replay-regression-fixture.php index f7ab8460..b6c41d59 100644 --- a/scripts/ci/run-replay-regression-fixture.php +++ b/scripts/ci/run-replay-regression-fixture.php @@ -12,6 +12,7 @@ use DurableWorkflow\Exception\NonDeterministicWorkflow; use DurableWorkflow\Exception\TransportException; use DurableWorkflow\Exception\WorkflowCancelled; +use DurableWorkflow\Exception\WorkflowEntryArgumentsMismatch; use DurableWorkflow\Model\WorkflowStreamAppendItem; use DurableWorkflow\Transport\Transport; use DurableWorkflow\Worker; @@ -209,6 +210,17 @@ public function send(string $method, string $uri, array $headers, ?array $body = return ['failed' => true]; } + if (is_array($failure) && ($failure['type'] ?? null) === WorkflowEntryArgumentsMismatch::class + && ($failure['reason'] ?? null) === 'workflow_entry_arguments_mismatch' + && !array_key_exists('sequence', $failure)) { + $this->completedCommands[] = [[ + 'type' => 'replay_error', + 'reason' => 'workflow_entry_arguments_mismatch', + 'sequence' => null, + ]]; + + return ['failed' => true]; + } if (!is_array($failure) || ($failure['type'] ?? null) !== NonDeterministicWorkflow::class || !is_string($failure['reason'] ?? null) @@ -605,6 +617,7 @@ private static function assertExpected(string $identity, array $fixture, array $ private static function workflow(string $workflowType): callable { return match ($workflowType) { + 'golden.entry-arguments-required' => static fn (WorkflowContext $context, int $amount, int $required): int => $amount + $required, 'golden.finally-disposal' => static function (WorkflowContext $context): void { try { $context->sleep(1); diff --git a/src/Exception/WorkflowEntryArgumentsMismatch.php b/src/Exception/WorkflowEntryArgumentsMismatch.php new file mode 100644 index 00000000..33c6295b --- /dev/null +++ b/src/Exception/WorkflowEntryArgumentsMismatch.php @@ -0,0 +1,23 @@ +reason = 'workflow_entry_arguments_mismatch'; + parent::__construct('replay failed: '.$this->reason.' for '.$workflowType + .' ('.$bindingReason.($parameter === null ? '' : ', parameter '.$parameter).').'); + } +} diff --git a/src/Worker.php b/src/Worker.php index 70c90a94..b882828c 100644 --- a/src/Worker.php +++ b/src/Worker.php @@ -10,6 +10,7 @@ use DurableWorkflow\Exception\InvalidWorkerDefinition; use DurableWorkflow\Exception\LocalActivityTimedOut; use DurableWorkflow\Exception\NonDeterministicWorkflow; +use DurableWorkflow\Exception\WorkflowEntryArgumentsMismatch; use DurableWorkflow\Exception\SagaCompensationFailed; use DurableWorkflow\Exception\ServerException; use DurableWorkflow\Exception\TransportException; @@ -40,6 +41,7 @@ use DurableWorkflow\Worker\WorkflowClaimDeferred; use DurableWorkflow\Worker\WorkflowClaimRevoked; use DurableWorkflow\Worker\WorkflowContext; +use DurableWorkflow\Worker\WorkflowEntryArguments; use DurableWorkflow\Worker\StickyWorkflowCache; use DurableWorkflow\Worker\WorkerSession; use DurableWorkflow\Worker\WorkerSessionOptions; @@ -1294,6 +1296,9 @@ private function executeWorkflowTask(array $task): void throw new \RuntimeException("No workflow handler is registered for {$workflowType}."); } $input = $this->decodeArguments($task['arguments'] ?? $task['input'] ?? null); + // Keep pre-entry refusal outside the application-error boundary. + // A valid-bound body throwing the same exception is still a body failure. + WorkflowEntryArguments::validate($handler, $input, $workflowType); try { $replay = $this->replayWorkflowClaim( $handler, @@ -1374,7 +1379,7 @@ function (Throwable $failure) use ($taskId, $leaseOwner, $attempt): void { $attempt, 'PHP workflow task execution failed: '.$failure->getMessage(), $failure::class, - $failure instanceof NonDeterministicWorkflow ? $failure->reason : null, + $failure instanceof NonDeterministicWorkflow || $failure instanceof WorkflowEntryArgumentsMismatch ? $failure->reason : null, $failure instanceof NonDeterministicWorkflow ? $failure->sequence : null, ); }, diff --git a/src/Worker/Replayer.php b/src/Worker/Replayer.php index e271890a..7dc709e4 100644 --- a/src/Worker/Replayer.php +++ b/src/Worker/Replayer.php @@ -65,6 +65,7 @@ public function replay( if ($allowScopedPreparedLocalActivities && (!$prepareLocalActivities || !$prepareCancellationScopeDelivery)) { throw new LogicException('Scoped local callbacks require prepared admission and canonical scope delivery.'); } + WorkflowEntryArguments::validate($handler, $input, (string) ($task['workflow_type'] ?? '')); $hasScopes = $this->assertCancellationScopeReplaySupported($history, $allowCancellationScopeAuthoring, $task, $replayCommittedCancellationScopes); $scopes = new CancellationScopeHistory($hasScopes ? $history : [], (string) ($task['run_id'] ?? '')); // Unfrozen source profile, enabled only by the Worker's explicit scope opt-in. diff --git a/src/Worker/WorkflowEntryArguments.php b/src/Worker/WorkflowEntryArguments.php new file mode 100644 index 00000000..7d8f0956 --- /dev/null +++ b/src/Worker/WorkflowEntryArguments.php @@ -0,0 +1,127 @@ + $input */ + public static function validate(callable $handler, array $input, string $workflowType): void + { + $contract = $handler instanceof HandlerDefinition ? $handler->contract() : Closure::fromCallable($handler); + $entry = new ReflectionFunction($contract); + $parameters = $entry->getParameters(); + if (($parameters[0] ?? null)?->isVariadic()) { + foreach ($input as $value) { + self::validateType($parameters[0], $value, $entry, $workflowType); + } + + return; + } + // The first parameter is the SDK context, validated at registration. + foreach (array_slice($parameters, 1) as $index => $parameter) { + if ($parameter->isVariadic()) { + foreach (array_slice($input, $index) as $value) { + self::validateType($parameter, $value, $entry, $workflowType); + } + + return; + } + if (!array_key_exists($index, $input)) { + if (!$parameter->isOptional()) { + throw new WorkflowEntryArgumentsMismatch($workflowType, 'required_argument_missing', $parameter->getName()); + } + continue; + } + self::validateType($parameter, $input[$index], $entry, $workflowType); + } + // PHP permits extra trailing positional arguments. Preserve that contract. + } + + private static function validateType(ReflectionParameter $parameter, mixed $value, ReflectionFunction $entry, string $workflowType): void + { + $type = $parameter->getType(); + if ($type !== null && !self::matches($type, $value, $entry->getClosureScopeClass())) { + throw new WorkflowEntryArgumentsMismatch($workflowType, 'argument_type_mismatch', $parameter->getName(), (string) $type, get_debug_type($value)); + } + } + + /** @param ReflectionClass|null $scope */ + private static function matches(ReflectionType $type, mixed $value, ?ReflectionClass $scope): bool + { + return match (true) { + $value === null && $type->allowsNull() => true, + $type instanceof ReflectionUnionType => self::matchesUnion($type, $value, $scope), + $type instanceof ReflectionIntersectionType => self::matchesIntersection($type, $value, $scope), + $type instanceof ReflectionNamedType => self::matchesNamed($type, $value, $scope), + default => false, + }; + } + + /** @param ReflectionClass|null $scope */ + private static function matchesUnion(ReflectionUnionType $type, mixed $value, ?ReflectionClass $scope): bool + { + foreach ($type->getTypes() as $member) { + if (self::matches($member, $value, $scope)) { + return true; + } + } + + return false; + } + + /** @param ReflectionClass|null $scope */ + private static function matchesIntersection(ReflectionIntersectionType $type, mixed $value, ?ReflectionClass $scope): bool + { + foreach ($type->getTypes() as $member) { + if (!self::matches($member, $value, $scope)) { + return false; + } + } + + return true; + } + + /** @param ReflectionClass|null $scope */ + private static function matchesNamed(ReflectionNamedType $type, mixed $value, ?ReflectionClass $scope): bool + { + if (!$type->isBuiltin()) { + $parent = $scope?->getParentClass(); + $name = match ($type->getName()) { + 'self', 'static' => $scope?->getName(), + 'parent' => $parent === false ? null : $parent?->getName(), + default => $type->getName(), + }; + + return is_string($name) && $value instanceof $name; + } + + return match ($type->getName()) { + 'mixed' => true, + 'int' => is_int($value), + 'float' => is_float($value) || is_int($value), + 'string' => is_string($value), + 'bool' => is_bool($value), + 'true' => $value === true, + 'false' => $value === false, + 'null' => $value === null, + 'array' => is_array($value), + 'object' => is_object($value), + 'callable' => is_callable($value), + 'iterable' => is_iterable($value), + default => false, + }; + } +} diff --git a/tests/Support/ReplayRegressionFixture.php b/tests/Support/ReplayRegressionFixture.php index 765a3608..9589e9fc 100644 --- a/tests/Support/ReplayRegressionFixture.php +++ b/tests/Support/ReplayRegressionFixture.php @@ -12,6 +12,7 @@ use DurableWorkflow\Exception\ChildWorkflowFailed; use DurableWorkflow\Exception\NonDeterministicWorkflow; use DurableWorkflow\Exception\WorkflowCancelled; +use DurableWorkflow\Exception\WorkflowEntryArgumentsMismatch; use DurableWorkflow\Model\WorkflowStreamAppendItem; use DurableWorkflow\Testing\WorkerTestHarness; use DurableWorkflow\Worker; @@ -175,12 +176,12 @@ public static function execute(array $fixture): array 'message' => $exception->getMessage(), 'exception_type' => $exception::class, ]]; - } catch (NonDeterministicWorkflow $exception) { + } catch (NonDeterministicWorkflow|WorkflowEntryArgumentsMismatch $exception) { $commands = [[ 'type' => 'replay_error', 'message' => $exception->getMessage(), 'reason' => $exception->reason, - 'sequence' => $exception->sequence, + 'sequence' => $exception instanceof NonDeterministicWorkflow ? $exception->sequence : null, ]]; } catch (\Throwable $exception) { $commands = [[ @@ -322,6 +323,7 @@ private static function executeWorkflowTasks( private static function workflow(string $workflowType): callable { return match ($workflowType) { + 'golden.entry-arguments-required' => static fn (WorkflowContext $context, int $amount, int $required): int => $amount + $required, 'golden.nexus-caller' => static function (WorkflowContext $context): mixed { return $context->serviceOperation('billing', 'invoices', 'create', ['invoice_id' => 42], ['idempotency_key' => 'call-key'])->responsePayload; diff --git a/tests/WorkflowEntryBindingTest.php b/tests/WorkflowEntryBindingTest.php new file mode 100644 index 00000000..dec39124 --- /dev/null +++ b/tests/WorkflowEntryBindingTest.php @@ -0,0 +1,134 @@ + $input */ + #[DataProvider('incompatibleEntries')] + public function testInputBindingRefusesBeforeEnteringApplication(callable $handler, array $input): void + { + try { + (new Replayer(new AvroPayloadCodec()))->replay($handler, [], $input, 'workers', ['workflow_type' => 'entry.test']); + self::fail('Incompatible entry executed.'); + } catch (\Throwable $failure) { + self::assertSame('workflow_entry_arguments_mismatch', $failure->reason ?? null); + self::assertStringContainsString('entry.test', $failure->getMessage()); + self::assertStringNotContainsString('private-recorded-value', $failure->getMessage()); + } + } + + /** @return iterable}> */ + public static function incompatibleEntries(): iterable + { + yield 'required input missing' => [static fn (WorkflowContext $ctx, int $amount, int $required): int => $required, [7]]; + yield 'strict scalar changed' => [static fn (WorkflowContext $ctx, string $amount): string => $amount, [7]]; + yield 'scalar to DTO' => [static fn (WorkflowContext $ctx, \stdClass $amount): object => $amount, [7]]; + yield 'nullable is not optional' => [static fn (WorkflowContext $ctx, ?int $amount): ?int => $amount, []]; + yield 'union mismatch' => [static fn (WorkflowContext $ctx, int|bool $amount): mixed => $amount, ['private-recorded-value']]; + yield 'intersection mismatch' => [static fn (WorkflowContext $ctx, \Countable&\Iterator $amount): object => $amount, [new \stdClass()]]; + yield 'variadic element mismatch' => [static fn (WorkflowContext $ctx, int ...$amounts): array => $amounts, [7, 'private-recorded-value']]; + yield 'variadic context also receives recorded values' => [static fn (WorkflowContext ...$contexts): int => count($contexts), [7]]; + yield 'discovered contract is not variadic wrapper' => [HandlerDefinition::shared(static fn (WorkflowContext $ctx, int $amount, int $required): int => $required), [7]]; + } + + /** @param list $input */ + #[DataProvider('compatibleEntries')] + public function testCompatiblePhpInvocationKeepsRecordedTypes(callable $handler, array $input, mixed $expected): void + { + $codec = new AvroPayloadCodec(); + $result = (new Replayer($codec))->replay($handler, [], $input, 'workers'); + self::assertSame('complete_workflow', $result->commands[0]['type']); + self::assertSame($expected, $codec->decodeEnvelope($result->commands[0]['result'])); + } + + /** @return iterable, mixed}> */ + public static function compatibleEntries(): iterable + { + yield 'optional addition' => [static fn (WorkflowContext $ctx, int $amount, ?int $extra = null): int => $amount, [7], 7]; + yield 'nullable value' => [static fn (WorkflowContext $ctx, ?int $amount): ?int => $amount, [null], null]; + yield 'int accepted for float' => [static fn (WorkflowContext $ctx, float $amount): float => $amount, [7], 7.0]; + yield 'union member' => [static fn (WorkflowContext $ctx, int|string $amount): mixed => $amount, [7], 7]; + yield 'variadic values' => [static fn (WorkflowContext $ctx, int ...$amounts): array => $amounts, [7, 8], [7, 8]]; + yield 'trailing argument removed' => [static fn (WorkflowContext $ctx): int => 7, [7], 7]; + } + + public function testWorkerReportsBindingAsTaskFailureWithOriginalClaim(): void + { + $requests = $this->executeWorker(static fn (WorkflowContext $ctx, int $amount, int $required): int => $required); + $failures = array_values(array_filter($requests, static fn (array $row): bool => str_ends_with($row['uri'], '/task-original/fail'))); + self::assertCount(1, $failures); + self::assertSame('owner-original', $failures[0]['body']['lease_owner']); + self::assertSame(3, $failures[0]['body']['workflow_task_attempt']); + self::assertSame('workflow_entry_arguments_mismatch', $failures[0]['body']['failure']['reason']); + self::assertSame([], array_filter($requests, static fn (array $row): bool => str_ends_with($row['uri'], '/task-original/complete'))); + } + + public function testValidBoundBodyTypeErrorIsStillBusinessFailure(): void + { + $requests = $this->executeWorker(static function (WorkflowContext $ctx, int $amount): never { + throw new \TypeError('application-body-failure'); + }); + $complete = array_values(array_filter($requests, static fn (array $row): bool => str_ends_with($row['uri'], '/task-original/complete'))); + self::assertCount(1, $complete); + self::assertSame('fail_workflow', $complete[0]['body']['commands'][0]['type']); + self::assertSame([], array_filter($requests, static fn (array $row): bool => str_ends_with($row['uri'], '/task-original/fail'))); + } + + public function testValidBoundBodyThrowingInternalExceptionIsStillBusinessFailure(): void + { + $requests = $this->executeWorker(static function (WorkflowContext $ctx, int $amount): never { + throw new WorkflowEntryArgumentsMismatch('application', 'body-controlled-error'); + }); + $complete = array_values(array_filter($requests, static fn (array $row): bool => str_ends_with($row['uri'], '/task-original/complete'))); + self::assertCount(1, $complete); + self::assertSame('fail_workflow', $complete[0]['body']['commands'][0]['type']); + self::assertSame([], array_filter($requests, static fn (array $row): bool => str_ends_with($row['uri'], '/task-original/fail'))); + } + + /** @return list, body: array|null}> */ + private function executeWorker(callable $handler): array + { + $codec = new AvroPayloadCodec(); + $transport = new FakeTransport(handler: static function (string $method, string $uri) use ($codec): array { + if (str_ends_with($uri, '/workflow-tasks/poll')) { + return ['poll_status' => 'leased', 'task' => [ + 'task_id' => 'task-original', 'workflow_task_attempt' => 3, 'lease_owner' => 'owner-original', + 'workflow_id' => 'workflow-original', 'run_id' => 'run-original', 'workflow_type' => 'entry.test', + 'payload_codec' => 'avro', 'arguments' => $codec->envelope([7]), 'history_events' => [], + ]]; + } + if (str_ends_with($uri, '/activity-tasks/poll')) { + return ['task' => null, 'poll_status' => 'stopped', 'reason' => 'worker_stopped']; + } + if (str_ends_with($uri, '/heartbeat')) { + return ['task_id' => 'task-original', 'workflow_task_attempt' => 3, 'lease_owner' => 'owner-original', 'renewed' => true]; + } + if (str_ends_with($uri, '/complete')) { + return ['completed' => true]; + } + if (str_ends_with($uri, '/fail')) { + return ['failed' => true, 'recorded' => true]; + } + self::fail('Unexpected worker request '.$method.' '.$uri); + }); + $worker = new Worker(new Client('https://server.example', transport: $transport), 'workers', workerId: 'owner-original'); + $worker->registerWorkflow('entry.test', $handler); + self::assertTrue($worker->tick(0)); + + return $transport->requests; + } +} diff --git a/tests/fixtures/replay-regressions/workflow-entry-arguments-required-after-wait.json b/tests/fixtures/replay-regressions/workflow-entry-arguments-required-after-wait.json new file mode 100644 index 00000000..7881195c --- /dev/null +++ b/tests/fixtures/replay-regressions/workflow-entry-arguments-required-after-wait.json @@ -0,0 +1,12 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "php-workflow-entry-arguments-required-after-wait", + "protocol_version": "1.18", + "bindings": ["php"], + "workflow": {"type": "golden.entry-arguments-required", "input": [7]}, + "history": [ + {"sequence": 3, "event_type": "ConditionWaitOpened", "payload": {"sequence": 1, "condition_wait_id": "original-condition-wait", "condition_key": "ready"}} + ], + "expected": {"type": "replay_error", "reason": "workflow_entry_arguments_mismatch"} +}