diff --git a/CHANGELOG.md b/CHANGELOG.md index 73cb341..908d1d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,17 @@ project follows [Semantic Versioning](https://semver.org/). ## [Unreleased] +## [2.0.9] - 2026-09-08 + +### Fixed + +- On supporting Servers, retry a draining refusal for a completion payload + with its exact activity, workflow, or query lease and immutable payload slot. + Client uploads, unknown capabilities, and hard storage fences remain blocked. +- Preserve computed outcomes during late upload pressure instead of executing + the handler again. The Server's bounded allowance and namespace quotas still + apply; stale leases remain errors. + ## [2.0.7] - 2026-09-08 ### Fixed diff --git a/composer.json b/composer.json index c1b9000..3788d95 100644 --- a/composer.json +++ b/composer.json @@ -74,7 +74,7 @@ } }, "durable-workflow": { - "product-train": "2.0.8", + "product-train": "2.0.9", "supported-server-versions": "2.1.0", "worker-protocol-version": "1.19", "control-plane-version": "2", diff --git a/docs/quickstart-contract.json b/docs/quickstart-contract.json index 414414b..86ca7ee 100644 --- a/docs/quickstart-contract.json +++ b/docs/quickstart-contract.json @@ -3,8 +3,8 @@ "schema_version": 2, "package": { "name": "durable-workflow/sdk", - "published_version": "2.0.8", - "composer_requirement": "2.0.8", + "published_version": "2.0.9", + "composer_requirement": "2.0.9", "onboarding_requirement": "^2.0" }, "runtime_targets": { diff --git a/src/Exception/ServerException.php b/src/Exception/ServerException.php index aaf1fca..dc298f7 100644 --- a/src/Exception/ServerException.php +++ b/src/Exception/ServerException.php @@ -47,7 +47,10 @@ public function isStorageAdmissionFailure(?string $pollRequestId = null): bool } if ($pollRequestId === null) { - return ($response['request_admitted'] ?? null) === false; + // Payload uploads are content-addressed and precede submission of + // the completion. A late upload refusal can safely reuse its bytes. + return ($response['request_admitted'] ?? null) === false + || ($this instanceof ExternalPayloadException && !array_key_exists('request_admitted', $response)); } // A refused claim does not resolve a prior request's uncertain outcome. diff --git a/src/Transport/RuntimePayloadUploads.php b/src/Transport/RuntimePayloadUploads.php index 5f1e1cf..dfffa14 100644 --- a/src/Transport/RuntimePayloadUploads.php +++ b/src/Transport/RuntimePayloadUploads.php @@ -11,6 +11,8 @@ /** Namespace-scoped discovery and request-local, content-addressed uploads. */ final class RuntimePayloadUploads { + private const COMPLETION_SCHEMA = 'durable-workflow.v2.payload-completion-context.v1'; + private const COMPLETION_HEADER = 'X-Durable-Workflow-Payload-Completion'; /** @var array> */ public const COMMAND_FIELDS = [ 'complete_workflow' => ['result'], @@ -95,13 +97,28 @@ public function request(array $body, string $method, string $path, bool $worker, $key = $expected['sha256'].':'.$expected['size_bytes']; if (!isset($uploaded[$key])) { try { - $response = $this->transport->uploadPayload($this->baseUri.'/api/external-payloads/v1', array_replace($headers, [ + $uploadHeaders = array_replace($headers, [ 'Content-Type' => 'application/octet-stream', 'Accept' => 'application/json', 'X-Durable-Workflow-Payload-Codec' => 'avro', 'X-Durable-Workflow-Payload-Size' => (string) strlen($blob), 'X-Durable-Workflow-Payload-SHA256' => $expected['sha256'], - ]), $blob, $policy['timeout_seconds']); + ]); + try { + $response = $this->transport->uploadPayload($this->baseUri.'/api/external-payloads/v1', $uploadHeaders, $blob, $policy['timeout_seconds']); + } catch (TransportException $refusal) { + $context = $worker && ($policy['completion_context'] ?? false) + ? $this->completionContext($body, $path, $payloads[$index]['path']) : null; + if ($context === null || $refusal->status !== 503 + || ($refusal->response['reason'] ?? null) !== 'storage_pressure' + || ($refusal->response['storage_state'] ?? null) !== 'draining') { + throw $refusal; + } + // One capability-negotiated retry; never turn other pressure + // errors into retries or repeat application activity code. + $uploadHeaders[self::COMPLETION_HEADER] = $context; + $response = $this->transport->uploadPayload($this->baseUri.'/api/external-payloads/v1', $uploadHeaders, $blob, $policy['timeout_seconds']); + } } catch (TransportException $exception) { $reason = $exception->response['reason'] ?? null; $reason = is_string($reason) ? $reason : match ($exception->status) { @@ -130,7 +147,7 @@ public function request(array $body, string $method, string $path, bool $worker, } /** @param array $headers - * @return array{threshold_bytes: int, max_bytes: int, request_bytes: int, timeout_seconds: int, status: string} + * @return array{threshold_bytes: int, max_bytes: int, request_bytes: int, timeout_seconds: int, status: string, completion_context?: bool} */ private function policy(array $headers, bool $worker): array { @@ -165,7 +182,9 @@ private function policy(array $headers, bool $worker): array throw new ExternalPayloadException('Unsupported namespace runtime payload transport discovery.', 422, 'external_payload_unsupported'); } $policy = ['threshold_bytes' => $threshold, 'max_bytes' => $max, 'request_bytes' => $requestLimit, - 'timeout_seconds' => $timeout, 'status' => is_string($storage['status'] ?? null) ? $storage['status'] : 'unavailable']; + 'timeout_seconds' => $timeout, 'status' => is_string($storage['status'] ?? null) ? $storage['status'] : 'unavailable', + 'completion_context' => ($manifest['upload']['completion_context']['schema'] ?? null) === self::COMPLETION_SCHEMA + && ($manifest['upload']['completion_context']['header'] ?? null) === self::COMPLETION_HEADER]; } $this->policies[$key] = ['expires' => time() + 60, 'policy' => $policy]; @@ -218,6 +237,28 @@ private function paths(array $body, string $path, bool $worker): array return $paths; } + /** @param array $body + * @param list $slot + */ + private function completionContext(array $body, string $path, array $slot): ?string + { + if (!preg_match('~\A/worker/(activity|workflow|query)-tasks/([^/]+)/(complete|fail)\z~', explode('?', $path)[0], $match)) { + return null; + } + $kind = $match[1]; + $attempt = $body[$kind === 'activity' ? 'activity_attempt_id' : $kind.'_task_attempt'] ?? null; + $owner = $body['lease_owner'] ?? null; + if (!is_string($owner) || $owner === '' || ($kind === 'activity' + ? !is_string($attempt) || $attempt === '' : !is_int($attempt) || $attempt < 1)) { + return null; + } + $context = json_encode(['schema' => self::COMPLETION_SCHEMA, 'kind' => $kind, + 'task_id' => rawurldecode($match[2]), 'attempt' => $attempt, 'lease_owner' => $owner, + 'operation' => $match[3], 'slot' => $slot], JSON_THROW_ON_ERROR); + + return strlen($context) <= 4096 ? $context : null; + } + /** @param array $body * @param array{path: list, blob: string} $payload * @return array{schema: string, reference_id: string, codec: string, size_bytes: int, sha256: string} diff --git a/tests/DependencyBoundaryTest.php b/tests/DependencyBoundaryTest.php index b2d0907..9422c6b 100644 --- a/tests/DependencyBoundaryTest.php +++ b/tests/DependencyBoundaryTest.php @@ -44,7 +44,7 @@ public function testStableMetadataDeclaresExactQualifiedArtifacts(): void $metadata = $this->manifest()['extra']['durable-workflow']; $quickstart = $this->quickstartContract(); - self::assertSame('2.0.8', $metadata['product-train']); + self::assertSame('2.0.9', $metadata['product-train']); self::assertSame('2.1.0', $metadata['supported-server-versions']); self::assertSame('1.19', $metadata['worker-protocol-version']); self::assertTrue($metadata['durable-selection']); diff --git a/tests/RuntimePayloadUploadTest.php b/tests/RuntimePayloadUploadTest.php index 38c7dee..5f7681f 100644 --- a/tests/RuntimePayloadUploadTest.php +++ b/tests/RuntimePayloadUploadTest.php @@ -261,6 +261,123 @@ public function testAdmissionFailurePreservesIdentityAndReasonOnRetry(): void self::assertSame($http->requests[1]->getHeaders(), $http->requests[2]->getHeaders()); } + #[DataProvider('completionProvider')] + public function testDrainingRetryBindsTheExactCompletionSlot(string $path, array $body, string $kind, array $slot): void + { + [, $http, $transport] = $this->client(self::completionDiscovery(), workerOnly: true, drainUnbound: true); + $prepared = (new RuntimePayloadUploads($transport, 'https://runtime.test'))->request($body, 'POST', $path, true, + ['Authorization' => 'Bearer fixture-worker', 'X-Namespace' => 'tenant-one']); + self::assertCount(3, $http->requests); + self::assertSame('', $http->requests[1]->getHeaderLine('X-Durable-Workflow-Payload-Completion')); + $context = json_decode($http->requests[2]->getHeaderLine('X-Durable-Workflow-Payload-Completion'), true, flags: JSON_THROW_ON_ERROR); + self::assertSame(['schema' => 'durable-workflow.v2.payload-completion-context.v1', + 'kind' => $kind, 'task_id' => 'task', 'attempt' => $kind === 'activity' ? 'attempt' : 2, + 'lease_owner' => 'worker', 'operation' => str_ends_with($path, '/fail') ? 'fail' : 'complete', 'slot' => $slot], $context); + self::assertSame((string) $http->requests[1]->getBody(), (string) $http->requests[2]->getBody()); + self::assertSame('Bearer fixture-worker', $http->requests[2]->getHeaderLine('Authorization')); + self::assertSame('tenant-one', $http->requests[2]->getHeaderLine('X-Namespace')); + self::assertStringContainsString(RuntimePayloads::SCHEMA, json_encode($prepared)); + } + + public static function completionProvider(): iterable + { + $envelope = (new AvroPayloadCodec())->envelope(str_repeat('x', 300)); + $activity = ['lease_owner' => 'worker', 'activity_attempt_id' => 'attempt']; + yield 'activity result' => ['/worker/activity-tasks/task/complete', $activity + ['result' => $envelope], 'activity', ['result']]; + yield 'activity failure' => ['/worker/activity-tasks/task/fail', $activity + ['failure' => ['details' => $envelope]], 'activity', ['failure', 'details']]; + yield 'query result' => ['/worker/query-tasks/task/complete', ['lease_owner' => 'worker', 'query_task_attempt' => 2, + 'result_envelope' => $envelope], 'query', ['result_envelope']]; + $workflow = ['lease_owner' => 'worker', 'workflow_task_attempt' => 2]; + foreach (RuntimePayloadUploads::COMMAND_FIELDS as $type => $fields) { + foreach ($fields as $field) { + yield $type.'.'.$field => ['/worker/workflow-tasks/task/complete', + $workflow + ['commands' => [['type' => $type, $field => $envelope]]], 'workflow', ['commands', 0, $field]]; + } + } + yield 'workflow failure' => ['/worker/workflow-tasks/task/complete', + $workflow + ['commands' => [['type' => 'fail_workflow', 'exception' => ['details' => $envelope]]]], + 'workflow', ['commands', 0, 'exception', 'details']]; + yield 'workflow stream' => ['/worker/workflow-tasks/task/complete', + $workflow + ['commands' => [['type' => 'complete_workflow', 'workflow_stream' => ['items' => [['payload' => $envelope]]]]]], + 'workflow', ['commands', 0, 'workflow_stream', 'items', 0, 'payload']]; + } + + public function testCompletionCapabilityDoesNotChangeNormalUploads(): void + { + [$client, $http] = $this->client(self::completionDiscovery()); + $client->completeActivityTask('task', 'attempt', 'worker', str_repeat('x', 200)); + self::assertCount(3, $http->requests); + self::assertSame('', $http->requests[1]->getHeaderLine('X-Durable-Workflow-Payload-Completion')); + } + + public function testLateUploadPressurePreservesTheWorkerResultRetryContract(): void + { + $body = ['reason' => 'storage_pressure', 'storage_state' => 'fenced', 'retryable' => true, 'retry_after_seconds' => 5]; + [$client] = $this->client(self::completionDiscovery(), uploadResponse: new Response(503, [], json_encode($body))); + try { + $client->completeActivityTask('task', 'attempt', 'worker', str_repeat('x', 200)); + self::fail('Expected late upload refusal.'); + } catch (ExternalPayloadException $exception) { + self::assertTrue($exception->isStorageAdmissionFailure()); + self::assertSame($body, $exception->details, 'Do not rewrite the Server response.'); + } + self::assertFalse((new ServerException('Late ordinary mutation', 503, 'storage_pressure', $body))->isStorageAdmissionFailure()); + self::assertFalse((new ExternalPayloadException('Invalid admitted claim', 503, 'storage_pressure', + $body + ['request_admitted' => true]))->isStorageAdmissionFailure()); + } + + #[DataProvider('unsupportedCompletionProvider')] + public function testDrainingDoesNotAuthorizeClientOrUnsupportedWorkerUploads(bool $worker, ?array $capability): void + { + $discovery = self::discovery(); + if ($capability !== null) { + $discovery['namespace']['external_payload_storage']['transport']['upload']['completion_context'] = $capability; + } + [$client, $http] = $this->client($discovery, drainUnbound: true); + try { + if ($worker) { + $client->completeActivityTask('task', 'attempt', 'worker', str_repeat('x', 200)); + } else { + $client->startWorkflow('echo', 'one', 'queue', [str_repeat('x', 200)]); + } + self::fail('Draining refusal must be preserved.'); + } catch (ExternalPayloadException $exception) { + self::assertSame('storage_pressure', $exception->reason); + } + self::assertCount(2, $http->requests); + } + + public static function unsupportedCompletionProvider(): iterable + { + yield 'older Server' => [true, null]; + yield 'unknown schema' => [true, ['schema' => 'unknown', 'header' => 'X-Durable-Workflow-Payload-Completion']]; + yield 'unexpected header' => [true, ['schema' => 'durable-workflow.v2.payload-completion-context.v1', 'header' => 'Authorization']]; + yield 'client input' => [false, ['schema' => 'durable-workflow.v2.payload-completion-context.v1', 'header' => 'X-Durable-Workflow-Payload-Completion']]; + } + + public function testRejectedBoundRetryDoesNotLoopOrSendCompletion(): void + { + [$client, $http] = $this->client(self::completionDiscovery(), drainUnbound: true, + uploadResponse: new Response(409, [], '{"reason":"external_payload_completion_lease_rejected","retryable":false}')); + try { + $client->completeActivityTask('task', 'attempt', 'worker', str_repeat('x', 200)); + self::fail('Rejected lease must not complete.'); + } catch (ExternalPayloadException $exception) { + self::assertSame(409, $exception->status); + self::assertSame('external_payload_completion_lease_rejected', $exception->reason); + } + self::assertCount(3, $http->requests); + } + + private static function completionDiscovery(): array + { + $discovery = self::discovery(); + $discovery['namespace']['external_payload_storage']['transport']['upload']['completion_context'] = [ + 'schema' => 'durable-workflow.v2.payload-completion-context.v1', 'header' => 'X-Durable-Workflow-Payload-Completion']; + + return $discovery; + } + public function testDiscoveryCannotRedirectUploadToAnotherHost(): void { $policy = self::discovery(); @@ -346,13 +463,13 @@ public function uploadPayload(string $uri, array $headers, string $blob, int $ti } } - private function client(?array $discovery = null, bool $workerOnly = false, ?Response $uploadResponse = null): array + private function client(?array $discovery = null, bool $workerOnly = false, ?Response $uploadResponse = null, bool $drainUnbound = false): array { - $http = new class($discovery ?? self::discovery(), $uploadResponse) { + $http = new class($discovery ?? self::discovery(), $uploadResponse, $drainUnbound) { public array $requests = []; public array $options = []; private string $uploadBody; - public function __construct(private array $discovery, private ?Response $uploadResponse) + public function __construct(private array $discovery, private ?Response $uploadResponse, private bool $drainUnbound) { $this->uploadBody = (string) $uploadResponse?->getBody(); } @@ -364,6 +481,9 @@ public function __invoke(RequestInterface $request, array $options): \GuzzleHttp if (str_ends_with($path, '/cluster/info')) { $response = $this->discovery; } elseif (str_ends_with($path, '/external-payloads/v1')) { + if ($this->drainUnbound && !$request->hasHeader('X-Durable-Workflow-Payload-Completion')) { + return Create::promiseFor(new Response(503, [], '{"reason":"storage_pressure","storage_state":"draining","request_admitted":false}')); + } if ($this->uploadResponse !== null) { return Create::promiseFor(new Response($this->uploadResponse->getStatusCode(), $this->uploadResponse->getHeaders(), $this->uploadBody)); }