Preserve worker poll progress and diagnose early cutoffs

CHANGELOG.md

@@ -6,6 +6,17 @@ project follows [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+## [3.5.1] - 2026-10-25
+
+### Fixed
+
+- Negotiate worker poll progress with supporting Servers, preserving terminal
+  status, refusal details and retry headers. Ordinary JSON remains compatible.
+  Progress bytes do not extend the complete cURL request deadline.
+- Report early worker poll connection failures and gateway cutoffs at warning
+  level with observed duration, configured wait and timeout remediation.
+  Retries retain the original poll identity.
+
 ## [3.5.0] - 2026-10-23
 
 ### Added

composer.json

@@ -93,7 +93,7 @@
             }
         },
         "durable-workflow": {
-            "product-train": "3.5.0",
+            "product-train": "3.5.1",
             "supported-server-versions": "2.5.0",
             "worker-protocol-version": "1.20",
             "control-plane-version": "2",

docs/portal/build/workers.md

@@ -97,6 +97,21 @@ An activity option such as `['queue' => 'payments-v2']` selects the queue of a n
 
 `Worker::run()` registers, negotiates heartbeat cadence, polls workflow/activity/query tasks, renews workflow-task leases, and reports terminal outcomes. Typed transient poll pressure is retried with capped backoff while registration heartbeats and shutdown checks remain responsive.
 
+### Poll timeouts and early disconnects
+
+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
+honor a shorter injected Guzzle timeout. Cooperative polls retain their explicit
+request budget. Older Servers continue returning ordinary JSON. Opaque PSR-18
+clients retain ordinary JSON and their configured policies.
+
+Keep proxy timeouts longer than the configured poll wait. A worker warning
+named `worker.poll_ended_early` reports the observed duration and configured
+wait when a connection failure or gateway cutoff ends a poll early. Inspect
+the proxy and client timeout settings named in its remediation. Configure the
+worker's PSR logger or diagnostic listener to collect these warnings.
+
 ### Large payloads on slower connections
 
 Cooperative workers supervise external payload downloads while bytes arrive and

docs/quickstart-contract.json

@@ -3,8 +3,8 @@
   "schema_version": 2,
   "package": {
     "name": "durable-workflow/sdk",
-    "published_version": "3.5.0",
-    "composer_requirement": "3.5.0",
+    "published_version": "3.5.1",
+    "composer_requirement": "3.5.1",
     "onboarding_requirement": "^3.5"
   },
   "runtime_targets": {

src/Client.php

@@ -86,6 +86,8 @@ final class Client implements WorkflowClientInterface
     private bool $observabilityContextSupported = false;
     /** @var (\Closure(string, array<string, mixed>|null, array<string, mixed>, RequestBudget|null): (?\Closure))|null */
     private ?\Closure $workerPayloadObserver = null;
+    /** @var (\Closure(array<string, mixed>): void)|null */
+    private ?\Closure $workerPollCutoffObserver = null;
 
     public function __construct(
         string $baseUri,
@@ -2612,6 +2614,50 @@ private function pollTaskResponse(string $path, array $body): array
         }
     }
 
+    /** @internal @param \Closure(array<string, mixed>): void $observer */
+    public function withWorkerPollCutoffObserver(\Closure $observer): self
+    {
+        $copy = clone $this;
+        $copy->workerPollCutoffObserver = $observer;
+
+        return $copy;
+    }
+
+    /**
+     * @param array<string, string> $headers
+     * @param array<string, mixed>|null $body
+     * @return array<string, mixed>|list<mixed>|null
+     */
+    private function sendTransportRequest(string $method, string $path, bool $worker, array $headers, ?array $body, ?RequestBudget $budget): ?array
+    {
+        $startedAt = hrtime(true);
+        try {
+            if ($worker && $this->boundedWorkerRequests) {
+                if (!$this->transport instanceof BoundedTransport || !$this->transport->supportsBoundedRequests()) {
+                    throw new InvalidArgumentException('The cooperative worker transport no longer supports bounded requests.');
+                }
+
+                return $this->transport->sendBounded($method, $this->baseUri.'/api'.$path, $headers, $body, $budget->remainingSeconds());
+            }
+
+            return $this->transport->send($method, $this->baseUri.'/api'.$path, $headers, $body);
+        } catch (TransportException $exception) {
+            $observed = (hrtime(true) - $startedAt) / 1_000_000_000;
+            $wait = $body['timeout_seconds'] ?? 0;
+            $kind = \DurableWorkflow\Transport\PollTransport::taskKind($method, $path);
+            if ($worker && $kind !== null && is_int($wait) && $wait > 0 && $observed < $wait
+                && (in_array($exception->status, [502, 504], true) || $exception->transientConnectionFailure)) {
+                ($this->workerPollCutoffObserver)?->__invoke([
+                    'task_kind' => $kind, 'observed_seconds' => round($observed, 3), 'configured_wait_seconds' => $wait,
+                    'status' => $exception->status,
+                    'remediation' => 'Check proxy idle timeouts and client request budgets against the configured worker poll wait.',
+                ]);
+            }
+
+            throw $exception;
+        }
+    }
+
     /** @return array<string, string> */
     private function requestHeaders(bool $worker): array
     {
@@ -2664,14 +2710,7 @@ private function request(
                 $activityLease->checkRequest($body);
                 $budget?->restrictAuthorityDeadline($activityLease->publicationDeadline());
             }
-            if ($worker && $this->boundedWorkerRequests) {
-                if (!$this->transport instanceof BoundedTransport || !$this->transport->supportsBoundedRequests()) {
-                    throw new InvalidArgumentException('The cooperative worker transport no longer supports bounded requests.');
-                }
-                $response = $this->transport->sendBounded($method, $this->baseUri.'/api'.$path, $headers, $body, $budget->remainingSeconds());
-            } else {
-                $response = $this->transport->send($method, $this->baseUri.'/api'.$path, $headers, $body);
-            }
+            $response = $this->sendTransportRequest($method, $path, $worker, $headers, $body, $budget);
 
             $budget?->remainingSeconds();
             // Control I/O and supervised payload I/O have separate bounds.

src/Exception/TransportException.php

@@ -15,17 +15,23 @@ public function __construct(
         public readonly ?array $response = null,
         ?Throwable $previous = null,
         public readonly bool $transientConnectionFailure = false,
+        /** @var array<string, string> */
+        public readonly array $headers = [],
     ) {
         parent::__construct($message, $status ?? 0, $previous);
     }
 
-    /** @param array<string, mixed>|list<mixed>|null $response */
-    public static function fromResponse(int $status, ?array $response, string $rawBody): self
+    /**
+     * @param array<string, mixed>|list<mixed>|null $response
+     * @param string $rawBody Retained for compatibility with existing positional and named calls.
+     * @param array<string, string> $headers
+     */
+    public static function fromResponse(int $status, ?array $response, string $rawBody, array $headers = []): self
     {
         $message = is_array($response)
             ? (string) ($response['message'] ?? $response['error'] ?? "Server returned HTTP {$status}.")
             : "Server returned HTTP {$status}.";
 
-        return new self($message, $status, $response);
+        return new self($message, $status, $response, headers: $headers);
     }
 }

src/Transport/PollTransport.php

@@ -0,0 +1,58 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Transport;
+
+use DurableWorkflow\Exception\TransportException;
+use stdClass;
+
+/** The negotiated envelope carries HTTP outcomes after response progress starts. */
+final class PollTransport
+{
+    public const MEDIA_TYPE = 'application/vnd.durable-workflow.poll.v1+json';
+    private const SCHEMA = 'durable-workflow.poll-transport.v1';
+    private const KINDS = ['workflow', 'activity', 'query', 'update-validation'];
+
+    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')) {
+                    return $kind;
+                }
+            }
+        }
+
+        return null;
+    }
+
+    /** @return array{status: int, headers: array<string, string>, body: array<string, mixed>} */
+    public static function decode(string $rawBody, int $outerStatus): array
+    {
+        $frame = json_decode($rawBody, flags: JSON_THROW_ON_ERROR);
+        if (!$frame instanceof stdClass || ($frame->schema ?? null) !== self::SCHEMA
+            || !is_int($frame->status ?? null) || $frame->status < 200 || $frame->status > 599
+            || !($frame->headers ?? null) instanceof stdClass || !($frame->body ?? null) instanceof stdClass
+            || ($outerStatus >= 400 && $frame->status < 400)) {
+            throw new TransportException('The server returned an invalid worker poll transport envelope.');
+        }
+        $headers = [];
+        foreach (get_object_vars($frame->headers) as $name => $value) {
+            $normalized = strtolower($name);
+            if (!in_array($normalized, ['retry-after', 'x-durable-workflow-protocol-version'], true)
+                || !is_string($value) || str_contains($value, "\r") || str_contains($value, "\n")
+                || array_key_exists($normalized, $headers)) {
+                throw new TransportException('The server returned invalid worker poll transport headers.');
+            }
+            $headers[$normalized] = $value;
+        }
+        // Decode the original bytes with the SDK's existing associative JSON
+        // semantics, without re-encoding payloads or changing their numbers.
+        /** @var array{body: array<string, mixed>} $decoded */
+        $decoded = json_decode($rawBody, true, flags: JSON_THROW_ON_ERROR);
+
+        return ['status' => $frame->status, 'headers' => $headers, 'body' => $decoded['body']];
+    }
+}

