Detect worker poll disconnects and report queue cutoffs

.github/workflows/server-perf-soak.yml

@@ -85,8 +85,8 @@ jobs:
           DW_PERF_MAX_SERVER_MEMORY_MB: "1024"
           DW_PERF_MAX_POLLING_KEYS: "2048"
           DW_PERF_MAX_FINAL_POLLING_KEYS: "0"
-          DW_PERF_MAX_SERVER_CACHE_KEYS_BY_POLICY: '{"polling_cache_availability_probe":0,"workflow_task_poll_requests":2048,"activity_task_poll_requests":2048,"query_task_poll_requests":2048,"sqlite_worker_poll_claim_gate":1,"long_poll_signals":2048,"long_poll_wait_slots":64,"workflow_query_tasks":128,"task_queue_admission_locks":256,"task_queue_dispatch_counters":256,"namespace_request_admission":256,"shared_service_boundary_admission":256,"namespace_durable_state_quota_rejections":256,"runtime_external_payload_quota_rejections":256,"workflow_task_expired_lease_recovery":256,"history_retention_inline":128,"worker_compatibility_heartbeat":128,"readiness_probe":32}'
-          DW_PERF_MAX_FINAL_SERVER_CACHE_KEYS_BY_POLICY: '{"polling_cache_availability_probe":0,"workflow_task_poll_requests":0,"activity_task_poll_requests":0,"query_task_poll_requests":0,"sqlite_worker_poll_claim_gate":0,"long_poll_signals":0,"long_poll_wait_slots":0,"workflow_query_tasks":0,"task_queue_admission_locks":0,"task_queue_dispatch_counters":0,"namespace_request_admission":0,"shared_service_boundary_admission":0,"namespace_durable_state_quota_rejections":0,"runtime_external_payload_quota_rejections":0,"workflow_task_expired_lease_recovery":0,"history_retention_inline":0,"worker_compatibility_heartbeat":0,"readiness_probe":0}'
+          DW_PERF_MAX_SERVER_CACHE_KEYS_BY_POLICY: '{"worker_poll_diagnostics":0,"polling_cache_availability_probe":0,"workflow_task_poll_requests":2048,"activity_task_poll_requests":2048,"query_task_poll_requests":2048,"sqlite_worker_poll_claim_gate":1,"long_poll_signals":2048,"long_poll_wait_slots":64,"workflow_query_tasks":128,"task_queue_admission_locks":256,"task_queue_dispatch_counters":256,"namespace_request_admission":256,"shared_service_boundary_admission":256,"namespace_durable_state_quota_rejections":256,"runtime_external_payload_quota_rejections":256,"workflow_task_expired_lease_recovery":256,"history_retention_inline":128,"worker_compatibility_heartbeat":128,"readiness_probe":32}'
+          DW_PERF_MAX_FINAL_SERVER_CACHE_KEYS_BY_POLICY: '{"worker_poll_diagnostics":0,"polling_cache_availability_probe":0,"workflow_task_poll_requests":0,"activity_task_poll_requests":0,"query_task_poll_requests":0,"sqlite_worker_poll_claim_gate":0,"long_poll_signals":0,"long_poll_wait_slots":0,"workflow_query_tasks":0,"task_queue_admission_locks":0,"task_queue_dispatch_counters":0,"namespace_request_admission":0,"shared_service_boundary_admission":0,"namespace_durable_state_quota_rejections":0,"runtime_external_payload_quota_rejections":0,"workflow_task_expired_lease_recovery":0,"history_retention_inline":0,"worker_compatibility_heartbeat":0,"readiness_probe":0}'
           DW_PERF_MAX_SERVER_MEMORY_SLOPE_MB_HOUR: "128"
           DW_PERF_REMOTE_WRITE_ENABLED: ${{ github.event_name != 'workflow_dispatch' || inputs.remote_write }}
           DW_PERF_REMOTE_WRITE_URL: ${{ vars.DW_PERF_REMOTE_WRITE_URL }}

.github/workflows/server-perf.yml

