Bound external payload HTTP memory and align SAPI limits

CONTRIBUTING.md

@@ -21,7 +21,7 @@ therefore join the PHP execution inventory automatically. Formats without an
 official PHP executor are rejected by corpus validation and cannot contribute
 guarded growth.
 
-A server codec-boundary fix also needs an append-only counterfactual proof under
+A wire-codec defect also needs an append-only counterfactual proof under
 `tests/Fixtures/CodecRegressionProofs/`. The proof pairs each new codec fixture
 with one changed boundary path and a changed Feature PHPUnit test. The test
 uses `ServerCodecRegressionFixtureExecutor::exercise()` exactly once. That shared
@@ -36,3 +36,14 @@ regression must pass on the candidate and fail on both defective revisions.
 Review the fixture and HTTP reproducer together; substituting an unrelated
 valid fixture cannot prove the causality of a request that deliberately submits
 malformed bytes.
+
+Transport and resource fixes are different: HTTP body limits, streaming storage,
+reference retention and metadata-only reads need their real HTTP, integrity,
+memory and recovery regressions. When no new wire fixture is appropriate, the
+validator reports changed payload paths for maintainer review without requiring
+artificial corpus growth. The PR must explain unchanged wire semantics and show
+the original failure and the corrected behavior at the actual affected surface.
+Review that evidence before merging. Existing fixtures remain immutable and run
+in the normal suite; wire-format, type-identity, framing or codec acceptance
+defects still require portable fixtures and the proof above. A green inventory
+check alone is not qualification.

app/Http/Controllers/Api/ActivityController.php

@@ -2,12 +2,13 @@
 
 namespace App\Http\Controllers\Api;
 
+use App\Support\AvroPayloadEnvelopeResolver;
 use App\Support\ControlPlaneProtocol;
 use App\Support\ExternalPayloadEnvelopeService;
 use App\Support\ExternalPayloadStorageUnavailable;
 use App\Support\NamespaceExternalPayloadStorage;
-use App\Support\PayloadCodecContract;
 use App\Support\NamespaceWorkflowScope;
+use App\Support\PayloadCodecContract;
 use App\Support\TaskQueueRoutingGate;
 use App\Support\WorkflowCommandContextFactory;
 use Carbon\CarbonInterface;
@@ -23,7 +24,6 @@
 use Workflow\V2\Models\WorkflowRun;
 use Workflow\V2\StandaloneActivity\StandaloneActivityHostType;
 use Workflow\V2\Support\ExternalPayloads;
-use App\Support\AvroPayloadEnvelopeResolver;
 use Workflow\V2\Support\RunActivityView;
 use Workflow\V2\Support\StandaloneActivityStartService;
 
@@ -142,6 +142,7 @@ public function start(Request $request): JsonResponse
                 $validated['input'] ?? null,
                 'input',
                 $this->externalPayloadStorage->driverFor($namespace),
+                retainExternal: true,
             );
             $payloadCodec = $envelope['codec'] ?? $defaultCodec;
             $arguments = $envelope['blob'] ?? null;
@@ -339,7 +340,6 @@ private function formatActivity(WorkflowRun $run, ?string $namespace): array
     }
 
     /**
-     * @param object $summary
      * @return array<string, mixed>
      */
     private function formatActivityListEntry(object $summary): array
@@ -406,7 +406,7 @@ private function activityViewForExecution(WorkflowRun $run, ?ActivityExecution $
             return [];
         }
 
-        foreach (RunActivityView::activitiesForRun($run) as $activity) {
+        foreach (RunActivityView::activitiesForRun($run, decodePayloads: false) as $activity) {
             if (is_array($activity) && ($activity['id'] ?? null) === $execution->id) {
                 return $activity;
             }
@@ -416,7 +416,7 @@ private function activityViewForExecution(WorkflowRun $run, ?ActivityExecution $
     }
 
     /**
-     * @param array<string, mixed> $activityView
+     * @param  array<string, mixed>  $activityView
      * @return list<array<string, mixed>>
      */
     private function formatAttempts(array $activityView, ?ActivityExecution $execution): array
@@ -430,7 +430,7 @@ private function formatAttempts(array $activityView, ?ActivityExecution $executi
     }
 
     /**
-     * @param array<string, mixed> $attempt
+     * @param  array<string, mixed>  $attempt
      * @return array<string, mixed>
      */
     private function formatAttempt(array $attempt, ?ActivityExecution $execution): array
@@ -468,7 +468,7 @@ private function formatAttempt(array $attempt, ?ActivityExecution $execution): a
     }
 
     /**
-     * @param list<array<string, mixed>> $attempts
+     * @param  list<array<string, mixed>>  $attempts
      * @return array<string, mixed>|null
      */
     private function currentAttempt(array $attempts, ?ActivityExecution $execution): ?array

app/Http/Controllers/Api/ActivityTaskController.php

@@ -244,6 +244,7 @@ public function complete(Request $request, string $taskId): JsonResponse
                 $validated['result'] ?? null,
                 'result',
                 $this->externalPayloadStorage->driverFor($namespace),
+                retainExternal: true,
             );
         } catch (ValidationException $exception) {
             throw $exception;

app/Http/Controllers/Api/RuntimeExternalPayloadController.php

@@ -42,7 +42,7 @@ public function store(Request $request): JsonResponse
         }
 
         $data = RuntimeExternalPayloadUploadBody::read($request, $maxBytes);
