From ed41f3ba7ee0102d0cf0077ec15ce99a35f31b66 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:01:43 +0000 Subject: [PATCH 1/2] Honor advertised workflow completion limits when planning payloads --- CHANGELOG.md | 11 ++ src/Transport/RuntimePayloadUploads.php | 25 ++++- tests/RuntimePayloadUploadTest.php | 143 +++++++++++++++++++++++- 3 files changed, 174 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b3dc708f..e28667ce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,17 @@ project follows [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Fixed + +- Workflow completion planning honors the Server's advertised + `limits.max_workflow_task_completion_bytes` without changing ordinary request + or per-value limits (.github#228, server#1136). When external storage is + explicitly unconfigured, disabled or unsupported, individually valid values + can be submitted inline. An oversized aggregate is sent once so the Server's + typed structural-limit refusal reaches the worker. Configured-storage + failures keep their existing behavior; older producers retain the + conservative ordinary-request fallback. + ## [3.18.0] - 2026-11-13 ### Added diff --git a/src/Transport/RuntimePayloadUploads.php b/src/Transport/RuntimePayloadUploads.php index 139280b9..841a57b5 100644 --- a/src/Transport/RuntimePayloadUploads.php +++ b/src/Transport/RuntimePayloadUploads.php @@ -11,7 +11,7 @@ /** Namespace-scoped discovery and request-local, content-addressed uploads. * @phpstan-type PayloadSlot array{path: list, blob: string} - * @phpstan-type UploadPolicy array{threshold_bytes: int, max_bytes: int, request_bytes: int, timeout_seconds: int, status: string, completion_context?: bool, prepared_completion_context?: bool, scope_completion_context?: bool} + * @phpstan-type UploadPolicy array{threshold_bytes: int, max_bytes: int, request_bytes: int, timeout_seconds: int, status: string, completion_request_bytes?: mixed, completion_context?: bool, prepared_completion_context?: bool, scope_completion_context?: bool} * @phpstan-type UploadRequest array{headers: array, path: string, worker: bool, budget: RequestBudget|null} */ final class RuntimePayloadUploads @@ -62,6 +62,22 @@ public function request(array $body, string $method, string $path, bool $worker, } $policy = $this->requestPolicy($body, $path, $worker, $headers, $budget); + if ($worker && preg_match('~\A/worker/workflow-tasks/[^/]+/complete\z~', $path) === 1 + && array_key_exists('completion_request_bytes', $policy)) { + $limit = $policy['completion_request_bytes']; + if (!is_int($limit) || $limit < 1) { + throw new ExternalPayloadException('Invalid workflow-task completion limit in runtime discovery.', 422, 'external_payload_unsupported'); + } + $inlineLimit = $policy['request_bytes']; + $policy['request_bytes'] = $limit; + if ($requestBytesLimit === null && in_array($policy['status'], ['unconfigured', 'disabled', 'unsupported'], true) + && array_filter($payloads, static fn (array $payload): bool => strlen($payload['blob']) > $inlineLimit) === []) { + // Without configured storage, individually valid values can still fit the + // completion's larger aggregate limit. Let the Server return its typed + // structural refusal if the actual serialized batch exceeds that limit. + return $body; + } + } if ($requestBytesLimit !== null) { $policy['request_bytes'] = min($policy['request_bytes'], $requestBytesLimit); } @@ -289,7 +305,7 @@ private function validateUploadReference(array $response, string $blob, array $e } /** @param array $headers - * @return array{threshold_bytes: int, max_bytes: int, request_bytes: int, timeout_seconds: int, status: string, completion_context?: bool, prepared_completion_context?: bool, scope_completion_context?: bool} + * @return UploadPolicy */ private function policy(array $headers, bool $worker, ?RequestBudget $budget = null): array { @@ -338,6 +354,11 @@ private function policy(array $headers, bool $worker, ?RequestBudget $budget = n 'prepared_completion_context' => ($manifest['upload']['completion_context']['prepared_schema'] ?? null) === self::PREPARED_COMPLETION_SCHEMA, 'scope_completion_context' => ($manifest['upload']['completion_context']['scope_schema'] ?? null) === self::PREPARED_COMPLETION_SCHEMA]; } + if (is_array($info['limits'] ?? null) && array_key_exists('max_workflow_task_completion_bytes', $info['limits'])) { + // Validate only on the completion route; this field grants no authority + // to widen an ordinary endpoint or an older producer's request limit. + $policy['completion_request_bytes'] = $info['limits']['max_workflow_task_completion_bytes']; + } $this->policies[$key] = ['expires' => time() + 60, 'policy' => $policy]; return $policy; diff --git a/tests/RuntimePayloadUploadTest.php b/tests/RuntimePayloadUploadTest.php index 37f37535..8ec2d44f 100644 --- a/tests/RuntimePayloadUploadTest.php +++ b/tests/RuntimePayloadUploadTest.php @@ -10,6 +10,7 @@ use DurableWorkflow\Exception\ExternalPayloadException; use DurableWorkflow\Exception\ServerException; use DurableWorkflow\Exception\TransportException; +use DurableWorkflow\Exception\WorkflowTaskCompletionTooLarge; use DurableWorkflow\Model\WorkflowStreamAppendItem; use DurableWorkflow\Tests\Support\FakeTransport; use DurableWorkflow\Transport\Psr18Transport; @@ -142,6 +143,140 @@ public function testAggregateInlineBatchIsOffloadedBeforeOrdinaryRequestLimit(): self::assertLessThanOrEqual(1100, strlen((string) $http->requests[2]->getBody())); } + public function testCompletionAggregateLimitDoesNotWidenTheCachedOrdinaryRequestLimit(): void + { + $policy = self::discovery(); + $policy['limits'] = ['max_payload_bytes' => 1100, 'max_workflow_task_completion_bytes' => 3000]; + $policy['namespace']['external_payload_storage']['threshold_bytes'] = 1500; + [$client, $http] = $this->client($policy); + $blob = $client->payloadCodec()->encode(str_repeat('a', 600)); + $commands = [['type' => 'record_side_effect', 'result' => $blob], ['type' => 'complete_workflow', 'result' => $blob]]; + $client->completeWorkflowTask('task', 'worker', 1, $commands); + + self::assertCount(2, $http->requests, 'The larger completion needs no upload.'); + self::assertGreaterThan(1100, strlen((string) $http->requests[1]->getBody())); + self::assertLessThanOrEqual(3000, strlen((string) $http->requests[1]->getBody())); + self::assertSame($commands, $this->body($http->requests[1])['commands']); + + $client->completeActivityTask('activity', 'attempt', 'worker', str_repeat('b', 900)); + self::assertCount(4, $http->requests, 'The same worker policy still offloads an ordinary oversized request.'); + self::assertArrayHasKey('external_payload', $this->body($http->requests[3])['result']); + self::assertLessThanOrEqual(1100, strlen((string) $http->requests[3]->getBody())); + } + + #[DataProvider('unconfiguredCompletionStorage')] + public function testExplicitCompletionLimitAllowsInlineValuesWithoutConfiguredStorage(string $status, bool $overLimit): void + { + $limit = $overLimit ? 1500 : 3000; + $policy = self::discovery(); + $policy['limits'] = ['max_payload_bytes' => 1100, 'max_workflow_task_completion_bytes' => $limit]; + $policy['namespace']['external_payload_storage']['status'] = $status; + $refusal = new Response(413, [], json_encode(['reason' => 'payload_too_large', + 'limit_kind' => 'workflow_task_completion_bytes', 'limit' => $limit], JSON_THROW_ON_ERROR)); + [$client, $http] = $this->client($policy, completionResponse: $overLimit ? $refusal : null); + $blob = $client->payloadCodec()->encode(str_repeat('a', 600)); + $commands = [['type' => 'record_side_effect', 'result' => $blob], ['type' => 'complete_workflow', 'result' => $blob]]; + try { + $client->completeWorkflowTask('task', 'worker', 1, $commands); + self::assertFalse($overLimit, 'An aggregate above the advertised limit must surface the Server refusal.'); + } catch (WorkflowTaskCompletionTooLarge $error) { + self::assertTrue($overLimit); + self::assertSame($limit, $error->limitBytes); + self::assertSame(strlen((string) $http->requests[1]->getBody()), $error->sizeBytes); + self::assertGreaterThan($limit, $error->sizeBytes); + } + self::assertCount(2, $http->requests, 'Discover once, submit once, with no upload or retry.'); + self::assertSame($commands, $this->body($http->requests[1])['commands']); + } + + public static function unconfiguredCompletionStorage(): iterable + { + foreach (['unconfigured', 'disabled', 'unsupported'] as $status) { + yield $status.' within aggregate' => [$status, false]; + yield $status.' above aggregate' => [$status, true]; + } + } + + public function testOlderProducerStillRequiresOffloadAtTheOrdinaryLimit(): void + { + $policy = self::discovery(); + $policy['limits']['max_payload_bytes'] = 1100; + $policy['namespace']['external_payload_storage']['status'] = 'unconfigured'; + [$client, $http] = $this->client($policy); + $blob = $client->payloadCodec()->encode(str_repeat('a', 600)); + try { + $client->completeWorkflowTask('task', 'worker', 1, + [['type' => 'record_side_effect', 'result' => $blob], ['type' => 'complete_workflow', 'result' => $blob]]); + self::fail('An older producer does not grant a larger aggregate limit.'); + } catch (ExternalPayloadException $error) { + self::assertSame('external_payload_unavailable', $error->reason); + } + self::assertCount(1, $http->requests); + } + + #[DataProvider('completionOffloadRefusals')] + public function testCompletionKeepsPerValueAndConfiguredStorageRefusals(string $status, int $valueBytes): void + { + $policy = self::discovery(); + $policy['limits'] = ['max_payload_bytes' => 1100, 'max_workflow_task_completion_bytes' => 3000]; + $policy['namespace']['external_payload_storage']['status'] = $status; + [$client, $http] = $this->client($policy); + try { + $client->completeWorkflowTask('task', 'worker', 1, + [['type' => 'complete_workflow', 'result' => $client->payloadCodec()->encode(str_repeat('a', $valueBytes))]]); + self::fail('The completion limit does not remove a payload/storage refusal.'); + } catch (ExternalPayloadException $error) { + self::assertSame('external_payload_unavailable', $error->reason); + } + self::assertCount(1, $http->requests); + } + + public static function completionOffloadRefusals(): iterable + { + yield 'individual value exceeds inline limit' => ['unconfigured', 1200]; + yield 'configured storage transient failure' => ['unavailable', 600]; + yield 'unknown storage state' => ['unknown', 600]; + } + + #[DataProvider('invalidCompletionLimits')] + public function testMalformedCompletionLimitRefusesOnlyTheCompletion(mixed $limit): void + { + $policy = self::discovery(); + $policy['limits']['max_workflow_task_completion_bytes'] = $limit; + [$client, $http] = $this->client($policy); + $client->startWorkflow('echo', 'one', 'queue'); + try { + $client->completeWorkflowTask('task', 'worker', 1, + [['type' => 'complete_workflow', 'result' => $client->payloadCodec()->encode('small')]]); + self::fail('An invalid aggregate limit cannot grant completion authority.'); + } catch (ExternalPayloadException $error) { + self::assertSame('external_payload_unsupported', $error->reason); + } + self::assertCount(3, $http->requests, 'The ordinary request succeeds; completion stops after worker discovery.'); + } + + public static function invalidCompletionLimits(): iterable + { + foreach ([null, 0, -1, true, false, 3000.5, '3000', []] as $value) { + yield [$value]; + } + } + + public function testCompletionMetadataUsesActualWholeFloatHttpBytes(): void + { + $policy = self::discovery(); + $policy['limits'] = ['max_payload_bytes' => 1100, 'max_workflow_task_completion_bytes' => 3000]; + [$client, $http] = $this->client($policy); + try { + $client->completeWorkflowTask('task', 'worker', 1, [['type' => 'complete_workflow', + 'result' => $client->payloadCodec()->encode('small'), 'metadata' => array_fill(0, 800, 1.0)]]); + self::fail('Whole-float metadata exceeds the actual completion limit even after offload.'); + } catch (ExternalPayloadException $error) { + self::assertSame('payload_too_large', $error->reason); + } + self::assertCount(1, $http->requests); + } + public function testOversizedMetadataIsRejectedBeforeAnyUpload(): void { $policy = self::discovery(); @@ -799,14 +934,14 @@ public function testMixedLocalFailureUsesTheExistingUploadPlannerWithoutChanging } } - private function client(?array $discovery = null, bool $workerOnly = false, ?Response $uploadResponse = null, bool $drainUnbound = false, ?Response $retryUploadResponse = null): array + private function client(?array $discovery = null, bool $workerOnly = false, ?Response $uploadResponse = null, bool $drainUnbound = false, ?Response $retryUploadResponse = null, ?Response $completionResponse = null): array { - $http = new class($discovery ?? self::discovery(), $uploadResponse, $drainUnbound, $retryUploadResponse) { + $http = new class($discovery ?? self::discovery(), $uploadResponse, $drainUnbound, $retryUploadResponse, $completionResponse) { public array $requests = []; public array $options = []; private string $uploadBody; private int $uploadAttempts = 0; - public function __construct(private array $discovery, private ?Response $uploadResponse, private bool $drainUnbound, private ?Response $retryUploadResponse) + public function __construct(private array $discovery, private ?Response $uploadResponse, private bool $drainUnbound, private ?Response $retryUploadResponse, private ?Response $completionResponse) { $this->uploadBody = (string) $uploadResponse?->getBody(); } @@ -829,6 +964,8 @@ public function __invoke(RequestInterface $request, array $options): \GuzzleHttp } $response = ['schema' => 'durable-workflow.v2.runtime-external-payload-upload.v1', 'transport_version' => 1, 'reference' => RuntimePayloadUploadTest::reference((string) $request->getBody())]; + } elseif (str_contains($path, '/workflow-tasks/') && str_ends_with($path, '/complete') && $this->completionResponse !== null) { + return Create::promiseFor($this->completionResponse); } else { $response = ['outcome' => 'completed']; } From da7eea724677019ade170d5edd75e4aee8a1c90c Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:10:01 +0000 Subject: [PATCH 2/2] Cover configured offload and missing transport manifests --- tests/RuntimePayloadUploadTest.php | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/tests/RuntimePayloadUploadTest.php b/tests/RuntimePayloadUploadTest.php index 8ec2d44f..a728ef87 100644 --- a/tests/RuntimePayloadUploadTest.php +++ b/tests/RuntimePayloadUploadTest.php @@ -164,6 +164,21 @@ public function testCompletionAggregateLimitDoesNotWidenTheCachedOrdinaryRequest self::assertLessThanOrEqual(1100, strlen((string) $http->requests[3]->getBody())); } + public function testConfiguredStorageOffloadsAboveTheAdvertisedCompletionLimit(): void + { + $policy = self::discovery(); + $policy['limits'] = ['max_payload_bytes' => 1100, 'max_workflow_task_completion_bytes' => 1500]; + $policy['namespace']['external_payload_storage']['threshold_bytes'] = 1500; + [$client, $http] = $this->client($policy); + $blob = $client->payloadCodec()->encode(str_repeat('a', 600)); + $client->completeWorkflowTask('task', 'worker', 1, + [['type' => 'record_side_effect', 'result' => $blob], ['type' => 'complete_workflow', 'result' => $blob]]); + + self::assertCount(3, $http->requests); + self::assertSame($blob, (string) $http->requests[1]->getBody()); + self::assertLessThanOrEqual(1500, strlen((string) $http->requests[2]->getBody())); + } + #[DataProvider('unconfiguredCompletionStorage')] public function testExplicitCompletionLimitAllowsInlineValuesWithoutConfiguredStorage(string $status, bool $overLimit): void { @@ -171,6 +186,9 @@ public function testExplicitCompletionLimitAllowsInlineValuesWithoutConfiguredSt $policy = self::discovery(); $policy['limits'] = ['max_payload_bytes' => 1100, 'max_workflow_task_completion_bytes' => $limit]; $policy['namespace']['external_payload_storage']['status'] = $status; + if ($status === 'unsupported') { + unset($policy['namespace']['external_payload_storage']['transport']); + } $refusal = new Response(413, [], json_encode(['reason' => 'payload_too_large', 'limit_kind' => 'workflow_task_completion_bytes', 'limit' => $limit], JSON_THROW_ON_ERROR)); [$client, $http] = $this->client($policy, completionResponse: $overLimit ? $refusal : null);