From 2261b3d823264c1005c9c77d8e485b0ad9e73694 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:49:08 +0000 Subject: [PATCH 1/2] Plan continuation payloads within existing slot authority --- CHANGELOG.md | 4 + src/Transport/RuntimePayloadUploads.php | 54 +++++- tests/RuntimePayloadUploadTest.php | 213 +++++++++++++++++++++++- 3 files changed, 263 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e28667ce..2b44d6e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,10 @@ project follows [Semantic Versioning](https://semver.org/). ### Fixed +- Activity continuations honor the completion aggregate and command per-value + limits. Commands stay inline; only the activity result may be offloaded. + An aggregate that cannot fit through an authorized rewrite reaches the + Server unchanged once, without partial uploads. - 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 diff --git a/src/Transport/RuntimePayloadUploads.php b/src/Transport/RuntimePayloadUploads.php index 841a57b5..b87ebda9 100644 --- a/src/Transport/RuntimePayloadUploads.php +++ b/src/Transport/RuntimePayloadUploads.php @@ -62,7 +62,15 @@ 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 + $continuation = $worker && preg_match('~\A/worker/activity-tasks/[^/]+/continuation/complete\z~', $path) === 1; + if ($continuation) { + foreach ($payloads as $payload) { + if ($payload['path'] !== ['result'] && strlen($payload['blob']) > $policy['request_bytes']) { + throw new ExternalPayloadException('Continuation command payload exceeds the inline value limit.', 413, 'external_payload_oversized'); + } + } + } + if ($worker && preg_match('~\A/worker/(?:workflow-tasks/[^/]+/complete|activity-tasks/[^/]+/continuation/complete)\z~', $path) === 1 && array_key_exists('completion_request_bytes', $policy)) { $limit = $policy['completion_request_bytes']; if (!is_int($limit) || $limit < 1) { @@ -81,6 +89,24 @@ public function request(array $body, string $method, string $path, bool $worker, if ($requestBytesLimit !== null) { $policy['request_bytes'] = min($policy['request_bytes'], $requestBytesLimit); } + if ($continuation) { + // The continuation contract authorizes result offload only. Commands + // stay inline, including when their bytes equal the activity result. + $payloads = array_values(array_filter($payloads, static fn (array $payload): bool => $payload['path'] === ['result'])); + $smallest = $body; + foreach ($payloads as $payload) { + if (strlen($payload['blob']) > $policy['max_bytes']) { + throw new ExternalPayloadException('Encoded payload exceeds the runtime upload limit.', 413, 'external_payload_oversized'); + } + $this->placeholder($smallest, $payload); + } + if ($requestBytesLimit === null && array_key_exists('completion_request_bytes', $policy) + && $this->jsonSize($smallest) > $policy['request_bytes']) { + // No authorized rewrite can fit. Avoid a futile result upload; + // submit unchanged once for the Server's structural settlement. + return $body; + } + } $selected = $this->plan($body, $payloads, $policy); return $this->uploadSelected($body, $payloads, $selected, $policy, ['headers' => $headers, 'path' => $path, 'worker' => $worker, 'budget' => $budget]); @@ -371,19 +397,25 @@ private function paths(array $body, string $path, bool $worker): array { $paths = []; if ($worker) { - if (preg_match('~\A/worker/workflow-tasks/[^/]+/(?:complete|local-activities/checkpoint(?:-group)?|cancellation-scopes/checkpoint)\z~', $path)) { - foreach ($body['commands'] ?? [] as $index => $command) { + $continuation = preg_match('~\A/worker/activity-tasks/[^/]+/continuation/complete\z~', $path) === 1; + if ($continuation || preg_match('~\A/worker/workflow-tasks/[^/]+/(?:complete|local-activities/checkpoint(?:-group)?|cancellation-scopes/checkpoint)\z~', $path)) { + $prefix = $continuation ? ['continuation', 'commands'] : ['commands']; + $commands = $continuation ? ($body['continuation']['commands'] ?? []) : ($body['commands'] ?? []); + if ($continuation) { + $paths[] = ['result']; + } + foreach ($commands as $index => $command) { if (!is_array($command)) { continue; } foreach (self::COMMAND_FIELDS[$command['type'] ?? ''] ?? [] as $field) { - $paths[] = ['commands', $index, $field]; + $paths[] = [...$prefix, $index, $field]; } if (($command['type'] ?? null) === 'fail_workflow') { - $paths[] = ['commands', $index, 'exception', 'details']; + $paths[] = [...$prefix, $index, 'exception', 'details']; } foreach ($command['workflow_stream']['items'] ?? [] as $item => $_) { - $paths[] = ['commands', $index, 'workflow_stream', 'items', $item, 'payload']; + $paths[] = [...$prefix, $index, 'workflow_stream', 'items', $item, 'payload']; } } } elseif (preg_match('~\A/worker/workflow-tasks/[^/]+/local-activities/(?:prepare|recover)\z~', $path)) { @@ -459,7 +491,15 @@ private function completionContext(array $body, string $path, array $slot, bool return strlen($context) <= 4096 ? $context : null; } - if (!preg_match('~\A/worker/(activity|workflow|query)-tasks/([^/]+)/(complete|fail)\z~', explode('?', $path)[0], $match)) { + if (preg_match('~\A/worker/activity-tasks/([^/]+)/continuation/complete\z~', $path, $match)) { + // V1 continuation uploads inherit only the activity result's lease. + // Proposed workflow commands have no completion upload authority. + if ($slot !== ['result']) { + return null; + } + $path = '/worker/activity-tasks/'.$match[1].'/complete'; + } + if (!preg_match('~\A/worker/(activity|workflow|query)-tasks/([^/]+)/(complete|fail)\z~', $path, $match)) { return null; } $kind = $match[1]; diff --git a/tests/RuntimePayloadUploadTest.php b/tests/RuntimePayloadUploadTest.php index a728ef87..f40caec6 100644 --- a/tests/RuntimePayloadUploadTest.php +++ b/tests/RuntimePayloadUploadTest.php @@ -232,6 +232,216 @@ public function testOlderProducerStillRequiresOffloadAtTheOrdinaryLimit(): void self::assertCount(1, $http->requests); } + #[DataProvider('unconfiguredCompletionStorage')] + public function testContinuationUsesTheAdvertisedAggregateWithoutStorage(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; + 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); + $value = str_repeat('a', 600); + $blob = $client->payloadCodec()->encode($value); + $continuation = self::continuation([['type' => 'record_side_effect', 'result' => $blob]]); + try { + $client->completeActivityTaskWithContinuation('task', 'attempt', 'worker', $value, self::workflowPoll(), $continuation); + self::assertFalse($overLimit); + } catch (WorkflowTaskCompletionTooLarge $error) { + self::assertTrue($overLimit); + self::assertSame($limit, $error->limitBytes); + self::assertCount(2, $http->requests); + self::assertSame(strlen((string) $http->requests[1]->getBody()), $error->sizeBytes); + self::assertGreaterThan($limit, $error->sizeBytes); + } + self::assertCount(2, $http->requests, 'Discover once, submit once; no upload or retry.'); + $sent = $this->body($http->requests[1]); + self::assertSame($continuation, $sent['continuation']); + self::assertSame(['codec' => 'avro', 'blob' => $blob], $sent['result']); + } + + public function testContinuationOffloadsOnlyResultWithoutWideningOrdinaryActivityCompletion(): void + { + $policy = self::discovery(); + $policy['limits'] = ['max_payload_bytes' => 1100, 'max_workflow_task_completion_bytes' => 1500]; + $policy['namespace']['external_payload_storage']['threshold_bytes'] = 512; + [$client, $http] = $this->client($policy); + $value = str_repeat('a', 600); + $blob = $client->payloadCodec()->encode($value); + $client->completeActivityTaskWithContinuation('task', 'attempt', 'worker', $value, self::workflowPoll(), + self::continuation([['type' => 'record_side_effect', 'result' => $blob]])); + self::assertCount(3, $http->requests, 'Only the authorized activity result is uploaded.'); + $sent = $this->body($http->requests[2]); + self::assertSame(self::reference($blob), $sent['result']['external_payload']); + self::assertSame($blob, $sent['continuation']['commands'][0]['result'], 'Identical bytes do not grant command reference-placement authority.'); + self::assertLessThanOrEqual(1500, strlen((string) $http->requests[2]->getBody())); + + $client->completeActivityTask('ordinary', 'attempt', 'worker', str_repeat('b', 900)); + self::assertCount(5, $http->requests); + self::assertArrayHasKey('external_payload', $this->body($http->requests[4])['result']); + self::assertLessThanOrEqual(1100, strlen((string) $http->requests[4]->getBody())); + } + + public function testContinuationKeepsOldProducerAndPerValueStorageRefusals(): void + { + foreach (['old producer', 'oversized result', 'oversized command', 'configured transient'] as $case) { + $policy = self::discovery(); + $policy['limits'] = ['max_payload_bytes' => 1100]; + if ($case !== 'old producer') { + $policy['limits']['max_workflow_task_completion_bytes'] = 5000; + } + $policy['namespace']['external_payload_storage']['status'] = $case === 'configured transient' ? 'unavailable' : 'unconfigured'; + [$client, $http] = $this->client($policy); + $result = str_repeat('a', $case === 'oversized result' ? 1200 : 600); + $blob = $client->payloadCodec()->encode(str_repeat('b', $case === 'oversized command' ? 1200 : 600)); + try { + $client->completeActivityTaskWithContinuation('task', 'attempt', 'worker', $result, self::workflowPoll(), + self::continuation([['type' => 'record_side_effect', 'result' => $blob]])); + self::fail($case.': no aggregate authority can remove this refusal.'); + } catch (ExternalPayloadException $error) { + self::assertSame(match ($case) { + 'oversized command' => 'external_payload_oversized', + 'old producer' => 'payload_too_large', + default => 'external_payload_unavailable', + }, $error->reason, $case); + } + self::assertCount(1, $http->requests, $case); + } + } + + public function testExplicitSmallerCompletionCapIsHonoredOnBothRoutes(): void + { + foreach ([false, true] as $continuation) { + $policy = self::discovery(); + $policy['limits']['max_workflow_task_completion_bytes'] = 64; + $refusal = new Response(413, [], json_encode(['reason' => 'payload_too_large', + 'limit_kind' => 'workflow_task_completion_bytes', 'limit' => 64], JSON_THROW_ON_ERROR)); + [$client, $http] = $this->client($policy, completionResponse: $refusal); + try { + if ($continuation) { + $client->completeActivityTaskWithContinuation('task', 'attempt', 'worker', 'small', self::workflowPoll(), self::continuation([])); + } else { + $client->completeWorkflowTask('task', 'worker', 1, + [['type' => 'complete_workflow', 'result' => $client->payloadCodec()->encode('small')]]); + } + self::fail('Explicit positive smaller caps cannot expand to the ordinary limit.'); + } catch (ExternalPayloadException $error) { + self::assertSame('payload_too_large', $error->reason); + } catch (WorkflowTaskCompletionTooLarge $error) { + self::assertTrue($continuation); + self::assertSame(64, $error->limitBytes); + } + self::assertCount($continuation ? 2 : 1, $http->requests); + } + } + + #[DataProvider('invalidCompletionLimits')] + public function testMalformedContinuationLimitStopsBeforeUploadsOrSubmission(mixed $limit): void + { + $policy = self::discovery(); + $policy['limits']['max_workflow_task_completion_bytes'] = $limit; + [$client, $http] = $this->client($policy); + try { + $client->completeActivityTaskWithContinuation('task', 'attempt', 'worker', 'small', self::workflowPoll(), self::continuation([])); + self::fail('Malformed continuation cap must refuse before any upload or completion.'); + } catch (ExternalPayloadException $error) { + self::assertSame('external_payload_unsupported', $error->reason); + } + self::assertCount(1, $http->requests); + } + + #[DataProvider('continuationLeaseSlots')] + public function testContinuationBoundUploadAuthorityIsOnlyForTheActivityResult(bool $resultSlot): void + { + [, $http, $transport] = $this->client(self::completionDiscovery(), workerOnly: true, drainUnbound: true); + $blob = (new AvroPayloadCodec())->encode(str_repeat('a', 600)); + $body = ['activity_attempt_id' => 'attempt', 'lease_owner' => 'worker', + 'result' => ['codec' => 'avro', 'blob' => $resultSlot ? $blob : (new AvroPayloadCodec())->encode('small')], + 'continuation' => self::continuation($resultSlot ? [] : [['type' => 'record_side_effect', 'result' => $blob]])]; + try { + $prepared = (new RuntimePayloadUploads($transport, 'https://runtime.test'))->request($body, 'POST', + '/worker/activity-tasks/task/continuation/complete', true, []); + if ($resultSlot) { + self::assertArrayHasKey('external_payload', $prepared['result']); + } else { + self::assertSame($body, $prepared); + } + } catch (ExternalPayloadException $error) { + self::assertFalse($resultSlot); + self::assertSame('storage_pressure', $error->reason); + } + self::assertCount($resultSlot ? 3 : 1, $http->requests); + if ($resultSlot) { + self::assertSame('', $http->requests[1]->getHeaderLine('X-Durable-Workflow-Payload-Completion')); + self::assertSame(['schema' => 'durable-workflow.v2.payload-completion-context.v1', 'kind' => 'activity', + 'task_id' => 'task', 'attempt' => 'attempt', 'lease_owner' => 'worker', 'operation' => 'complete', 'slot' => ['result']], + json_decode($http->requests[2]->getHeaderLine('X-Durable-Workflow-Payload-Completion'), true, flags: JSON_THROW_ON_ERROR)); + } + } + + public static function continuationLeaseSlots(): iterable + { + yield 'activity result' => [true]; + yield 'proposed command' => [false]; + } + + public function testEqualContinuationCommandBytesNeverReuseAResultBoundReference(): void + { + [, $http, $transport] = $this->client(self::completionDiscovery(), workerOnly: true, drainUnbound: true); + $blob = (new AvroPayloadCodec())->encode(str_repeat('a', 600)); + $commands = [['type' => 'record_side_effect', 'result' => $blob]]; + $body = ['activity_attempt_id' => 'attempt', 'lease_owner' => 'worker', + 'result' => ['codec' => 'avro', 'blob' => $blob], 'continuation' => self::continuation($commands)]; + $prepared = (new RuntimePayloadUploads($transport, 'https://runtime.test'))->request($body, 'POST', + '/worker/activity-tasks/task/continuation/complete', true, []); + self::assertCount(3, $http->requests, 'One result upload, then its authorized bound retry; no command upload.'); + self::assertArrayHasKey('external_payload', $prepared['result']); + self::assertSame($commands, $prepared['continuation']['commands']); + $context = json_decode($http->requests[2]->getHeaderLine('X-Durable-Workflow-Payload-Completion'), true, flags: JSON_THROW_ON_ERROR); + self::assertSame(['result'], $context['slot']); + } + + public function testImpossibleContinuationRewriteSubmitsOriginalOnceWithoutPartialUploads(): void + { + foreach (['available', 'unavailable'] as $status) { + $policy = self::discovery(); + $policy['limits'] = ['max_payload_bytes' => 1100, 'max_workflow_task_completion_bytes' => 500]; + $policy['namespace']['external_payload_storage']['status'] = $status; + $refusal = new Response(413, [], json_encode(['reason' => 'payload_too_large', + 'limit_kind' => 'workflow_task_completion_bytes', 'limit' => 500], JSON_THROW_ON_ERROR)); + [$client, $http] = $this->client($policy, completionResponse: $refusal); + $value = str_repeat('a', 600); + $blob = $client->payloadCodec()->encode($value); + $continuation = self::continuation([['type' => 'record_side_effect', 'result' => $blob]]); + try { + $client->completeActivityTaskWithContinuation('task', 'attempt', 'worker', $value, self::workflowPoll(), $continuation); + self::fail('Only the actual Server response can settle this impossible aggregate.'); + } catch (WorkflowTaskCompletionTooLarge $error) { + self::assertSame(500, $error->limitBytes); + } + self::assertCount(2, $http->requests); + $sent = $this->body($http->requests[1]); + self::assertSame($continuation, $sent['continuation']); + self::assertSame(['codec' => 'avro', 'blob' => $blob], $sent['result']); + self::assertSame('', $http->requests[1]->getHeaderLine('X-Durable-Workflow-Payload-Completion')); + } + } + + private static function workflowPoll(): array + { + return ['worker_id' => 'worker', 'task_queue' => 'queue', 'poll_request_id' => 'poll', + 'timeout_seconds' => 0, 'history_page_size' => 100, 'task_kinds' => ['workflow']]; + } + + private static function continuation(array $commands): array + { + return ['version' => 1, 'history_through_sequence' => 1, 'history_last_event_id' => 'event', 'commands' => $commands]; + } + #[DataProvider('completionOffloadRefusals')] public function testCompletionKeepsPerValueAndConfiguredStorageRefusals(string $status, int $valueBytes): void { @@ -982,7 +1192,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) { + } elseif ((str_contains($path, '/workflow-tasks/') || str_ends_with($path, '/continuation/complete')) + && str_ends_with($path, '/complete') && $this->completionResponse !== null) { return Create::promiseFor($this->completionResponse); } else { $response = ['outcome' => 'completed']; From 697fb5e816f7adb015195c9628391c8dd0bb0fe5 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 11:01:07 +0000 Subject: [PATCH 2/2] Preflight completion discovery without inline payloads --- src/Transport/RuntimePayloadUploads.php | 109 +++++++++++++++--------- tests/RuntimePayloadUploadTest.php | 42 ++++++++- 2 files changed, 110 insertions(+), 41 deletions(-) diff --git a/src/Transport/RuntimePayloadUploads.php b/src/Transport/RuntimePayloadUploads.php index b87ebda9..4bfdee97 100644 --- a/src/Transport/RuntimePayloadUploads.php +++ b/src/Transport/RuntimePayloadUploads.php @@ -57,53 +57,23 @@ public function request(array $body, string $method, string $path, bool $worker, $body = $this->failureMetadata($body, $payloads, $path, $failureReports); $payloads = $this->payloads($body, $slots); } - if ($payloads === []) { + $commandCompletion = $worker && preg_match('~\A/worker/(?:workflow-tasks/[^/]+/complete|activity-tasks/[^/]+/continuation/complete)\z~', $path) === 1; + if ($payloads === [] && !$commandCompletion) { return $body; } $policy = $this->requestPolicy($body, $path, $worker, $headers, $budget); - $continuation = $worker && preg_match('~\A/worker/activity-tasks/[^/]+/continuation/complete\z~', $path) === 1; - if ($continuation) { - foreach ($payloads as $payload) { - if ($payload['path'] !== ['result'] && strlen($payload['blob']) > $policy['request_bytes']) { - throw new ExternalPayloadException('Continuation command payload exceeds the inline value limit.', 413, 'external_payload_oversized'); - } - } - } - if ($worker && preg_match('~\A/worker/(?:workflow-tasks/[^/]+/complete|activity-tasks/[^/]+/continuation/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; - } + $inlineLimit = $policy['request_bytes']; + $policy = $this->completionPolicy($policy, $commandCompletion); + if ($this->maySubmitInlineCompletion($policy, $payloads, $inlineLimit, $requestBytesLimit, $commandCompletion)) { + return $body; } if ($requestBytesLimit !== null) { $policy['request_bytes'] = min($policy['request_bytes'], $requestBytesLimit); } - if ($continuation) { - // The continuation contract authorizes result offload only. Commands - // stay inline, including when their bytes equal the activity result. - $payloads = array_values(array_filter($payloads, static fn (array $payload): bool => $payload['path'] === ['result'])); - $smallest = $body; - foreach ($payloads as $payload) { - if (strlen($payload['blob']) > $policy['max_bytes']) { - throw new ExternalPayloadException('Encoded payload exceeds the runtime upload limit.', 413, 'external_payload_oversized'); - } - $this->placeholder($smallest, $payload); - } - if ($requestBytesLimit === null && array_key_exists('completion_request_bytes', $policy) - && $this->jsonSize($smallest) > $policy['request_bytes']) { - // No authorized rewrite can fit. Avoid a futile result upload; - // submit unchanged once for the Server's structural settlement. + if ($commandCompletion && str_starts_with($path, '/worker/activity-tasks/')) { + $payloads = $this->continuationUploads($body, $payloads, $policy, $inlineLimit, $requestBytesLimit); + if ($payloads === null) { return $body; } } @@ -112,6 +82,67 @@ public function request(array $body, string $method, string $path, bool $worker, ['headers' => $headers, 'path' => $path, 'worker' => $worker, 'budget' => $budget]); } + /** @param UploadPolicy $policy + * @return UploadPolicy + */ + private function completionPolicy(array $policy, bool $completion): array + { + if (!$completion || !array_key_exists('completion_request_bytes', $policy)) { + return $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'); + } + $policy['request_bytes'] = $limit; + + return $policy; + } + + /** @param UploadPolicy $policy + * @param list $payloads + */ + private function maySubmitInlineCompletion(array $policy, array $payloads, int $inlineLimit, ?int $reportLimit, bool $completion): bool + { + // Individually valid values need no unavailable external storage. The + // Server settles an oversized aggregate through its typed refusal. + return $completion && $reportLimit === null && array_key_exists('completion_request_bytes', $policy) + && in_array($policy['status'], ['unconfigured', 'disabled', 'unsupported'], true) + && array_filter($payloads, static fn (array $payload): bool => strlen($payload['blob']) > $inlineLimit) === []; + } + + /** @param array $body + * @param list $payloads + * @param UploadPolicy $policy + * @return list|null Null means no authorized rewrite fits; submit unchanged once. + */ + private function continuationUploads(array $body, array $payloads, array $policy, int $inlineLimit, ?int $reportLimit): ?array + { + $results = []; + $smallest = $body; + foreach ($payloads as $payload) { + // Commands stay inline even when their bytes equal the result. + if ($payload['path'] !== ['result']) { + if (strlen($payload['blob']) > $inlineLimit) { + throw new ExternalPayloadException('Continuation command payload exceeds the inline value limit.', 413, 'external_payload_oversized'); + } + continue; + } + if (strlen($payload['blob']) > $policy['max_bytes']) { + throw new ExternalPayloadException('Encoded payload exceeds the runtime upload limit.', 413, 'external_payload_oversized'); + } + $results[] = $payload; + $this->placeholder($smallest, $payload); + } + if ($reportLimit === null && array_key_exists('completion_request_bytes', $policy) + && $this->jsonSize($smallest) > $policy['request_bytes']) { + // Avoid a futile partial result upload before structural settlement. + return null; + } + + return $results; + } + /** @param array $body * @param array $headers * @return UploadPolicy diff --git a/tests/RuntimePayloadUploadTest.php b/tests/RuntimePayloadUploadTest.php index f40caec6..f1e5881c 100644 --- a/tests/RuntimePayloadUploadTest.php +++ b/tests/RuntimePayloadUploadTest.php @@ -485,11 +485,49 @@ public function testMalformedCompletionLimitRefusesOnlyTheCompletion(mixed $limi public static function invalidCompletionLimits(): iterable { - foreach ([null, 0, -1, true, false, 3000.5, '3000', []] as $value) { + foreach ([null, 0, -1, true, false, 3000.0, 3000.5, '3000', []] as $value) { yield [$value]; } } + #[DataProvider('referencedCompletionPreflightCases')] + public function testReferencedAndPayloadFreeCompletionsStillValidateDiscovery(bool $continuation, string $shape, mixed $limit): void + { + $policy = self::discovery(); + $policy['limits']['max_workflow_task_completion_bytes'] = $limit; + [, $http, $transport] = $this->client($policy, workerOnly: true); + $reference = ['codec' => 'avro', 'external_payload' => self::reference('existing')]; + $commands = match ($shape) { + 'empty' => [], + 'timer' => [['type' => 'start_timer', 'sequence' => 1, 'delay_seconds' => 1]], + default => [['type' => 'complete_workflow', 'result' => $reference]], + }; + $body = $continuation + ? ['activity_attempt_id' => 'attempt', 'lease_owner' => 'worker', 'result' => $reference, + 'continuation' => self::continuation($commands)] + : ['workflow_task_attempt' => 1, 'lease_owner' => 'worker', 'commands' => $commands]; + $path = $continuation ? '/worker/activity-tasks/task/continuation/complete' : '/worker/workflow-tasks/task/complete'; + try { + (new RuntimePayloadUploads($transport, 'https://runtime.test'))->request($body, 'POST', $path, true, []); + self::fail('Explicit malformed discovery must refuse even when no inline value needs uploading.'); + } catch (ExternalPayloadException $error) { + self::assertSame('external_payload_unsupported', $error->reason); + } + self::assertCount(1, $http->requests, 'Exactly one discovery, no upload or completion.'); + self::assertStringEndsWith('/cluster/info', $http->requests[0]->getUri()->getPath()); + } + + public static function referencedCompletionPreflightCases(): iterable + { + foreach ([false, true] as $continuation) { + foreach (['empty', 'timer', 'reference'] as $shape) { + foreach (self::invalidCompletionLimits() as [$limit]) { + yield [$continuation, $shape, $limit]; + } + } + } + } + public function testCompletionMetadataUsesActualWholeFloatHttpBytes(): void { $policy = self::discovery(); @@ -1199,7 +1237,7 @@ public function __invoke(RequestInterface $request, array $options): \GuzzleHttp $response = ['outcome' => 'completed']; } - return Create::promiseFor(new Response(200, [], json_encode($response, JSON_THROW_ON_ERROR))); + return Create::promiseFor(new Response(200, [], json_encode($response, JSON_THROW_ON_ERROR | JSON_PRESERVE_ZERO_FRACTION))); } }; $transport = new Psr18Transport(new GuzzleClient(['handler' => $http]));