@@ -24,6 +24,7 @@ on:
       - "app/Support/ServerReadiness.php"
       - "app/Support/RemoteScheduleStarter.php"
       - "app/Support/ServerPollingCache.php"
+      - "app/Support/WorkerPollDiagnostics.php"
       - "app/Support/SharedServiceBoundaryPolicy.php"
       - "app/Support/TaskQueueAdmission.php"
       - "app/Support/NamespaceRequestAdmission.php"
@@ -82,6 +83,7 @@ on:
       - "app/Support/ServerReadiness.php"
       - "app/Support/RemoteScheduleStarter.php"
       - "app/Support/ServerPollingCache.php"
+      - "app/Support/WorkerPollDiagnostics.php"
       - "app/Support/SharedServiceBoundaryPolicy.php"
       - "app/Support/TaskQueueAdmission.php"
       - "app/Support/NamespaceRequestAdmission.php"
@@ -272,8 +274,8 @@ jobs:
           # limit per poll. Allow the 1,000 synthetic starts plus SDK canary;
           # total keys and zero-after-drain remain independently enforced.
           DW_PERF_MIN_SAMPLE_COVERAGE: "0.75"
-          DW_PERF_MAX_SERVER_CACHE_KEYS_BY_POLICY: '{"polling_cache_availability_probe":0,"workflow_task_poll_requests":3072,"activity_task_poll_requests":512,"query_task_poll_requests":512,"sqlite_worker_poll_claim_gate":1,"long_poll_signals":2048,"workflow_query_tasks":64,"task_queue_admission_locks":128,"task_queue_dispatch_counters":128,"namespace_request_admission":128,"shared_service_boundary_admission":128,"namespace_durable_state_quota_rejections":128,"runtime_external_payload_quota_rejections":128,"workflow_task_expired_lease_recovery":1536,"history_retention_inline":64,"worker_compatibility_heartbeat":64,"readiness_probe":16}'
-          DW_PERF_MAX_FINAL_SERVER_CACHE_KEYS_BY_POLICY: '{"polling_cache_availability_probe":0,"workflow_task_poll_requests":0,"activity_task_poll_requests":0,"query_task_poll_requests":0,"sqlite_worker_poll_claim_gate":0,"long_poll_signals":0,"workflow_query_tasks":0,"task_queue_admission_locks":0,"task_queue_dispatch_counters":0,"namespace_request_admission":0,"shared_service_boundary_admission":0,"namespace_durable_state_quota_rejections":0,"runtime_external_payload_quota_rejections":0,"workflow_task_expired_lease_recovery":0,"history_retention_inline":0,"worker_compatibility_heartbeat":0,"readiness_probe":0}'
+          DW_PERF_MAX_SERVER_CACHE_KEYS_BY_POLICY: '{"worker_poll_diagnostics":0,"polling_cache_availability_probe":0,"workflow_task_poll_requests":3072,"activity_task_poll_requests":512,"query_task_poll_requests":512,"sqlite_worker_poll_claim_gate":1,"long_poll_signals":2048,"workflow_query_tasks":64,"task_queue_admission_locks":128,"task_queue_dispatch_counters":128,"namespace_request_admission":128,"shared_service_boundary_admission":128,"namespace_durable_state_quota_rejections":128,"runtime_external_payload_quota_rejections":128,"workflow_task_expired_lease_recovery":1536,"history_retention_inline":64,"worker_compatibility_heartbeat":64,"readiness_probe":16}'
+          DW_PERF_MAX_FINAL_SERVER_CACHE_KEYS_BY_POLICY: '{"worker_poll_diagnostics":0,"polling_cache_availability_probe":0,"workflow_task_poll_requests":0,"activity_task_poll_requests":0,"query_task_poll_requests":0,"sqlite_worker_poll_claim_gate":0,"long_poll_signals":0,"workflow_query_tasks":0,"task_queue_admission_locks":0,"task_queue_dispatch_counters":0,"namespace_request_admission":0,"shared_service_boundary_admission":0,"namespace_durable_state_quota_rejections":0,"runtime_external_payload_quota_rejections":0,"workflow_task_expired_lease_recovery":0,"history_retention_inline":0,"worker_compatibility_heartbeat":0,"readiness_probe":0}'
           DW_PERF_COMPOSE_PROJECT: server-perf-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}
           RUNNER_ENVIRONMENT: "github-hosted"
         if: ${{ always() && success() }}

