Negotiate replay-safe task delivery and retain uncertain acceptances

docs/portal/build/workers.md

@@ -99,6 +99,14 @@ An activity option such as `['queue' => 'payments-v2']` selects the queue of a n
 
 ### Poll timeouts and early disconnects
 
+When the Server advertises `task_delivery`, the managed worker negotiates
+offer/accept delivery. Receiving an offer does not reserve a task or start an
+activity attempt. The SDK accepts the selected offer before running a handler.
+If that response is lost, it retries the same packet and poll identity until
+the Server confirms the claim, expiry, or loss of registration authority.
+An older node returning an error cannot switch that poll to legacy delivery.
+Servers without this capability continue using the existing poll protocol.
+
 With the default cURL transport, supporting Servers send progress during idle
 worker polls. Progress keeps an idle connection active while the complete
 request remains bounded to the requested wait plus five seconds. Ordinary polls

docs/protocol.md

@@ -198,6 +198,16 @@ may have committed a claim before the connection disappeared, so `task=null`
 does not establish that no lease exists. This does not make arbitrary database
 errors or unrelated structured HTTP 500/503 responses retryable.
 
+With negotiated `task_delivery`, worker polls use `/offers`. An offer alone
+grants no execution authority. Positive acceptance includes the original
+registration token and selected offer packet; its request body remains fixed
+through uncertain replies. Managed workers retain this state across client
+clones and honor the normal retry backoff. If using the low-level
+`poll*TaskResponse()` methods, catch `TaskDeliveryUncertain` and retry with its
+`details['poll_request_id']` on the same Client. An ordinary empty response to
+an acceptance does not prove that no claim committed. A confirmed offer expiry
+permits a new poll; lost registration authority stops that worker.
+
 Before replaying or completing a workflow task, the managed worker requires a
 successful lease-renewal response whose task ID, attempt, and lease owner match
 the current claim. A typed `renewed=false`, `retryable=true` response is retried

src/Client.php

@@ -62,6 +62,9 @@
 use DurableWorkflow\Worker\WorkerSessionOptions;
 use DurableWorkflow\Worker\ActivityLeaseBinding;
 use DurableWorkflow\Worker\ActivityLeaseProcess;
+use DurableWorkflow\Worker\TaskDelivery;
+use DurableWorkflow\Worker\TaskDeliveryAttempt;
+use DurableWorkflow\Transport\PollTransport;
 use InvalidArgumentException;
 
 /** Synchronous control-plane and worker-plane client for the standalone server. */
@@ -81,6 +84,7 @@ final class Client implements WorkflowClientInterface
     private FailureReportLimiter $failureReports;
     private bool $boundedWorkerRequests = false;
     private ActivityLeaseBinding $activityLeaseBinding;
+    private TaskDelivery $taskDelivery;
     private ?string $activityLeaseBootstrap = null;
     private ?ExecutionTelemetry $executionTelemetry = null;
     private bool $observabilityContextSupported = false;
@@ -143,6 +147,7 @@ public function __construct(
         $this->payloadUploads = new RuntimePayloadUploads($this->transport, $this->baseUri);
         $this->failureReports = new FailureReportLimiter(new FailureReportLimits(), $this->transport);
         $this->activityLeaseBinding = new ActivityLeaseBinding();
+        $this->taskDelivery = new TaskDelivery();
     }
 
     /** Narrow the encoded failure-report request budget for an application transport. */
@@ -1576,6 +1581,7 @@ public function registerWorker(
             'memo_upserts',
             'typed_search_attributes',
             'durable_selection',
+            TaskDelivery::CAPABILITY,
         ],
         int $maxConcurrentWorkflowTasks = 1,
         int $maxConcurrentActivityTasks = 1,