-        $observedSize = strlen($data);
+        $observedSize = $data->sizeBytes;
         if ($observedSize > $maxBytes) {
             throw new RuntimeExternalPayloadException(
                 'external_payload_oversized',
@@ -91,7 +91,7 @@ public function show(Request $request, string $referenceId): Response
             'codec' => (string) $request->header('X-Durable-Workflow-Payload-Codec'),
             'size_bytes' => $this->declaredSize($request),
             'sha256' => strtolower((string) $request->header('X-Durable-Workflow-Payload-SHA256')),
-        ]);
+        ], stream: true);
         $reference = $result['reference'];
 
         $this->audit->record($request, 'external_payload.fetched', [
@@ -100,7 +100,21 @@ public function show(Request $request, string $referenceId): Response
             'size_bytes' => $reference['size_bytes'],
         ]);
 
-        return response($result['data'], 200, [
+        return response()->stream(static function () use ($result): void {
+            $data = $result['data'];
+            try {
+                $stream = $data->rewind();
+                while (! feof($stream)) {
+                    $chunk = fread($stream, 8192);
+                    if ($chunk === false) {
+                        throw new \RuntimeException('External payload snapshot could not be read.');
+                    }
+                    echo $chunk;
+                }
+            } finally {
+                $data->close();
+            }
+        }, 200, [
             'Content-Type' => 'application/octet-stream',
             'Content-Length' => (string) $reference['size_bytes'],
             'X-Durable-Workflow-Payload-Codec' => $reference['codec'],

app/Http/Controllers/Api/WorkerController.php

@@ -2921,6 +2921,7 @@ private function resolveWorkflowTaskCommandPayloadReferences(array $commands, st
                     $command[$field],
                     "commands.{$index}.{$field}",
                     $driver,
+                    retainExternal: in_array($commandType, ['complete_workflow', 'schedule_activity'], true),
                 );
 
                 if ($resolved['codec'] === null) {

app/Http/Controllers/Api/WorkflowController.php

@@ -39,6 +39,7 @@
 use Workflow\V2\Models\WorkflowInstance;
 use Workflow\V2\Models\WorkflowRun;
 use Workflow\V2\Models\WorkflowUpdate;
+use Workflow\V2\Support\ExternalPayloads;
 use Workflow\V2\Support\FailureSnapshots;
 use Workflow\V2\Support\RunCommandContract;
 use Workflow\V2\Support\TypeRegistry;
@@ -1169,6 +1170,13 @@ private function formatRun(WorkflowRun $run, string $namespace, array $descripti
         }
         $terminalFailure = $this->terminalFailurePayload($run);
         $outputEnvelope = $run->outputEnvelope();
+        $inputEnvelope = $this->workerEnvelope(
+            $namespace,
+            $run->payload_codec,
+            is_string($run->arguments) ? $run->arguments : null,
+        );
+        $omitInputPreview = $this->oversizedPayloadPreview($inputEnvelope);
+        $omitOutputPreview = $this->oversizedPayloadPreview($outputEnvelope);
 
         $payload = [
             'workflow_id' => $run->workflow_instance_id,
@@ -1193,14 +1201,15 @@ private function formatRun(WorkflowRun $run, string $namespace, array $descripti
             'run_timeout_seconds' => $runDescription['run_timeout_seconds'] ?? null,
             'execution_deadline_at' => $runDescription['execution_deadline_at'] ?? null,
             'run_deadline_at' => $runDescription['run_deadline_at'] ?? null,
-            'input' => AvroValueJsonProjection::project($this->workflowArguments($run)),
-            'output' => AvroValueJsonProjection::project($run->workflowOutput()),
-            'input_envelope' => $this->workerEnvelope(
-                $namespace,
-                $run->payload_codec,
-                is_string($run->arguments) ? $run->arguments : null,
-            ),
+            'input' => $omitInputPreview ? null : AvroValueJsonProjection::project($this->workflowArguments($run)),
+            'output' => $omitOutputPreview ? null : AvroValueJsonProjection::project($run->workflowOutput()),
+            'input_envelope' => $inputEnvelope,
             'output_envelope' => $outputEnvelope,
+            'payload_previews' => [
+                'input_omitted' => $omitInputPreview,
+                'output_omitted' => $omitOutputPreview,
+                'max_encoded_bytes' => (int) config('server.limits.max_payload_bytes', 2 * 1024 * 1024),
+            ],
             'started_at' => $run->started_at?->toJSON(),
             'closed_at' => $run->closed_at?->toJSON(),
             'last_progress_at' => $runDescription['last_progress_at'] ?? $run->last_progress_at?->toJSON(),
@@ -1232,6 +1241,13 @@ private function formatRun(WorkflowRun $run, string $namespace, array $descripti
         return $payload;
     }
 
+    /** @param array<string, mixed>|null $envelope */
+    private function oversizedPayloadPreview(?array $envelope): bool
+    {
+        return isset($envelope['external_storage']['size_bytes'])
+            && $envelope['external_storage']['size_bytes'] > (int) config('server.limits.max_payload_bytes', 2 * 1024 * 1024);
+    }
+
     private function rejectProjectedV1Operation(
         Request $request,
         string $namespace,
@@ -1281,6 +1297,9 @@ private function runCommandDiagnostics(WorkflowRun $run): array
             ->map(function (WorkflowCommand $command) use ($updatesByCommandId): array {
                 $update = $updatesByCommandId->get($command->id);
                 $context = $command->commandContext();
+                $omitPayloadPreview = $this->oversizedPayloadPreview(
+                    is_string($command->payload) ? ExternalPayloads::storedEnvelope($command->payload) : null,
+                );
 
                 return $this->withoutNullOrEmptyArrays([
                     'id' => $command->id,
@@ -1289,7 +1308,7 @@ private function runCommandDiagnostics(WorkflowRun $run): array
                     'target_scope' => $command->target_scope,
                     'requested_run_id' => $command->requestedRunId(),
                     'resolved_run_id' => $command->resolvedRunId(),
-                    'target_name' => $command->targetName(),
+                    'target_name' => $omitPayloadPreview ? null : $command->targetName(),
                     'source' => $command->source,
                     'context' => $this->commandPublicContext($context),
                     'caller_label' => $this->contextString($context, ['caller', 'label']),
@@ -1307,9 +1326,10 @@ private function runCommandDiagnostics(WorkflowRun $run): array
                     'correlation_id' => $this->contextString($context, ['request', 'correlation_id']),
                     'status' => $this->enumOrString($command->status),
                     'outcome' => $this->enumOrString($command->outcome),
-                    'reason' => $command->commandReason(),
+                    'reason' => $omitPayloadPreview ? $command->rejection_reason : $command->commandReason(),
                     'rejection_reason' => $command->rejection_reason,
-                    'validation_errors' => $command->validationErrors(),
+                    'validation_errors' => $omitPayloadPreview ? [] : $command->validationErrors(),
+                    'payload_preview_omitted' => $omitPayloadPreview,
                     'workflow_type' => $command->workflow_type,
                     'workflow_class' => $command->workflow_class,
                     'accepted_at' => $command->accepted_at?->toJSON(),

app/Http/Middleware/EnforcePayloadLimits.php

@@ -3,6 +3,7 @@
 namespace App\Http\Middleware;
 
 use App\Support\ControlPlaneProtocol;
+use App\Support\ExternalPayloadObjectOversized;
 use App\Support\RuntimeExternalPayloadAudit;
 use App\Support\RuntimeExternalPayloadUploadBody;
 use App\Support\WorkerProtocol;
@@ -26,9 +27,28 @@ public function handle(Request $request, Closure $next): Response
             return $this->tooLarge($request, $maxBytes);
         }
 
-        $body = $externalPayloadUpload
-            ? RuntimeExternalPayloadUploadBody::read($request, $maxBytes)
-            : $request->getContent();
+        if ($externalPayloadUpload) {
+            // Reject non-binary bodies before any middleware can JSON-decode
+            // an upload using the much larger external transport allowance.
+            if (! $this->usesOctetStreamMediaType($request)) {
+                return $this->unsupportedMediaType($request);
+            }
+
+            try {
+                RuntimeExternalPayloadUploadBody::read($request, $maxBytes);
+            } catch (ExternalPayloadObjectOversized) {
+                return $this->tooLarge($request, $maxBytes);
+            }
+
+            return $next($request);
+        }
+
+        $stream = $request->getContent(true);
+        $readLimit = $maxBytes === PHP_INT_MAX ? PHP_INT_MAX : max(1, $maxBytes) + 1;
+        $body = stream_get_contents($stream, $readLimit);
+        if ($body === false) {
+            throw new \RuntimeException('Request body could not be read.');
+        }
         $bodySize = strlen($body);
 
         if ($bodySize > $maxBytes) {
@@ -37,10 +57,6 @@ public function handle(Request $request, Closure $next): Response
 
         if ($this->methodCanHaveBody($request)
             && $this->hasBody($contentLength, $bodySize)) {
-            if ($externalPayloadUpload && $this->usesOctetStreamMediaType($request)) {
-                return $next($request);
-            }
-
             if (! $this->usesJsonMediaType($request)) {
                 return $this->unsupportedMediaType($request);
             }
@@ -95,11 +111,7 @@ private function methodCanHaveBody(Request $request): bool
 
     private function hasBody(?string $contentLength, int $bodySize): bool
     {
-        if (is_numeric($contentLength)) {
-            return (int) $contentLength > 0;
-        }
-
-        return $bodySize > 0;
+        return $bodySize > 0 || (is_numeric($contentLength) && (int) $contentLength > 0);
     }
 
     private function usesJsonMediaType(Request $request): bool

app/Support/AvroExternalPayloadValidator.php

@@ -0,0 +1,88 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Support;
+
+use Apache\Avro\Datum\AvroIODatumReader;
+use Illuminate\Validation\ValidationException;
+use Throwable;
+use UnexpectedValueException;
+use Workflow\Serializers\Avro;
+
+final class AvroExternalPayloadValidator
+{
+    public static function validate(ExternalPayloadStream $payload, string $field): void
+    {
+        $decoded = tmpfile();
+        if ($decoded === false) {
+            throw new ExternalPayloadStorageUnavailable('Cannot create Avro validation stream.');
+        }
+        try {
+            $size = self::decodeBase64($payload->rewind(), $decoded);
+            rewind($decoded);
+            $input = new AvroStreamInput($decoded, $size);
+            if ($input->read(10) !== Avro::SINGLE_OBJECT_MAGIC.Avro::VALUE_SCHEMA_FINGERPRINT) {
+                throw new UnexpectedValueException('Invalid Avro Value single-object frame.');
+            }
+            AvroIODatumReader::skipData(
+                Avro::parseSchema(Avro::valueSchemaJson()),
+                new AvroValidationDecoder($input),
+            );
+            if (! $input->isEof()) {
+                throw new UnexpectedValueException('Trailing bytes after Avro Value datum.');
+            }
+        } catch (ExternalPayloadStorageUnavailable $exception) {
+            throw $exception;
+        } catch (Throwable) {
+            throw ValidationException::withMessages([
+                $field => ['The payload must contain one complete Avro Value encoded with the official codec. JSON is only the HTTP document transport.'],
+            ]);
+        } finally {
+            fclose($decoded);
+        }
+    }
+
+    /** @param resource $source
+     * @param  resource  $destination
+     */
+    private static function decodeBase64($source, $destination): int
+    {
+        $buffer = '';
+        $size = 0;
+        while (! feof($source)) {
+            $chunk = fread($source, 8192);
+            if ($chunk === false || ($chunk === '' && ! feof($source))) {
+                throw new ExternalPayloadStorageUnavailable('Cannot read Avro validation source.');
+            }
+            $buffer .= str_replace(["\r", "\n", "\t", ' '], '', $chunk);
+            // Keep the last quartet for the native decoder's final padding check.
+            $length = intdiv(max(0, strlen($buffer) - 1), 4) * 4;
+            if ($length === 0) {
+                continue;
+            }
+            $block = substr($buffer, 0, $length);
+            if (str_contains($block, '=')) {
+                throw new UnexpectedValueException('Base64 padding must end the payload.');
+            }
+            $size += self::writeDecoded($destination, $block);
+            $buffer = substr($buffer, $length);
+        }
+
+        return $size + self::writeDecoded($destination, $buffer);
+    }
+
+    /** @param resource $destination */
+    private static function writeDecoded($destination, string $encoded): int
+    {
+        $bytes = base64_decode($encoded, true);
+        if ($bytes === false) {
+            throw new UnexpectedValueException('Invalid base64 Avro payload.');
+        }
+        if (fwrite($destination, $bytes) !== strlen($bytes)) {
+            throw new ExternalPayloadStorageUnavailable('Cannot write Avro validation stream.');
+        }
+
+        return strlen($bytes);
+    }
+}