Bind draining payload upload retries to the current completion lease

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

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",

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": {

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.

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<string, list<string>> */
     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<string, string> $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<string, mixed> $body
+     * @param list<int|string> $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<string, mixed> $body
      * @param array{path: list<int|string>, blob: string} $payload
      * @return array{schema: string, reference_id: string, codec: string, size_bytes: int, sha256: string}

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']);

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));
                     }