@@ -1585,6 +1591,10 @@ public function registerWorker(
         int $maxConcurrentWorkerSessions = 10,
         ?array $workflowDefinitionFingerprints = null,
     ): array {
+        if ($this->workerProtocolVersion !== '1.20') {
+            $capabilities = array_values(array_filter($capabilities,
+                static fn (string $capability): bool => $capability !== TaskDelivery::CAPABILITY));
+        }
         if (in_array('message_streams', $capabilities, true) && !Version::supportsMessageStreams()) {
             throw new InvalidArgumentException('Message streams require worker protocol 1.15 or newer.');
         }
@@ -1606,7 +1616,7 @@ public function registerWorker(
             }
         }
 
-        return $this->worker('POST', '/worker/register', $this->withoutNulls([
+        $response = $this->worker('POST', '/worker/register', $this->withoutNulls([
             'worker_id' => $workerId,
             'task_queue' => $taskQueue,
             'runtime' => 'php',
@@ -1621,7 +1631,11 @@ public function registerWorker(
             'max_concurrent_activity_tasks' => $maxConcurrentActivityTasks,
             'max_concurrent_worker_sessions' => $maxConcurrentWorkerSessions,
             'build_id' => $buildId,
+            'process_metrics' => ['process_id' => getmypid(), 'process_uptime_seconds' => 0],
         ]));
+        $this->taskDelivery->registered($workerId, $taskQueue, $this->workerProtocolVersion, $capabilities, $response);
+
+        return $response;
     }
 
     /** @return array<string, mixed> */
@@ -1678,6 +1692,7 @@ public function heartbeatWorker(string $workerId, array $taskSlots = [], ?Reques
     public function deregisterWorker(string $workerId): void
     {
         $this->control('DELETE', '/workers/'.$this->segment($workerId));
+        $this->taskDelivery->forget($workerId);
     }
 
     /** @return array{worker_id: string, outcome: string, recovered_workflow_task_count: int} */
@@ -1702,6 +1717,8 @@ public function deregisterWorkerRegistration(string $workerId, ?string $registra
                 'invalid_worker_deregistration_receipt', $response);
         }
 
+        $this->taskDelivery->forget($workerId);
+
         /** @var array{worker_id: string, outcome: string, recovered_workflow_task_count: int} $response */
         return $response;
     }
@@ -2592,7 +2609,12 @@ private function pollTaskResponse(string $path, array $body): array
         $retryTransientFailure = true;
         while (true) {
             try {
-                return $this->worker('POST', $path, $body);
+                return $this->taskDelivery->poll($path, $body,
+                    fn (string $wirePath, array $wireBody, ?TaskDeliveryAttempt $selected): array =>
+                        $this->request('POST', $wirePath, true, $wireBody,
+                            responseValidator: static fn (array $response) =>
+                                TaskDelivery::assertResponse($response, $wireBody, $wirePath, $selected)))
+                    ?? $this->worker('POST', $path, $body);
             } catch (ServerException $exception) {
                 if ($retryTransientFailure && $exception->isTransientConnectionFailure()) {
                     $retryTransientFailure = false;
@@ -2659,7 +2681,7 @@ private function sendTransportRequest(string $method, string $path, bool $worker
     }
 
     /** @return array<string, string> */
-    private function requestHeaders(bool $worker): array
+    private function requestHeaders(bool $worker, string $path = ''): array
     {
         $headers = [
             'Accept' => 'application/json',
@@ -2671,13 +2693,17 @@ private function requestHeaders(bool $worker): array
         if ($this->authentication !== null) {
             $headers = array_merge($headers, $this->authentication->headers($worker));
         }
+        if ($worker && PollTransport::taskKind('POST', $path) !== null && str_ends_with($path, '/offers')) {
+            $headers[TaskDelivery::HEADER] = TaskDelivery::VERSION;
+        }
 
         return $headers;
     }
 
     /**
      * @param array<string, mixed>|null $body
      * @param (\Closure(array<string, mixed>): list<list<int|string>>)|null $payloadPaths
+     * @param (\Closure(array<string, mixed>): void)|null $responseValidator
      * @return array<string, mixed>
      */
     private function request(
@@ -2689,11 +2715,13 @@ private function request(
         ?RequestBudget $budget = null,
         ?\Closure $payloadPaths = null,
         ?string $requestId = null,
+        ?\Closure $responseValidator = null,
     ): array {
-        $headers = $this->requestHeaders($worker);
+        $headers = $this->requestHeaders($worker, $path);
+        $logicalPath = PollTransport::logicalPath($path);
         try {
             $callerBudget = $budget;
-            $poll = str_ends_with($path, '/poll') && is_int($body['timeout_seconds'] ?? null);
+            $poll = PollTransport::taskKind($method, $path) !== null && is_int($body['timeout_seconds'] ?? null);
             $pollSeconds = $poll ? max(0, min(60, $body['timeout_seconds'])) : 0;
             $budget ??= $worker && $this->boundedWorkerRequests ? new RequestBudget($pollSeconds + 5) : null;
             if ($body !== null) {
@@ -2703,7 +2731,7 @@ private function request(
                 $body = $this->failureReports->request($path, $body);
             }
             $activityLease = $worker ? $this->activityLeaseBinding->current() : null;
-            if ($activityLease !== null && $activityLease->governs($path)) {
+            if ($activityLease !== null && $activityLease->governs($logicalPath)) {
                 // Uploading an external result is not publication authority.
                 // Check again before its final Worker POST and bound the send
                 // by the latest acknowledged grant, not an old cached lease.
@@ -2713,21 +2741,27 @@ private function request(
             $response = $this->sendTransportRequest($method, $path, $worker, $headers, $body, $budget);
 
             $budget?->remainingSeconds();
+            if ($responseValidator !== null) {
+                if (!is_array($response) || array_is_list($response)) {
+                    throw new ServerException('The task-delivery response is invalid.', 200, 'invalid_task_delivery_response');
+                }
+                $responseValidator($response);
+            }
             // Control I/O and supervised payload I/O have separate bounds.
             // Explicit caller/cleanup budgets still cover their entire request.
             $responseBudget = $budget !== null && $poll && $callerBudget === null ? new RequestBudget(5) : $budget;
-            $canStreamBounded = $worker && ($this->boundedWorkerRequests || $path === '/worker/query-tasks/poll')
+            $canStreamBounded = $worker && ($this->boundedWorkerRequests || $logicalPath === '/worker/query-tasks/poll')
                 && $this->transport instanceof \DurableWorkflow\Transport\BoundedProgressPayloadTransport
                 && $this->transport instanceof BoundedTransport && $this->transport->supportsBoundedRequests();
             $observerBudget = $canStreamBounded && $callerBudget === null ? null : $responseBudget;
             $progress = $worker && $this->workerPayloadObserver !== null && is_array($response) && !array_is_list($response)
-                ? ($this->workerPayloadObserver)($path, $body, $response, $observerBudget) : null;
+                ? ($this->workerPayloadObserver)($logicalPath, $body, $response, $observerBudget) : null;
             $streamBounded = $canStreamBounded && $progress !== null;
             if ($streamBounded && $callerBudget === null) {
                 $responseBudget = null;
             }
             $result = is_array($response) && !array_is_list($response)
-                ? (new RuntimePayloads($this->transport, $this->baseUri, $headers, $this->maxExternalPayloadBytes, $responseBudget, $progress, $streamBounded))->response($response, $path, $worker, $payloadPaths?->__invoke($response))
+                ? (new RuntimePayloads($this->transport, $this->baseUri, $headers, $this->maxExternalPayloadBytes, $responseBudget, $progress, $streamBounded))->response($response, $logicalPath, $worker, $payloadPaths?->__invoke($response))
                 : [];
             $responseBudget?->remainingSeconds();
 

src/Exception/TaskDeliveryUncertain.php

@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Exception;
+
+use Throwable;
+
+/** Retry only the original logical poll; this exception grants no execution. */
+final class TaskDeliveryUncertain extends ServerException
+{
+    public function __construct(string $pollRequestId, int $status, ?Throwable $previous = null)
+    {
+        parent::__construct('Task delivery is not confirmed. Retry the same poll and selected offer.',
+            $status, 'task_delivery_outcome_unknown', [
+                'task' => null, 'poll_request_id' => $pollRequestId,
+                'retry_same_poll_request_id' => true,
+                ...($previous instanceof ServerException && is_int($previous->details['retry_after_seconds'] ?? null)
+                    && $previous->details['retry_after_seconds'] > 0
+                    ? ['retry_after_seconds' => $previous->details['retry_after_seconds']] : []),
+            ], $previous);
+    }
+}

src/Transport/PollTransport.php

@@ -19,7 +19,8 @@ public static function taskKind(string $method, string $uri): ?string
         $path = parse_url($uri, PHP_URL_PATH);
         if (strtoupper($method) === 'POST' && is_string($path)) {
             foreach (self::KINDS as $kind) {
-                if (str_ends_with($path, '/worker/'.$kind.'-tasks/poll')) {
+                if (str_ends_with($path, '/worker/'.$kind.'-tasks/poll')
+                    || str_ends_with($path, '/worker/'.$kind.'-tasks/offers')) {
                     return $kind;
                 }
             }
@@ -28,6 +29,13 @@ public static function taskKind(string $method, string $uri): ?string
         return null;
     }
 
+    /** Payload observers and claim supervision use the original logical poll. */
+    public static function logicalPath(string $path): string
+    {
+        return self::taskKind('POST', $path) !== null && str_ends_with($path, '/offers')
+            ? substr($path, 0, -7).'/poll' : $path;
+    }
+
     /** @return array{status: int, headers: array<string, string>, body: array<string, mixed>} */
     public static function decode(string $rawBody, int $outerStatus): array
     {

src/Worker.php

@@ -656,6 +656,7 @@ private function registerWithRetry(): ?array
                         'worker_sessions',
                         'sticky_execution',
                         'cross_kind_poll_wake',
+                        ...($this->client->workerProtocolVersion === '1.20' ? [\DurableWorkflow\Worker\TaskDelivery::CAPABILITY] : []),
                         ...($this->enableCooperativeCancellation ? ['cooperative_cancellation', 'nexus_cooperative_cancellation'] : []),
                         ...($this->enablePreparedLocalActivities ? ['prepared_local_activities'] : []),
                         ...($this->preparedLocalActivityGroupsSupported ? ['prepared_local_activity_groups'] : []),
@@ -929,6 +930,7 @@ private function pollWithRetry(string $taskKind, \Closure $poll, int $requestedT
                 }
 
                 if (!$exception->isTransientConnectionFailure() && !$exception->isTransientUpstreamFailure()
+                    && !$exception instanceof \DurableWorkflow\Exception\TaskDeliveryUncertain
                     && !PollResponse::isTransientPayloadFailure($exception)
                     && !$databaseUnavailable && !$storageAdmission) {
                     $requestId = 'php-'.$taskKind.'-poll-'.bin2hex(random_bytes(16));

src/Worker/PollResponse.php

@@ -6,6 +6,7 @@
 
 use DurableWorkflow\Exception\ExternalPayloadException;
 use DurableWorkflow\Exception\ServerException;
+use DurableWorkflow\Exception\TaskDeliveryUncertain;
 
 /** Classifies full worker poll response envelopes using protocol fields. */
 final class PollResponse
@@ -23,6 +24,7 @@ final class PollResponse
         'worker_heartbeat_stale',
         'worker_draining',
         'worker_stopped',
+        'worker_registration_lost_authority',
     ];
 
     /** @param array<string, mixed> $response */
@@ -42,7 +44,8 @@ public static function isTerminal(array $response): bool
      */
     public static function isTransientFailure(ServerException $exception): bool
     {
-        if ($exception->isTransientConnectionFailure() || $exception->isTransientUpstreamFailure()
+        if ($exception instanceof TaskDeliveryUncertain
+            || $exception->isTransientConnectionFailure() || $exception->isTransientUpstreamFailure()
             || self::isTransientPayloadFailure($exception)) {
             return true;
         }