app/Http/Controllers/Api/TaskQueueController.php

@@ -10,6 +10,7 @@
 use App\Support\TaskQueueAdmission;
 use App\Support\TaskQueueBuildIdRolloutSnapshot;
 use App\Support\TaskQueuePriorityFairnessSurface;
+use App\Support\WorkerPollDiagnostics;
 use App\Support\WorkflowDefinitionRouting;
 use App\Support\WorkflowQueryTaskBroker;
 use Illuminate\Http\JsonResponse;
@@ -30,6 +31,7 @@ public function __construct(
         private readonly TaskQueuePriorityFairnessSurface $priorityFairness,
         private readonly DeploymentLifecycleService $deployments,
         private readonly WorkflowDefinitionRouting $definitionRouting,
+        private readonly WorkerPollDiagnostics $pollDiagnostics,
     ) {}
 
     public function index(Request $request): JsonResponse
@@ -439,6 +441,12 @@ private function withAdmission(string $namespace, array $payload): array
             'query_tasks' => $this->queryTasks->queueAdmission($namespace, $taskQueue),
         ];
 
+        $payload['poll_transport'] = $this->pollDiagnostics->snapshot($namespace, $taskQueue);
+        if ($payload['poll_transport']['status'] === 'warning') {
+            $payload['repair']['needs_attention'] = true;
+            $payload['repair']['transport_warnings'] = [$payload['poll_transport']['last_disconnect']];
+        }
+
         return $payload;
     }
 

app/Http/Middleware/WorkerPollResponseProgress.php

@@ -0,0 +1,36 @@
+<?php
+
+namespace App\Http\Middleware;
+
+use App\Support\LongPollResponseProgress;
+use Closure;
+use Illuminate\Http\Request;
+use Symfony\Component\HttpFoundation\AcceptHeader;
+use Symfony\Component\HttpFoundation\Response;
+
+final class WorkerPollResponseProgress
+{
+    private const PATHS = [
+        'api/worker/workflow-tasks/poll',
+        'api/worker/activity-tasks/poll',
+        'api/worker/query-tasks/poll',
+        'api/worker/update-validation-tasks/poll',
+    ];
+
+    public function handle(Request $request, Closure $next): Response
+    {
+        $accept = AcceptHeader::fromString(strtolower($request->headers->get('Accept', '')));
+        $transport = $accept->get(LongPollResponseProgress::MEDIA_TYPE);
+
+        if (! $request->isMethod('POST') || ! in_array($request->path(), self::PATHS, true)
+            || $transport === null || $transport->getQuality() <= 0
+            || $transport->getQuality() < ($accept->get('application/json')?->getQuality() ?? 0)) {
+            return $next($request);
+        }
+
+        $progress = app()->make(LongPollResponseProgress::class, ['request' => $request]);
+        $request->attributes->set(LongPollResponseProgress::REQUEST_ATTRIBUTE, $progress);
+
+        return $progress->response(static fn (): Response => $next($request));
+    }
+}

app/Support/ActivityTaskPollRequestStore.php

@@ -72,6 +72,7 @@ public function waitForResult(
     ): array {
         $pendingKey = $this->pendingKey($namespace, $taskQueue, $buildId, $leaseOwner, $pollRequestId);
         $deadline = microtime(true) + (($timeoutMilliseconds ?? $this->waitTimeoutMilliseconds()) / 1000);
+        $progress = LongPollResponseProgress::current();
 
         while (microtime(true) < $deadline) {
             $result = $this->result($namespace, $taskQueue, $buildId, $leaseOwner, $pollRequestId);
@@ -88,6 +89,7 @@ public function waitForResult(
                 ];
             }
 
+            $progress?->checkpoint();
             $this->pause($this->pollIntervalMilliseconds());
         }
 

