From 086b3839624a112b0d71703200df033bc335c57d Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 25 Oct 2026 08:17:19 +0000 Subject: [PATCH 1/3] Negotiate replay-safe worker task offers and preserve uncertain acceptances --- docs/portal/build/workers.md | 8 + docs/protocol.md | 10 + src/Client.php | 52 ++++- src/Exception/TaskDeliveryUncertain.php | 23 +++ src/Transport/PollTransport.php | 10 +- src/Worker.php | 2 + src/Worker/PollResponse.php | 5 +- src/Worker/TaskDelivery.php | 179 ++++++++++++++++ src/Worker/TaskDeliveryAttempt.php | 53 +++++ tests/ClientContractTest.php | 2 + tests/WorkerAuthoringTest.php | 2 + tests/WorkerPollTransportTest.php | 21 ++ tests/WorkerTaskDeliveryTest.php | 259 ++++++++++++++++++++++++ 13 files changed, 615 insertions(+), 11 deletions(-) create mode 100644 src/Exception/TaskDeliveryUncertain.php create mode 100644 src/Worker/TaskDelivery.php create mode 100644 src/Worker/TaskDeliveryAttempt.php create mode 100644 tests/WorkerTaskDeliveryTest.php diff --git a/docs/portal/build/workers.md b/docs/portal/build/workers.md index 41e9e64b..ca88fbc5 100644 --- a/docs/portal/build/workers.md +++ b/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 diff --git a/docs/protocol.md b/docs/protocol.md index 6ce822cb..7814f0c1 100644 --- a/docs/protocol.md +++ b/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 diff --git a/src/Client.php b/src/Client.php index d4d3aa4e..cf5ba4a1 100644 --- a/src/Client.php +++ b/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 */ @@ -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 */ - private function requestHeaders(bool $worker): array + private function requestHeaders(bool $worker, string $path = ''): array { $headers = [ 'Accept' => 'application/json', @@ -2671,6 +2693,9 @@ 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; } @@ -2678,6 +2703,7 @@ private function requestHeaders(bool $worker): array /** * @param array|null $body * @param (\Closure(array): list>)|null $payloadPaths + * @param (\Closure(array): void)|null $responseValidator * @return array */ 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(); diff --git a/src/Exception/TaskDeliveryUncertain.php b/src/Exception/TaskDeliveryUncertain.php new file mode 100644 index 00000000..035f351c --- /dev/null +++ b/src/Exception/TaskDeliveryUncertain.php @@ -0,0 +1,23 @@ + 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); + } +} diff --git a/src/Transport/PollTransport.php b/src/Transport/PollTransport.php index 4aaf2088..3f719441 100644 --- a/src/Transport/PollTransport.php +++ b/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, body: array} */ public static function decode(string $rawBody, int $outerStatus): array { diff --git a/src/Worker.php b/src/Worker.php index 57192c9e..92995267 100644 --- a/src/Worker.php +++ b/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)); diff --git a/src/Worker/PollResponse.php b/src/Worker/PollResponse.php index e09477df..62a4c261 100644 --- a/src/Worker/PollResponse.php +++ b/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 $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; } diff --git a/src/Worker/TaskDelivery.php b/src/Worker/TaskDelivery.php new file mode 100644 index 00000000..9ab67953 --- /dev/null +++ b/src/Worker/TaskDelivery.php @@ -0,0 +1,179 @@ + */ + private array $registrations = []; + /** @var array */ + private array $pending = []; + + /** + * @param list $capabilities + * @param array $response + */ + public function registered(string $workerId, string $queue, string $protocol, array $capabilities, array $response): void + { + $server = $response['server_capabilities'] ?? []; + $feature = is_array($server) ? ($server['task_delivery'] ?? null) : null; + if ($protocol !== '1.20' || !in_array(self::CAPABILITY, $capabilities, true) + || $feature === null || (is_array($feature) && ($feature['supported'] ?? null) === false)) { + foreach ($this->pending as $attempt) { + if ($attempt->workerId === $workerId) { + throw new ServerException('An unresolved delivery cannot change transport modes.', 201, 'invalid_task_delivery_capability'); + } + } + $this->forget($workerId); + return; + } + $token = $response['registration_token'] ?? null; + if (!is_array($feature) || array_is_list($feature) || ($feature['supported'] ?? null) !== true + || ($feature['schema'] ?? null) !== self::SCHEMA || ($feature['version'] ?? null) !== self::VERSION + || ($feature['header'] ?? null) !== self::HEADER || ($feature['worker_capability'] ?? null) !== self::CAPABILITY + || ($feature['minimum_protocol_version'] ?? null) !== '1.20' || ($feature['endpoint_suffix'] ?? null) !== '/offers' + || ($feature['unaccepted_offer_reserves_task'] ?? null) !== false + || !is_int($feature['acceptance_timeout_seconds'] ?? null) + || $feature['acceptance_timeout_seconds'] < 1 || $feature['acceptance_timeout_seconds'] > 10 + || !is_string($token) || preg_match('/^[a-f0-9]{32}$/D', $token) !== 1) { + throw new ServerException('The task-delivery registration contract is invalid.', 201, 'invalid_task_delivery_capability'); + } + if (($this->registrations[$workerId] ?? null) !== ['queue' => $queue, 'token' => $token]) { + $this->forget($workerId); + } + $this->registrations[$workerId] = ['queue' => $queue, 'token' => $token]; + } + + /** + * @param array $body + * @param Closure(string, array, ?TaskDeliveryAttempt): array $send + * @return array|null Null means negotiation never selected this mode. + */ + public function poll(string $path, array $body, Closure $send): ?array + { + $workerId = (string) $body['worker_id']; + $registration = $this->registrations[$workerId] ?? null; + if ($registration === null || $registration['queue'] !== ($body['task_queue'] ?? null)) { + return null; + } + $key = hash('sha256', json_encode([$path, $workerId, $body['task_queue'], $body['poll_request_id']], JSON_THROW_ON_ERROR)); + $selected = $this->pending[$key] ?? null; + try { + if ($selected === null) { + $wirePath = substr($path, 0, -5).'/offers'; + $body['registration_token'] = $registration['token']; + $response = $send($wirePath, $body, null); + if (!array_key_exists('task_offer', $response)) { + return $response; + } + $candidate = TaskDeliveryAttempt::fromResponse($wirePath, $body, $response, + (string) PollTransport::taskKind('POST', $path)); + // Concurrent callers for one logical poll select only its first packet. + $selected = $this->pending[$key] ??= $candidate; + } + $response = $send($selected->path, $selected->requestBody(), $selected); + if ($response['task'] === null && !PollResponse::isTerminal($response)) { + throw new TaskDeliveryUncertain((string) $body['poll_request_id'], 200); + } + $this->resolved($key, $selected); + + return $response; + } catch (ServerException $exception) { + if ($this->definitiveRefusal($exception)) { + $this->resolved($key, $selected); + $response = $exception->details ?? []; + if ($exception->reason === 'worker_registration_lost_authority') { + $this->forget($workerId); + } else { + $response['poll_status'] = 'empty'; + } + + return $response; + } + if ($exception instanceof TaskDeliveryUncertain) { + throw $exception; + } + if (($selected !== null && ($exception->status >= 500 || $exception->status === 409 || $exception->status === 429 + || $exception->reason === 'invalid_task_delivery_response' + || in_array($exception->status, [0, 404, 405], true))) + || ($selected === null && in_array($exception->status, [404, 405], true))) { + if (PollResponse::isTransientPayloadFailure($exception)) { + throw $exception; + } + throw new TaskDeliveryUncertain((string) $body['poll_request_id'], $exception->status, $exception); + } + + throw $exception; + } + } + + /** + * Validate before payload hydration can initialize a claim supervisor. + * @param array $response + * @param array $body + */ + public static function assertResponse(array $response, array $body, string $path, ?TaskDeliveryAttempt $selected): void + { + $task = $response['task'] ?? null; + $kind = PollTransport::taskKind('POST', $path); + $idField = match ($kind) { + 'query' => 'query_task_id', 'update-validation' => 'update_validation_task_id', default => 'task_id', + }; + if (!array_key_exists('task', $response) + || (($selected !== null || $task !== null) && array_key_exists('task_offer', $response)) + || (($response['poll_status'] ?? null) === 'offered' && !array_key_exists('task_offer', $response)) + || ($task !== null && (!is_array($task) || array_is_list($task) + || !is_string($task[$idField] ?? null) || $task[$idField] === '' + || ($task['lease_owner'] ?? null) !== $body['worker_id'] + || ($selected !== null && $task[$idField] !== $selected->taskId) + || (isset($task['task_kind']) && $task['task_kind'] !== str_replace('-', '_', (string) $kind))))) { + throw new ServerException('The task-delivery response is invalid.', 200, 'invalid_task_delivery_response'); + } + } + + public function forget(string $workerId): void + { + unset($this->registrations[$workerId]); + foreach ($this->pending as $key => $attempt) { + if ($attempt->workerId === $workerId) { + unset($this->pending[$key]); + } + } + } + + /** @return array */ + public function __debugInfo(): array + { + return ['registered_workers' => count($this->registrations), 'unresolved_deliveries' => count($this->pending)]; + } + + private function definitiveRefusal(ServerException $exception): bool + { + $response = $exception->details; + + return $exception->status === 409 && $response !== null && !array_is_list($response) + && array_key_exists('task', $response) && $response['task'] === null + && ($response['reason'] ?? null) === $exception->reason + && in_array($exception->reason, ['task_offer_expired', 'worker_registration_lost_authority'], true); + } + + private function resolved(string $key, ?TaskDeliveryAttempt $selected): void + { + if (($this->pending[$key] ?? null) === $selected) { + unset($this->pending[$key]); + } + } +} diff --git a/src/Worker/TaskDeliveryAttempt.php b/src/Worker/TaskDeliveryAttempt.php new file mode 100644 index 00000000..64cff2ca --- /dev/null +++ b/src/Worker/TaskDeliveryAttempt.php @@ -0,0 +1,53 @@ + $body */ + private function __construct( + public readonly string $workerId, + public readonly string $path, + public readonly string $taskId, + private readonly array $body, + ) { + } + + /** + * @param array $body + * @param array $response + */ + public static function fromResponse(string $path, array $body, array $response, string $kind): self + { + $offer = $response['task_offer'] ?? null; + if (!array_key_exists('task', $response) || $response['task'] !== null || ($response['poll_status'] ?? null) !== 'offered' + || !is_array($offer) || array_is_list($offer) + || ($offer['schema'] ?? null) !== TaskDelivery::SCHEMA + || ($offer['task_kind'] ?? null) !== str_replace('-', '_', $kind) + || !is_string($offer['task_id'] ?? null) || $offer['task_id'] === '' + || !is_string($offer['token'] ?? null) || $offer['token'] === '' || strlen($offer['token']) > 8192 + || !is_int($offer['expires_at_ms'] ?? null) || $offer['expires_at_ms'] < 1) { + throw new ServerException('The server returned an invalid task offer.', 200, 'invalid_task_delivery_offer'); + } + + return new self((string) $body['worker_id'], $path, $offer['task_id'], + array_replace($body, ['task_offer_token' => $offer['token'], 'timeout_seconds' => 0])); + } + + /** @return array */ + public function requestBody(): array + { + return $this->body; + } + + /** @return array */ + public function __debugInfo(): array + { + return ['worker_id' => $this->workerId, 'path' => $this->path, 'task_id' => $this->taskId]; + } +} diff --git a/tests/ClientContractTest.php b/tests/ClientContractTest.php index 439a9c90..eef3ec12 100644 --- a/tests/ClientContractTest.php +++ b/tests/ClientContractTest.php @@ -119,6 +119,7 @@ public function testDefaultWorkerRegistrationAdvertisesTheCompletePhpCapabilityS 'memo_upserts', 'typed_search_attributes', 'durable_selection', + 'task_delivery_offers', ], 'capability_manifest' => [ 'local_activities' => [ @@ -140,6 +141,7 @@ public function testDefaultWorkerRegistrationAdvertisesTheCompletePhpCapabilityS 'max_concurrent_workflow_tasks' => 1, 'max_concurrent_activity_tasks' => 1, 'max_concurrent_worker_sessions' => 10, + 'process_metrics' => ['process_id' => getmypid(), 'process_uptime_seconds' => 0], ], ], $transport->requests[0]); } diff --git a/tests/WorkerAuthoringTest.php b/tests/WorkerAuthoringTest.php index b8918bfe..930c2af6 100644 --- a/tests/WorkerAuthoringTest.php +++ b/tests/WorkerAuthoringTest.php @@ -182,6 +182,7 @@ public function testHighLevelWorkerAdvertisesTheCompleteManagedCapabilitySet(): 'worker_sessions', 'sticky_execution', 'cross_kind_poll_wake', + 'task_delivery_offers', ], 'capability_manifest' => [ 'local_activities' => [ @@ -204,6 +205,7 @@ public function testHighLevelWorkerAdvertisesTheCompleteManagedCapabilitySet(): 'max_concurrent_activity_tasks' => 1, 'max_concurrent_worker_sessions' => 10, 'build_id' => 'release-a', + 'process_metrics' => ['process_id' => getmypid(), 'process_uptime_seconds' => 0], ], ], $transport->requests[0]); } diff --git a/tests/WorkerPollTransportTest.php b/tests/WorkerPollTransportTest.php index eda12b19..2ecf2df0 100644 --- a/tests/WorkerPollTransportTest.php +++ b/tests/WorkerPollTransportTest.php @@ -51,6 +51,27 @@ public static function pollKinds(): array return [['workflow'], ['activity'], ['query'], ['update-validation']]; } + #[DataProvider('pollKinds')] + public function testOfferAndAcceptanceKeepProgressNegotiationAndTheirAbsoluteBudgets(string $kind): void + { + $calls = []; + $http = new GuzzleClient(['handler' => static function (RequestInterface $request, array $options) use (&$calls) { + $calls[] = [$request, $options]; + return Create::promiseFor(new Response(200, ['Content-Type' => self::MEDIA_TYPE], self::frame(200))); + }]); + $transport = new Psr18Transport($http); + $uri = 'https://server.example/api/worker/'.$kind.'-tasks/offers'; + $transport->send('POST', $uri, [], ['timeout_seconds' => 4, 'poll_request_id' => 'same-poll']); + $transport->send('POST', $uri, [], ['timeout_seconds' => 0, 'poll_request_id' => 'same-poll', 'task_offer_token' => 'same-packet']); + self::assertSame(9, $calls[0][1]['timeout']); + self::assertSame(5, $calls[1][1]['timeout']); + foreach ($calls as [$request, $options]) { + self::assertSame(self::MEDIA_TYPE.', application/json;q=0.9', $request->getHeaderLine('Accept')); + self::assertFalse($options['stream']); + } + self::assertSame('same-packet', json_decode((string) $calls[1][0]->getBody(), true)['task_offer_token']); + } + public function testAShorterInjectedGuzzleDeadlineIsPreserved(): void { $observed = null; diff --git a/tests/WorkerTaskDeliveryTest.php b/tests/WorkerTaskDeliveryTest.php new file mode 100644 index 00000000..66a81788 --- /dev/null +++ b/tests/WorkerTaskDeliveryTest.php @@ -0,0 +1,259 @@ +withWorkerPayloadObserver( + static function (string $path, ?array $body, array $response) use (&$observed): ?\Closure { + if ($response['task'] ?? null) { + $observed[] = $path; + } + return null; + }); + $client->registerWorker('worker', 'orders', [], []); + self::assertSame(self::claimed($kind), self::poll($client, $kind)); + self::assertCount(3, $transport->requests); + self::assertSame(['/worker/'.$kind.'-tasks/poll'], $observed); + $initial = $transport->requests[1]; + $accept = $transport->requests[2]; + self::assertStringEndsWith('/worker/'.$kind.'-tasks/offers', $initial['uri']); + self::assertSame($initial['uri'], $accept['uri']); + self::assertSame(TaskDelivery::VERSION, $initial['headers'][TaskDelivery::HEADER]); + self::assertSame($initial['headers'], $accept['headers']); + self::assertSame(4, $initial['body']['timeout_seconds']); + self::assertSame(array_replace($initial['body'], ['timeout_seconds' => 0, 'task_offer_token' => 'opaque-first-packet']), $accept['body']); + self::assertSame(str_repeat('a', 32), $initial['body']['registration_token']); + self::assertSame(getmypid(), $transport->requests[0]['body']['process_metrics']['process_id']); + self::assertArrayNotHasKey(TaskDelivery::HEADER, $transport->requests[0]['headers']); + } + + #[DataProvider('uncertainOutcomes')] + public function testUncertainRepliesRetryOnlyTheFirstSelectedPacket(TransportException|array|null $outcome): void + { + $transport = new FakeTransport([self::registration(), self::offer('workflow'), $outcome, self::claimed('workflow')]); + $client = new Client('https://server.example', transport: $transport); + $client->registerWorker('worker', 'orders', [], []); + try { + self::poll($client, 'workflow'); + self::fail('The uncertain acceptance must not authorize execution or start another poll.'); + } catch (TaskDeliveryUncertain $error) { + self::assertTrue(PollResponse::isTransientFailure($error)); + self::assertSame('fixed-poll', $error->details['poll_request_id']); + self::assertArrayNotHasKey('task_offer_token', $error->details); + } + // A clone and changed wait/page inputs cannot replace an unresolved packet. + $clone = $client->withWorkerPollCutoffObserver(static function (): void {}); + self::assertSame(self::claimed('workflow'), $clone->pollWorkflowTaskResponse('worker', 'orders', 19, 'fixed-poll', 30)); + self::assertCount(4, $transport->requests); + self::assertSame($transport->requests[2], $transport->requests[3]); + self::assertSame('opaque-first-packet', $transport->requests[3]['body']['task_offer_token']); + } + + public static function uncertainOutcomes(): iterable + { + yield 'lost response' => [new TransportException('Connection reset.', transientConnectionFailure: true)]; + yield 'gateway timeout' => [new TransportException('Gateway timeout.', 504)]; + yield 'older node' => [new TransportException('Not found.', 404)]; + yield 'structured backend failure' => [new TransportException('Backend unavailable.', 503, + ['task' => null, 'reason' => 'backend_unavailable', 'retryable' => true])]; + yield 'lock pressure' => [new TransportException('Lock pressure.', 503, + ['task' => null, 'reason' => 'backend_lock_pressure', 'poll_status' => 'backend_lock_pressure', 'retry_after_seconds' => 1])]; + yield 'packet unavailable on another node' => [new TransportException('Unavailable.', 409, + ['task' => null, 'reason' => 'task_offer_unavailable'])]; + yield 'empty is not proof of no claim' => [['task' => null, 'poll_status' => 'empty']]; + yield 'malformed response' => [null]; + yield 'wrong task' => [['task' => ['task_id' => 'other', 'lease_owner' => 'worker']]]; + yield 'wrong owner' => [['task' => ['task_id' => 'task', 'lease_owner' => 'other']]]; + } + + public function testAnOlderNodeCannotDowngradeTheNegotiatedEndpoint(): void + { + $transport = new FakeTransport([self::registration(), new TransportException('Not found.', 404), ['task' => null]]); + $client = new Client('https://server.example', transport: $transport); + $client->registerWorker('worker', 'orders', [], []); + try { + self::poll($client, 'activity'); + self::fail('Negotiated delivery cannot fall back to a legacy immediate claim.'); + } catch (TaskDeliveryUncertain) { + self::assertSame(['task' => null], self::poll($client, 'activity')); + } + self::assertSame($transport->requests[1], $transport->requests[2]); + self::assertStringEndsWith('/activity-tasks/offers', $transport->requests[2]['uri']); + } + + public function testAConfirmedExpiryReleasesTheSelectedPacket(): void + { + $transport = new FakeTransport([self::registration(), self::offer('workflow'), + new TransportException('Expired.', 409, ['task' => null, 'poll_status' => 'task_offer_expired', 'reason' => 'task_offer_expired']), + self::offer('workflow', 'second-packet'), self::claimed('workflow')]); + $client = new Client('https://server.example', transport: $transport); + $client->registerWorker('worker', 'orders', [], []); + self::assertSame('empty', self::poll($client, 'workflow')['poll_status']); + self::assertSame(self::claimed('workflow'), self::poll($client, 'workflow')); + self::assertArrayNotHasKey('task_offer_token', $transport->requests[3]['body']); + self::assertSame('second-packet', $transport->requests[4]['body']['task_offer_token']); + } + + public function testRegistrationCannotDowngradeAnUnresolvedAcceptance(): void + { + $transport = new FakeTransport([self::registration(), self::offer('workflow'), + new TransportException('Gateway timeout.', 504), ['registered' => true], self::claimed('workflow')]); + $client = new Client('https://server.example', transport: $transport); + $client->registerWorker('worker', 'orders', [], []); + try { self::poll($client, 'workflow'); self::fail('Expected an uncertain acceptance.'); } + catch (TaskDeliveryUncertain) {} + try { $client->registerWorker('worker', 'orders', [], []); self::fail('Expected a rejected downgrade.'); } + catch (ServerException $error) { self::assertSame('invalid_task_delivery_capability', $error->reason); } + self::assertSame(self::claimed('workflow'), self::poll($client, 'workflow')); + self::assertSame($transport->requests[2], $transport->requests[4]); + } + + public function testReplacementRegistrationDiscardsOnlyTheOldIncarnationsPacket(): void + { + $replacement = self::registration(); + $replacement['registration_token'] = str_repeat('b', 32); + $transport = new FakeTransport([self::registration(), self::offer('activity'), + new TransportException('Gateway timeout.', 504), $replacement, self::offer('activity', 'replacement-packet'), self::claimed('activity')]); + $client = new Client('https://server.example', transport: $transport); + $client->registerWorker('worker', 'orders', [], []); + try { self::poll($client, 'activity'); self::fail('Expected an uncertain acceptance.'); } + catch (TaskDeliveryUncertain) {} + $client->registerWorker('worker', 'orders', [], []); + self::assertSame(self::claimed('activity'), self::poll($client, 'activity')); + self::assertSame(str_repeat('b', 32), $transport->requests[4]['body']['registration_token']); + self::assertArrayNotHasKey('task_offer_token', $transport->requests[4]['body']); + self::assertSame('replacement-packet', $transport->requests[5]['body']['task_offer_token']); + } + + #[DataProvider('legacyRegistrations')] + public function testLegacyDeliveryIsChosenOnlyAtRegistration(array $registration, string $protocol, ?array $capabilities): void + { + $transport = new FakeTransport([$registration, ['task' => null]]); + $client = new Client('https://server.example', transport: $transport, workerProtocolVersion: $protocol); + if ($capabilities === null) { + $client->registerWorker('worker', 'orders', [], []); + } else { + $client->registerWorker('worker', 'orders', [], [], $capabilities); + } + self::poll($client, 'workflow'); + self::assertStringEndsWith('/workflow-tasks/poll', $transport->requests[1]['uri']); + self::assertArrayNotHasKey(TaskDelivery::HEADER, $transport->requests[1]['headers']); + if ($protocol === '1.19') { + self::assertNotContains(TaskDelivery::CAPABILITY, $transport->requests[0]['body']['capabilities']); + } + } + + public static function legacyRegistrations(): iterable + { + yield 'no server feature' => [['registered' => true], '1.20', null]; + yield 'explicit worker opt-out' => [self::registration(), '1.20', ['query_tasks']]; + yield 'older protocol' => [self::registration(), '1.19', null]; + } + + public function testAContradictoryOfferCannotReachPayloadOrExecutionObservers(): void + { + $contradictory = self::claimed('workflow') + ['task_offer' => self::offer('workflow')['task_offer']]; + $transport = new FakeTransport([self::registration(), $contradictory]); + $claims = 0; + $client = (new Client('https://server.example', transport: $transport))->withWorkerPayloadObserver( + static function (string $path, ?array $body, array $response) use (&$claims): ?\Closure { + if ($response['task'] ?? null) { ++$claims; } + return null; + }); + $client->registerWorker('worker', 'orders', [], []); + try { + self::poll($client, 'workflow'); + self::fail('A contradictory offer cannot authorize a task.'); + } catch (ServerException $error) { + self::assertSame('invalid_task_delivery_response', $error->reason); + } + self::assertSame(0, $claims); + self::assertCount(2, $transport->requests); + } + + public function testTheManagedWorkerRetainsItsLogicalPollAfterAnUncertainAcceptance(): void + { + $now = 0.0; + $retries = []; + $transport = new FakeTransport([self::registration(), self::offer('workflow'), + new TransportException('Not found.', 404), + ['task' => null, 'poll_status' => 'stopped', 'reason' => 'worker_stopped']]); + $worker = new Worker(new Client('https://server.example', transport: $transport), 'orders', workerId: 'worker', + clock: static function () use (&$now): float { return $now; }, + sleeper: static function (int $microseconds) use (&$now): void { $now += $microseconds / 1_000_000; }, + transientPollRetryObserver: static function (string $kind, int $attempt, float $delay, ServerException $error) use (&$retries): void { + $retries[] = [$kind, $attempt, $error::class]; + }); + $worker->run(4); + self::assertSame([['workflow', 1, TaskDeliveryUncertain::class]], $retries); + self::assertSame($transport->requests[2], $transport->requests[3]); + self::assertContains(TaskDelivery::CAPABILITY, $transport->requests[0]['body']['capabilities']); + self::assertCount(4, $transport->requests); + } + + public function testOfferTransportUsesTheSameProgressAndLogicalPathForAllKinds(): void + { + foreach (['workflow', 'activity', 'query', 'update-validation'] as $kind) { + $path = '/worker/'.$kind.'-tasks/offers'; + self::assertSame($kind, PollTransport::taskKind('POST', $path)); + self::assertSame('/worker/'.$kind.'-tasks/poll', PollTransport::logicalPath($path)); + self::assertNull(PollTransport::taskKind('GET', $path)); + } + } + + public static function kinds(): array { return [['workflow'], ['activity'], ['query']]; } + + private static function poll(Client $client, string $kind): array + { + return match ($kind) { + 'workflow' => $client->pollWorkflowTaskResponse('worker', 'orders', 4, 'fixed-poll'), + 'activity' => $client->pollActivityTaskResponse('worker', 'orders', 4, 'fixed-poll'), + 'query' => $client->pollQueryTaskResponse('worker', 'orders', 4, 'fixed-poll'), + }; + } + + private static function offer(string $kind, string $token = 'opaque-first-packet'): array + { + return ['task' => null, 'poll_status' => 'offered', 'task_offer' => [ + 'schema' => TaskDelivery::SCHEMA, 'task_kind' => $kind, 'task_id' => 'task', + 'token' => $token, 'expires_at_ms' => 1000, + ]]; + } + + private static function claimed(string $kind): array + { + return ['task' => [($kind === 'query' ? 'query_task_id' : 'task_id') => 'task', 'lease_owner' => 'worker', 'task_kind' => $kind]]; + } + + private static function registration(): array + { + return ['registered' => true, 'registration_token' => str_repeat('a', 32), 'heartbeat_interval_seconds' => 10, + 'server_capabilities' => ['task_delivery' => [ + 'schema' => TaskDelivery::SCHEMA, 'supported' => true, 'minimum_protocol_version' => '1.20', + 'worker_capability' => TaskDelivery::CAPABILITY, 'header' => TaskDelivery::HEADER, + 'version' => TaskDelivery::VERSION, 'endpoint_suffix' => '/offers', + 'acceptance_timeout_seconds' => 2, 'unaccepted_offer_reserves_task' => false, + ], 'worker_deregistration_fencing' => ['schema' => 'durable-workflow.v2.worker-deregistration.v1', + 'supported' => true, 'endpoint' => '/worker/registrations/{workerId}/deregister', 'receipt_retention_seconds' => 600]]]; + } +} From 3ab9d59d02f2c81f84a47ccf73a87fd58d12857f Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 25 Oct 2026 08:26:57 +0000 Subject: [PATCH 2/3] Honor fenced terminal worker responses during offer acceptance --- src/Worker/TaskDelivery.php | 9 +++++---- tests/WorkerTaskDeliveryTest.php | 12 ++++++++++++ 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/src/Worker/TaskDelivery.php b/src/Worker/TaskDelivery.php index 9ab67953..590fffb3 100644 --- a/src/Worker/TaskDelivery.php +++ b/src/Worker/TaskDelivery.php @@ -95,10 +95,10 @@ public function poll(string $path, array $body, Closure $send): ?array if ($this->definitiveRefusal($exception)) { $this->resolved($key, $selected); $response = $exception->details ?? []; - if ($exception->reason === 'worker_registration_lost_authority') { - $this->forget($workerId); - } else { + if ($exception->reason === 'task_offer_expired') { $response['poll_status'] = 'empty'; + } else { + $this->forget($workerId); } return $response; @@ -167,7 +167,8 @@ private function definitiveRefusal(ServerException $exception): bool return $exception->status === 409 && $response !== null && !array_is_list($response) && array_key_exists('task', $response) && $response['task'] === null && ($response['reason'] ?? null) === $exception->reason - && in_array($exception->reason, ['task_offer_expired', 'worker_registration_lost_authority'], true); + && (in_array($exception->reason, ['task_offer_expired', 'worker_registration_lost_authority'], true) + || PollResponse::isTerminal($response)); } private function resolved(string $key, ?TaskDeliveryAttempt $selected): void diff --git a/tests/WorkerTaskDeliveryTest.php b/tests/WorkerTaskDeliveryTest.php index 66a81788..5ab31ba0 100644 --- a/tests/WorkerTaskDeliveryTest.php +++ b/tests/WorkerTaskDeliveryTest.php @@ -128,6 +128,18 @@ public function testRegistrationCannotDowngradeAnUnresolvedAcceptance(): void self::assertSame($transport->requests[2], $transport->requests[4]); } + public function testAnAuthoritativeWorkerStopIsNotRetriedAsAnUncertainAcceptance(): void + { + $terminal = ['task' => null, 'poll_status' => 'draining', 'reason' => 'worker_draining']; + $transport = new FakeTransport([self::registration(), self::offer('workflow'), + new TransportException('Worker is draining.', 409, $terminal)]); + $client = new Client('https://server.example', transport: $transport); + $client->registerWorker('worker', 'orders', [], []); + self::assertSame($terminal, self::poll($client, 'workflow')); + self::assertTrue(PollResponse::isTerminal($terminal)); + self::assertCount(3, $transport->requests); + } + public function testReplacementRegistrationDiscardsOnlyTheOldIncarnationsPacket(): void { $replacement = self::registration(); From 13700088131c74a00fb3b656eb5d2c1c8c543d5a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 25 Oct 2026 08:34:27 +0000 Subject: [PATCH 3/3] Separate delivery refusal reconciliation from uncertain retry classification --- src/Worker/TaskDelivery.php | 57 +++++++++++++++++++++---------------- 1 file changed, 32 insertions(+), 25 deletions(-) diff --git a/src/Worker/TaskDelivery.php b/src/Worker/TaskDelivery.php index 590fffb3..22287169 100644 --- a/src/Worker/TaskDelivery.php +++ b/src/Worker/TaskDelivery.php @@ -88,36 +88,18 @@ public function poll(string $path, array $body, Closure $send): ?array if ($response['task'] === null && !PollResponse::isTerminal($response)) { throw new TaskDeliveryUncertain((string) $body['poll_request_id'], 200); } - $this->resolved($key, $selected); - - return $response; } catch (ServerException $exception) { - if ($this->definitiveRefusal($exception)) { - $this->resolved($key, $selected); - $response = $exception->details ?? []; - if ($exception->reason === 'task_offer_expired') { - $response['poll_status'] = 'empty'; - } else { - $this->forget($workerId); + $response = $this->refusedResponse($exception, $workerId); + if ($response === null) { + if ($this->isUncertain($exception, $selected)) { + throw new TaskDeliveryUncertain((string) $body['poll_request_id'], $exception->status, $exception); } - - return $response; - } - if ($exception instanceof TaskDeliveryUncertain) { throw $exception; } - if (($selected !== null && ($exception->status >= 500 || $exception->status === 409 || $exception->status === 429 - || $exception->reason === 'invalid_task_delivery_response' - || in_array($exception->status, [0, 404, 405], true))) - || ($selected === null && in_array($exception->status, [404, 405], true))) { - if (PollResponse::isTransientPayloadFailure($exception)) { - throw $exception; - } - throw new TaskDeliveryUncertain((string) $body['poll_request_id'], $exception->status, $exception); - } - - throw $exception; } + $this->resolved($key, $selected); + + return $response; } /** @@ -171,6 +153,31 @@ private function definitiveRefusal(ServerException $exception): bool || PollResponse::isTerminal($response)); } + /** @return array|null */ + private function refusedResponse(ServerException $exception, string $workerId): ?array + { + if (!$this->definitiveRefusal($exception)) { + return null; + } + $response = $exception->details ?? []; + if ($exception->reason === 'task_offer_expired') { + $response['poll_status'] = 'empty'; + } else { + $this->forget($workerId); + } + + return $response; + } + + private function isUncertain(ServerException $exception, ?TaskDeliveryAttempt $selected): bool + { + return !$exception instanceof TaskDeliveryUncertain && !PollResponse::isTransientPayloadFailure($exception) + && (($selected !== null && ($exception->status >= 500 || $exception->status === 409 || $exception->status === 429 + || $exception->reason === 'invalid_task_delivery_response' + || in_array($exception->status, [0, 404, 405], true))) + || ($selected === null && in_array($exception->status, [404, 405], true))); + } + private function resolved(string $key, ?TaskDeliveryAttempt $selected): void { if (($this->pending[$key] ?? null) === $selected) {