diff --git a/CHANGELOG.md b/CHANGELOG.md index aa9e7b1..252338c 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/composer.json b/composer.json index 1640f09..f43720d 100644 --- a/composer.json +++ b/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", diff --git a/docs/portal/build/workers.md b/docs/portal/build/workers.md index 09b605f..41e9e64 100644 --- a/docs/portal/build/workers.md +++ b/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 diff --git a/docs/quickstart-contract.json b/docs/quickstart-contract.json index 824e1d1..072ba3f 100644 --- a/docs/quickstart-contract.json +++ b/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": { diff --git a/package-lock.json b/package-lock.json index 575e1a3..9992655 100644 --- a/package-lock.json +++ b/package-lock.json @@ -644,9 +644,9 @@ "license": "MIT" }, "node_modules/fast-uri": { - "version": "3.1.8", - "resolved": "https://registry.npmjs.org/fast-uri/-/fast-uri-3.1.8.tgz", - "integrity": "sha512-GZMtZUTNRpOVIECoXwLNZS5xUGE+mVNbTB8h/7Rwh2TFWcBQiPzTgyZi05BF9UMZKkLJv8XBRJTlU7zg8+ZfMg==", + "version": "3.1.9", + "resolved": "https://registry.npmjs.org/fast-uri/-/fast-uri-3.1.9.tgz", + "integrity": "sha512-hS/0Wt3Ya043oKXs0sIxrMpKQITXhMLIrVAKKXi4gozGC9v7YRWVnXIZzjNPTO02M9BdxLCWqK1yRXgNziVyJQ==", "dev": true, "funding": [ { diff --git a/src/Client.php b/src/Client.php index b086be9..d4d3aa4 100644 --- a/src/Client.php +++ b/src/Client.php @@ -86,6 +86,8 @@ final class Client implements WorkflowClientInterface private bool $observabilityContextSupported = false; /** @var (\Closure(string, array|null, array, RequestBudget|null): (?\Closure))|null */ private ?\Closure $workerPayloadObserver = null; + /** @var (\Closure(array): 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): void $observer */ + public function withWorkerPollCutoffObserver(\Closure $observer): self + { + $copy = clone $this; + $copy->workerPollCutoffObserver = $observer; + + return $copy; + } + + /** + * @param array $headers + * @param array|null $body + * @return array|list|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 */ 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. diff --git a/src/Exception/TransportException.php b/src/Exception/TransportException.php index 9784aa9..c21f464 100644 --- a/src/Exception/TransportException.php +++ b/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 */ + public readonly array $headers = [], ) { parent::__construct($message, $status ?? 0, $previous); } - /** @param array|list|null $response */ - public static function fromResponse(int $status, ?array $response, string $rawBody): self + /** + * @param array|list|null $response + * @param string $rawBody Retained for compatibility with existing positional and named calls. + * @param array $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); } } diff --git a/src/Transport/PollTransport.php b/src/Transport/PollTransport.php new file mode 100644 index 0000000..4aaf208 --- /dev/null +++ b/src/Transport/PollTransport.php @@ -0,0 +1,58 @@ +, body: array} */ + 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} $decoded */ + $decoded = json_decode($rawBody, true, flags: JSON_THROW_ON_ERROR); + + return ['status' => $frame->status, 'headers' => $headers, 'body' => $decoded['body']]; + } +} diff --git a/src/Transport/Psr18Transport.php b/src/Transport/Psr18Transport.php index 1cb593a..6fb6a0b 100644 --- a/src/Transport/Psr18Transport.php +++ b/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|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|list|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); diff --git a/src/Worker.php b/src/Worker.php index 8e59779..57192c9 100644 --- a/src/Worker.php +++ b/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; diff --git a/tests/WorkerPollTransportTest.php b/tests/WorkerPollTransportTest.php new file mode 100644 index 0000000..eda12b1 --- /dev/null +++ b/tests/WorkerPollTransportTest.php @@ -0,0 +1,232 @@ + static function (RequestInterface $request, array $options) use (&$calls) { + $calls[] = [$request, $options]; + + return Create::promiseFor(new Response(200, ['Content-Type' => strtoupper(self::MEDIA_TYPE).'; charset=utf-8'], + "\n \n".self::frame(200, '{"task":null,"poll_status":"empty","decimal":1.0,"large":9223372036854775808}'))); + }]); + $result = (new Psr18Transport($http))->send('POST', 'https://server.example/namespaces/example/api/worker/'.$kind.'-tasks/poll', + ['Accept' => 'application/json'], ['timeout_seconds' => 4]); + self::assertCount(1, $calls); + self::assertSame(self::MEDIA_TYPE.', application/json;q=0.9', $calls[0][0]->getHeaderLine('Accept')); + self::assertSame(9, $calls[0][1]['timeout']); + self::assertFalse($calls[0][1]['stream']); + self::assertFalse($calls[0][1]['allow_redirects']); + self::assertSame(json_decode('{"task":null,"poll_status":"empty","decimal":1.0,"large":9223372036854775808}', true), $result); + } + + public static function pollKinds(): array + { + return [['workflow'], ['activity'], ['query'], ['update-validation']]; + } + + public function testAShorterInjectedGuzzleDeadlineIsPreserved(): void + { + $observed = null; + $http = new GuzzleClient(['timeout' => 0.25, 'handler' => static function (RequestInterface $request, array $options) use (&$observed) { + $observed = $options['timeout']; + + return Create::promiseFor(new Response(200, [], '{"task":null}')); + }]); + (new Psr18Transport($http))->send('POST', 'https://server.example/api/worker/workflow-tasks/poll', [], ['timeout_seconds' => 30]); + self::assertSame(0.25, $observed); + } + + #[DataProvider('lateStatuses')] + public function testLateErrorsKeepStatusDetailsAndRetryHeaders(int $status): void + { + $http = new GuzzleClient(['handler' => static fn () => Create::promiseFor( + new Response(200, ['Content-Type' => self::MEDIA_TYPE], self::frame($status, + '{"task":null,"reason":"fixture_reason","retry_after_seconds":2}', '{"Retry-After":"2","X-Durable-Workflow-Protocol-Version":"1.20"}')), + )]); + try { + (new Client('https://server.example', transport: new Psr18Transport($http))) + ->pollActivityTaskResponse('worker', 'queue', 4, 'same-poll'); + self::fail('A late HTTP error became a successful task response.'); + } catch (ServerException $error) { + self::assertSame($status, $error->status); + self::assertSame('fixture_reason', $error->reason); + self::assertSame(2, $error->details['retry_after_seconds']); + self::assertInstanceOf(TransportException::class, $error->getPrevious()); + self::assertSame(['retry-after' => '2', 'x-durable-workflow-protocol-version' => '1.20'], $error->getPrevious()->headers); + } + } + + public static function lateStatuses(): array + { + return [[409], [422], [429], [500], [503]]; + } + + #[DataProvider('invalidFrames')] + public function testMalformedOrContradictoryFramesNeverBecomeSuccessfulPolls(string $frame, int $outer): void + { + $http = new GuzzleClient(['handler' => static fn () => Create::promiseFor(new Response($outer, ['Content-Type' => self::MEDIA_TYPE], $frame))]); + $this->expectException(TransportException::class); + (new Psr18Transport($http))->send('POST', 'https://server.example/api/worker/workflow-tasks/poll', [], ['timeout_seconds' => 4]); + } + + public static function invalidFrames(): array + { + return [ + 'unknown schema' => [str_replace('transport.v1', 'transport.v2', self::frame(200)), 200], + 'boolean status' => [str_replace('"status":200', '"status":true', self::frame(200)), 200], + 'invalid status' => [self::frame(600), 200], + 'body list' => [self::frame(200, '[]'), 200], + 'headers list' => [self::frame(200, '{}', '[]'), 200], + 'header injection' => [self::frame(200, '{}', '{"retry-after":"2\\r\\nX-Other: no"}'), 200], + 'unknown header' => [self::frame(200, '{}', '{"set-cookie":"fixture"}'), 200], + 'duplicate normalized header' => [self::frame(200, '{}', '{"Retry-After":"2","retry-after":"3"}'), 200], + 'contradictory gateway status' => [self::frame(200), 504], + 'truncated frame' => [substr(self::frame(200), 0, -1), 200], + ]; + } + + public function testLegacyJsonAndOpaqueClientPoliciesStayUsable(): void + { + $http = new class implements ClientInterface { + public function sendRequest(RequestInterface $request): ResponseInterface + { + self::assertLegacy($request); + + return new Response(200, ['Content-Type' => 'application/json'], '{"task":null,"poll_status":"empty"}'); + } + + private static function assertLegacy(RequestInterface $request): void + { + WorkerPollTransportTest::assertSame('application/json', $request->getHeaderLine('Accept')); + } + }; + $client = new Client('https://server.example', transport: new Psr18Transport($http)); + self::assertSame(['task' => null, 'poll_status' => 'empty'], $client->pollWorkflowTaskResponse('worker', 'queue', 4)); + + $http = new GuzzleClient(['handler' => static function (RequestInterface $request) { + self::assertSame('application/json', $request->getHeaderLine('Accept')); + + return Create::promiseFor(new Response(200, [], '{"renewed":true}')); + }]); + self::assertSame(['renewed' => true], (new Psr18Transport($http))->send('POST', 'https://server.example/api/worker/workflow-tasks/task/heartbeat', ['Accept' => 'application/json'])); + } + + public function testAConnectionRetryKeepsTheSamePollIdentityAndWarnsBeforeTheRetrySucceeds(): void + { + $ids = []; + $http = new GuzzleClient(['handler' => static function (RequestInterface $request, array $options) use (&$ids) { + $ids[] = json_decode((string) $request->getBody(), true)['poll_request_id']; + if (count($ids) === 1) { + $options['on_stats'](new TransferStats($request, null, 0, 7)); + + return Create::rejectionFor(new ConnectException('Fixture connection failure.', $request)); + } + + return Create::promiseFor(new Response(200, ['Content-Type' => 'application/json'], '{"task":null,"poll_status":"stopped"}')); + }]); + $handler = new TestHandler(); + $worker = new Worker(new Client('https://server.example', transport: new Psr18Transport($http)), 'queue', logger: new Logger('fixture', [$handler])); + self::assertFalse($worker->tick(30)); + self::assertCount(2, $ids); + self::assertSame($ids[0], $ids[1]); + $warnings = array_values(array_filter($handler->getRecords(), static fn ($record) => $record['message'] === 'worker.poll_ended_early')); + self::assertCount(1, $warnings); + self::assertSame(Logger::WARNING, $warnings[0]['level']); + self::assertSame(30, $warnings[0]['context']['configured_wait_seconds']); + self::assertLessThan(30, $warnings[0]['context']['observed_seconds']); + self::assertArrayNotHasKey('exception', $warnings[0]['context']); + } + + public function testProgressBytesCannotExtendTheRealCurlDeadline(): void + { + if (!extension_loaded('curl') || !function_exists('pcntl_fork') || !function_exists('posix_kill')) { + self::markTestSkipped('Real cURL and process control are required.'); + } + $listener = stream_socket_server('tcp://127.0.0.1:0', $errno, $error); + self::assertIsResource($listener); + $address = stream_socket_get_name($listener, false); + $pid = pcntl_fork(); + self::assertGreaterThanOrEqual(0, $pid); + if ($pid === 0) { + $connection = stream_socket_accept($listener, 3); + if ($connection !== false) { + while (($line = fgets($connection)) !== false && $line !== "\r\n") {} + fwrite($connection, "HTTP/1.1 200 OK\r\nContent-Type: ".self::MEDIA_TYPE."\r\nConnection: close\r\n\r\n"); + for ($i = 0; $i < 30; ++$i) { + @fwrite($connection, "\n"); + @fflush($connection); + usleep(100000); + } + @fwrite($connection, self::frame(200)); + fclose($connection); + } + posix_kill(getmypid(), SIGKILL); + } + fclose($listener); + $started = hrtime(true); + try { + (new Psr18Transport())->sendBounded('POST', 'http://'.$address.'/api/worker/query-tasks/poll', [], ['timeout_seconds' => 0], 1); + self::fail('Progress extended the absolute request deadline.'); + } catch (TransportException) { + $observed = (hrtime(true) - $started) / 1_000_000_000; + self::assertGreaterThanOrEqual(0.8, $observed); + self::assertLessThan(2.5, $observed); + } finally { + posix_kill($pid, SIGKILL); + self::assertSame($pid, pcntl_waitpid($pid, $status)); + } + } + + #[DataProvider('cutoffStatuses')] + public function testOnlyEarlyGatewayFailuresAreDiagnosedAsCutoffs(int $status, int $wait, bool $warn): void + { + $http = new GuzzleClient(['handler' => static fn () => Create::promiseFor(new Response($status, [], '{"task":null,"reason":"fixture"}'))]); + $warnings = []; + $client = (new Client('https://server.example', transport: new Psr18Transport($http))) + ->withWorkerPollCutoffObserver(static function (array $context) use (&$warnings): void { $warnings[] = $context; }); + try { + $client->pollWorkflowTaskResponse('worker', 'queue', $wait); + self::fail('Fixture HTTP error became success.'); + } catch (ServerException $error) { + self::assertSame($status, $error->status); + } + self::assertCount($warn ? 1 : 0, $warnings); + } + + public static function cutoffStatuses(): array + { + return [[502, 4, true], [504, 4, true], [429, 4, false], [503, 4, false], [500, 4, false], [504, 0, false]]; + } + + private static function frame(int $status, string $body = '{}', string $headers = '{}'): string + { + return '{"schema":"durable-workflow.poll-transport.v1","status":'.$status.',"headers":'.$headers.',"body":'.$body.'}'; + } +}