Prevent abandoned polls from claiming through explicit task offers

app/Http/Controllers/Api/ActivityTaskController.php

@@ -23,6 +23,7 @@
 use App\Support\WorkerProtocol;
 use App\Support\WorkerProtocolMutationRetrier;
 use App\Support\WorkerSessionRegistry;
+use App\Support\WorkerTaskDelivery;
 use App\Support\WorkflowObservabilityPoll;
 use Carbon\CarbonImmutable;
 use Carbon\CarbonInterface;
@@ -133,6 +134,7 @@ public function poll(Request $request): JsonResponse
                 ),
                 timeoutSeconds: $timeoutSeconds,
                 protocolVersion: WorkerProtocol::requestVersion($request),
+                delivery: WorkerTaskDelivery::forRequest($request, $worker, $namespace, $validated['task_queue'], $validated['poll_request_id'] ?? null),
             );
         } catch (\Throwable $exception) {
             if ($exception instanceof InvalidArgumentException
@@ -164,6 +166,9 @@ public function poll(Request $request): JsonResponse
 
             throw $exception;
         }
+        if (is_array($poll['task']['delivery_offer'] ?? null)) {
+            return WorkerProtocol::json(['task' => null, 'task_offer' => $poll['task']['delivery_offer'], 'poll_status' => 'offered']);
+        }
         $claim = $poll['task'] ?? null;
 
         if (is_array($claim)) {

app/Http/Controllers/Api/WorkerController.php

@@ -42,6 +42,7 @@
 use App\Support\WorkerPollFence;
 use App\Support\WorkerProtocol;
 use App\Support\WorkerProtocolMutationRetrier;
+use App\Support\WorkerTaskDelivery;
 use App\Support\WorkerTerminalEventAttribution;
 use App\Support\WorkflowHistoryPageToken;
 use App\Support\WorkflowMetadataCapabilityPolicy;
@@ -1391,6 +1392,7 @@ public function pollWorkflowTasks(Request $request): JsonResponse
                 ),
                 timeoutSeconds: $timeoutSeconds,
                 taskKinds: $taskKinds,
+                delivery: WorkerTaskDelivery::forRequest($request, $worker, $namespace, $validated['task_queue'], $validated['poll_request_id'] ?? null),
             );
         } catch (\Throwable $exception) {
             if ($exception instanceof \InvalidArgumentException
@@ -1448,6 +1450,9 @@ public function pollWorkflowTasks(Request $request): JsonResponse
             throw $exception;
         }
 
+        if (is_array($poll['task']['delivery_offer'] ?? null)) {
+            return WorkerProtocol::json(['task' => null, 'task_offer' => $poll['task']['delivery_offer'], 'poll_status' => 'offered']);
+        }
         $task = $this->formatTaskHistoryPagination($poll['task'] ?? null);
 
         // PHP SDKs before 2.2.0 reject the valid StartAccepted prefix when
@@ -4223,6 +4228,7 @@ public function pollQueryTasks(Request $request): JsonResponse
                 $worker,
                 $validated['poll_request_id'] ?? null,
                 $timeoutSeconds,
+                WorkerTaskDelivery::forRequest($request, $worker, $namespace, $validated['task_queue'], $validated['poll_request_id'] ?? null),
             );
         } catch (QueryTaskQueueUnavailableException $exception) {
             return WorkerProtocol::json([
@@ -4255,6 +4261,10 @@ public function pollQueryTasks(Request $request): JsonResponse
             throw $exception;
         }
 
+        if (isset($poll['task']['delivery_offer'])) {
+            return WorkerProtocol::json(['task' => null, 'task_offer' => $poll['task']['delivery_offer'], 'poll_status' => 'offered']);
+        }
+
         $pollContext = [
             'namespace' => $namespace,
             'worker_id' => $worker->worker_id,
@@ -4264,18 +4274,24 @@ public function pollQueryTasks(Request $request): JsonResponse
             'query_task_id' => $poll['task']['query_task_id'] ?? null,
         ];
 
-        if ($poll['poll_status'] === 'empty') {
+        $this->logQueryPollResponse($pollContext);
+
+        return WorkerProtocol::json([
+            'task' => WorkflowObservabilityPoll::filterOptionalDelivered($poll['task']),
+            'poll_status' => $poll['poll_status'],
+        ]);
+    }
+
+    private function logQueryPollResponse(array $pollContext): void
+    {
+        if ($pollContext['poll_status'] === 'empty') {
             if (config('app.debug')) {
                 Log::debug('worker_query_poll_response_ready', $pollContext);
             }
         } else {
             Log::info('worker_query_poll_response_ready', $pollContext);
         }
 
-        return WorkerProtocol::json([
-            'task' => WorkflowObservabilityPoll::filterOptionalDelivered($poll['task']),
-            'poll_status' => $poll['poll_status'],
-        ]);
     }
 
     public function queryTaskStatus(Request $request, string $queryTaskId): JsonResponse
@@ -4477,6 +4493,7 @@ public function pollUpdateValidationTasks(Request $request): JsonResponse
         $validated = $request->validate([
             'worker_id' => ['required', 'string'],
             'task_queue' => ['required', 'string'],
+            'poll_request_id' => ['nullable', 'string', 'max:255'],
             'timeout_seconds' => [
                 'nullable',
                 'integer',
@@ -4512,6 +4529,7 @@ public function pollUpdateValidationTasks(Request $request): JsonResponse
                 $namespace,
                 $worker,
                 isset($validated['timeout_seconds']) ? (int) $validated['timeout_seconds'] : null,
+                WorkerTaskDelivery::forRequest($request, $worker, $namespace, $validated['task_queue'], $validated['poll_request_id'] ?? null),
             );
         } catch (LongPollCapacityExhaustedException $exception) {
             return WorkerPollBackpressure::response(
@@ -4532,6 +4550,10 @@ public function pollUpdateValidationTasks(Request $request): JsonResponse
             );
         }
 
+        if (isset($poll['task']['delivery_offer'])) {
+            return WorkerProtocol::json(['task' => null, 'task_offer' => $poll['task']['delivery_offer'], 'poll_status' => 'offered']);
+        }
+
         return WorkerProtocol::json($poll);
     }
 

app/Http/Middleware/WorkerPollResponseProgress.php

@@ -15,6 +15,10 @@ final class WorkerPollResponseProgress
         'api/worker/activity-tasks/poll',
         'api/worker/query-tasks/poll',
         'api/worker/update-validation-tasks/poll',
+        'api/worker/workflow-tasks/offers',
+        'api/worker/activity-tasks/offers',
+        'api/worker/update-validation-tasks/offers',
+        'api/worker/query-tasks/offers',
     ];
 
     public function handle(Request $request, Closure $next): Response

app/Models/WorkflowUpdateValidationTask.php

@@ -37,6 +37,7 @@ protected function casts(): array
             'attempt_count' => 'integer',
             'command_context' => 'array',
             'validation_errors' => 'array',
+            'delivery_receipt' => 'array',
             'lease_expires_at' => 'datetime',
             'approved_at' => 'datetime',
             'rejected_at' => 'datetime',

app/Support/ActivityTaskPoller.php

@@ -47,6 +47,7 @@ public function poll(
         bool $workerSessionsAvailable = true,
         ?int $timeoutSeconds = null,
         ?string $protocolVersion = null,
+        ?WorkerTaskDelivery $delivery = null,
     ): array {
         $pollRequestId = $this->nonEmptyString($pollRequestId);
 
@@ -57,6 +58,21 @@ public function poll(
             ];
         }
 
+        if ($delivery !== null) {
+            $result = $this->performPoll(
+                namespace: $namespace, taskQueue: $taskQueue, leaseOwner: $leaseOwner,
+                buildId: $buildId, worker: $worker, pollRequestId: $pollRequestId,
+                supportedActivityTypes: $supportedActivityTypes, workerSessionsAvailable: $workerSessionsAvailable,
+                timeoutSeconds: $delivery->accepting() ? 0 : $timeoutSeconds,
+                protocolVersion: $protocolVersion, delivery: $delivery,
+            );
+            if (is_array($result['task'] ?? null) && ! isset($result['task']['delivery_offer'])) {
+                $delivery->assertReplay($result['task']['task_id']);
+            }
+
+            return $result;
+        }
+
         if ($pollRequestId !== null) {
             $task = $this->activeLeasedTaskForWorker(
                 namespace: $namespace,
@@ -440,6 +456,7 @@ private function performPoll(
         bool $workerSessionsAvailable = true,
         ?int $timeoutSeconds = null,
         ?string $protocolVersion = null,
+        ?WorkerTaskDelivery $delivery = null,
     ): array {
         $limit = max(10, max(1, (int) config('server.polling.max_tasks_per_poll', 1)) * 10);
         $nextProbeAt = null;
@@ -464,12 +481,14 @@ function () use (
                 $workerSessionsAvailable,
                 $limit,
                 $workerPollFence,
+                $delivery,
                 &$nextProbeAt,
                 &$resolvedResult,
             ): ?array {
                 app(StoragePressure::class)->requireNewWork();
 
-                if (! WorkerPollFence::isCurrent($workerPollFence)) {
+                $delivery?->lockWorker();
+                if ($delivery === null && ! WorkerPollFence::isCurrent($workerPollFence)) {
                     $resolvedResult = [
                         'task' => null,
                         'poll_status' => 'stale_worker_registration',
@@ -490,6 +509,7 @@ function () use (
                     $supportedActivityTypes,
                     $workerSessionsAvailable,
                     $workerPollFence,
+                    $delivery,
                 );
                 $nextProbeAt = $resolvedResult['next_probe_at'] ?? null;
 
@@ -694,6 +714,7 @@ private function nextTask(
         array $supportedActivityTypes = [],
         bool $workerSessionsAvailable = true,
         array $workerPollFence = [],
+        ?WorkerTaskDelivery $delivery = null,
     ): array {
         return $this->withWorkerCompatibility(
             $namespace,
@@ -709,6 +730,7 @@ function () use (
                 $supportedActivityTypes,
                 $workerSessionsAvailable,
                 $workerPollFence,
+                $delivery,
             ): array {
                 if ($pollRequestId !== null) {
                     $task = $this->activeLeasedTaskForWorker(
@@ -751,6 +773,7 @@ function () use (
                                 $supportedActivityTypes,
                                 $workerSessionsAvailable,
                                 $workerPollFence,
+                                $delivery,
                             ),
                         ),
                     );
@@ -821,8 +844,9 @@ private function claimReadyTask(
         array $supportedActivityTypes = [],
         bool $workerSessionsAvailable = true,
         array $workerPollFence = [],
+        ?WorkerTaskDelivery $delivery = null,
     ): ?array {
-        $readyTasks = $this->bridge->poll(
+        $readyTasks = $delivery?->accepting() ? $delivery->candidates('activity') : $this->bridge->poll(
             connection: null,
             queue: $taskQueue,
             limit: $limit,
@@ -881,6 +905,15 @@ private function claimReadyTask(
                 continue;
             }
 
+            if ($delivery !== null && ! $delivery->accepting()) {
+                $offer = $delivery->offerTask($readyTask, 'activity');
+                if ($offer !== null) {
+                    return $offer;
+                }
+
+                continue;
+            }
+
             try {
                 $claim = DB::transaction(function () use (
                     $namespace,
@@ -892,8 +925,10 @@ private function claimReadyTask(
                     $pollRequestId,
                     $taskQueue,
                     $buildId,
+                    $delivery,
                 ): ?array {
-                    if ($workerPollFence !== [] && ! WorkerPollFence::isCurrentForUpdate($workerPollFence)) {
+                    $delivery?->lockWorker();
+                    if ($delivery === null && $workerPollFence !== [] && ! WorkerPollFence::isCurrentForUpdate($workerPollFence)) {
                         throw new ActivityTaskClaimRolledBack;
                     }
 
@@ -906,6 +941,7 @@ private function claimReadyTask(
                         $pollRequestId,
                     );
 
+                    $delivery?->assertClaimable($taskId, 'activity');
                     $claim = $this->claimStatus($taskId, $leaseOwner);
 
                     if (($claim['claimed'] ?? false) !== true) {
@@ -942,6 +978,7 @@ private function claimReadyTask(
                         $pollRequestId,
                         $workerPollFence,
                     );
+                    $delivery?->bindAccepted($taskId);
 
                     return $claim;
                 });

app/Support/BackendUnavailable.php

@@ -72,10 +72,10 @@ public static function workerResponse(Request $request): ?JsonResponse
         }
 
         $operation = match (true) {
-            $request->is('api/worker/workflow-tasks/poll') => 'poll_workflow_task',
-            $request->is('api/worker/activity-tasks/poll') => 'poll_activity_task',
-            $request->is('api/worker/query-tasks/poll') => 'poll_query_task',
-            $request->is('api/worker/update-validation-tasks/poll') => 'poll_update_validation_task',
+            $request->is('api/worker/workflow-tasks/poll', 'api/worker/workflow-tasks/offers') => 'poll_workflow_task',
+            $request->is('api/worker/activity-tasks/poll', 'api/worker/activity-tasks/offers') => 'poll_activity_task',
+            $request->is('api/worker/query-tasks/poll', 'api/worker/query-tasks/offers') => 'poll_query_task',
+            $request->is('api/worker/update-validation-tasks/poll', 'api/worker/update-validation-tasks/offers') => 'poll_update_validation_task',
             $request->is('api/worker/register') => 'register_worker',
             $request->isMethod('POST') && $request->is('api/worker/registrations/*/deregister') => 'deregister_worker',
             $request->is('api/worker/heartbeat') => 'heartbeat_worker',

app/Support/TaskDeliveryOffer.php

@@ -0,0 +1,61 @@
+<?php
+
+namespace App\Support;
+
+use Illuminate\Contracts\Encryption\DecryptException;
+use Illuminate\Support\Facades\Crypt;
+use JsonException;
+
+/** Bounded, sealed offer packets; these packets never grant execution. */
+final class TaskDeliveryOffer
+{
+    public static function encode(array $data): string
+    {
+        $token = Crypt::encryptString(json_encode($data, JSON_THROW_ON_ERROR));
+        if (strlen($token) > 8192) {
+            throw new TaskDeliveryOfferRejected('task_offer_too_large');
+        }
+
+        return $token;
+    }
+
+    public static function decode(mixed $token): array
+    {
+        try {
+            $data = is_string($token) && strlen($token) <= 8192
+                ? json_decode(Crypt::decryptString($token), true, flags: JSON_THROW_ON_ERROR) : null;
+        } catch (DecryptException|JsonException) {
+            $data = null;
+        }
+        if (! is_array($data) || ($data['schema'] ?? null) !== WorkerTaskDelivery::SCHEMA
+            || ! is_string($data['task_id'] ?? null) || ! is_string($data['task_kind'] ?? null)
+            || ! is_array($data['worker'] ?? null) || ! is_array($data['candidate'] ?? null)
+            || ! is_int($data['offered_attempt'] ?? null) || ! is_int($data['expires_at_ms'] ?? null)
+            || ! is_string($data['fingerprint'] ?? null) || ! is_string($data['delivery_id'] ?? null)) {
+            throw new TaskDeliveryOfferRejected('task_offer_invalid');
+        }
+
+        return $data;
+    }
+
+    public static function identifier(array $data): string
+    {
+        return hash('sha256', json_encode([
+            $data['schema'], $data['namespace'], $data['queue'], $data['route'], $data['poll_request_id'],
+            $data['task_kinds'], $data['worker']['registration_id'], $data['worker']['registration_incarnation'],
+            $data['task_id'], $data['task_kind'], $data['offered_attempt'], $data['fingerprint'],
+        ], JSON_THROW_ON_ERROR));
+    }
+
+    public static function assertUnexpired(array $data): void
+    {
+        if (now()->getTimestampMs() >= $data['expires_at_ms']) {
+            throw new TaskDeliveryOfferRejected('task_offer_expired');
+        }
+    }
+
+    public static function acceptanceMilliseconds(): int
+    {
+        return max(1000, min(10000, (int) config('server.polling.task_offer_accept_seconds', 2) * 1000));
+    }
+}

app/Support/TaskDeliveryOfferRejected.php

@@ -0,0 +1,13 @@
+<?php
+
+namespace App\Support;
+
+use RuntimeException;
+
+final class TaskDeliveryOfferRejected extends RuntimeException
+{
+    public function __construct(public readonly string $reason)
+    {
+        parent::__construct('The worker task offer could not be accepted.');
+    }
+}

app/Support/WorkerProtocol.php

@@ -510,6 +510,17 @@ public static function serverCapabilities(): array
             'query_task_poll_request_idempotency' => true,
             'query_task_authority' => true,
             'query_task_timeouts' => self::queryTaskTimeouts(),
+            'task_delivery' => [
+                'schema' => WorkerTaskDelivery::SCHEMA,
+                'supported' => true,
+                'minimum_protocol_version' => '1.20',
+                'worker_capability' => WorkerTaskDelivery::CAPABILITY,
+                'header' => WorkerTaskDelivery::HEADER,
+                'version' => WorkerTaskDelivery::VERSION,
+                'endpoint_suffix' => '/offers',
+                'acceptance_timeout_seconds' => max(1, min(10, (int) config('server.polling.task_offer_accept_seconds', 2))),
+                'unaccepted_offer_reserves_task' => false,
+            ],
             'update_validation_tasks' => true,
             'synchronous_update_validation' => [
                 'supported' => true,