app/Support/LongPollResponseProgress.php

@@ -0,0 +1,189 @@
+<?php
+
+namespace App\Support;
+
+use Closure;
+use Fiber;
+use Illuminate\Http\Request;
+use Symfony\Component\HttpFoundation\JsonResponse;
+use Symfony\Component\HttpFoundation\Response;
+use Symfony\Component\HttpFoundation\StreamedResponse;
+
+class LongPollResponseProgress
+{
+    public const MEDIA_TYPE = 'application/vnd.durable-workflow.poll.v1+json';
+
+    public const SCHEMA = 'durable-workflow.poll-transport.v1';
+
+    public const REQUEST_ATTRIBUTE = '_server_poll_response_progress';
+
+    private ?Fiber $fiber = null;
+
+    private ?int $waitSeconds = null;
+
+    private float $lastProgressAt = 0;
+
+    private readonly float $startedAt;
+
+    private bool $abandonmentRecorded = false;
+
+    public function __construct(private readonly Request $request)
+    {
+        $this->startedAt = hrtime(true) / 1_000_000_000;
+    }
+
+    public static function current(): ?self
+    {
+        if (! app()->bound('request')) {
+            return null;
+        }
+
+        $progress = request()->attributes->get(self::REQUEST_ATTRIBUTE);
+
+        return $progress instanceof self ? $progress : null;
+    }
+
+    public function beginWait(int $waitSeconds): void
+    {
+        $this->waitSeconds = $waitSeconds;
+        $this->checkpoint();
+    }
+
+    public function checkpoint(): void
+    {
+        if ($this->fiber === null || Fiber::getCurrent() !== $this->fiber
+            || hrtime(true) / 1_000_000_000 - $this->lastProgressAt < 1) {
+            return;
+        }
+
+        $this->waitSeconds ??= max(0, (int) $this->request->input('timeout_seconds', config('server.polling.timeout', 30)));
+        $this->lastProgressAt = hrtime(true) / 1_000_000_000;
+        Fiber::suspend();
+    }
+
+    public function response(Closure $handler): Response
+    {
+        $this->fiber = new Fiber($handler);
+        $this->fiber->start();
+        $status = $this->fiber->isTerminated() ? $this->terminalResponse()->getStatusCode() : 200;
+
+        return new StreamedResponse($this->send(...), $status, [
+            'Content-Type' => self::MEDIA_TYPE,
+            'Cache-Control' => 'no-store',
+            'X-Accel-Buffering' => 'no',
+            'Vary' => 'Accept',
+        ]);
+    }
+
+    private function send(): void
+    {
+        $previous = ignore_user_abort(true);
+
+        try {
+            while (! $this->fiber->isTerminated()) {
+                echo "\n";
+                $this->flushOutput();
+
+                if ($this->clientDisconnected()) {
+                    $observedSeconds = hrtime(true) / 1_000_000_000 - $this->startedAt;
+                    try {
+                        $this->fiber->throw(new PollClientDisconnected);
+                    } finally {
+                        // Logging and optional cache I/O follow exact wait
+                        // cleanup, so a slow observer cannot retain capacity.
+                        $this->recordAbandonment($observedSeconds);
+                    }
+
+                    return;
+                }
+
+                $this->fiber->resume();
+            }
+
+            $response = $this->terminalResponse();
+            $headers = array_filter([
+                'retry-after' => $response->headers->get('Retry-After'),
+                strtolower(WorkerProtocol::HEADER) => $response->headers->get(WorkerProtocol::HEADER),
+            ], static fn (?string $value): bool => $value !== null);
+
+            // Keep the already encoded worker body intact, including numeric
+            // representation and payload envelopes. Progress is JSON whitespace.
+            echo '{"schema":"'.self::SCHEMA.'","status":'.$response->getStatusCode().',"headers":';
+            echo json_encode((object) $headers, JSON_THROW_ON_ERROR);
+            echo ',"body":', $response->getContent(), '}';
+            $this->flushOutput();
+
+            if ($this->clientDisconnected()) {
+                $this->recordAbandonment();
+            }
+        } catch (PollClientDisconnected) {
+            // The exact wait owner has unwound through its normal finally.
+        } finally {
+            try {
+                if ($this->fiber->isSuspended()) {
+                    try {
+                        $this->fiber->throw(new PollClientDisconnected);
+                    } catch (PollClientDisconnected) {
+                        // Do not leave an admitted wait suspended on output failure.
+                    }
+                }
+            } finally {
+                ignore_user_abort($previous);
+            }
+        }
+    }
+
+    private function terminalResponse(): JsonResponse
+    {
+        $response = $this->fiber->getReturn();
+
+        if ($response instanceof JsonResponse) {
+            return $response;
+        }
+
+        logger()->error('Worker poll returned an unsupported progress-transport response.');
+
+        return new JsonResponse([
+            'task' => null,
+            'reason' => 'worker_poll_transport_unavailable',
+            'retryable' => true,
+        ], 500);
+    }
+
+    private function recordAbandonment(?float $observedSeconds = null): void
+    {
+        if ($this->abandonmentRecorded || $this->waitSeconds === null) {
+            return;
+        }
+
+        $this->abandonmentRecorded = true;
+        $observedSeconds ??= hrtime(true) / 1_000_000_000 - $this->startedAt;
+        logger()->warning('Worker long-poll client disconnected before its response was delivered.', [
+            'namespace' => $this->request->attributes->get('namespace'),
+            'task_queue' => $this->request->input('task_queue'),
+            'observed_seconds' => round($observedSeconds, 3),
+            'configured_wait_seconds' => $this->waitSeconds,
+        ]);
+        app(WorkerPollDiagnostics::class)->recordDisconnect(
+            (string) $this->request->attributes->get('namespace'),
+            (string) $this->request->input('task_queue'),
+            basename(dirname($this->request->path())),
+            $observedSeconds,
+            $this->waitSeconds,
+        );
+    }
+
+    protected function clientDisconnected(): bool
+    {
+        return connection_aborted() !== 0;
+    }
+
+    protected function flushOutput(): void
+    {
+        if (ob_get_level() > 0) {
+            ob_flush();
+        }
+
+        flush();
+    }
+}

