From eebf0d93849e87af4f046aea32a02f2f362db137 Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Sun, 23 Aug 2026 19:01:14 +0530 Subject: [PATCH 01/18] feat: add streamed response format 0.12.0 The multipart response is serialised in full before any of it reaches the caller, so a site that renders progressively still arrives as one block once the runtime has finished. This adds a response format whose parts are length prefixed, letting a caller read them as they are framed. Content is prefixed with a hex length in the same shape HTTP uses for chunked transfer encoding, and each part is marked Content-Transfer-Encoding: chunked. The prefix is what makes the incremental read tractable on the other side: content is never scanned for the boundary, so a body carrying the boundary string cannot split the envelope and the reader needs no lookahead. statusCode and headers are known from the runtime's response headers before its first body byte, so they are emitted first and the body streams after them. logs, errors and duration are only known once the body is done, so they trail it. Streaming commits on the first body byte: a response with no body never commits and falls back to the buffered document, as does any caller below 0.12.0 or asking for JSON. --- app/controllers.php | 63 ++++++- app/init.php | 4 + src/Executor/BodyMultipartStream.php | 121 +++++++++++++ src/Executor/Runner/Adapter.php | 1 + src/Executor/Runner/Docker.php | 55 +++++- .../unit/Executor/BodyMultipartStreamTest.php | 163 ++++++++++++++++++ 6 files changed, 400 insertions(+), 7 deletions(-) create mode 100644 src/Executor/BodyMultipartStream.php create mode 100644 tests/unit/Executor/BodyMultipartStreamTest.php diff --git a/app/controllers.php b/app/controllers.php index 12208bdc..02cc9a05 100644 --- a/app/controllers.php +++ b/app/controllers.php @@ -4,6 +4,7 @@ use OpenRuntimes\Executor\Exception; use OpenRuntimes\Executor\BodyMultipart; +use OpenRuntimes\Executor\BodyMultipartStream; use OpenRuntimes\Executor\Runner\Adapter as Runner; use Utopia\System\System; use Utopia\Http\Request; @@ -223,6 +224,45 @@ function ( $variables = array_map(strval(...), $variables); + $responseFormat = $request->getHeaderLine('x-executor-response-format') ?: '0.10.0'; // Last version without support for array value for headers + + $acceptTypes = \explode(', ', $request->getHeaderLine('accept') ?: 'multipart/form-data'); + $isJson = array_any($acceptTypes, fn ($acceptType): bool => \str_starts_with((string) $acceptType, 'application/json') || \str_starts_with((string) $acceptType, 'application/*')); + + // A caller on 0.12.0 or later can read parts as they are framed, so the runtime's body + // can be forwarded while it is still being produced. Anything older, and the JSON + // shape, still get the complete document they expect. + $stream = null; + $onStream = null; + + if (!$isJson && \version_compare($responseFormat, RESPONSE_FORMAT_STREAM, '>=')) { + $stream = new BodyMultipartStream( + BodyMultipart::generateBoundary(), + function (string $bytes) use ($response): void { + $response->chunk($bytes); + } + ); + + $onStream = function (string $event, mixed $data) use ($response, $stream): void { + if ($event !== 'headers') { + $stream->writeContent(\strval($data)); + + return; + } + + // Last chance to set response headers: the first content run follows straight + // after, and chunk() commits them on its first call. + $response + ->setStatusCode(Response::STATUS_CODE_OK) + ->addHeader('content-type', $stream->exportHeader()) + ->addHeader('x-executor-response-format', RESPONSE_FORMAT_STREAM); + + $stream->part('statusCode', $data['statusCode']); + $stream->part('headers', $data['headers']); + $stream->startPart('body'); + }; + } + $execution = $runner->createExecution( $runtimeId, $payload, @@ -240,10 +280,28 @@ function ( $runtimeEntrypoint, $logging, $restartPolicy, + onStream: $onStream, ); + $streamed = ($execution['streamed'] ?? false) === true; + unset($execution['streamed']); + + // The body left the executor already, so only the trailing metadata is still owed. A + // response with no body at all never commits to streaming and falls through below. + if ($stream !== null && $streamed) { + $stream->endPart(); + + foreach (['logs', 'errors', 'duration', 'startTime'] as $key) { + $stream->part($key, $execution[$key] ?? ''); + } + + $stream->end(); + $response->chunk('', true); + + return; + } + // Backwards compatibility for headers - $responseFormat = $request->getHeaderLine('x-executor-response-format') ?: '0.10.0'; // Last version without support for array value for headers if (version_compare($responseFormat, '0.11.0', '<')) { foreach ($execution['headers'] as $key => $value) { if (\is_array($value)) { @@ -253,9 +311,6 @@ function ( } } - $acceptTypes = \explode(', ', $request->getHeaderLine('accept') ?: 'multipart/form-data'); - $isJson = array_any($acceptTypes, fn ($acceptType): bool => \str_starts_with((string) $acceptType, 'application/json') || \str_starts_with((string) $acceptType, 'application/*')); - if ($isJson) { $executionString = \json_encode($execution, JSON_UNESCAPED_UNICODE); if (!$executionString) { diff --git a/app/init.php b/app/init.php index a8c15f5e..fe825a38 100644 --- a/app/init.php +++ b/app/init.php @@ -17,6 +17,10 @@ const MAX_LOG_SIZE = 5 * 1024 * 1024; const MAX_BUILD_LOG_SIZE = 1000 * 1000; +// First response format whose parts are length prefixed, so a caller can read them as they are +// framed instead of waiting for the whole document to be serialised. +const RESPONSE_FORMAT_STREAM = '0.12.0'; + Config::load('errors', __DIR__ . '/config/errors.php'); $container = new Container(); diff --git a/src/Executor/BodyMultipartStream.php b/src/Executor/BodyMultipartStream.php new file mode 100644 index 00000000..0c4bc11a --- /dev/null +++ b/src/Executor/BodyMultipartStream.php @@ -0,0 +1,121 @@ +\r\n\r\n (repeated for each run of content) + * 0\r\n\r\n (terminates the part) + * --BOUNDARY-- (after the last part) + * + * The length prefix is what makes an incremental read tractable on the other side: content is + * never scanned for the boundary, so content that happens to contain the boundary string cannot + * split the envelope, and the reader needs no lookahead between reads. + */ +class BodyMultipartStream +{ + private bool $ended = false; + private bool $inPart = false; + + /** + * @param string $boundary + * @param callable(string): void $write Receives raw bytes to put on the wire. + */ + public function __construct( + private readonly string $boundary, + private readonly mixed $write, + ) { + } + + public function exportHeader(): string + { + return 'multipart/form-data; boundary=' . $this->boundary; + } + + /** + * Open a part. Content follows via writeContent(), and the part is closed by endPart(). + */ + public function startPart(string $name): void + { + if ($this->ended || $this->inPart) { + return; + } + + $this->inPart = true; + + ($this->write)( + '--' . $this->boundary . "\r\n" + . 'Content-Disposition: form-data; name="' . $name . '"' . "\r\n" + . 'Content-Transfer-Encoding: chunked' . "\r\n\r\n" + ); + } + + /** + * Append a run of content to the open part. + */ + public function writeContent(string $content): void + { + // A zero length run is the part terminator, so an empty write has to be dropped rather + // than forwarded. Callers stream whatever curl hands them, and curl can hand over nothing. + if ($this->ended || !$this->inPart || $content === '') { + return; + } + + ($this->write)(\dechex(\strlen($content)) . "\r\n" . $content . "\r\n"); + } + + public function endPart(): void + { + if ($this->ended || !$this->inPart) { + return; + } + + $this->inPart = false; + + ($this->write)("0\r\n\r\n"); + } + + /** + * Write a part whose content is already known in full. + * + * @param mixed $value Scalars are stringified; arrays are JSON encoded, matching BodyMultipart. + */ + public function part(string $name, mixed $value): void + { + $this->startPart($name); + $this->writeContent(\is_array($value) ? (\json_encode($value) ?: '') : \strval($value)); + $this->endPart(); + } + + /** + * Close the envelope. Any part still open is terminated first. + */ + public function end(): void + { + if ($this->ended) { + return; + } + + $this->endPart(); + $this->ended = true; + + ($this->write)('--' . $this->boundary . '--'); + } + + public function isEnded(): bool + { + return $this->ended; + } +} diff --git a/src/Executor/Runner/Adapter.php b/src/Executor/Runner/Adapter.php index c7bac752..42375bd2 100644 --- a/src/Executor/Runner/Adapter.php +++ b/src/Executor/Runner/Adapter.php @@ -55,6 +55,7 @@ abstract public function createExecution( bool $logging, string $restartPolicy, string $region = '', + ?callable $onStream = null, ): mixed; abstract public function getRuntimes(): mixed; diff --git a/src/Executor/Runner/Docker.php b/src/Executor/Runner/Docker.php index 00db6382..54dc2842 100644 --- a/src/Executor/Runner/Docker.php +++ b/src/Executor/Runner/Docker.php @@ -681,6 +681,7 @@ public function createExecution( bool $logging, string $restartPolicy, string $region = '', + ?callable $onStream = null, ): mixed { $runtimeName = System::getHostname() . '-' . $runtimeId; @@ -889,7 +890,11 @@ public function createExecution( ]; }; - $executeV5 = function () use ($path, $method, $headers, $payload, $secret, $hostname, $timeout, $runtimeName, $logging): array { + // Set once the runtime's body has begun leaving the executor. From that point the response + // is committed: it can be neither retried nor turned into an error. + $streamed = false; + + $executeV5 = function () use ($path, $method, $headers, $payload, $secret, $hostname, $timeout, $runtimeName, $logging, $onStream, &$streamed): array { $statusCode = 0; $errNo = -1; $executorResponse = ''; @@ -910,7 +915,10 @@ public function createExecution( \curl_setopt($ch, CURLOPT_POSTFIELDS, $payload); } - \curl_setopt($ch, CURLOPT_RETURNTRANSFER, true); + if ($onStream === null) { + \curl_setopt($ch, CURLOPT_RETURNTRANSFER, true); + } + \curl_setopt($ch, CURLOPT_HEADERFUNCTION, function ($curl, $header) use (&$responseHeaders): int { $len = strlen($header); $header = explode(':', $header, 2); @@ -938,6 +946,35 @@ public function createExecution( return $len; }); + if ($onStream !== null) { + // Forward the runtime's body as it arrives instead of accumulating it. The header + // callback above has already run by the time the first body byte lands, so status + // and headers are known and can be announced before any content goes out. + \curl_setopt($ch, CURLOPT_WRITEFUNCTION, function ($curl, $data) use ($onStream, &$responseHeaders, &$streamed): int { + if (!$streamed) { + $streamed = true; + + $outputHeaders = []; + foreach ($responseHeaders as $key => $value) { + if (\str_starts_with($key, 'x-open-runtimes-')) { + continue; + } + + $outputHeaders[$key] = $value; + } + + $onStream('headers', [ + 'statusCode' => \intval(\curl_getinfo($curl, CURLINFO_HTTP_CODE)), + 'headers' => $outputHeaders, + ]); + } + + $onStream('body', $data); + + return \strlen($data); + }); + } + \curl_setopt($ch, CURLOPT_TIMEOUT, (int) $timeout + 5); // Gives extra 5s after safe timeout to recieve response \curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, 5); $headers['x-open-runtimes-logging'] = $logging ? 'enabled' : 'disabled'; @@ -954,7 +991,10 @@ public function createExecution( \curl_setopt($ch, CURLOPT_HEADEROPT, CURLHEADER_UNIFIED); \curl_setopt($ch, CURLOPT_HTTPHEADER, $headersArr); - $executorResponse = \curl_exec($ch); + // With a write callback installed curl_exec returns a bool and the body has already + // been forwarded, so there is nothing left to keep here. + $result = \curl_exec($ch); + $executorResponse = \is_string($result) ? $result : ''; $statusCode = \curl_getinfo($ch, CURLINFO_HTTP_CODE); @@ -1086,6 +1126,13 @@ public function createExecution( break; } + // Content is already on the wire, so a second attempt would append a second response + // to it. The retryable errors here all happen before a connection is established, so + // this is a guard rather than an expected path. + if ($streamed) { + break; + } + usleep($retryDelayMs * 1000); } while ((++$attempts < $retryAttempts) || (\microtime(true) - $startTime < $timeout)); @@ -1118,6 +1165,8 @@ public function createExecution( 'errors' => $errors, 'duration' => $duration, 'startTime' => $startTime, + // Lets the caller tell an empty body apart from one that has already been forwarded. + 'streamed' => $streamed, ]; // Update swoole table diff --git a/tests/unit/Executor/BodyMultipartStreamTest.php b/tests/unit/Executor/BodyMultipartStreamTest.php new file mode 100644 index 00000000..5ad35ee1 --- /dev/null +++ b/tests/unit/Executor/BodyMultipartStreamTest.php @@ -0,0 +1,163 @@ +wire = ''; + + return new BodyMultipartStream($boundary, function (string $bytes): void { + $this->wire .= $bytes; + }); + } + + public function testPartIsLengthPrefixedAndChunkMarked(): void + { + $stream = $this->writer(); + $stream->part('statusCode', 200); + + $this->assertSame( + "--BOUNDARY\r\n" + . "Content-Disposition: form-data; name=\"statusCode\"\r\n" + . "Content-Transfer-Encoding: chunked\r\n\r\n" + . "3\r\n200\r\n" + . "0\r\n\r\n", + $this->wire + ); + } + + public function testLengthPrefixIsHexadecimal(): void + { + $stream = $this->writer(); + $stream->startPart('body'); + $stream->writeContent(\str_repeat('x', 255)); + $stream->endPart(); + + $this->assertStringContainsString("\r\nff\r\n" . \str_repeat('x', 255) . "\r\n", $this->wire); + } + + public function testArrayPartIsJsonEncoded(): void + { + $stream = $this->writer(); + $stream->part('headers', ['content-type' => 'text/html']); + + $this->assertStringContainsString('{"content-type":"text\/html"}', $this->wire); + } + + public function testEachContentRunGetsItsOwnPrefix(): void + { + $stream = $this->writer(); + $stream->startPart('body'); + $stream->writeContent('abc'); + $stream->writeContent('de'); + $stream->endPart(); + + $this->assertStringContainsString("3\r\nabc\r\n2\r\nde\r\n0\r\n\r\n", $this->wire); + } + + public function testEmptyContentRunIsDroppedRatherThanTerminatingThePart(): void + { + $stream = $this->writer(); + $stream->startPart('body'); + $stream->writeContent(''); + $stream->writeContent('after'); + $stream->endPart(); + + // A zero length run would read as the part terminator, orphaning everything after it. + $this->assertStringNotContainsString("0\r\n\r\nafter", $this->wire); + $this->assertStringContainsString("5\r\nafter\r\n0\r\n\r\n", $this->wire); + } + + public function testEmptyPartStillFramesSoTheKeyExists(): void + { + $stream = $this->writer(); + $stream->part('errors', ''); + + $this->assertStringContainsString('name="errors"', $this->wire); + $this->assertStringEndsWith("0\r\n\r\n", $this->wire); + } + + public function testEndClosesEnvelopeWithTerminalDelimiter(): void + { + $stream = $this->writer(); + $stream->part('logs', 'hello'); + $stream->end(); + + $this->assertStringEndsWith('--BOUNDARY--', $this->wire); + $this->assertTrue($stream->isEnded()); + } + + public function testEndTerminatesAnOpenPartFirst(): void + { + $stream = $this->writer(); + $stream->startPart('body'); + $stream->writeContent('partial'); + $stream->end(); + + $this->assertStringEndsWith("7\r\npartial\r\n0\r\n\r\n--BOUNDARY--", $this->wire); + } + + public function testWritesAfterEndAreIgnored(): void + { + $stream = $this->writer(); + $stream->part('logs', 'x'); + $stream->end(); + $after = $this->wire; + + $stream->part('errors', 'ignored'); + $stream->end(); + + $this->assertSame($after, $this->wire); + } + + public function testNestedStartPartIsIgnored(): void + { + $stream = $this->writer(); + $stream->startPart('body'); + $stream->startPart('other'); + + $this->assertStringNotContainsString('name="other"', $this->wire); + } + + public function testExportHeaderCarriesBoundary(): void + { + $stream = $this->writer('abc123'); + + $this->assertSame('multipart/form-data; boundary=abc123', $stream->exportHeader()); + } + + public function testContentContainingBoundaryDoesNotSplitEnvelope(): void + { + $stream = $this->writer(); + $payload = "--BOUNDARY--\r\n0\r\n\r\nnot a terminator"; + + $stream->startPart('body'); + $stream->writeContent($payload); + $stream->endPart(); + $stream->end(); + + // Content is never scanned for the boundary, so the length prefix is the only framing. + $this->assertStringContainsString(\dechex(\strlen($payload)) . "\r\n" . $payload . "\r\n", $this->wire); + } + + public function testBinaryContentIsLengthedInBytes(): void + { + $stream = $this->writer(); + $binary = \random_bytes(1024); + + $stream->startPart('body'); + $stream->writeContent($binary); + $stream->endPart(); + + $this->assertStringContainsString("400\r\n" . $binary . "\r\n", $this->wire); + } +} From aac5ea3bbe1bd6e68f81ae6fd362b8a5e7f2aea9 Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Sun, 23 Aug 2026 21:24:46 +0530 Subject: [PATCH 02/18] test: cover the streamed response format end to end Asserts against the raw wire rather than a decoded body, because the framing is the thing being tested: that parts arrive length prefixed and chunk marked, that statusCode and headers land before the body, and that a body too large for one socket read is framed as several runs instead of one document. A single run would mean the executor buffered after all, so that assertion is what separates this from the existing execution tests. Adds a node resource that returns a body sized by request header to produce that case, and checks a 0.11.0 caller still receives the buffered document with no format echo. --- tests/e2e/StreamedResponseTest.php | 264 ++++++++++++++++++ .../functions/node-large-response/index.js | 14 + .../node-large-response/package.json | 13 + 3 files changed, 291 insertions(+) create mode 100644 tests/e2e/StreamedResponseTest.php create mode 100644 tests/resources/functions/node-large-response/index.js create mode 100644 tests/resources/functions/node-large-response/package.json diff --git a/tests/e2e/StreamedResponseTest.php b/tests/e2e/StreamedResponseTest.php new file mode 100644 index 00000000..d82295f1 --- /dev/null +++ b/tests/e2e/StreamedResponseTest.php @@ -0,0 +1,264 @@ +client = new Client($this->endpoint, [ + 'content-type' => 'application/json', + ]); + $this->client->setKey($this->key); + } + + /** + * Reads the 0.12.0 envelope. Mirrors the reader on the caller's side: content is located by its + * length prefix and never by scanning for the boundary. + * + * @return array{parts: array, runs: array, order: array} + */ + private function parse(string $boundary, string $wire): array + { + $parts = []; + $runs = []; + $order = []; + $offset = 0; + $delimiter = '--' . $boundary; + + while (true) { + $this->assertSame($delimiter, \substr($wire, $offset, \strlen($delimiter)), 'expected delimiter at ' . $offset); + $offset += \strlen($delimiter); + + if (\substr($wire, $offset, 2) === '--') { + break; // closing delimiter + } + + $this->assertSame("\r\n", \substr($wire, $offset, 2)); + $offset += 2; + + $end = \strpos($wire, "\r\n\r\n", $offset); + $this->assertNotFalse($end, 'unterminated part headers'); + $headers = \substr($wire, $offset, $end - $offset); + $offset = $end + 4; + + $this->assertMatchesRegularExpression('/Content-Transfer-Encoding:\s*chunked/i', $headers, 'part is not chunked: ' . $headers); + $this->assertSame(1, \preg_match('/name="([^"]+)"/', $headers, $m), 'part has no name: ' . $headers); + $name = $m[1]; + + $order[] = $name; + $parts[$name] = ''; + $runs[$name] = 0; + + while (true) { + $eol = \strpos($wire, "\r\n", $offset); + $this->assertNotFalse($eol, 'unterminated chunk size'); + $size = \substr($wire, $offset, $eol - $offset); + $this->assertMatchesRegularExpression('/^[0-9a-fA-F]+$/', $size, 'bad chunk size "' . $size . '"'); + $offset = $eol + 2; + + $length = \intval($size, 16); + + if ($length === 0) { + $this->assertSame("\r\n", \substr($wire, $offset, 2), 'part terminator must be followed by CRLF'); + $offset += 2; + break; + } + + $parts[$name] .= \substr($wire, $offset, $length); + $runs[$name]++; + $offset += $length; + + $this->assertSame("\r\n", \substr($wire, $offset, 2), 'chunk must be followed by CRLF'); + $offset += 2; + } + } + + return ['parts' => $parts, 'runs' => $runs, 'order' => $order]; + } + + private function build(string $folder, string $image, string $entrypoint, string $buildCommand, string $runtimeId): string + { + $output = ''; + $stderr = ''; + Console::execute('cd /app/tests/resources/functions/' . $folder . ' && tar --exclude code.tar.gz -czf code.tar.gz .', '', $output, $stderr); + + $response = $this->client->call(Client::METHOD_POST, '/runtimes', [], [ + 'runtimeId' => $runtimeId . '-build', + 'source' => '/storage/functions/' . $folder . '/code.tar.gz', + 'destination' => '/storage/builds/' . $runtimeId, + 'entrypoint' => $entrypoint, + 'image' => $image, + 'command' => 'tar -zxf /tmp/code.tar.gz -C /mnt/code && bash helpers/build.sh "' . $buildCommand . '"', + 'remove' => true, + ]); + + $this->assertEquals(201, $response['headers']['status-code'], 'build failed: ' . \json_encode($response)); + + return $response['body']['path']; + } + + /** + * @return array{wire: string, headers: array, chunks: int} + */ + private function executeStreamed(string $runtimeId, array $params, string $format = '0.12.0'): array + { + $wire = ''; + $chunks = 0; + + $response = $this->client->call( + Client::METHOD_POST, + '/runtimes/' . $runtimeId . '/executions', + [ + 'x-executor-response-format' => $format, + 'accept' => 'multipart/form-data', + ], + $params, + false, + function (string $data) use (&$wire, &$chunks): void { + $wire .= $data; + $chunks++; + } + ); + + return ['wire' => $wire, 'headers' => $response['headers'], 'chunks' => $chunks]; + } + + public function testStreamedEnvelopeIsFramedAndOrdered(): void + { + $runtimeId = 'stream-' . \bin2hex(\random_bytes(4)); + $buildPath = $this->build('php-mock', 'openruntimes/php:v5-8.1', 'index.php', 'composer install', $runtimeId); + + $response = $this->client->call(Client::METHOD_POST, '/runtimes', [], [ + 'runtimeId' => $runtimeId, + 'source' => $buildPath, + 'entrypoint' => 'index.php', + 'image' => 'openruntimes/php:v5-8.1', + 'runtimeEntrypoint' => 'cp /tmp/code.tar.gz /mnt/code/code.tar.gz && nohup helpers/start.sh "php src/server.php"', + ]); + $this->assertEquals(201, $response['headers']['status-code']); + + $result = $this->executeStreamed($runtimeId, []); + + // The executor must announce the format, otherwise a caller cannot tell it apart from a + // buffered document and will not attempt an incremental read. + $this->assertEquals('0.12.0', $result['headers']['x-executor-response-format'] ?? null, 'format not echoed: ' . \json_encode($result['headers'])); + + $contentType = $result['headers']['content-type'] ?? ''; + $this->assertStringStartsWith('multipart/form-data', (string) $contentType); + $this->assertSame(1, \preg_match('/boundary=(.+)$/', (string) $contentType, $m), 'no boundary in ' . $contentType); + $boundary = \trim($m[1], '"'); + + $parsed = $this->parse($boundary, $result['wire']); + + $this->assertSame( + ['statusCode', 'headers', 'body', 'logs', 'errors', 'duration', 'startTime'], + $parsed['order'], + 'unexpected part order' + ); + + // statusCode and headers must precede the body: the proxy cannot set either once content + // has started going out to the visitor. + $this->assertSame(0, \array_search('statusCode', $parsed['order'], true)); + $this->assertSame(1, \array_search('headers', $parsed['order'], true)); + + $this->assertSame('200', $parsed['parts']['statusCode']); + + $headers = \json_decode($parsed['parts']['headers'], true); + $this->assertIsArray($headers); + $this->assertArrayHasKey('content-type', $headers); + + $body = \json_decode($parsed['parts']['body'], true); + $this->assertIsArray($body); + $this->assertTrue($body['isTest']); + $this->assertSame('Hello Open Runtimes 👋', $body['message']); + + $this->assertStringContainsString('Sample Log', $parsed['parts']['logs']); + $this->assertGreaterThan(0, (float) $parsed['parts']['duration']); + $this->assertGreaterThan(0, (float) $parsed['parts']['startTime']); + + $this->client->call(Client::METHOD_DELETE, '/runtimes/' . $runtimeId, [], []); + } + + public function testLargeBodyIsFramedAsSeveralRuns(): void + { + $runtimeId = 'stream-large-' . \bin2hex(\random_bytes(4)); + $buildPath = $this->build('node-large-response', 'openruntimes/node:v5-18.0', 'index.js', 'npm install', $runtimeId); + + $response = $this->client->call(Client::METHOD_POST, '/runtimes', [], [ + 'runtimeId' => $runtimeId, + 'source' => $buildPath, + 'entrypoint' => 'index.js', + 'image' => 'openruntimes/node:v5-18.0', + 'runtimeEntrypoint' => 'cp /tmp/code.tar.gz /mnt/code/code.tar.gz && nohup helpers/start.sh "node src/server.js"', + ]); + $this->assertEquals(201, $response['headers']['status-code']); + + $result = $this->executeStreamed($runtimeId, [ + 'headers' => ['x-response-size' => '1048576'], + ]); + + $this->assertEquals('0.12.0', $result['headers']['x-executor-response-format'] ?? null); + + \preg_match('/boundary=(.+)$/', (string) ($result['headers']['content-type'] ?? ''), $m); + $parsed = $this->parse(\trim($m[1] ?? '', '"'), $result['wire']); + + $this->assertGreaterThanOrEqual(1048576, \strlen($parsed['parts']['body'])); + $this->assertSame('200', $parsed['parts']['statusCode']); + + // The whole point: a body this size is forwarded as it arrives, so it is framed as many + // runs. One run would mean the executor buffered it after all. + $this->assertGreaterThan( + 1, + $parsed['runs']['body'], + 'body was framed as a single run, so it was buffered rather than streamed' + ); + + // And it arrived over several socket reads on the caller's side too. + $this->assertGreaterThan(1, $result['chunks'], 'response arrived in one read'); + + $this->client->call(Client::METHOD_DELETE, '/runtimes/' . $runtimeId, [], []); + } + + public function testOlderFormatStillGetsBufferedDocument(): void + { + $runtimeId = 'stream-compat-' . \bin2hex(\random_bytes(4)); + $buildPath = $this->build('php-mock', 'openruntimes/php:v5-8.1', 'index.php', 'composer install', $runtimeId); + + $response = $this->client->call(Client::METHOD_POST, '/runtimes', [], [ + 'runtimeId' => $runtimeId, + 'source' => $buildPath, + 'entrypoint' => 'index.php', + 'image' => 'openruntimes/php:v5-8.1', + 'runtimeEntrypoint' => 'cp /tmp/code.tar.gz /mnt/code/code.tar.gz && nohup helpers/start.sh "php src/server.php"', + ]); + $this->assertEquals(201, $response['headers']['status-code']); + + $result = $this->executeStreamed($runtimeId, [], '0.11.0'); + + // No echo, so a caller keeps buffering, and no part carries a chunked encoding. + $this->assertArrayNotHasKey('x-executor-response-format', $result['headers']); + $this->assertStringNotContainsStringIgnoringCase('Content-Transfer-Encoding: chunked', $result['wire']); + $this->assertStringContainsString('Hello Open Runtimes', $result['wire']); + + $this->client->call(Client::METHOD_DELETE, '/runtimes/' . $runtimeId, [], []); + } +} diff --git a/tests/resources/functions/node-large-response/index.js b/tests/resources/functions/node-large-response/index.js new file mode 100644 index 00000000..19d9ce7e --- /dev/null +++ b/tests/resources/functions/node-large-response/index.js @@ -0,0 +1,14 @@ +// Returns a body far larger than a single socket read, so the executor's streaming format has to +// frame it as several length prefixed runs rather than one. +module.exports = async (context) => { + const size = parseInt(context.req.headers['x-response-size'] ?? '1048576', 10); + const line = 'A'.repeat(63) + '\n'; + let body = ''; + while (body.length < size) { + body += line; + } + + context.log('Produced ' + body.length + ' bytes'); + + return context.res.text(body, 200, { 'content-type': 'text/plain' }); +} diff --git a/tests/resources/functions/node-large-response/package.json b/tests/resources/functions/node-large-response/package.json new file mode 100644 index 00000000..aee98b8b --- /dev/null +++ b/tests/resources/functions/node-large-response/package.json @@ -0,0 +1,13 @@ +{ + "name": "code", + "version": "1.0.0", + "description": "", + "main": "index.js", + "scripts": { + "test": "echo \"Error: no test specified\" && exit 1" + }, + "keywords": [], + "author": "", + "license": "ISC", + "dependencies": {} +} From fe24b5092f832ba1cb096079d4da3200750bd51d Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Sun, 23 Aug 2026 21:50:27 +0530 Subject: [PATCH 03/18] style: satisfy phpstan and rector on the streamed response tests Extract the regex capture into a helper that returns a string, so the offset access is typed rather than assumed, and take rector's instanceof narrowing and spacing. --- app/controllers.php | 2 +- src/Executor/BodyMultipartStream.php | 2 +- tests/e2e/StreamedResponseTest.php | 23 ++++++++++++++----- .../unit/Executor/BodyMultipartStreamTest.php | 1 + 4 files changed, 20 insertions(+), 8 deletions(-) diff --git a/app/controllers.php b/app/controllers.php index 02cc9a05..3addaaea 100644 --- a/app/controllers.php +++ b/app/controllers.php @@ -288,7 +288,7 @@ function (string $bytes) use ($response): void { // The body left the executor already, so only the trailing metadata is still owed. A // response with no body at all never commits to streaming and falls through below. - if ($stream !== null && $streamed) { + if ($stream instanceof BodyMultipartStream && $streamed) { $stream->endPart(); foreach (['logs', 'errors', 'duration', 'startTime'] as $key) { diff --git a/src/Executor/BodyMultipartStream.php b/src/Executor/BodyMultipartStream.php index 0c4bc11a..0931853b 100644 --- a/src/Executor/BodyMultipartStream.php +++ b/src/Executor/BodyMultipartStream.php @@ -27,10 +27,10 @@ class BodyMultipartStream { private bool $ended = false; + private bool $inPart = false; /** - * @param string $boundary * @param callable(string): void $write Receives raw bytes to put on the wire. */ public function __construct( diff --git a/tests/e2e/StreamedResponseTest.php b/tests/e2e/StreamedResponseTest.php index d82295f1..5a42c25c 100644 --- a/tests/e2e/StreamedResponseTest.php +++ b/tests/e2e/StreamedResponseTest.php @@ -30,6 +30,17 @@ protected function setUp(): void $this->client->setKey($this->key); } + /** + * Returns the first capture group, failing the test if the pattern does not match. + */ + private function captureGroup(string $pattern, string $subject, string $message): string + { + $matches = []; + $this->assertSame(1, \preg_match($pattern, $subject, $matches), $message); + + return $matches[1] ?? ''; + } + /** * Reads the 0.12.0 envelope. Mirrors the reader on the caller's side: content is located by its * length prefix and never by scanning for the boundary. @@ -61,8 +72,7 @@ private function parse(string $boundary, string $wire): array $offset = $end + 4; $this->assertMatchesRegularExpression('/Content-Transfer-Encoding:\s*chunked/i', $headers, 'part is not chunked: ' . $headers); - $this->assertSame(1, \preg_match('/name="([^"]+)"/', $headers, $m), 'part has no name: ' . $headers); - $name = $m[1]; + $name = $this->captureGroup('/name="([^"]+)"/', $headers, 'part has no name: ' . $headers); $order[] = $name; $parts[$name] = ''; @@ -117,6 +127,7 @@ private function build(string $folder, string $image, string $entrypoint, string } /** + * @param array $params * @return array{wire: string, headers: array, chunks: int} */ private function executeStreamed(string $runtimeId, array $params, string $format = '0.12.0'): array @@ -164,8 +175,7 @@ public function testStreamedEnvelopeIsFramedAndOrdered(): void $contentType = $result['headers']['content-type'] ?? ''; $this->assertStringStartsWith('multipart/form-data', (string) $contentType); - $this->assertSame(1, \preg_match('/boundary=(.+)$/', (string) $contentType, $m), 'no boundary in ' . $contentType); - $boundary = \trim($m[1], '"'); + $boundary = \trim($this->captureGroup('/boundary=(.+)$/', (string) $contentType, 'no boundary in ' . $contentType), '"'); $parsed = $this->parse($boundary, $result['wire']); @@ -218,8 +228,9 @@ public function testLargeBodyIsFramedAsSeveralRuns(): void $this->assertEquals('0.12.0', $result['headers']['x-executor-response-format'] ?? null); - \preg_match('/boundary=(.+)$/', (string) ($result['headers']['content-type'] ?? ''), $m); - $parsed = $this->parse(\trim($m[1] ?? '', '"'), $result['wire']); + $contentType = (string) ($result['headers']['content-type'] ?? ''); + $boundary = \trim($this->captureGroup('/boundary=(.+)$/', $contentType, 'no boundary in ' . $contentType), '"'); + $parsed = $this->parse($boundary, $result['wire']); $this->assertGreaterThanOrEqual(1048576, \strlen($parsed['parts']['body'])); $this->assertSame('200', $parsed['parts']['statusCode']); diff --git a/tests/unit/Executor/BodyMultipartStreamTest.php b/tests/unit/Executor/BodyMultipartStreamTest.php index 5ad35ee1..66250593 100644 --- a/tests/unit/Executor/BodyMultipartStreamTest.php +++ b/tests/unit/Executor/BodyMultipartStreamTest.php @@ -111,6 +111,7 @@ public function testWritesAfterEndAreIgnored(): void $stream = $this->writer(); $stream->part('logs', 'x'); $stream->end(); + $after = $this->wire; $stream->part('errors', 'ignored'); From be5f67fe51f29da9d96af7912644c9ccd4ab79d4 Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Sun, 23 Aug 2026 23:17:07 +0530 Subject: [PATCH 04/18] style: trim comments to the density of surrounding code The writer carried a twenty-two line class docblock and method docblocks that restated their own names, at roughly eight times the comment density of BodyMultipart beside it. Keep the wire format, the reason content is length prefixed, and the contracts that are not visible from a signature. --- src/Executor/BodyMultipartStream.php | 35 ++++------------------------ tests/e2e/StreamedResponseTest.php | 3 --- 2 files changed, 4 insertions(+), 34 deletions(-) diff --git a/src/Executor/BodyMultipartStream.php b/src/Executor/BodyMultipartStream.php index 0931853b..0e207a55 100644 --- a/src/Executor/BodyMultipartStream.php +++ b/src/Executor/BodyMultipartStream.php @@ -3,26 +3,10 @@ namespace OpenRuntimes\Executor; /** - * Incremental writer for the streaming multipart response format (x-executor-response-format - * 0.12.0 and above). - * - * BodyMultipart serialises a finished document, which means the whole response has to exist in - * memory before a single byte reaches the client. This writes the envelope as parts are produced, - * so a response can leave the executor while the runtime is still generating it. - * - * Part content is length prefixed, in the same shape HTTP uses for chunked transfer encoding: - * - * --BOUNDARY\r\n - * Content-Disposition: form-data; name="body"\r\n - * Content-Transfer-Encoding: chunked\r\n - * \r\n - * \r\n\r\n (repeated for each run of content) - * 0\r\n\r\n (terminates the part) - * --BOUNDARY-- (after the last part) - * - * The length prefix is what makes an incremental read tractable on the other side: content is - * never scanned for the boundary, so content that happens to contain the boundary string cannot - * split the envelope, and the reader needs no lookahead between reads. + * Writes the multipart response format of x-executor-response-format 0.12.0, where BodyMultipart + * serialises a finished document. Part content is length prefixed as + * `\r\n\r\n`, closed by `0\r\n\r\n`, so the reader locates content by its + * length and never scans for the boundary. */ class BodyMultipartStream { @@ -44,9 +28,6 @@ public function exportHeader(): string return 'multipart/form-data; boundary=' . $this->boundary; } - /** - * Open a part. Content follows via writeContent(), and the part is closed by endPart(). - */ public function startPart(string $name): void { if ($this->ended || $this->inPart) { @@ -62,9 +43,6 @@ public function startPart(string $name): void ); } - /** - * Append a run of content to the open part. - */ public function writeContent(string $content): void { // A zero length run is the part terminator, so an empty write has to be dropped rather @@ -88,8 +66,6 @@ public function endPart(): void } /** - * Write a part whose content is already known in full. - * * @param mixed $value Scalars are stringified; arrays are JSON encoded, matching BodyMultipart. */ public function part(string $name, mixed $value): void @@ -99,9 +75,6 @@ public function part(string $name, mixed $value): void $this->endPart(); } - /** - * Close the envelope. Any part still open is terminated first. - */ public function end(): void { if ($this->ended) { diff --git a/tests/e2e/StreamedResponseTest.php b/tests/e2e/StreamedResponseTest.php index 5a42c25c..81254a79 100644 --- a/tests/e2e/StreamedResponseTest.php +++ b/tests/e2e/StreamedResponseTest.php @@ -30,9 +30,6 @@ protected function setUp(): void $this->client->setKey($this->key); } - /** - * Returns the first capture group, failing the test if the pattern does not match. - */ private function captureGroup(string $pattern, string $subject, string $message): string { $matches = []; From 6b57df4cf373aa596f680bb25f87e6dae3319b35 Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Sun, 23 Aug 2026 23:24:20 +0530 Subject: [PATCH 05/18] fix: abort the connection when an execution fails mid-stream A curl failure after the body has begun streaming throws out of createExecution, and the error hook then appends its JSON document to a response whose headers and content are already on the wire. The caller reads those bytes as envelope content, so the error leaks into the visitor-facing body before the truncated envelope is finally rejected. Once the first byte is out, close the connection instead: a broken transfer is the one remaining way to report the failure, and the caller already treats an envelope without its closing delimiter as a failed execution. --- app/controllers.php | 57 +++++++++++++++++++++++++++++---------------- 1 file changed, 37 insertions(+), 20 deletions(-) diff --git a/app/controllers.php b/app/controllers.php index 3addaaea..a59e8aab 100644 --- a/app/controllers.php +++ b/app/controllers.php @@ -10,6 +10,7 @@ use Utopia\Http\Request; use Utopia\Http\Http; use Utopia\Http\Response; +use Utopia\Http\Adapter\Swoole\Response as SwooleResponse; use Utopia\Validator\AnyOf; use Utopia\Validator\Assoc; use Utopia\Validator\Boolean; @@ -234,11 +235,13 @@ function ( // shape, still get the complete document they expect. $stream = null; $onStream = null; + $streamStarted = false; if (!$isJson && \version_compare($responseFormat, RESPONSE_FORMAT_STREAM, '>=')) { $stream = new BodyMultipartStream( BodyMultipart::generateBoundary(), - function (string $bytes) use ($response): void { + function (string $bytes) use ($response, &$streamStarted): void { + $streamStarted = true; $response->chunk($bytes); } ); @@ -263,25 +266,39 @@ function (string $bytes) use ($response): void { }; } - $execution = $runner->createExecution( - $runtimeId, - $payload, - $path, - $method, - $headers, - $timeout, - $image, - $source, - $entrypoint, - $variables, - $cpus, - $memory, - $version, - $runtimeEntrypoint, - $logging, - $restartPolicy, - onStream: $onStream, - ); + try { + $execution = $runner->createExecution( + $runtimeId, + $payload, + $path, + $method, + $headers, + $timeout, + $image, + $source, + $entrypoint, + $variables, + $cpus, + $memory, + $version, + $runtimeEntrypoint, + $logging, + $restartPolicy, + onStream: $onStream, + ); + } catch (\Throwable $throwable) { + if (!$streamStarted) { + throw $throwable; + } + + // Content is committed, so the error hook's JSON document would be read as part + // of the envelope. A broken transfer is the remaining way to report the failure. + if ($response instanceof SwooleResponse) { + $response->getSwooleResponse()->close(); + } + + return; + } $streamed = ($execution['streamed'] ?? false) === true; unset($execution['streamed']); From 6fe38c386683568bae03862b4c1133e5f5702836 Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Sun, 23 Aug 2026 23:27:45 +0530 Subject: [PATCH 06/18] style: match the large-response fixture to its siblings --- .../functions/node-large-response/index.js | 4 ---- .../node-large-response/package.json | 21 +++++++++---------- 2 files changed, 10 insertions(+), 15 deletions(-) diff --git a/tests/resources/functions/node-large-response/index.js b/tests/resources/functions/node-large-response/index.js index 19d9ce7e..b1fdf95c 100644 --- a/tests/resources/functions/node-large-response/index.js +++ b/tests/resources/functions/node-large-response/index.js @@ -1,5 +1,3 @@ -// Returns a body far larger than a single socket read, so the executor's streaming format has to -// frame it as several length prefixed runs rather than one. module.exports = async (context) => { const size = parseInt(context.req.headers['x-response-size'] ?? '1048576', 10); const line = 'A'.repeat(63) + '\n'; @@ -8,7 +6,5 @@ module.exports = async (context) => { body += line; } - context.log('Produced ' + body.length + ' bytes'); - return context.res.text(body, 200, { 'content-type': 'text/plain' }); } diff --git a/tests/resources/functions/node-large-response/package.json b/tests/resources/functions/node-large-response/package.json index aee98b8b..1d470015 100644 --- a/tests/resources/functions/node-large-response/package.json +++ b/tests/resources/functions/node-large-response/package.json @@ -1,13 +1,12 @@ { - "name": "code", - "version": "1.0.0", - "description": "", - "main": "index.js", - "scripts": { - "test": "echo \"Error: no test specified\" && exit 1" - }, - "keywords": [], - "author": "", - "license": "ISC", - "dependencies": {} + "name": "large-response", + "version": "1.0.0", + "description": "", + "main": "index.js", + "scripts": { + "test": "echo \"Error: no test specified\" && exit 1" + }, + "keywords": [], + "author": "", + "license": "ISC" } From d7bc4d27691adefdceef91a8fd9809ac583671e0 Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Sun, 23 Aug 2026 23:40:06 +0530 Subject: [PATCH 07/18] fix: fail a part that cannot be encoded instead of writing it empty A header value that is not valid UTF-8 makes json_encode return false, which was written as a well-formed zero-length headers part. The caller then served the page with none of its headers. Encode before anything is written, so the failure surfaces while the response is still recoverable. --- src/Executor/BodyMultipartStream.php | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/src/Executor/BodyMultipartStream.php b/src/Executor/BodyMultipartStream.php index 0e207a55..8a5fe863 100644 --- a/src/Executor/BodyMultipartStream.php +++ b/src/Executor/BodyMultipartStream.php @@ -70,8 +70,17 @@ public function endPart(): void */ public function part(string $name, mixed $value): void { + if (\is_array($value)) { + $value = \json_encode($value); + + // Nothing has been written yet, so the failure can still surface as an error. + if ($value === false) { + throw new \Exception('Part "' . $name . '" could not be encoded'); + } + } + $this->startPart($name); - $this->writeContent(\is_array($value) ? (\json_encode($value) ?: '') : \strval($value)); + $this->writeContent(\strval($value)); $this->endPart(); } From 605f532321ca96bfa6286c931e4049ac1af535c7 Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Mon, 24 Aug 2026 18:29:19 +0530 Subject: [PATCH 08/18] style: keep added comments to one line, as the surrounding files do --- app/controllers.php | 12 +++--------- src/Executor/BodyMultipartStream.php | 3 +-- src/Executor/Runner/Docker.php | 14 ++++---------- 3 files changed, 8 insertions(+), 21 deletions(-) diff --git a/app/controllers.php b/app/controllers.php index a59e8aab..287fc930 100644 --- a/app/controllers.php +++ b/app/controllers.php @@ -230,9 +230,6 @@ function ( $acceptTypes = \explode(', ', $request->getHeaderLine('accept') ?: 'multipart/form-data'); $isJson = array_any($acceptTypes, fn ($acceptType): bool => \str_starts_with((string) $acceptType, 'application/json') || \str_starts_with((string) $acceptType, 'application/*')); - // A caller on 0.12.0 or later can read parts as they are framed, so the runtime's body - // can be forwarded while it is still being produced. Anything older, and the JSON - // shape, still get the complete document they expect. $stream = null; $onStream = null; $streamStarted = false; @@ -253,8 +250,7 @@ function (string $bytes) use ($response, &$streamStarted): void { return; } - // Last chance to set response headers: the first content run follows straight - // after, and chunk() commits them on its first call. + // Last chance to set headers: chunk() commits them on its first call. $response ->setStatusCode(Response::STATUS_CODE_OK) ->addHeader('content-type', $stream->exportHeader()) @@ -291,8 +287,7 @@ function (string $bytes) use ($response, &$streamStarted): void { throw $throwable; } - // Content is committed, so the error hook's JSON document would be read as part - // of the envelope. A broken transfer is the remaining way to report the failure. + // Content is committed, so the error hook's JSON would land inside the envelope. if ($response instanceof SwooleResponse) { $response->getSwooleResponse()->close(); } @@ -303,8 +298,7 @@ function (string $bytes) use ($response, &$streamStarted): void { $streamed = ($execution['streamed'] ?? false) === true; unset($execution['streamed']); - // The body left the executor already, so only the trailing metadata is still owed. A - // response with no body at all never commits to streaming and falls through below. + // The body already left, so only the trailing metadata is still owed. if ($stream instanceof BodyMultipartStream && $streamed) { $stream->endPart(); diff --git a/src/Executor/BodyMultipartStream.php b/src/Executor/BodyMultipartStream.php index 8a5fe863..06dabe5f 100644 --- a/src/Executor/BodyMultipartStream.php +++ b/src/Executor/BodyMultipartStream.php @@ -45,8 +45,7 @@ public function startPart(string $name): void public function writeContent(string $content): void { - // A zero length run is the part terminator, so an empty write has to be dropped rather - // than forwarded. Callers stream whatever curl hands them, and curl can hand over nothing. + // A zero length run is the part terminator, so an empty write has to be dropped. if ($this->ended || !$this->inPart || $content === '') { return; } diff --git a/src/Executor/Runner/Docker.php b/src/Executor/Runner/Docker.php index 54dc2842..279eb5ee 100644 --- a/src/Executor/Runner/Docker.php +++ b/src/Executor/Runner/Docker.php @@ -890,8 +890,7 @@ public function createExecution( ]; }; - // Set once the runtime's body has begun leaving the executor. From that point the response - // is committed: it can be neither retried nor turned into an error. + // Once set, the response is committed: it can be neither retried nor turned into an error. $streamed = false; $executeV5 = function () use ($path, $method, $headers, $payload, $secret, $hostname, $timeout, $runtimeName, $logging, $onStream, &$streamed): array { @@ -947,9 +946,7 @@ public function createExecution( }); if ($onStream !== null) { - // Forward the runtime's body as it arrives instead of accumulating it. The header - // callback above has already run by the time the first body byte lands, so status - // and headers are known and can be announced before any content goes out. + // The header callback has already run, so status and headers are known here. \curl_setopt($ch, CURLOPT_WRITEFUNCTION, function ($curl, $data) use ($onStream, &$responseHeaders, &$streamed): int { if (!$streamed) { $streamed = true; @@ -991,8 +988,7 @@ public function createExecution( \curl_setopt($ch, CURLOPT_HEADEROPT, CURLHEADER_UNIFIED); \curl_setopt($ch, CURLOPT_HTTPHEADER, $headersArr); - // With a write callback installed curl_exec returns a bool and the body has already - // been forwarded, so there is nothing left to keep here. + // With a write callback the body is already forwarded, so curl_exec returns a bool. $result = \curl_exec($ch); $executorResponse = \is_string($result) ? $result : ''; @@ -1126,9 +1122,7 @@ public function createExecution( break; } - // Content is already on the wire, so a second attempt would append a second response - // to it. The retryable errors here all happen before a connection is established, so - // this is a guard rather than an expected path. + // A retry would append a second response to content already on the wire. if ($streamed) { break; } From 4da6cf66bda48e2b21fc378007fdc0db34678133 Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Mon, 24 Aug 2026 18:38:09 +0530 Subject: [PATCH 09/18] refactor: drop the redundant streamed flag and unused isEnded The runner returned a 'streamed' key that the controller read once and then unset before serialising, while a local already tracked the same fact from the one place that can know it: whether any byte reached the write callback. That local is also the more accurate of the two, since a part that fails to encode throws before writing anything. isEnded() had no caller outside the assertion that read it. --- app/controllers.php | 5 +---- src/Executor/BodyMultipartStream.php | 5 ----- src/Executor/Runner/Docker.php | 2 -- tests/unit/Executor/BodyMultipartStreamTest.php | 1 - 4 files changed, 1 insertion(+), 12 deletions(-) diff --git a/app/controllers.php b/app/controllers.php index 287fc930..966ed1a5 100644 --- a/app/controllers.php +++ b/app/controllers.php @@ -295,11 +295,8 @@ function (string $bytes) use ($response, &$streamStarted): void { return; } - $streamed = ($execution['streamed'] ?? false) === true; - unset($execution['streamed']); - // The body already left, so only the trailing metadata is still owed. - if ($stream instanceof BodyMultipartStream && $streamed) { + if ($stream instanceof BodyMultipartStream && $streamStarted) { $stream->endPart(); foreach (['logs', 'errors', 'duration', 'startTime'] as $key) { diff --git a/src/Executor/BodyMultipartStream.php b/src/Executor/BodyMultipartStream.php index 06dabe5f..b0048a07 100644 --- a/src/Executor/BodyMultipartStream.php +++ b/src/Executor/BodyMultipartStream.php @@ -94,9 +94,4 @@ public function end(): void ($this->write)('--' . $this->boundary . '--'); } - - public function isEnded(): bool - { - return $this->ended; - } } diff --git a/src/Executor/Runner/Docker.php b/src/Executor/Runner/Docker.php index 279eb5ee..0bd806d2 100644 --- a/src/Executor/Runner/Docker.php +++ b/src/Executor/Runner/Docker.php @@ -1159,8 +1159,6 @@ public function createExecution( 'errors' => $errors, 'duration' => $duration, 'startTime' => $startTime, - // Lets the caller tell an empty body apart from one that has already been forwarded. - 'streamed' => $streamed, ]; // Update swoole table diff --git a/tests/unit/Executor/BodyMultipartStreamTest.php b/tests/unit/Executor/BodyMultipartStreamTest.php index 66250593..900b3c5c 100644 --- a/tests/unit/Executor/BodyMultipartStreamTest.php +++ b/tests/unit/Executor/BodyMultipartStreamTest.php @@ -93,7 +93,6 @@ public function testEndClosesEnvelopeWithTerminalDelimiter(): void $stream->end(); $this->assertStringEndsWith('--BOUNDARY--', $this->wire); - $this->assertTrue($stream->isEnded()); } public function testEndTerminatesAnOpenPartFirst(): void From ffaa796ea8ee6f1edeea8ab2527bb87dcba50bef Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Mon, 24 Aug 2026 18:44:23 +0530 Subject: [PATCH 10/18] refactor: type the writer callback as Closure Every caller already passes a closure, and mixed only existed because a property cannot be declared callable. --- src/Executor/BodyMultipartStream.php | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Executor/BodyMultipartStream.php b/src/Executor/BodyMultipartStream.php index b0048a07..caf6eb2e 100644 --- a/src/Executor/BodyMultipartStream.php +++ b/src/Executor/BodyMultipartStream.php @@ -15,11 +15,11 @@ class BodyMultipartStream private bool $inPart = false; /** - * @param callable(string): void $write Receives raw bytes to put on the wire. + * @param \Closure(string): void $write Receives raw bytes to put on the wire. */ public function __construct( private readonly string $boundary, - private readonly mixed $write, + private readonly \Closure $write, ) { } From 5dedc977fee421100462d9f0eb17103a1c9c943a Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Mon, 24 Aug 2026 18:52:42 +0530 Subject: [PATCH 11/18] test: assert on the wire where the looser checks could not fail Removing the empty-run guard from writeContent left the suite green: a substring check for the orphaned content could not match, because the mutant frames a spurious terminator ahead of the next length prefix rather than raw content. Asserting the whole wire catches it, and matches how the neighbouring tests read. The two array_search checks restated indices the full order assertion above them already fixes. The chunk counter measured how curl handed bytes to the client, which is per socket read whether or not the executor streamed, so it held for a buffered response too. --- tests/e2e/StreamedResponseTest.php | 17 +++++------------ .../unit/Executor/BodyMultipartStreamTest.php | 19 +++++++++++++++---- 2 files changed, 20 insertions(+), 16 deletions(-) diff --git a/tests/e2e/StreamedResponseTest.php b/tests/e2e/StreamedResponseTest.php index 81254a79..d1fb1b81 100644 --- a/tests/e2e/StreamedResponseTest.php +++ b/tests/e2e/StreamedResponseTest.php @@ -125,12 +125,11 @@ private function build(string $folder, string $image, string $entrypoint, string /** * @param array $params - * @return array{wire: string, headers: array, chunks: int} + * @return array{wire: string, headers: array} */ private function executeStreamed(string $runtimeId, array $params, string $format = '0.12.0'): array { $wire = ''; - $chunks = 0; $response = $this->client->call( Client::METHOD_POST, @@ -141,13 +140,12 @@ private function executeStreamed(string $runtimeId, array $params, string $forma ], $params, false, - function (string $data) use (&$wire, &$chunks): void { + function (string $data) use (&$wire): void { $wire .= $data; - $chunks++; } ); - return ['wire' => $wire, 'headers' => $response['headers'], 'chunks' => $chunks]; + return ['wire' => $wire, 'headers' => $response['headers']]; } public function testStreamedEnvelopeIsFramedAndOrdered(): void @@ -179,14 +177,11 @@ public function testStreamedEnvelopeIsFramedAndOrdered(): void $this->assertSame( ['statusCode', 'headers', 'body', 'logs', 'errors', 'duration', 'startTime'], $parsed['order'], + // statusCode and headers precede the body: the proxy cannot set either once content + // has started going out to the visitor. 'unexpected part order' ); - // statusCode and headers must precede the body: the proxy cannot set either once content - // has started going out to the visitor. - $this->assertSame(0, \array_search('statusCode', $parsed['order'], true)); - $this->assertSame(1, \array_search('headers', $parsed['order'], true)); - $this->assertSame('200', $parsed['parts']['statusCode']); $headers = \json_decode($parsed['parts']['headers'], true); @@ -240,8 +235,6 @@ public function testLargeBodyIsFramedAsSeveralRuns(): void 'body was framed as a single run, so it was buffered rather than streamed' ); - // And it arrived over several socket reads on the caller's side too. - $this->assertGreaterThan(1, $result['chunks'], 'response arrived in one read'); $this->client->call(Client::METHOD_DELETE, '/runtimes/' . $runtimeId, [], []); } diff --git a/tests/unit/Executor/BodyMultipartStreamTest.php b/tests/unit/Executor/BodyMultipartStreamTest.php index 900b3c5c..0e08c67e 100644 --- a/tests/unit/Executor/BodyMultipartStreamTest.php +++ b/tests/unit/Executor/BodyMultipartStreamTest.php @@ -73,8 +73,14 @@ public function testEmptyContentRunIsDroppedRatherThanTerminatingThePart(): void $stream->endPart(); // A zero length run would read as the part terminator, orphaning everything after it. - $this->assertStringNotContainsString("0\r\n\r\nafter", $this->wire); - $this->assertStringContainsString("5\r\nafter\r\n0\r\n\r\n", $this->wire); + $this->assertSame( + "--BOUNDARY\r\n" + . "Content-Disposition: form-data; name=\"body\"\r\n" + . "Content-Transfer-Encoding: chunked\r\n\r\n" + . "5\r\nafter\r\n" + . "0\r\n\r\n", + $this->wire + ); } public function testEmptyPartStillFramesSoTheKeyExists(): void @@ -82,8 +88,13 @@ public function testEmptyPartStillFramesSoTheKeyExists(): void $stream = $this->writer(); $stream->part('errors', ''); - $this->assertStringContainsString('name="errors"', $this->wire); - $this->assertStringEndsWith("0\r\n\r\n", $this->wire); + $this->assertSame( + "--BOUNDARY\r\n" + . "Content-Disposition: form-data; name=\"errors\"\r\n" + . "Content-Transfer-Encoding: chunked\r\n\r\n" + . "0\r\n\r\n", + $this->wire + ); } public function testEndClosesEnvelopeWithTerminalDelimiter(): void From 1f0a825f99d391aced8a2442fd0b78067c4d47e8 Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Mon, 24 Aug 2026 19:20:50 +0530 Subject: [PATCH 12/18] test: pin that an abandoned envelope stays unclosed An execution that dies mid-part never calls end(), and the caller reads a missing closing delimiter as a failed execution. Nothing may close the envelope on its behalf, so hold that: adding a destructor that finalises makes this fail. --- tests/unit/Executor/BodyMultipartStreamTest.php | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tests/unit/Executor/BodyMultipartStreamTest.php b/tests/unit/Executor/BodyMultipartStreamTest.php index 0e08c67e..03697893 100644 --- a/tests/unit/Executor/BodyMultipartStreamTest.php +++ b/tests/unit/Executor/BodyMultipartStreamTest.php @@ -139,6 +139,18 @@ public function testNestedStartPartIsIgnored(): void $this->assertStringNotContainsString('name="other"', $this->wire); } + public function testAbandonedEnvelopeCarriesNoClosingDelimiter(): void + { + $stream = $this->writer(); + $stream->startPart('body'); + $stream->writeContent('half a page'); + unset($stream); + + // An execution that dies here never calls end(). The caller reads a missing closing + // delimiter as a failed execution, so nothing may close the envelope on its behalf. + $this->assertStringNotContainsString('--BOUNDARY--', $this->wire); + } + public function testExportHeaderCarriesBoundary(): void { $stream = $this->writer('abc123'); From 1e4d5b59a4604f69281f9d46e402e69d55e07765 Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Mon, 24 Aug 2026 19:31:16 +0530 Subject: [PATCH 13/18] test: inline the regex capture helper A generic capture-with-assertion wrapper had no counterpart anywhere in the suite, and two of its three callers were extracting the boundary, which the code under test does with a plain explode. --- tests/e2e/StreamedResponseTest.php | 16 +++++----------- 1 file changed, 5 insertions(+), 11 deletions(-) diff --git a/tests/e2e/StreamedResponseTest.php b/tests/e2e/StreamedResponseTest.php index d1fb1b81..989b977f 100644 --- a/tests/e2e/StreamedResponseTest.php +++ b/tests/e2e/StreamedResponseTest.php @@ -30,14 +30,6 @@ protected function setUp(): void $this->client->setKey($this->key); } - private function captureGroup(string $pattern, string $subject, string $message): string - { - $matches = []; - $this->assertSame(1, \preg_match($pattern, $subject, $matches), $message); - - return $matches[1] ?? ''; - } - /** * Reads the 0.12.0 envelope. Mirrors the reader on the caller's side: content is located by its * length prefix and never by scanning for the boundary. @@ -69,7 +61,9 @@ private function parse(string $boundary, string $wire): array $offset = $end + 4; $this->assertMatchesRegularExpression('/Content-Transfer-Encoding:\s*chunked/i', $headers, 'part is not chunked: ' . $headers); - $name = $this->captureGroup('/name="([^"]+)"/', $headers, 'part has no name: ' . $headers); + $matches = []; + $this->assertSame(1, \preg_match('/name="([^"]+)"/', $headers, $matches), 'part has no name: ' . $headers); + $name = $matches[1] ?? ''; $order[] = $name; $parts[$name] = ''; @@ -170,7 +164,7 @@ public function testStreamedEnvelopeIsFramedAndOrdered(): void $contentType = $result['headers']['content-type'] ?? ''; $this->assertStringStartsWith('multipart/form-data', (string) $contentType); - $boundary = \trim($this->captureGroup('/boundary=(.+)$/', (string) $contentType, 'no boundary in ' . $contentType), '"'); + $boundary = \trim(\explode('boundary=', (string) $contentType)[1] ?? '', '"'); $parsed = $this->parse($boundary, $result['wire']); @@ -221,7 +215,7 @@ public function testLargeBodyIsFramedAsSeveralRuns(): void $this->assertEquals('0.12.0', $result['headers']['x-executor-response-format'] ?? null); $contentType = (string) ($result['headers']['content-type'] ?? ''); - $boundary = \trim($this->captureGroup('/boundary=(.+)$/', $contentType, 'no boundary in ' . $contentType), '"'); + $boundary = \trim(\explode('boundary=', $contentType)[1] ?? '', '"'); $parsed = $this->parse($boundary, $result['wire']); $this->assertGreaterThanOrEqual(1048576, \strlen($parsed['parts']['body'])); From 9aa7efa3221200005bb289ec63e5c6122fedace7 Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Tue, 25 Aug 2026 12:29:00 +0530 Subject: [PATCH 14/18] refactor: name the writer for what it does Stream said nothing about direction, which is how the same name ended up on both this and the reader that consumes it. Every class here is a plain noun or names its role, as StorageFactory and ImagePuller do, and none is a Stream. --- app/controllers.php | 6 +++--- .../{BodyMultipartStream.php => BodyMultipartWriter.php} | 2 +- ...ultipartStreamTest.php => BodyMultipartWriterTest.php} | 8 ++++---- 3 files changed, 8 insertions(+), 8 deletions(-) rename src/Executor/{BodyMultipartStream.php => BodyMultipartWriter.php} (98%) rename tests/unit/Executor/{BodyMultipartStreamTest.php => BodyMultipartWriterTest.php} (96%) diff --git a/app/controllers.php b/app/controllers.php index 966ed1a5..af94082d 100644 --- a/app/controllers.php +++ b/app/controllers.php @@ -4,7 +4,7 @@ use OpenRuntimes\Executor\Exception; use OpenRuntimes\Executor\BodyMultipart; -use OpenRuntimes\Executor\BodyMultipartStream; +use OpenRuntimes\Executor\BodyMultipartWriter; use OpenRuntimes\Executor\Runner\Adapter as Runner; use Utopia\System\System; use Utopia\Http\Request; @@ -235,7 +235,7 @@ function ( $streamStarted = false; if (!$isJson && \version_compare($responseFormat, RESPONSE_FORMAT_STREAM, '>=')) { - $stream = new BodyMultipartStream( + $stream = new BodyMultipartWriter( BodyMultipart::generateBoundary(), function (string $bytes) use ($response, &$streamStarted): void { $streamStarted = true; @@ -296,7 +296,7 @@ function (string $bytes) use ($response, &$streamStarted): void { } // The body already left, so only the trailing metadata is still owed. - if ($stream instanceof BodyMultipartStream && $streamStarted) { + if ($stream instanceof BodyMultipartWriter && $streamStarted) { $stream->endPart(); foreach (['logs', 'errors', 'duration', 'startTime'] as $key) { diff --git a/src/Executor/BodyMultipartStream.php b/src/Executor/BodyMultipartWriter.php similarity index 98% rename from src/Executor/BodyMultipartStream.php rename to src/Executor/BodyMultipartWriter.php index caf6eb2e..e26cdc32 100644 --- a/src/Executor/BodyMultipartStream.php +++ b/src/Executor/BodyMultipartWriter.php @@ -8,7 +8,7 @@ * `\r\n\r\n`, closed by `0\r\n\r\n`, so the reader locates content by its * length and never scans for the boundary. */ -class BodyMultipartStream +class BodyMultipartWriter { private bool $ended = false; diff --git a/tests/unit/Executor/BodyMultipartStreamTest.php b/tests/unit/Executor/BodyMultipartWriterTest.php similarity index 96% rename from tests/unit/Executor/BodyMultipartStreamTest.php rename to tests/unit/Executor/BodyMultipartWriterTest.php index 03697893..3bd53e1a 100644 --- a/tests/unit/Executor/BodyMultipartStreamTest.php +++ b/tests/unit/Executor/BodyMultipartWriterTest.php @@ -4,18 +4,18 @@ namespace Tests\Unit\Executor; -use OpenRuntimes\Executor\BodyMultipartStream; +use OpenRuntimes\Executor\BodyMultipartWriter; use PHPUnit\Framework\TestCase; -final class BodyMultipartStreamTest extends TestCase +final class BodyMultipartWriterTest extends TestCase { private string $wire = ''; - private function writer(string $boundary = 'BOUNDARY'): BodyMultipartStream + private function writer(string $boundary = 'BOUNDARY'): BodyMultipartWriter { $this->wire = ''; - return new BodyMultipartStream($boundary, function (string $bytes): void { + return new BodyMultipartWriter($boundary, function (string $bytes): void { $this->wire .= $bytes; }); } From 1009d69840fdbe07118d46d7cb30ec7c23d9c67c Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Tue, 25 Aug 2026 12:38:01 +0530 Subject: [PATCH 15/18] refactor: name the whole-part write for what it does part() was the only bare noun among startPart, writeContent and endPart, and the document class beside it calls its equivalent setPart. --- app/controllers.php | 6 +++--- src/Executor/BodyMultipartWriter.php | 2 +- tests/unit/Executor/BodyMultipartWriterTest.php | 12 ++++++------ 3 files changed, 10 insertions(+), 10 deletions(-) diff --git a/app/controllers.php b/app/controllers.php index af94082d..f227e0c2 100644 --- a/app/controllers.php +++ b/app/controllers.php @@ -256,8 +256,8 @@ function (string $bytes) use ($response, &$streamStarted): void { ->addHeader('content-type', $stream->exportHeader()) ->addHeader('x-executor-response-format', RESPONSE_FORMAT_STREAM); - $stream->part('statusCode', $data['statusCode']); - $stream->part('headers', $data['headers']); + $stream->writePart('statusCode', $data['statusCode']); + $stream->writePart('headers', $data['headers']); $stream->startPart('body'); }; } @@ -300,7 +300,7 @@ function (string $bytes) use ($response, &$streamStarted): void { $stream->endPart(); foreach (['logs', 'errors', 'duration', 'startTime'] as $key) { - $stream->part($key, $execution[$key] ?? ''); + $stream->writePart($key, $execution[$key] ?? ''); } $stream->end(); diff --git a/src/Executor/BodyMultipartWriter.php b/src/Executor/BodyMultipartWriter.php index e26cdc32..56335e17 100644 --- a/src/Executor/BodyMultipartWriter.php +++ b/src/Executor/BodyMultipartWriter.php @@ -67,7 +67,7 @@ public function endPart(): void /** * @param mixed $value Scalars are stringified; arrays are JSON encoded, matching BodyMultipart. */ - public function part(string $name, mixed $value): void + public function writePart(string $name, mixed $value): void { if (\is_array($value)) { $value = \json_encode($value); diff --git a/tests/unit/Executor/BodyMultipartWriterTest.php b/tests/unit/Executor/BodyMultipartWriterTest.php index 3bd53e1a..8be2935d 100644 --- a/tests/unit/Executor/BodyMultipartWriterTest.php +++ b/tests/unit/Executor/BodyMultipartWriterTest.php @@ -23,7 +23,7 @@ private function writer(string $boundary = 'BOUNDARY'): BodyMultipartWriter public function testPartIsLengthPrefixedAndChunkMarked(): void { $stream = $this->writer(); - $stream->part('statusCode', 200); + $stream->writePart('statusCode', 200); $this->assertSame( "--BOUNDARY\r\n" @@ -48,7 +48,7 @@ public function testLengthPrefixIsHexadecimal(): void public function testArrayPartIsJsonEncoded(): void { $stream = $this->writer(); - $stream->part('headers', ['content-type' => 'text/html']); + $stream->writePart('headers', ['content-type' => 'text/html']); $this->assertStringContainsString('{"content-type":"text\/html"}', $this->wire); } @@ -86,7 +86,7 @@ public function testEmptyContentRunIsDroppedRatherThanTerminatingThePart(): void public function testEmptyPartStillFramesSoTheKeyExists(): void { $stream = $this->writer(); - $stream->part('errors', ''); + $stream->writePart('errors', ''); $this->assertSame( "--BOUNDARY\r\n" @@ -100,7 +100,7 @@ public function testEmptyPartStillFramesSoTheKeyExists(): void public function testEndClosesEnvelopeWithTerminalDelimiter(): void { $stream = $this->writer(); - $stream->part('logs', 'hello'); + $stream->writePart('logs', 'hello'); $stream->end(); $this->assertStringEndsWith('--BOUNDARY--', $this->wire); @@ -119,12 +119,12 @@ public function testEndTerminatesAnOpenPartFirst(): void public function testWritesAfterEndAreIgnored(): void { $stream = $this->writer(); - $stream->part('logs', 'x'); + $stream->writePart('logs', 'x'); $stream->end(); $after = $this->wire; - $stream->part('errors', 'ignored'); + $stream->writePart('errors', 'ignored'); $stream->end(); $this->assertSame($after, $this->wire); From cb9f69bf4d9b010d75b459498f4f879e2945c003 Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Tue, 25 Aug 2026 13:47:43 +0530 Subject: [PATCH 16/18] refactor: take separate headers and body callbacks One callback tagged with an event string meant the payload type changed with the tag, so it had to be typed mixed and unpacked by the receiver. Two callbacks carry their own types: an int status with an array of headers, and a string chunk. --- app/controllers.php | 25 ++++++++++++++----------- src/Executor/Runner/Adapter.php | 3 ++- src/Executor/Runner/Docker.php | 20 +++++++++----------- 3 files changed, 25 insertions(+), 23 deletions(-) diff --git a/app/controllers.php b/app/controllers.php index f227e0c2..f1a44443 100644 --- a/app/controllers.php +++ b/app/controllers.php @@ -231,7 +231,8 @@ function ( $isJson = array_any($acceptTypes, fn ($acceptType): bool => \str_starts_with((string) $acceptType, 'application/json') || \str_starts_with((string) $acceptType, 'application/*')); $stream = null; - $onStream = null; + $onHeaders = null; + $onBody = null; $streamStarted = false; if (!$isJson && \version_compare($responseFormat, RESPONSE_FORMAT_STREAM, '>=')) { @@ -243,23 +244,24 @@ function (string $bytes) use ($response, &$streamStarted): void { } ); - $onStream = function (string $event, mixed $data) use ($response, $stream): void { - if ($event !== 'headers') { - $stream->writeContent(\strval($data)); - - return; - } - + /** + * @param array $headers + */ + $onHeaders = function (int $statusCode, array $headers) use ($response, $stream): void { // Last chance to set headers: chunk() commits them on its first call. $response ->setStatusCode(Response::STATUS_CODE_OK) ->addHeader('content-type', $stream->exportHeader()) ->addHeader('x-executor-response-format', RESPONSE_FORMAT_STREAM); - $stream->writePart('statusCode', $data['statusCode']); - $stream->writePart('headers', $data['headers']); + $stream->writePart('statusCode', $statusCode); + $stream->writePart('headers', $headers); $stream->startPart('body'); }; + + $onBody = function (string $chunk) use ($stream): void { + $stream->writeContent($chunk); + }; } try { @@ -280,7 +282,8 @@ function (string $bytes) use ($response, &$streamStarted): void { $runtimeEntrypoint, $logging, $restartPolicy, - onStream: $onStream, + onHeaders: $onHeaders, + onBody: $onBody, ); } catch (\Throwable $throwable) { if (!$streamStarted) { diff --git a/src/Executor/Runner/Adapter.php b/src/Executor/Runner/Adapter.php index 42375bd2..1c2f15db 100644 --- a/src/Executor/Runner/Adapter.php +++ b/src/Executor/Runner/Adapter.php @@ -55,7 +55,8 @@ abstract public function createExecution( bool $logging, string $restartPolicy, string $region = '', - ?callable $onStream = null, + ?callable $onHeaders = null, + ?callable $onBody = null, ): mixed; abstract public function getRuntimes(): mixed; diff --git a/src/Executor/Runner/Docker.php b/src/Executor/Runner/Docker.php index 0bd806d2..27cf44ca 100644 --- a/src/Executor/Runner/Docker.php +++ b/src/Executor/Runner/Docker.php @@ -681,7 +681,8 @@ public function createExecution( bool $logging, string $restartPolicy, string $region = '', - ?callable $onStream = null, + ?callable $onHeaders = null, + ?callable $onBody = null, ): mixed { $runtimeName = System::getHostname() . '-' . $runtimeId; @@ -893,7 +894,7 @@ public function createExecution( // Once set, the response is committed: it can be neither retried nor turned into an error. $streamed = false; - $executeV5 = function () use ($path, $method, $headers, $payload, $secret, $hostname, $timeout, $runtimeName, $logging, $onStream, &$streamed): array { + $executeV5 = function () use ($path, $method, $headers, $payload, $secret, $hostname, $timeout, $runtimeName, $logging, $onHeaders, $onBody, &$streamed): array { $statusCode = 0; $errNo = -1; $executorResponse = ''; @@ -914,7 +915,7 @@ public function createExecution( \curl_setopt($ch, CURLOPT_POSTFIELDS, $payload); } - if ($onStream === null) { + if ($onHeaders === null || $onBody === null) { \curl_setopt($ch, CURLOPT_RETURNTRANSFER, true); } @@ -945,9 +946,8 @@ public function createExecution( return $len; }); - if ($onStream !== null) { - // The header callback has already run, so status and headers are known here. - \curl_setopt($ch, CURLOPT_WRITEFUNCTION, function ($curl, $data) use ($onStream, &$responseHeaders, &$streamed): int { + if ($onHeaders !== null && $onBody !== null) { + \curl_setopt($ch, CURLOPT_WRITEFUNCTION, function ($curl, $data) use ($onHeaders, $onBody, &$responseHeaders, &$streamed): int { if (!$streamed) { $streamed = true; @@ -960,13 +960,11 @@ public function createExecution( $outputHeaders[$key] = $value; } - $onStream('headers', [ - 'statusCode' => \intval(\curl_getinfo($curl, CURLINFO_HTTP_CODE)), - 'headers' => $outputHeaders, - ]); + // The header callback has already run, so the status is known here. + $onHeaders(\intval(\curl_getinfo($curl, CURLINFO_HTTP_CODE)), $outputHeaders); } - $onStream('body', $data); + $onBody($data); return \strlen($data); }); From ced418e1df2837c0d2ff33b916ab7b31b2e56ccf Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Tue, 25 Aug 2026 13:51:53 +0530 Subject: [PATCH 17/18] refactor: inline the multipart framing The class was five methods and two guards around what is, written out, three string expressions: a part header, a length prefixed run, and the zero length run that closes a part. The guards existed to keep a reusable object honest about its own state; inline, the call order is fixed and readable in one place, so they are not needed. Boundary generation still comes from BodyMultipart, so that logic stays in one place. The rule worth keeping in mind is that an empty write emits no run at all, since a zero length run is what closes a part. --- app/controllers.php | 63 ++++-- src/Executor/BodyMultipartWriter.php | 97 --------- .../unit/Executor/BodyMultipartWriterTest.php | 186 ------------------ 3 files changed, 43 insertions(+), 303 deletions(-) delete mode 100644 src/Executor/BodyMultipartWriter.php delete mode 100644 tests/unit/Executor/BodyMultipartWriterTest.php diff --git a/app/controllers.php b/app/controllers.php index f1a44443..6735abd6 100644 --- a/app/controllers.php +++ b/app/controllers.php @@ -4,7 +4,6 @@ use OpenRuntimes\Executor\Exception; use OpenRuntimes\Executor\BodyMultipart; -use OpenRuntimes\Executor\BodyMultipartWriter; use OpenRuntimes\Executor\Runner\Adapter as Runner; use Utopia\System\System; use Utopia\Http\Request; @@ -230,37 +229,61 @@ function ( $acceptTypes = \explode(', ', $request->getHeaderLine('accept') ?: 'multipart/form-data'); $isJson = array_any($acceptTypes, fn ($acceptType): bool => \str_starts_with((string) $acceptType, 'application/json') || \str_starts_with((string) $acceptType, 'application/*')); - $stream = null; + $boundary = BodyMultipart::generateBoundary(); $onHeaders = null; $onBody = null; $streamStarted = false; - if (!$isJson && \version_compare($responseFormat, RESPONSE_FORMAT_STREAM, '>=')) { - $stream = new BodyMultipartWriter( - BodyMultipart::generateBoundary(), - function (string $bytes) use ($response, &$streamStarted): void { - $streamStarted = true; - $response->chunk($bytes); - } + // Writes one whole part. Content is length prefixed as `\r\n\r\n` and the + // part is closed by a zero length run, so a reader locates content by its length and + // never scans for the boundary. An empty value writes no run at all, since a zero + // length run is what closes the part. + $writePart = function (string $name, string $content) use ($boundary, $response, &$streamStarted): void { + $streamStarted = true; + $response->chunk( + '--' . $boundary . "\r\n" + . 'Content-Disposition: form-data; name="' . $name . '"' . "\r\n" + . "Content-Transfer-Encoding: chunked\r\n\r\n" + . ($content === '' ? '' : \dechex(\strlen($content)) . "\r\n" . $content . "\r\n") + . "0\r\n\r\n" ); + }; + if (!$isJson && \version_compare($responseFormat, RESPONSE_FORMAT_STREAM, '>=')) { /** * @param array $headers */ - $onHeaders = function (int $statusCode, array $headers) use ($response, $stream): void { + $onHeaders = function (int $statusCode, array $headers) use ($response, $boundary, $writePart): void { + $encoded = \json_encode($headers); + + // Nothing has been written yet, so this can still surface as an error. + if ($encoded === false) { + throw new Exception(Exception::GENERAL_UNKNOWN, 'Response headers could not be encoded', 500); + } + // Last chance to set headers: chunk() commits them on its first call. $response ->setStatusCode(Response::STATUS_CODE_OK) - ->addHeader('content-type', $stream->exportHeader()) + ->addHeader('content-type', 'multipart/form-data; boundary=' . $boundary) ->addHeader('x-executor-response-format', RESPONSE_FORMAT_STREAM); - $stream->writePart('statusCode', $statusCode); - $stream->writePart('headers', $headers); - $stream->startPart('body'); + $writePart('statusCode', \strval($statusCode)); + $writePart('headers', $encoded); + + $response->chunk( + '--' . $boundary . "\r\n" + . 'Content-Disposition: form-data; name="body"' . "\r\n" + . "Content-Transfer-Encoding: chunked\r\n\r\n" + ); }; - $onBody = function (string $chunk) use ($stream): void { - $stream->writeContent($chunk); + $onBody = function (string $chunk) use ($response): void { + // curl hands over empty writes, and a zero length run would close the part. + if ($chunk === '') { + return; + } + + $response->chunk(\dechex(\strlen($chunk)) . "\r\n" . $chunk . "\r\n"); }; } @@ -299,14 +322,14 @@ function (string $bytes) use ($response, &$streamStarted): void { } // The body already left, so only the trailing metadata is still owed. - if ($stream instanceof BodyMultipartWriter && $streamStarted) { - $stream->endPart(); + if ($streamStarted) { + $response->chunk("0\r\n\r\n"); // closes the body part foreach (['logs', 'errors', 'duration', 'startTime'] as $key) { - $stream->writePart($key, $execution[$key] ?? ''); + $writePart($key, \strval($execution[$key] ?? '')); } - $stream->end(); + $response->chunk('--' . $boundary . '--'); $response->chunk('', true); return; diff --git a/src/Executor/BodyMultipartWriter.php b/src/Executor/BodyMultipartWriter.php deleted file mode 100644 index 56335e17..00000000 --- a/src/Executor/BodyMultipartWriter.php +++ /dev/null @@ -1,97 +0,0 @@ -\r\n\r\n`, closed by `0\r\n\r\n`, so the reader locates content by its - * length and never scans for the boundary. - */ -class BodyMultipartWriter -{ - private bool $ended = false; - - private bool $inPart = false; - - /** - * @param \Closure(string): void $write Receives raw bytes to put on the wire. - */ - public function __construct( - private readonly string $boundary, - private readonly \Closure $write, - ) { - } - - public function exportHeader(): string - { - return 'multipart/form-data; boundary=' . $this->boundary; - } - - public function startPart(string $name): void - { - if ($this->ended || $this->inPart) { - return; - } - - $this->inPart = true; - - ($this->write)( - '--' . $this->boundary . "\r\n" - . 'Content-Disposition: form-data; name="' . $name . '"' . "\r\n" - . 'Content-Transfer-Encoding: chunked' . "\r\n\r\n" - ); - } - - public function writeContent(string $content): void - { - // A zero length run is the part terminator, so an empty write has to be dropped. - if ($this->ended || !$this->inPart || $content === '') { - return; - } - - ($this->write)(\dechex(\strlen($content)) . "\r\n" . $content . "\r\n"); - } - - public function endPart(): void - { - if ($this->ended || !$this->inPart) { - return; - } - - $this->inPart = false; - - ($this->write)("0\r\n\r\n"); - } - - /** - * @param mixed $value Scalars are stringified; arrays are JSON encoded, matching BodyMultipart. - */ - public function writePart(string $name, mixed $value): void - { - if (\is_array($value)) { - $value = \json_encode($value); - - // Nothing has been written yet, so the failure can still surface as an error. - if ($value === false) { - throw new \Exception('Part "' . $name . '" could not be encoded'); - } - } - - $this->startPart($name); - $this->writeContent(\strval($value)); - $this->endPart(); - } - - public function end(): void - { - if ($this->ended) { - return; - } - - $this->endPart(); - $this->ended = true; - - ($this->write)('--' . $this->boundary . '--'); - } -} diff --git a/tests/unit/Executor/BodyMultipartWriterTest.php b/tests/unit/Executor/BodyMultipartWriterTest.php deleted file mode 100644 index 8be2935d..00000000 --- a/tests/unit/Executor/BodyMultipartWriterTest.php +++ /dev/null @@ -1,186 +0,0 @@ -wire = ''; - - return new BodyMultipartWriter($boundary, function (string $bytes): void { - $this->wire .= $bytes; - }); - } - - public function testPartIsLengthPrefixedAndChunkMarked(): void - { - $stream = $this->writer(); - $stream->writePart('statusCode', 200); - - $this->assertSame( - "--BOUNDARY\r\n" - . "Content-Disposition: form-data; name=\"statusCode\"\r\n" - . "Content-Transfer-Encoding: chunked\r\n\r\n" - . "3\r\n200\r\n" - . "0\r\n\r\n", - $this->wire - ); - } - - public function testLengthPrefixIsHexadecimal(): void - { - $stream = $this->writer(); - $stream->startPart('body'); - $stream->writeContent(\str_repeat('x', 255)); - $stream->endPart(); - - $this->assertStringContainsString("\r\nff\r\n" . \str_repeat('x', 255) . "\r\n", $this->wire); - } - - public function testArrayPartIsJsonEncoded(): void - { - $stream = $this->writer(); - $stream->writePart('headers', ['content-type' => 'text/html']); - - $this->assertStringContainsString('{"content-type":"text\/html"}', $this->wire); - } - - public function testEachContentRunGetsItsOwnPrefix(): void - { - $stream = $this->writer(); - $stream->startPart('body'); - $stream->writeContent('abc'); - $stream->writeContent('de'); - $stream->endPart(); - - $this->assertStringContainsString("3\r\nabc\r\n2\r\nde\r\n0\r\n\r\n", $this->wire); - } - - public function testEmptyContentRunIsDroppedRatherThanTerminatingThePart(): void - { - $stream = $this->writer(); - $stream->startPart('body'); - $stream->writeContent(''); - $stream->writeContent('after'); - $stream->endPart(); - - // A zero length run would read as the part terminator, orphaning everything after it. - $this->assertSame( - "--BOUNDARY\r\n" - . "Content-Disposition: form-data; name=\"body\"\r\n" - . "Content-Transfer-Encoding: chunked\r\n\r\n" - . "5\r\nafter\r\n" - . "0\r\n\r\n", - $this->wire - ); - } - - public function testEmptyPartStillFramesSoTheKeyExists(): void - { - $stream = $this->writer(); - $stream->writePart('errors', ''); - - $this->assertSame( - "--BOUNDARY\r\n" - . "Content-Disposition: form-data; name=\"errors\"\r\n" - . "Content-Transfer-Encoding: chunked\r\n\r\n" - . "0\r\n\r\n", - $this->wire - ); - } - - public function testEndClosesEnvelopeWithTerminalDelimiter(): void - { - $stream = $this->writer(); - $stream->writePart('logs', 'hello'); - $stream->end(); - - $this->assertStringEndsWith('--BOUNDARY--', $this->wire); - } - - public function testEndTerminatesAnOpenPartFirst(): void - { - $stream = $this->writer(); - $stream->startPart('body'); - $stream->writeContent('partial'); - $stream->end(); - - $this->assertStringEndsWith("7\r\npartial\r\n0\r\n\r\n--BOUNDARY--", $this->wire); - } - - public function testWritesAfterEndAreIgnored(): void - { - $stream = $this->writer(); - $stream->writePart('logs', 'x'); - $stream->end(); - - $after = $this->wire; - - $stream->writePart('errors', 'ignored'); - $stream->end(); - - $this->assertSame($after, $this->wire); - } - - public function testNestedStartPartIsIgnored(): void - { - $stream = $this->writer(); - $stream->startPart('body'); - $stream->startPart('other'); - - $this->assertStringNotContainsString('name="other"', $this->wire); - } - - public function testAbandonedEnvelopeCarriesNoClosingDelimiter(): void - { - $stream = $this->writer(); - $stream->startPart('body'); - $stream->writeContent('half a page'); - unset($stream); - - // An execution that dies here never calls end(). The caller reads a missing closing - // delimiter as a failed execution, so nothing may close the envelope on its behalf. - $this->assertStringNotContainsString('--BOUNDARY--', $this->wire); - } - - public function testExportHeaderCarriesBoundary(): void - { - $stream = $this->writer('abc123'); - - $this->assertSame('multipart/form-data; boundary=abc123', $stream->exportHeader()); - } - - public function testContentContainingBoundaryDoesNotSplitEnvelope(): void - { - $stream = $this->writer(); - $payload = "--BOUNDARY--\r\n0\r\n\r\nnot a terminator"; - - $stream->startPart('body'); - $stream->writeContent($payload); - $stream->endPart(); - $stream->end(); - - // Content is never scanned for the boundary, so the length prefix is the only framing. - $this->assertStringContainsString(\dechex(\strlen($payload)) . "\r\n" . $payload . "\r\n", $this->wire); - } - - public function testBinaryContentIsLengthedInBytes(): void - { - $stream = $this->writer(); - $binary = \random_bytes(1024); - - $stream->startPart('body'); - $stream->writeContent($binary); - $stream->endPart(); - - $this->assertStringContainsString("400\r\n" . $binary . "\r\n", $this->wire); - } -} From a4f5bd41dc03ceb89614e2c6a41ac16dee41a88a Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Tue, 25 Aug 2026 14:18:36 +0530 Subject: [PATCH 18/18] refactor: name the response format versions The negotiation compared against three protocol versions with only the newest named, leaving '0.10.0' explained by a trailing comment and '0.11.0' bare. Name all three next to each other so the protocol history reads in one place and each comparison says what it gates. --- app/controllers.php | 4 ++-- app/init.php | 8 ++++++++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/app/controllers.php b/app/controllers.php index 6735abd6..635943c2 100644 --- a/app/controllers.php +++ b/app/controllers.php @@ -224,7 +224,7 @@ function ( $variables = array_map(strval(...), $variables); - $responseFormat = $request->getHeaderLine('x-executor-response-format') ?: '0.10.0'; // Last version without support for array value for headers + $responseFormat = $request->getHeaderLine('x-executor-response-format') ?: RESPONSE_FORMAT_STRING_HEADERS; $acceptTypes = \explode(', ', $request->getHeaderLine('accept') ?: 'multipart/form-data'); $isJson = array_any($acceptTypes, fn ($acceptType): bool => \str_starts_with((string) $acceptType, 'application/json') || \str_starts_with((string) $acceptType, 'application/*')); @@ -336,7 +336,7 @@ function ( } // Backwards compatibility for headers - if (version_compare($responseFormat, '0.11.0', '<')) { + if (version_compare($responseFormat, RESPONSE_FORMAT_ARRAY_HEADERS, '<')) { foreach ($execution['headers'] as $key => $value) { if (\is_array($value)) { $lastKey = \array_key_last($value); diff --git a/app/init.php b/app/init.php index fe825a38..38447d58 100644 --- a/app/init.php +++ b/app/init.php @@ -17,6 +17,14 @@ const MAX_LOG_SIZE = 5 * 1024 * 1024; const MAX_BUILD_LOG_SIZE = 1000 * 1000; +// Response formats negotiated via x-executor-response-format. A caller sends +// the newest format it understands; no header means the oldest. +const RESPONSE_FORMAT_STRING_HEADERS = '0.10.0'; + +// First response format whose header values may be arrays, keeping every value +// of a duplicated response header instead of only the last. +const RESPONSE_FORMAT_ARRAY_HEADERS = '0.11.0'; + // First response format whose parts are length prefixed, so a caller can read them as they are // framed instead of waiting for the whole document to be serialised. const RESPONSE_FORMAT_STREAM = '0.12.0';