src/Transport/Psr18Transport.php

@@ -102,13 +102,22 @@ public function sendBounded(string $method, string $uri, array $headers, ?array
      */
     private function sendJson(string $method, string $uri, array $headers, ?array $body, ?int $timeoutSeconds = null): ?array
     {
+        $poll = PollTransport::taskKind($method, $uri) !== null && $this->supportsBoundedRequests();
+        if ($poll && $timeoutSeconds === null) {
+            // cURL's total timeout includes all progress bytes. Opaque PSR-18
+            // clients retain ordinary JSON because their deadline is unknown.
+            $timeoutSeconds = $this->pollTimeoutSeconds($body);
+        }
         $controlTimeout = $timeoutSeconds === null && $this->isControlRequest($headers) ? $this->requestTimeoutSeconds : null;
         try {
             $request = $this->createJsonRequest($method, $uri, $headers, $body);
+            if ($poll) {
+                $request = $request->withHeader('Accept', PollTransport::MEDIA_TYPE.', application/json;q=0.9');
+            }
             $response = $this->sendRequest($request, timeoutSeconds: $timeoutSeconds ?? $controlTimeout ?? 30,
                 bounded: $timeoutSeconds !== null || $controlTimeout !== null, controlTimeout: $controlTimeout);
 
-            return $this->decodeJsonResponse($response);
+            return $this->decodeJsonResponse($response, $poll);
         } catch (TransportException $exception) {
             throw $exception;
         } catch (JsonException $exception) {
@@ -121,15 +130,35 @@ private function sendJson(string $method, string $uri, array $headers, ?array $b
         }
     }
 
+    /** @param array<string, mixed>|null $body */
+    private function pollTimeoutSeconds(?array $body): int|float
+    {
+        $timeout = max(0, min(60, (int) ($body['timeout_seconds'] ?? 30))) + 5;
+        $configured = $this->client instanceof GuzzleClient ? $this->client->getConfig('timeout') : null;
+
+        return is_numeric($configured) && (float) $configured > 0
+            ? min($timeout, (float) $configured) : $timeout;
+    }
+
     /** @return array<string, mixed>|list<mixed>|null */
-    private function decodeJsonResponse(ResponseInterface $response): ?array
+    private function decodeJsonResponse(ResponseInterface $response, bool $poll = false): ?array
     {
         $rawBody = (string) $response->getBody();
-        if ($response->getStatusCode() < 200 || $response->getStatusCode() >= 300) {
+        $status = $response->getStatusCode();
+        $mediaType = strtolower(trim(explode(';', $response->getHeaderLine('Content-Type'), 2)[0]));
+        if ($poll && $mediaType === PollTransport::MEDIA_TYPE) {
+            $frame = PollTransport::decode($rawBody, $status);
+            if ($frame['status'] >= 300) {
+                throw TransportException::fromResponse($frame['status'], $frame['body'], $rawBody, $frame['headers']);
+            }
+
+            return $frame['body'];
+        }
+        if ($status < 200 || $status >= 300) {
             // An upstream proxy can return HTML or an empty body. Preserve
             // its status without interpreting it as a successful protocol reply.
             $decoded = json_decode($rawBody, true);
-            throw TransportException::fromResponse($response->getStatusCode(), is_array($decoded) ? $decoded : null, $rawBody);
+            throw TransportException::fromResponse($status, is_array($decoded) ? $decoded : null, $rawBody);
         }
 
         $decoded = $rawBody === '' ? null : json_decode($rawBody, true, 512, JSON_THROW_ON_ERROR);

src/Worker.php

@@ -158,7 +158,8 @@ public function __construct(
         }
         $workerClient = $telemetry === null ? $client : $client->withTelemetry($telemetry);
         $workerClient = $enableCooperativeCancellation ? $workerClient->withBoundedWorkerRequests() : $workerClient;
-        $this->client = $workerClient->withWorkerPayloadObserver($this->workerPayloadProgress(...));
+        $this->client = $workerClient->withWorkerPayloadObserver($this->workerPayloadProgress(...))
+            ->withWorkerPollCutoffObserver(fn (array $context) => $this->diagnostic('worker.poll_ended_early', $context, 'warning'));
         $this->workerId = $workerId ?? 'php-worker-'.bin2hex(random_bytes(8));
         $this->heartbeatIntervalSeconds = $this->validHeartbeatInterval($heartbeatIntervalSeconds)
             ?? self::DEFAULT_HEARTBEAT_INTERVAL_SECONDS;