feat: add streamed response format 0.12.0

app/controllers.php

@@ -9,6 +9,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;
@@ -224,28 +225,119 @@ function (
 
             $variables = array_map(strval(...), $variables);
 
-            $execution = $runner->createExecution(
-                $runtimeId,
-                $payload,
-                $path,
-                $method,
-                $headers,
-                $timeout,
-                $image,
-                $source,
-                $entrypoint,
-                $variables,
-                $cpus,
-                $memory,
-                $version,
-                $runtimeEntrypoint,
-                $logging,
-                $restartPolicy,
-            );
+            $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/*'));
+
+            $boundary = BodyMultipart::generateBoundary();
+            $onHeaders = null;
+            $onBody = null;
+            $streamStarted = false;
+
+            // Writes one whole part. Content is length prefixed as `<hex>\r\n<content>\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<string, mixed> $headers
+                 */
+                $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', 'multipart/form-data; boundary=' . $boundary)
+                        ->addHeader('x-executor-response-format', RESPONSE_FORMAT_STREAM);
+
+                    $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 ($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");
+                };
+            }
+
+            try {
+                $execution = $runner->createExecution(
+                    $runtimeId,
+                    $payload,
+                    $path,
+                    $method,
+                    $headers,
+                    $timeout,
+                    $image,
+                    $source,
+                    $entrypoint,
+                    $variables,
+                    $cpus,
+                    $memory,
+                    $version,
+                    $runtimeEntrypoint,
+                    $logging,
+                    $restartPolicy,
+                    onHeaders: $onHeaders,
+                    onBody: $onBody,
+                );
+            } catch (\Throwable $throwable) {
+                if (!$streamStarted) {
+                    throw $throwable;
+                }
+
+                // Content is committed, so the error hook's JSON would land inside the envelope.
+                if ($response instanceof SwooleResponse) {
+                    $response->getSwooleResponse()->close();
+                }
+
+                return;
+            }
+
+            // The body already left, so only the trailing metadata is still owed.
+            if ($streamStarted) {
+                $response->chunk("0\r\n\r\n"); // closes the body part
+
+                foreach (['logs', 'errors', 'duration', 'startTime'] as $key) {
+                    $writePart($key, \strval($execution[$key] ?? ''));
+                }
+
+                $response->chunk('--' . $boundary . '--');
+                $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', '<')) {
+            if (version_compare($responseFormat, RESPONSE_FORMAT_ARRAY_HEADERS, '<')) {
                 foreach ($execution['headers'] as $key => $value) {
                     if (\is_array($value)) {
                         $lastKey = \array_key_last($value);
@@ -254,9 +346,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) {

app/init.php

@@ -17,6 +17,18 @@
 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';
+
 Config::load('errors', __DIR__ . '/config/errors.php');
 
 $container = new Container();

src/Executor/Runner/Adapter.php

@@ -55,6 +55,8 @@ abstract public function createExecution(
         bool $logging,
         string $restartPolicy,
         string $region = '',
+        ?callable $onHeaders = null,
+        ?callable $onBody = null,
     ): mixed;
 
     abstract public function getRuntimes(): mixed;

src/Executor/Runner/Docker.php

@@ -693,6 +693,8 @@ public function createExecution(
         bool $logging,
         string $restartPolicy,
         string $region = '',
+        ?callable $onHeaders = null,
+        ?callable $onBody = null,
     ): mixed {
         $runtimeName = System::getHostname() . '-' . $runtimeId;
 
@@ -901,7 +903,10 @@ public function createExecution(
             ];
         };
 
-        $executeV5 = function () use ($path, $method, $headers, $payload, $secret, $hostname, $timeout, $runtimeName, $logging): array {
+        // 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, $onHeaders, $onBody, &$streamed): array {
             $statusCode = 0;
             $errNo = -1;
             $executorResponse = '';
@@ -922,7 +927,10 @@ public function createExecution(
                 \curl_setopt($ch, CURLOPT_POSTFIELDS, $payload);
             }
 
-            \curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
+            if ($onHeaders === null || $onBody === 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);
@@ -950,6 +958,30 @@ public function createExecution(
                 return $len;
             });
 
+            if ($onHeaders !== null && $onBody !== null) {
+                \curl_setopt($ch, CURLOPT_WRITEFUNCTION, function ($curl, $data) use ($onHeaders, $onBody, &$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;
+                        }
+
+                        // The header callback has already run, so the status is known here.
+                        $onHeaders(\intval(\curl_getinfo($curl, CURLINFO_HTTP_CODE)), $outputHeaders);
+                    }
+
+                    $onBody($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';
@@ -966,7 +998,9 @@ public function createExecution(
             \curl_setopt($ch, CURLOPT_HEADEROPT, CURLHEADER_UNIFIED);
             \curl_setopt($ch, CURLOPT_HTTPHEADER, $headersArr);
 
-            $executorResponse = \curl_exec($ch);
+            // With a write callback the body is already forwarded, so curl_exec returns a bool.
+            $result = \curl_exec($ch);
+            $executorResponse = \is_string($result) ? $result : '';
 
             $statusCode = \curl_getinfo($ch, CURLINFO_HTTP_CODE);
 
@@ -1098,6 +1132,11 @@ public function createExecution(
                 break;
             }
 
+            // A retry would append a second response to content already on the wire.
+            if ($streamed) {
+                break;
+            }
+
             usleep($retryDelayMs * 1000);
         } while ((++$attempts < $retryAttempts) || (\microtime(true) - $startTime < $timeout));