app/Support/LongPoller.php

@@ -87,6 +87,9 @@ public function until(
         );
 
         try {
+            $progress = LongPollResponseProgress::current();
+            $progress?->beginWait($timeoutSeconds);
+
             while (microtime(true) < $deadline) {
                 $now = microtime(true);
                 $sleepUntil = min(
@@ -97,6 +100,7 @@ public function until(
                 $sleepMilliseconds = max(1, (int) ceil(max(0, $sleepUntil - $now) * 1000));
 
                 $this->pause($sleepMilliseconds);
+                $progress?->checkpoint();
 
                 $now = microtime(true);
                 $wakeChanged = $wakeChannels !== [] && $this->signals->changed($wakeSnapshot);

app/Support/PollClientDisconnected.php

@@ -0,0 +1,7 @@
+<?php
+
+namespace App\Support;
+
+use RuntimeException;
+
+final class PollClientDisconnected extends RuntimeException {}

app/Support/QueryTaskPollRequestStore.php

@@ -117,6 +117,7 @@ public function waitForResult(
     ): array {
         $pendingKey = $this->pendingKey($namespace, $taskQueue, $buildId, $leaseOwner, $pollRequestId);
         $deadline = microtime(true) + (($timeoutMilliseconds ?? $this->waitTimeoutMilliseconds()) / 1000);
+        $progress = LongPollResponseProgress::current();
 
         while (microtime(true) < $deadline) {
             $result = $this->result($namespace, $taskQueue, $buildId, $leaseOwner, $pollRequestId);
@@ -133,6 +134,7 @@ public function waitForResult(
                 ];
             }
 
+            $progress?->checkpoint();
             $this->pause($this->pollIntervalMilliseconds());
         }