fix(cloudfront): stream origin responses and viewer request bodies (#5278)

A distribution read every origin response whole before answering the
viewer: an S3 origin's object through getObject, a custom origin's body
through the HTTP client into a byte array. A viewer's POST, PUT, PATCH
or DELETE body was likewise read whole before it was forwarded. So an
object or upload of about 2 GiB or more failed, and a smaller one could
run the heap out of memory.

S3 origins now open the object as a stream, the snapshot GetObject
serves, and custom origins return their response before its body is
read; either body is written to the viewer as it arrives, with its
Content-Length, or chunked as the origin sent it. A viewer's body is
streamed to the custom origin with its Content-Length, or chunked when
the viewer sent it chunked or, as HTTP/2 allows, with no length.

An origin response closed before its end, as when the viewer goes away
or a custom error page replaces it, cancels its request instead of
reading the rest of the body to reuse the connection. Since a streamed
response holds its connection for the whole transfer, one origin may
now use the whole connection pool, as CloudFront sets no per-origin
limit.

(cherry picked from commit 1f1d4f94dd8a179972bda124bc3aa47b9704f69b)

docs/services/cloudfront.md

@@ -156,9 +156,12 @@ PUT, PATCH and DELETE).
   method; any other method returns 403 `Invalid method.`. A behavior without `AllowedMethods` allows
   GET and HEAD. Origin forwarding preserves the raw path; custom-origin redirects are not followed.
 - POST, PUT, PATCH and DELETE are forwarded to custom origins with the viewer request body and its
-  `Content-Type`; `Content-Length` is set from the body. Their responses are never cached, as on AWS.
-  Signed URL and signed cookie enforcement applies to them like any other method. Request bodies are
-  bounded by the emulator-wide `floci.protocols.max-request-size` limit.
+  `Content-Type`. The body is streamed to the origin with the viewer's `Content-Length`, or chunked
+  when the viewer sent it chunked or, as HTTP/2 allows, with no length, so it is never held in
+  memory. Their responses are never cached, as on AWS. Signed URL and signed cookie enforcement
+  applies to them like any other method. Request bodies are bounded by the emulator-wide
+  `floci.protocols.max-request-size` limit (`FLOCI_PROTOCOLS_MAX_REQUEST_SIZE`, 2048 MB by default);
+  CloudFront accepts up to 64 GB, so raise it for larger uploads.
 - Every distribution is also served as `{id}.cloudfront.{host}` for each endpoint host Floci
   resolves: `localhost`, `localhost.floci.io`, `localhost.localstack.cloud`, `FLOCI_HOSTNAME` and
   every `FLOCI_DNS_EXTRA_SUFFIXES` entry. `{id}.cloudfront.localhost.floci.io` and
@@ -223,6 +226,10 @@ PUT, PATCH and DELETE).
   `Elemental-MediaPackage` or the `UseOriginCacheControlHeaders` policies.
 - In-process S3 origins do not apply forwarding settings: viewer query strings, cookies and headers
   do not change the S3 read.
+- Origin responses are streamed to the viewer, from in-process S3 origins and custom origins alike,
+  so a large object is never held in memory. A custom origin's `Content-Length` is passed on, and a
+  response it sends chunked stays chunked. Floci does not cache, so it serves objects larger than
+  50 GB the way CloudFront does with caching disabled; with caching enabled, CloudFront refuses them.
 - Origin `Set-Cookie` headers always reach the viewer. With legacy `Forward=none`, AWS strips them
   from the response; Floci does not.
 - Custom origins that resolve to loopback, private, link-local, carrier-grade NAT, or other non-routable addresses are rejected by default. Development-only private origins must be explicitly allowlisted by exact hostname.

src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontOriginHttpClient.java

@@ -3,6 +3,7 @@ package io.github.hectorvent.floci.services.cloudfront;
 import io.github.hectorvent.floci.core.common.SsrfProtection;
 import org.apache.hc.client5.http.DnsResolver;
 import org.apache.hc.client5.http.SystemDefaultDnsResolver;
+import org.apache.hc.client5.http.classic.methods.HttpUriRequestBase;
 import org.apache.hc.client5.http.config.ConnectionConfig;
 import org.apache.hc.client5.http.config.RequestConfig;
 import org.apache.hc.client5.http.config.TlsConfig;
@@ -13,32 +14,32 @@ import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
 import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder;
 import org.apache.hc.client5.http.impl.routing.DefaultRoutePlanner;
 import org.apache.hc.client5.http.protocol.HttpClientContext;
+import org.apache.hc.client5.http.routing.RoutingSupport;
 import org.apache.hc.client5.http.ssl.ClientTlsStrategyBuilder;
+import org.apache.hc.core5.http.ClassicHttpResponse;
 import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpEntity;
+import org.apache.hc.core5.http.HttpException;
 import org.apache.hc.core5.http.HttpVersion;
-import org.apache.hc.core5.http.io.entity.ByteArrayEntity;
-import org.apache.hc.core5.http.io.entity.EntityUtils;
-import org.apache.hc.core5.http.io.support.ClassicRequestBuilder;
+import org.apache.hc.core5.http.io.entity.InputStreamEntity;
 import org.apache.hc.core5.http2.HttpVersionPolicy;
 import org.apache.hc.core5.util.Timeout;
+import org.jboss.logging.Logger;
 
 import javax.net.ssl.SSLContext;
-import javax.net.ssl.SSLSession;
+import java.io.FilterInputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.InterruptedIOException;
 import java.net.InetAddress;
-import java.net.URI;
 import java.net.UnknownHostException;
-import java.net.http.HttpClient;
 import java.net.http.HttpHeaders;
 import java.net.http.HttpRequest;
-import java.net.http.HttpResponse;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.TreeSet;
@@ -51,8 +52,11 @@ import java.util.stream.Collectors;
  */
 final class CloudFrontOriginHttpClient implements AutoCloseable {
 
+    private static final Logger LOG = Logger.getLogger(CloudFrontOriginHttpClient.class);
+
     private static final Duration CONNECT_TIMEOUT = Duration.ofSeconds(10);
     private static final Duration RESPONSE_TIMEOUT = Duration.ofSeconds(30);
+    static final int MAX_CONNECTIONS = 100;
 
     private final CloseableHttpClient client;
 
@@ -86,8 +90,10 @@ final class CloudFrontOriginHttpClient implements AutoCloseable {
                         .setDnsResolver(validatingResolver)
                         .setDefaultConnectionConfig(connectionConfig)
                         .setDefaultTlsConfig(tlsConfig)
-                        .setMaxConnTotal(100)
-                        .setMaxConnPerRoute(20);
+                        // CloudFront has no per-origin connection limit, and a streamed response holds its
+                        // connection for the whole transfer, so one origin may use the whole pool.
+                        .setMaxConnTotal(MAX_CONNECTIONS)
+                        .setMaxConnPerRoute(MAX_CONNECTIONS);
         if (sslContext != null) {
             connectionManagerBuilder.setTlsSocketStrategy(ClientTlsStrategyBuilder.create()
                     .setSslContext(sslContext)
@@ -113,39 +119,139 @@ final class CloudFrontOriginHttpClient implements AutoCloseable {
                 .build();
     }
 
-    HttpResponse<byte[]> send(HttpRequest request, HttpResponse.BodyHandler<byte[]> bodyHandler)
-            throws IOException, InterruptedException {
-        return send(request, Map.of(), bodyHandler);
-    }
-
-    HttpResponse<byte[]> send(HttpRequest request, Map<String, String> originHeaders,
-                              HttpResponse.BodyHandler<byte[]> bodyHandler)
-            throws IOException, InterruptedException {
-        return send(request, List.of(), originHeaders, null, bodyHandler);
+    /**
+     * An origin response whose body has not been read. The caller reads {@code body} as it needs it
+     * and closes it, which releases the connection. {@code contentLength} is the length the body
+     * arrives with, {@code -1} when it comes chunked or with none.
+     */
+    record OpenResponse(int statusCode, HttpHeaders headers, long contentLength, InputStream body) {
     }
 
     /**
-     * Sends {@code request} with {@code forwardedHeaders} added and {@code requestBody} as its
-     * entity, or with no entity when it is {@code null}. Each origin header replaces every
-     * same-named request or forwarded header, so a viewer cannot repeat a header to smuggle its own
-     * value past an origin custom header. The request's own body publisher must be empty: the body
-     * travels as bytes so its {@code Content-Length} is exact.
+     * Sends {@code request} with {@code forwardedHeaders} added and {@code requestBody}, when it is
+     * not {@code null}, streamed as the entity, with {@code requestBodyLength} as its
+     * {@code Content-Length}, or chunked when the length is negative. Each origin header replaces
+     * every same-named request or forwarded header, so a viewer cannot repeat a header to smuggle its
+     * own value past an origin custom header. The response comes back before its body is read, so
+     * neither body has to fit in memory.
      */
-    HttpResponse<byte[]> send(HttpRequest request,
-                              List<CloudFrontOriginRequestBuilder.Header> forwardedHeaders,
-                              Map<String, String> originHeaders, byte[] requestBody,
-                              HttpResponse.BodyHandler<byte[]> bodyHandler)
+    OpenResponse open(HttpRequest request,
+                      List<CloudFrontOriginRequestBuilder.Header> forwardedHeaders,
+                      Map<String, String> originHeaders, InputStream requestBody, long requestBodyLength)
+            throws IOException, InterruptedException {
+        // No entity content type: the caller's Content-Type request header is sent as is.
+        HttpUriRequestBase originRequest = originRequest(request, forwardedHeaders, originHeaders,
+                requestBody != null ? new InputStreamEntity(requestBody, requestBodyLength, null) : null);
+        ClassicHttpResponse response;
+        try {
+            response = client.executeOpen(RoutingSupport.determineHost(originRequest), originRequest, context(request));
+        } catch (HttpException e) {
+            throw new IOException("CloudFront origin request has no target host", e);
+        } catch (InterruptedIOException e) {
+            if (Thread.currentThread().isInterrupted()) {
+                throw new InterruptedException("CloudFront origin request interrupted");
+            }
+            throw e;
+        }
+        try {
+            HttpEntity entity = response.getEntity();
+            InputStream body = entity != null
+                    ? new ResponseBody(entity.getContent(), originRequest, response)
+                    : new ResponseBody(InputStream.nullInputStream(), null, response);
+            // The entity's length follows the framing the body arrives with; with no entity, as for a
+            // HEAD, the header describes the body a GET would get.
+            long contentLength = entity != null ? entity.getContentLength() : headerContentLength(response);
+            return new OpenResponse(response.getCode(), responseHeaders(response), contentLength, body);
+        } catch (IOException | RuntimeException e) {
+            originRequest.cancel();
+            closeAborted(response);
+            throw e;
+        }
+    }
+
+    /**
+     * An origin response body that releases its connection when it is closed. Closed at its end, the
+     * connection goes back to the pool. Closed before its end, as when a viewer goes away or a custom
+     * error page replaces it, the request is cancelled and the connection closed, since closing the
+     * response normally reads the rest of the body to reuse the connection, however long it is.
+     */
+    private static final class ResponseBody extends FilterInputStream {
+
+        private final HttpUriRequestBase request;
+        private final ClassicHttpResponse response;
+        private boolean finished;
+
+        ResponseBody(InputStream content, HttpUriRequestBase request, ClassicHttpResponse response) {
+            super(content);
+            this.request = request;
+            this.response = response;
+        }
+
+        @Override
+        public int read() throws IOException {
+            int b = super.read();
+            finished |= b < 0;
+            return b;
+        }
+
+        @Override
+        public int read(byte[] buffer, int offset, int length) throws IOException {
+            int read = super.read(buffer, offset, length);
+            finished |= read < 0;
+            return read;
+        }
+
+        @Override
+        public void close() throws IOException {
+            if (finished || request == null) {
+                try {
+                    super.close();
+                } finally {
+                    response.close();
+                }
+                return;
+            }
+            request.cancel();
+            closeAborted(response);
+        }
+    }
+
+    /** Closes a response whose request was cancelled; reading the rest of it fails, as it should. */
+    private static void closeAborted(ClassicHttpResponse response) {
+        try {
+            response.close();
+        } catch (IOException e) {
+            LOG.debugv("Closed an unfinished CloudFront origin response: {0}", e.getMessage());
+        }
+    }
+
+    private static long headerContentLength(ClassicHttpResponse response) {
+        Header header = response.getFirstHeader("Content-Length");
+        if (header == null) {
+            return -1;
+        }
+        try {
+            return Long.parseLong(header.getValue().trim());
+        } catch (NumberFormatException e) {
+            LOG.debugv("Ignoring an invalid custom-origin Content-Length: {0}", e.getMessage());
+            return -1;
+        }
+    }
+
+    private HttpUriRequestBase originRequest(HttpRequest request,
+                                             List<CloudFrontOriginRequestBuilder.Header> forwardedHeaders,
+                                             Map<String, String> originHeaders, HttpEntity entity)
             throws IOException, InterruptedException {
         if (Thread.currentThread().isInterrupted()) {
             throw new InterruptedException("CloudFront origin request interrupted");
         }
         if (request.bodyPublisher().filter(publisher -> publisher.contentLength() != 0).isPresent()) {
-            throw new IOException("CloudFront origin request bodies must be passed as bytes");
+            throw new IOException("CloudFront origin request bodies must be passed separately");
         }
 
-        ClassicRequestBuilder builder = ClassicRequestBuilder.create(request.method())
-                .setUri(request.uri())
-                .setVersion(HttpVersion.HTTP_1_1);
+        // A request that can be cancelled, so a response closed before its end is not read to the end.
+        HttpUriRequestBase builder = new HttpUriRequestBase(request.method(), request.uri());
+        builder.setVersion(HttpVersion.HTTP_1_1);
         Set<String> replaced = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
         replaced.addAll(originHeaders.keySet());
         request.headers().map().forEach((name, values) -> {
@@ -159,11 +265,13 @@ final class CloudFrontOriginHttpClient implements AutoCloseable {
             }
         }
         originHeaders.forEach(builder::addHeader);
-        if (requestBody != null) {
-            // No entity content type: the caller's Content-Type request header is sent as is.
-            builder.setEntity(new ByteArrayEntity(requestBody, null));
+        if (entity != null) {
+            builder.setEntity(entity);
         }
+        return builder;
+    }
 
+    private static HttpClientContext context(HttpRequest request) {
         HttpClientContext context = HttpClientContext.create();
         Duration responseTimeout = request.timeout().orElse(RESPONSE_TIMEOUT);
         context.setRequestConfig(RequestConfig.custom()
@@ -173,28 +281,16 @@ final class CloudFrontOriginHttpClient implements AutoCloseable {
                 .setContentCompressionEnabled(false)
                 .setProtocolUpgradeEnabled(false)
                 .build());
+        return context;
+    }
 
-        try {
-            return client.execute(builder.build(), context, response -> {
-                Map<String, List<String>> responseHeaders = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
-                for (Header header : response.getHeaders()) {
-                    responseHeaders.computeIfAbsent(header.getName(), ignored -> new ArrayList<>())
-                            .add(header.getValue());
-                }
-                byte[] body = response.getEntity() != null
-                        ? EntityUtils.toByteArray(response.getEntity())
-                        : new byte[0];
-                return new PinnedHttpResponse(
-                        response.getCode(), request,
-                        HttpHeaders.of(responseHeaders, (name, value) -> true),
-                        body, request.uri());
-            });
-        } catch (InterruptedIOException e) {
-            if (Thread.currentThread().isInterrupted()) {
-                throw new InterruptedException("CloudFront origin request interrupted");
-            }
-            throw e;
+    private static HttpHeaders responseHeaders(ClassicHttpResponse response) {
+        Map<String, List<String>> responseHeaders = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+        for (Header header : response.getHeaders()) {
+            responseHeaders.computeIfAbsent(header.getName(), ignored -> new ArrayList<>())
+                    .add(header.getValue());
         }
+        return HttpHeaders.of(responseHeaders, (name, value) -> true);
     }
 
     @Override
@@ -202,29 +298,6 @@ final class CloudFrontOriginHttpClient implements AutoCloseable {
         client.close();
     }
 
-    private record PinnedHttpResponse(
-            int statusCode,
-            HttpRequest request,
-            HttpHeaders headers,
-            byte[] body,
-            URI uri) implements HttpResponse<byte[]> {
-
-        @Override
-        public Optional<HttpResponse<byte[]>> previousResponse() {
-            return Optional.empty();
-        }
-
-        @Override
-        public Optional<SSLSession> sslSession() {
-            return Optional.empty();
-        }
-
-        @Override
-        public HttpClient.Version version() {
-            return HttpClient.Version.HTTP_1_1;
-        }
-    }
-
     private record ValidatingDnsResolver(DnsResolver delegate, Set<String> allowedHosts)
             implements DnsResolver {
 

src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontServingController.java

@@ -29,13 +29,16 @@ import jakarta.ws.rs.core.Context;
 import jakarta.ws.rs.core.HttpHeaders;
 import jakarta.ws.rs.core.MediaType;
 import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.core.StreamingOutput;
 import jakarta.ws.rs.core.UriInfo;
 import org.jboss.logging.Logger;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PushbackInputStream;
 import java.net.IDN;
 import java.net.URI;
 import java.net.http.HttpRequest;
-import java.net.http.HttpResponse;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.time.Instant;
@@ -169,7 +172,7 @@ public class CloudFrontServingController {
     @Path("/{proxy:.*}")
     @Consumes(MediaType.WILDCARD)
     public Response post(@PathParam("distId") String distId, @PathParam("proxy") String proxy,
-                         @Context HttpHeaders headers, @Context UriInfo uriInfo, byte[] body) {
+                         @Context HttpHeaders headers, @Context UriInfo uriInfo, InputStream body) {
         return serveWithBody(distId, "POST", headers, uriInfo, body);
     }
 
@@ -177,7 +180,7 @@ public class CloudFrontServingController {
     @Path("/{proxy:.*}")
     @Consumes(MediaType.WILDCARD)
     public Response put(@PathParam("distId") String distId, @PathParam("proxy") String proxy,
-                        @Context HttpHeaders headers, @Context UriInfo uriInfo, byte[] body) {
+                        @Context HttpHeaders headers, @Context UriInfo uriInfo, InputStream body) {
         return serveWithBody(distId, "PUT", headers, uriInfo, body);
     }
 
@@ -185,7 +188,7 @@ public class CloudFrontServingController {
     @Path("/{proxy:.*}")
     @Consumes(MediaType.WILDCARD)
     public Response patch(@PathParam("distId") String distId, @PathParam("proxy") String proxy,
-                          @Context HttpHeaders headers, @Context UriInfo uriInfo, byte[] body) {
+                          @Context HttpHeaders headers, @Context UriInfo uriInfo, InputStream body) {
         return serveWithBody(distId, "PATCH", headers, uriInfo, body);
     }
 
@@ -193,12 +196,12 @@ public class CloudFrontServingController {
     @Path("/{proxy:.*}")
     @Consumes(MediaType.WILDCARD)
     public Response delete(@PathParam("distId") String distId, @PathParam("proxy") String proxy,
-                           @Context HttpHeaders headers, @Context UriInfo uriInfo, byte[] body) {
+                           @Context HttpHeaders headers, @Context UriInfo uriInfo, InputStream body) {
         return serveWithBody(distId, "DELETE", headers, uriInfo, body);
     }
 
     private Response serveWithBody(String distId, String method, HttpHeaders headers,
-                                   UriInfo uriInfo, byte[] body) {
+                                   UriInfo uriInfo, InputStream body) {
         HttpServerRequest request = currentVertxRequest.getCurrent().request();
         String rawViewerPath = rawViewerPath(request.uri());
         return serve(distId, rawViewerPath, decodedViewerPath(rawViewerPath),
@@ -207,7 +210,41 @@ public class CloudFrontServingController {
                 headers.getHeaderString(HttpHeaders.AUTHORIZATION),
                 request.getHeader("Origin"), method, null, null,
                 request.getHeader("Pragma"),
-                body != null ? body : new byte[0]);
+                viewerBody(body, headers.getHeaderString("Transfer-Encoding"),
+                        headers.getHeaderString(HttpHeaders.CONTENT_LENGTH)));
+    }
+
+    /**
+     * The viewer's request body and its length. It is {@code -1} when the body comes chunked, which
+     * takes precedence over any {@code Content-Length} as HTTP has it, and otherwise the
+     * {@code Content-Length}. With neither header an HTTP/1.1 request has no body, but an HTTP/2 one
+     * can still send one, so the body is looked at: one that is already at its end is empty, and one
+     * that is not is forwarded chunked.
+     */
+    static ViewerBody viewerBody(InputStream body, String transferEncoding, String contentLength) {
+        InputStream stream = body != null ? body : InputStream.nullInputStream();
+        if (transferEncoding != null) {
+            return new ViewerBody(stream, -1);
+        }
+        if (contentLength != null && !contentLength.isBlank()) {
+            try {
+                return new ViewerBody(stream, Long.parseLong(contentLength.trim()));
+            } catch (NumberFormatException e) {
+                LOG.debugv("Treating an invalid viewer Content-Length as chunked: {0}", e.getMessage());
+                return new ViewerBody(stream, -1);
+            }
+        }
+        PushbackInputStream peekable = new PushbackInputStream(stream, 1);
+        try {
+            int first = peekable.read();
+            if (first < 0) {
+                return new ViewerBody(InputStream.nullInputStream(), 0);
+            }
+            peekable.unread(first);
+        } catch (IOException e) {
+            LOG.debugv("Could not read the start of a viewer request body: {0}", e.getMessage());
+        }
+        return new ViewerBody(peekable, -1);
     }
 
     private Response serve(String distId, String rawViewerPath, String decodedViewerPath,
@@ -217,7 +254,7 @@ public class CloudFrontServingController {
                            String accessControlRequestMethod,
                            String accessControlRequestHeaders,
                            String pragma,
-                           byte[] viewerBody) {
+                           ViewerBody viewerBody) {
         boolean includeBody = !"HEAD".equals(method);
         boolean preflightRequest = "OPTIONS".equals(method)
                 && viewerOrigin != null && !viewerOrigin.isBlank()
@@ -263,8 +300,14 @@ public class CloudFrontServingController {
                 accessControlRequestMethod, accessControlRequestHeaders, viewerBody);
 
         if (origin.status() >= 400) {
-            Response fallback = applyCustomError(
-                    dist, origin, viewerScheme, viewerAuthorization, includeBody, directives);
+            Response fallback;
+            try {
+                fallback = applyCustomError(
+                        dist, origin, viewerScheme, viewerAuthorization, includeBody, directives);
+            } catch (RuntimeException e) {
+                origin.closeStream();
+                throw e;
+            }
             if (fallback != null) {
                 return fallback;
             }
@@ -407,7 +450,7 @@ public class CloudFrontServingController {
                                  String method, String viewerOrigin,
                                  String accessControlRequestMethod,
                                  String accessControlRequestHeaders,
-                                 byte[] viewerBody) {
+                                 ViewerBody viewerBody) {
         DistributionConfig config = distribution.getConfig();
         String originId = CloudFrontRequestRouter.matchTargetOriginId(config, normalized);
         Origin origin = CloudFrontRequestRouter.findOrigin(config, originId);
@@ -531,10 +574,12 @@ public class CloudFrontServingController {
             authorizeS3OriginRead(
                     distribution, origin, bucket, key, viewerAuthorization);
             if (includeBody) {
-                S3Object obj = s3Service.getObject(bucket, key);
-                byte[] data = obj.getData() != null ? obj.getData() : new byte[0];
+                // The body is streamed to the viewer, so an object of any size is served without
+                // being read into the heap; the metadata and the stream come from one snapshot.
+                S3Service.ObjectRead read = s3Service.openObject(bucket, key, null);
+                S3Object obj = read.object();
                 response = new OriginResponse(
-                        200, contentType(obj), data, data.length, s3ObjectHeaders(obj));
+                        200, contentType(obj), null, read.body(), obj.getSize(), s3ObjectHeaders(obj));
             } else {
                 S3Object meta = s3Service.headObject(bucket, key);
                 response = new OriginResponse(
@@ -543,8 +588,13 @@ public class CloudFrontServingController {
         } catch (AwsException e) {
             response = OriginResponse.error(e.getHttpStatus(), e.getMessage());
         }
-        return withS3CorsHeaders(
-                bucket, origin, includeBody ? "GET" : "HEAD", response);
+        try {
+            return withS3CorsHeaders(
+                    bucket, origin, includeBody ? "GET" : "HEAD", response);
+        } catch (RuntimeException e) {
+            response.closeStream();
+            throw e;
+        }
     }
 
     private void authorizeS3OriginRead(
@@ -641,7 +691,7 @@ public class CloudFrontServingController {
      */
     private OriginResponse fetchFromCustomOrigin(Origin origin, String forwardUri,
                                                  CloudFrontOriginRequestBuilder.OriginRequest originRequest,
-                                                 String viewerScheme, String method, byte[] viewerBody) {
+                                                 String viewerScheme, String method, ViewerBody viewerBody) {
         boolean includeBody = !"HEAD".equals(method);
         try {
             URI target = buildCustomOriginUri(
@@ -664,15 +714,19 @@ public class CloudFrontServingController {
                 }
             }
             rb.method(method, HttpRequest.BodyPublishers.noBody());
-            // The transport derives Content-Length from the body itself.
-            HttpResponse<byte[]> resp = httpClient.send(rb.build(), originRequest.headers(),
-                    originHeaders, originRequestBody(method, viewerBody),
-                    HttpResponse.BodyHandlers.ofByteArray());
+            // Both bodies are streamed, so neither has to fit in memory: the viewer's goes out with its
+            // own Content-Length, or chunked as it came, and the origin's comes back as it arrives.
+            ViewerBody requestBody = originRequestBody(method, viewerBody);
+            CloudFrontOriginHttpClient.OpenResponse resp = httpClient.open(rb.build(), originRequest.headers(),
+                    originHeaders, requestBody != null ? requestBody.stream() : null,
+                    requestBody != null ? requestBody.length() : -1);
             String ct = resp.headers().firstValue("content-type").orElse(DEFAULT_CONTENT_TYPE);
-            byte[] body = resp.body() != null ? resp.body() : new byte[0];
-            long contentLength = includeBody ? body.length : responseContentLength(resp);
-            return new OriginResponse(resp.statusCode(), ct, includeBody ? body : null,
-                    contentLength, resp.headers().map());
+            long contentLength = resp.contentLength();
+            if (!includeBody) {
+                resp.body().close();
+                return new OriginResponse(resp.statusCode(), ct, null, contentLength, resp.headers().map());
+            }
+            return new OriginResponse(resp.statusCode(), ct, null, resp.body(), contentLength, resp.headers().map());
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             LOG.warnv("CloudFront custom origin fetch was interrupted for host {0}",
@@ -690,16 +744,20 @@ public class CloudFrontServingController {
      * The body CloudFront sends to the origin: POST, PUT and PATCH always carry one (possibly empty, so
      * the origin still sees {@code Content-Length: 0}); DELETE carries one only when the viewer sent it.
      */
-    private static byte[] originRequestBody(String method, byte[] viewerBody) {
+    private static ViewerBody originRequestBody(String method, ViewerBody viewerBody) {
         if (viewerBody == null || !BODY_METHODS.contains(method)) {
             return null;
         }
-        if ("DELETE".equals(method) && viewerBody.length == 0) {
+        if ("DELETE".equals(method) && viewerBody.length() == 0) {
             return null;
         }
         return viewerBody;
     }
 
+    /** A viewer request body, read as it is forwarded, and its length, {@code -1} when it comes chunked. */
+    record ViewerBody(InputStream stream, long length) {
+    }
+
     private static String originAccessIdentityId(Origin origin) {
         Map<String, String> config = origin.getS3OriginConfig();
         String value = config != null ? config.get("OriginAccessIdentity") : null;
@@ -797,15 +855,6 @@ public class CloudFrontServingController {
         return normalized.toLowerCase(Locale.ROOT);
     }
 
-    private static long responseContentLength(HttpResponse<?> response) {
-        try {
-            return response.headers().firstValueAsLong("content-length").orElse(-1L);
-        } catch (NumberFormatException e) {
-            LOG.debugv("Ignoring invalid custom-origin Content-Length: {0}", e.getMessage());
-            return -1L;
-        }
-    }
-
     private static String safeOriginName(Origin origin) {
         try {
             return normalizeOriginHost(origin.getDomainName());
@@ -846,8 +895,7 @@ public class CloudFrontServingController {
         String pagePath = str(cer.get("ResponsePagePath"));
         if (pagePath.isBlank()) {
             // ResponseCode override with no custom page: keep the origin body, change the status.
-            return toResponse(new OriginResponse(responseCode, origin.contentType(), origin.body(),
-                    origin.contentLength(), origin.headers()), includeBody, directives);
+            return toResponse(origin.withStatus(responseCode), includeBody, directives);
         }
 
         String errNormalized = CloudFrontRequestRouter.normalizePath(pagePath);
@@ -857,6 +905,8 @@ public class CloudFrontServingController {
             return null;
         }
 
+        // The custom error page replaces the origin's error response, which is not sent.
+        origin.closeStream();
         OriginResponse page;
         if (CloudFrontRequestRouter.isS3Origin(errOrigin)) {
             String key = CloudFrontRequestRouter.resolveOriginKey(errOrigin.getOriginPath(), errNormalized, null);
@@ -873,8 +923,7 @@ public class CloudFrontServingController {
             // (AWS behavior), without recursively applying custom error handling.
             return toResponse(page, includeBody, directives);
         }
-        return toResponse(new OriginResponse(responseCode, page.contentType(), page.body(),
-                page.contentLength(), page.headers()), includeBody, directives);
+        return toResponse(page.withStatus(responseCode), includeBody, directives);
     }
 
     private Map<String, Object> matchCustomError(DistributionConfig config, int status) {
@@ -892,21 +941,40 @@ public class CloudFrontServingController {
 
     private Response toResponse(OriginResponse origin, boolean includeBody,
                                 ResponseHeadersPolicyConfigCodec.Directives directives) {
-        Response.ResponseBuilder rb = Response.status(origin.status());
-        HeaderCollection collected = collectResponseHeaders(origin, includeBody);
-        applyResponseHeadersPolicy(collected.headers(), directives, collected.forbiddenPolicyHeaders());
-        for (HeaderValues header : collected.headers().values()) {
-            if ("date".equalsIgnoreCase(header.name())) {
-                currentVertxRequest.getCurrent().response()
-                        .putHeader(header.name(), header.values());
-            } else {
-                header.values().forEach(value -> rb.header(header.name(), value));
+        try {
+            Response.ResponseBuilder rb = Response.status(origin.status());
+            HeaderCollection collected = collectResponseHeaders(origin, includeBody);
+            applyResponseHeadersPolicy(collected.headers(), directives, collected.forbiddenPolicyHeaders());
+            for (HeaderValues header : collected.headers().values()) {
+                if ("date".equalsIgnoreCase(header.name())) {
+                    currentVertxRequest.getCurrent().response()
+                            .putHeader(header.name(), header.values());
+                } else {
+                    header.values().forEach(value -> rb.header(header.name(), value));
+                }
             }
+            if (includeBody && origin.stream() != null) {
+                InputStream body = origin.stream();
+                StreamingOutput entity = output -> {
+                    try (InputStream in = body) {
+                        in.transferTo(output);
+                    }
+                };
+                rb.entity(entity);
+            } else {
+                if (origin.stream() != null) {
+                    // A body that will not be sent, as for a HEAD, still holds its source open.
+                    origin.closeStream();
+                }
+                if (includeBody && origin.body() != null) {
+                    rb.entity(origin.body());
+                }
+            }
+            return rb.build();
+        } catch (RuntimeException e) {
+            origin.closeStream();
+            throw e;
         }
-        if (includeBody && origin.body() != null) {
-            rb.entity(origin.body());
-        }
-        return rb.build();
     }
 
     /**
@@ -942,7 +1010,8 @@ public class CloudFrontServingController {
         if (origin.contentType() != null) {
             setHeader(headers, "Content-Type", origin.contentType());
         }
-        if (!includeBody && origin.contentLength() >= 0) {
+        // A byte body sets its own length; a HEAD response and a streamed body need it stated.
+        if ((!includeBody || origin.stream() != null) && origin.contentLength() >= 0) {
             setHeader(headers, "Content-Length", Long.toString(origin.contentLength()));
         }
         return new HeaderCollection(headers, forbiddenPolicyHeaders);
@@ -1093,12 +1162,7 @@ public class CloudFrontServingController {
                                 "Access-Control-Request-Method"));
             }
         });
-        return new OriginResponse(
-                response.status(),
-                response.contentType(),
-                response.body(),
-                response.contentLength(),
-                headers);
+        return response.withHeaders(headers);
     }
 
     private static String originCustomHeader(Origin origin, String requestedName) {
@@ -1191,12 +1255,42 @@ public class CloudFrontServingController {
                                     Set<String> forbiddenPolicyHeaders) {
     }
 
-    /** The result of fetching from an origin, including end-to-end response headers. */
-    private record OriginResponse(int status, String contentType, byte[] body, long contentLength,
-                                  Map<String, List<String>> headers) {
+    /**
+     * The result of fetching from an origin, including end-to-end response headers. An origin's
+     * content, from an S3 object or a custom origin, is an open {@code stream} instead of
+     * {@code body}, so it never passes through the heap; the viewer response writes and closes it.
+     * {@code body} holds the bytes of a response Floci makes itself, such as an error.
+     */
+    private record OriginResponse(int status, String contentType, byte[] body, InputStream stream,
+                                  long contentLength, Map<String, List<String>> headers) {
+        OriginResponse(int status, String contentType, byte[] body, long contentLength,
+                       Map<String, List<String>> headers) {
+            this(status, contentType, body, null, contentLength, headers);
+        }
+
         static OriginResponse error(int status, String message) {
             byte[] b = message == null ? new byte[0] : message.getBytes(StandardCharsets.UTF_8);
             return new OriginResponse(status, MediaType.TEXT_PLAIN, b, b.length, Map.of());
         }
+
+        OriginResponse withStatus(int newStatus) {
+            return new OriginResponse(newStatus, contentType, body, stream, contentLength, headers);
+        }
+
+        OriginResponse withHeaders(Map<String, List<String>> newHeaders) {
+            return new OriginResponse(status, contentType, body, stream, contentLength, newHeaders);
+        }
+
+        /** Closes the stream of a response that will not be written. */
+        void closeStream() {
+            if (stream == null) {
+                return;
+            }
+            try {
+                stream.close();
+            } catch (IOException e) {
+                LOG.debugv("Could not close an unsent origin body: {0}", e.getMessage());
+            }
+        }
     }
 }

src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontCustomOriginServingTest.java

@@ -16,22 +16,29 @@ import io.github.hectorvent.floci.services.cloudfront.model.ResponseHeadersPolic
 import io.quarkus.test.junit.QuarkusTest;
 import io.quarkus.test.junit.QuarkusTestProfile;
 import io.quarkus.test.junit.TestProfile;
+import io.restassured.RestAssured;
 import io.restassured.response.Response;
 import jakarta.inject.Inject;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 
+import java.io.BufferedReader;
 import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
 import java.net.InetSocketAddress;
+import java.net.Socket;
 import java.nio.charset.StandardCharsets;
 import java.security.KeyPair;
 import java.security.KeyPairGenerator;
 import java.security.Signature;
+import java.time.Duration;
 import java.time.Instant;
 import java.util.Base64;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Random;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
@@ -43,6 +50,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @QuarkusTest
@@ -316,6 +324,106 @@ class CloudFrontCustomOriginServingTest {
         assertEquals(5, hits.get());
     }
 
+    @Test
+    void streamsALargeOriginResponseWithItsLength() throws Exception {
+        byte[] video = randomBytes(3 * 1024 * 1024 + 1);
+        startServingOrigin(video, true);
+        Distribution created = cloudFrontService.createDistribution(
+                customOriginDistribution(allMethodsBehavior()), Map.of());
+
+        Response response = given().header("Host", created.getDomainName()).when().get("/video.bin");
+
+        assertEquals(200, response.statusCode());
+        assertEquals(Integer.toString(video.length), response.header("Content-Length"));
+        assertArrayEquals(video, response.asByteArray());
+    }
+
+    @Test
+    void passesAChunkedOriginResponseThroughChunked() throws Exception {
+        byte[] video = randomBytes(3 * 1024 * 1024 + 1);
+        startServingOrigin(video, false);
+        Distribution created = cloudFrontService.createDistribution(
+                customOriginDistribution(allMethodsBehavior()), Map.of());
+
+        Response response = given().header("Host", created.getDomainName()).when().get("/video.bin");
+
+        assertEquals(200, response.statusCode());
+        assertNull(response.header("Content-Length"), "an origin that sent no length was given one");
+        assertArrayEquals(video, response.asByteArray());
+    }
+
+    @Test
+    void streamsALargeViewerBodyToTheOriginWithItsLength() throws Exception {
+        AtomicReference<ReceivedRequest> received = new AtomicReference<>();
+        startRecordingOrigin(received, new AtomicInteger());
+        Distribution created = cloudFrontService.createDistribution(
+                customOriginDistribution(allMethodsBehavior()), Map.of());
+        byte[] upload = randomBytes(3 * 1024 * 1024 + 1);
+
+        given()
+            .header("Host", created.getDomainName())
+            .contentType("application/octet-stream")
+            .body(upload)
+        .when()
+            .put("/uploads/video.bin")
+        .then()
+            .statusCode(200);
+
+        assertEquals(Integer.toString(upload.length), received.get().header("Content-Length"));
+        assertArrayEquals(upload, received.get().body());
+    }
+
+    @Test
+    void forwardsAChunkedViewerBodyChunked() throws Exception {
+        AtomicReference<ReceivedRequest> received = new AtomicReference<>();
+        startRecordingOrigin(received, new AtomicInteger());
+        Distribution created = cloudFrontService.createDistribution(
+                customOriginDistribution(allMethodsBehavior()), Map.of());
+        byte[] upload = randomBytes(200_001);
+
+        int status = assertTimeoutPreemptively(Duration.ofSeconds(60),
+                () -> postChunked(created.getDomainName(), "/uploads", upload));
+
+        assertEquals(201, status);
+        assertEquals("chunked", received.get().header("Transfer-Encoding"));
+        assertNull(received.get().header("Content-Length"));
+        assertArrayEquals(upload, received.get().body());
+    }
+
+    @Test
+    void releasesTheOriginConnectionOfEveryResponseItDoesNotSend() throws Exception {
+        // The origin client keeps a fixed number of connections. An error response replaced by a custom
+        // error page, or a HEAD's, that kept its connection would leave a request past that number
+        // waiting for one.
+        originServer = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        originServer.createContext("/", exchange -> {
+            boolean errorPage = "/errors/404.html".equals(exchange.getRequestURI().getRawPath());
+            byte[] body = (errorPage ? "error-page" : "x".repeat(100_000)).getBytes(StandardCharsets.UTF_8);
+            exchange.getResponseHeaders().add("Content-Type", "text/plain");
+            boolean head = "HEAD".equals(exchange.getRequestMethod());
+            exchange.sendResponseHeaders(errorPage ? 200 : 404, head ? -1 : body.length);
+            if (!head) {
+                exchange.getResponseBody().write(body);
+            }
+            exchange.close();
+        });
+        originServer.start();
+        Map<String, Object> errorPage = new LinkedHashMap<>(Map.of(
+                "ErrorCode", "404", "ResponseCode", "200", "ResponsePagePath", "/errors/404.html"));
+        Distribution distribution = customOriginDistribution(allMethodsBehavior());
+        distribution.getConfig().setCustomErrorResponses(List.of(errorPage));
+        Distribution created = cloudFrontService.createDistribution(distribution, Map.of());
+
+        assertTimeoutPreemptively(Duration.ofSeconds(60), () -> {
+            for (int i = 0; i < CloudFrontOriginHttpClient.MAX_CONNECTIONS + 10; i++) {
+                given().header("Host", created.getDomainName()).when().get("/missing/" + i)
+                        .then().statusCode(200).body(equalTo("error-page"));
+                given().header("Host", created.getDomainName()).when().head("/missing/" + i)
+                        .then().statusCode(200);
+            }
+        });
+    }
+
     @Test
     void forwardsTheViewerAuthorizationOnWriteMethodsOnly() throws Exception {
         AtomicReference<ReceivedRequest> received = new AtomicReference<>();
@@ -705,6 +813,51 @@ class CloudFrontCustomOriginServingTest {
         originServer.start();
     }
 
+    /** An origin that answers every GET with {@code body}, with its length or chunked. */
+    private void startServingOrigin(byte[] body, boolean withLength) throws IOException {
+        originServer = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        originServer.createContext("/", exchange -> {
+            exchange.getRequestBody().readAllBytes();
+            exchange.getResponseHeaders().add("Content-Type", "application/octet-stream");
+            exchange.sendResponseHeaders(200, withLength ? body.length : 0);
+            exchange.getResponseBody().write(body);
+            exchange.close();
+        });
+        originServer.start();
+    }
+
+    /**
+     * POSTs {@code body} with {@code Transfer-Encoding: chunked} over a raw connection, since an HTTP
+     * client library buffers a stream it is given and sends its length instead.
+     */
+    private static int postChunked(String host, String path, byte[] body) throws IOException {
+        try (Socket socket = new Socket("localhost", RestAssured.port)) {
+            // A response that never comes fails the read instead of leaving it waiting.
+            socket.setSoTimeout(30_000);
+            OutputStream out = socket.getOutputStream();
+            out.write(("POST " + path + " HTTP/1.1\r\nHost: " + host
+                    + "\r\nContent-Type: application/octet-stream\r\nTransfer-Encoding: chunked"
+                    + "\r\nConnection: close\r\n\r\n").getBytes(StandardCharsets.US_ASCII));
+            for (int offset = 0; offset < body.length; offset += 8192) {
+                int size = Math.min(8192, body.length - offset);
+                out.write((Integer.toHexString(size) + "\r\n").getBytes(StandardCharsets.US_ASCII));
+                out.write(body, offset, size);
+                out.write("\r\n".getBytes(StandardCharsets.US_ASCII));
+            }
+            out.write("0\r\n\r\n".getBytes(StandardCharsets.US_ASCII));
+            out.flush();
+            String statusLine = new BufferedReader(
+                    new InputStreamReader(socket.getInputStream(), StandardCharsets.US_ASCII)).readLine();
+            return Integer.parseInt(statusLine.split(" ")[1]);
+        }
+    }
+
+    private static byte[] randomBytes(int size) {
+        byte[] bytes = new byte[size];
+        new Random(size).nextBytes(bytes);
+        return bytes;
+    }
+
     private Distribution customOriginDistribution(DefaultCacheBehavior behavior) {
         return customOriginDistribution(behavior, null);
     }

src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontDistributionServingTest.java

@@ -273,6 +273,7 @@ class CloudFrontDistributionServingTest {
                 .header("x-amz-meta-source", equalTo("origin"))
                 .header("x-amz-server-side-encryption", nullValue())
                 .header("x-amz-checksum-sha256", nullValue())
+                .header("Content-Length", equalTo(Integer.toString(body.getBytes(StandardCharsets.UTF_8).length)))
                 .body(equalTo(body));
 
         given().header("Host", dist.getDomainName()).when().head("/asset.txt")

src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontOriginHttpClientTest.java

@@ -16,12 +16,15 @@ import javax.net.ssl.SNIHostName;
 import javax.net.ssl.SNIServerName;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.TrustManagerFactory;
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.URI;
 import java.net.UnknownHostException;
 import java.net.http.HttpRequest;
-import java.net.http.HttpResponse;
 import java.nio.charset.StandardCharsets;
 import java.security.KeyStore;
 import java.security.Security;
@@ -29,12 +32,18 @@ import java.security.cert.X509Certificate;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 class CloudFrontOriginHttpClientTest {
 
@@ -46,8 +55,7 @@ class CloudFrontOriginHttpClientTest {
                 resolver(InetAddress.getByName("127.0.0.1")), List.of())) {
             HttpRequest request = request("http://blocked.invalid:" + server.getAddress().getPort() + "/");
 
-            assertThrows(UnknownHostException.class,
-                    () -> client.send(request, HttpResponse.BodyHandlers.ofByteArray()));
+            assertThrows(UnknownHostException.class, () -> get(client, request));
             assertEquals(0, hits.get());
         } finally {
             server.stop(0);
@@ -60,9 +68,7 @@ class CloudFrontOriginHttpClientTest {
                 InetAddress.getByName("8.8.8.8"),
                 InetAddress.getByName("127.0.0.1"));
         try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient(resolver, List.of())) {
-            assertThrows(UnknownHostException.class, () -> client.send(
-                    request("http://mixed.invalid:8080/"),
-                    HttpResponse.BodyHandlers.ofByteArray()));
+            assertThrows(UnknownHostException.class, () -> get(client, request("http://mixed.invalid:8080/")));
         }
     }
 
@@ -99,12 +105,11 @@ class CloudFrontOriginHttpClientTest {
 
         try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient(
                 resolver, List.of("rebind.invalid"))) {
-            HttpResponse<byte[]> response = client.send(
-                    request("http://rebind.invalid:" + server.getAddress().getPort() + "/"),
-                    HttpResponse.BodyHandlers.ofByteArray());
+            CloudFrontOriginHttpClient.OpenResponse response = get(client,
+                    request("http://rebind.invalid:" + server.getAddress().getPort() + "/"));
 
             assertEquals(200, response.statusCode());
-            assertArrayEquals("origin-ok".getBytes(StandardCharsets.UTF_8), response.body());
+            assertArrayEquals("origin-ok".getBytes(StandardCharsets.UTF_8), readAll(response));
             assertEquals(1, resolutions.get());
             assertEquals(1, hits.get());
             assertEquals("rebind.invalid:" + server.getAddress().getPort(), hostHeader.get());
@@ -131,9 +136,9 @@ class CloudFrontOriginHttpClientTest {
 
         try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient(
                 resolver(InetAddress.getByName("127.0.0.1")), List.of("redirect.invalid"))) {
-            HttpResponse<byte[]> response = client.send(
-                    request("http://redirect.invalid:" + server.getAddress().getPort() + "/start"),
-                    HttpResponse.BodyHandlers.ofByteArray());
+            CloudFrontOriginHttpClient.OpenResponse response = get(client,
+                    request("http://redirect.invalid:" + server.getAddress().getPort() + "/start"));
+            readAll(response);
 
             assertEquals(302, response.statusCode());
             assertEquals(0, redirectedHits.get());
@@ -162,10 +167,7 @@ class CloudFrontOriginHttpClientTest {
                     .GET()
                     .build();
 
-            client.send(
-                    request,
-                    Map.of("X-Origin-Verify", "configured-value"),
-                    HttpResponse.BodyHandlers.ofByteArray());
+            readAll(client.open(request, List.of(), Map.of("X-Origin-Verify", "configured-value"), null, -1));
 
             assertEquals(List.of("configured-value"), receivedValues.get());
         } finally {
@@ -200,8 +202,9 @@ class CloudFrontOriginHttpClientTest {
                     .method("PUT", HttpRequest.BodyPublishers.noBody())
                     .build();
 
-            HttpResponse<byte[]> response = client.send(
-                    request, List.of(), Map.of(), payload, HttpResponse.BodyHandlers.ofByteArray());
+            CloudFrontOriginHttpClient.OpenResponse response = client.open(
+                    request, List.of(), Map.of(), new ByteArrayInputStream(payload), payload.length);
+            readAll(response);
 
             assertEquals(204, response.statusCode());
             assertEquals("PUT", method.get());
@@ -249,21 +252,125 @@ class CloudFrontOriginHttpClientTest {
         try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient(
                 resolver, List.of("origin.invalid", "wrong.invalid"), clientContext)) {
             int port = server.getAddress().getPort();
-            HttpResponse<byte[]> response = client.send(
-                    request("https://origin.invalid:" + port + "/"),
-                    HttpResponse.BodyHandlers.ofByteArray());
+            CloudFrontOriginHttpClient.OpenResponse response = get(client,
+                    request("https://origin.invalid:" + port + "/"));
+            readAll(response);
 
             assertEquals(200, response.statusCode());
             assertEquals("origin.invalid:" + port, hostHeader.get());
             assertEquals("origin.invalid", requestedSni.get());
-            assertThrows(IOException.class, () -> client.send(
-                    request("https://wrong.invalid:" + port + "/"),
-                    HttpResponse.BodyHandlers.ofByteArray()));
+            assertThrows(IOException.class, () -> get(client, request("https://wrong.invalid:" + port + "/")));
         } finally {
             server.stop(0);
         }
     }
 
+    @Test
+    void streamsARequestBodyChunkedWhenItsLengthIsUnknown() throws Exception {
+        AtomicReference<String> transferEncoding = new AtomicReference<>();
+        AtomicReference<byte[]> body = new AtomicReference<>();
+        HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        server.createContext("/", exchange -> {
+            transferEncoding.set(exchange.getRequestHeaders().getFirst("Transfer-Encoding"));
+            body.set(exchange.getRequestBody().readAllBytes());
+            exchange.sendResponseHeaders(204, -1);
+            exchange.close();
+        });
+        server.start();
+
+        byte[] payload = new byte[100_000];
+        new Random(1).nextBytes(payload);
+        try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient(
+                resolver(InetAddress.getByName("127.0.0.1")), List.of("origin.invalid"))) {
+            HttpRequest request = HttpRequest.newBuilder(URI.create(
+                            "http://origin.invalid:" + server.getAddress().getPort() + "/items"))
+                    .timeout(Duration.ofSeconds(5))
+                    .method("POST", HttpRequest.BodyPublishers.noBody())
+                    .build();
+
+            readAll(client.open(request, List.of(), Map.of(), new ByteArrayInputStream(payload), -1));
+
+            assertEquals("chunked", transferEncoding.get());
+            assertArrayEquals(payload, body.get());
+        } finally {
+            server.stop(0);
+        }
+    }
+
+    @Test
+    void closingAResponseBeforeItsEndDoesNotReadTheRestOfIt() throws Exception {
+        long size = 512L * 1024 * 1024;
+        AtomicLong sent = new AtomicLong();
+        CountDownLatch done = new CountDownLatch(1);
+        HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        server.createContext("/", exchange -> {
+            byte[] chunk = new byte[64 * 1024];
+            exchange.sendResponseHeaders(200, size);
+            try (OutputStream out = exchange.getResponseBody()) {
+                for (long written = 0; written < size; written += chunk.length) {
+                    out.write(chunk);
+                    sent.addAndGet(chunk.length);
+                }
+            } catch (IOException expected) {
+                // The client went away mid-body, which is what the test does.
+            } finally {
+                done.countDown();
+            }
+        });
+        server.start();
+
+        try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient(
+                resolver(InetAddress.getByName("127.0.0.1")), List.of("origin.invalid"))) {
+            CloudFrontOriginHttpClient.OpenResponse response = get(client,
+                    request("http://origin.invalid:" + server.getAddress().getPort() + "/big"));
+            assertEquals(size, response.contentLength());
+            response.body().readNBytes(1024);
+            response.body().close();
+
+            assertTrue(done.await(30, TimeUnit.SECONDS), "the origin is still sending");
+            assertTrue(sent.get() < size / 2, "closing read " + (sent.get() >> 20) + " MiB of " + (size >> 20));
+        } finally {
+            server.stop(0);
+        }
+    }
+
+    @Test
+    void aResponseReadToItsEndLeavesItsConnectionForTheNextRequest() throws Exception {
+        List<InetSocketAddress> peers = new CopyOnWriteArrayList<>();
+        HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        server.createContext("/", exchange -> {
+            peers.add(exchange.getRemoteAddress());
+            byte[] body = "origin-ok".getBytes(StandardCharsets.UTF_8);
+            exchange.sendResponseHeaders(200, body.length);
+            exchange.getResponseBody().write(body);
+            exchange.close();
+        });
+        server.start();
+
+        try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient(
+                resolver(InetAddress.getByName("127.0.0.1")), List.of("origin.invalid"))) {
+            String uri = "http://origin.invalid:" + server.getAddress().getPort() + "/";
+            readAll(get(client, request(uri)));
+            readAll(get(client, request(uri)));
+
+            assertEquals(2, peers.size());
+            assertEquals(peers.get(0), peers.get(1), "the second request opened a new connection");
+        } finally {
+            server.stop(0);
+        }
+    }
+
+    private static CloudFrontOriginHttpClient.OpenResponse get(CloudFrontOriginHttpClient client,
+                                                               HttpRequest request) throws Exception {
+        return client.open(request, List.of(), Map.of(), null, -1);
+    }
+
+    private static byte[] readAll(CloudFrontOriginHttpClient.OpenResponse response) throws IOException {
+        try (InputStream body = response.body()) {
+            return body.readAllBytes();
+        }
+    }
+
     private static HttpServer server(AtomicInteger hits) throws Exception {
         HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
         server.createContext("/", exchange -> {

src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontS3DiskOriginServingTest.java

@@ -0,0 +1,108 @@
+package io.github.hectorvent.floci.services.cloudfront;
+
+import io.github.hectorvent.floci.services.cloudfront.model.DefaultCacheBehavior;
+import io.github.hectorvent.floci.services.cloudfront.model.Distribution;
+import io.github.hectorvent.floci.services.cloudfront.model.DistributionConfig;
+import io.github.hectorvent.floci.services.cloudfront.model.Origin;
+import io.github.hectorvent.floci.services.s3.S3Service;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.QuarkusTestProfile;
+import io.quarkus.test.junit.TestProfile;
+import io.restassured.response.Response;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+import static io.restassured.RestAssured.given;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * An S3 origin whose objects live on disk, where CloudFront streams the object file to the viewer
+ * instead of reading it into memory first.
+ */
+@QuarkusTest
+@TestProfile(CloudFrontS3DiskOriginServingTest.S3OnDiskProfile.class)
+class CloudFrontS3DiskOriginServingTest {
+
+    private static final String REGION = "us-east-1";
+
+    @Inject
+    S3Service s3Service;
+
+    @Inject
+    CloudFrontService cloudFrontService;
+
+    @Test
+    void streamsAnObjectFileWithItsLength() {
+        String suffix = Long.toString(System.nanoTime(), 36);
+        String bucket = "cf-disk-origin-" + suffix;
+        // Larger than any one buffer on the way, so the body arrives over many writes.
+        byte[] body = new byte[3 * 1024 * 1024 + 1];
+        new Random(5080).nextBytes(body);
+        s3Service.createBucket(bucket, REGION);
+        try {
+            s3Service.putObject(bucket, "video.bin", body, "application/octet-stream", Map.of());
+            s3Service.putObject(bucket, "index.html", ("INDEX-ON-DISK-" + suffix).getBytes(StandardCharsets.UTF_8),
+                    "text/html", Map.of());
+            Distribution dist = distributionFor(bucket);
+
+            Response get = given().header("Host", dist.getDomainName()).when().get("/video.bin");
+
+            assertEquals(200, get.statusCode());
+            assertEquals(Integer.toString(body.length), get.header("Content-Length"));
+            assertArrayEquals(body, get.asByteArray());
+            given().header("Host", dist.getDomainName()).when().head("/video.bin")
+                    .then().statusCode(200)
+                    .header("Content-Length", equalTo(Integer.toString(body.length)));
+            // The single-page-app fallback serves its error page through the same stream.
+            given().header("Host", dist.getDomainName()).when().get("/missing/route")
+                    .then().statusCode(200)
+                    .header("Content-Length", equalTo(Integer.toString(("INDEX-ON-DISK-" + suffix).length())))
+                    .body(containsString("INDEX-ON-DISK-" + suffix));
+        } finally {
+            // The bucket is on disk and outlives the test, so it is removed rather than left for the next run.
+            s3Service.deleteObject(bucket, "video.bin");
+            s3Service.deleteObject(bucket, "index.html");
+            s3Service.deleteBucket(bucket);
+        }
+    }
+
+    private Distribution distributionFor(String bucket) {
+        Origin origin = new Origin();
+        origin.setId("disk-origin");
+        origin.setDomainName(bucket + ".s3." + REGION + ".amazonaws.com");
+        origin.setS3OriginConfig(new LinkedHashMap<>(Map.of("OriginAccessIdentity", "")));
+        DefaultCacheBehavior behavior = new DefaultCacheBehavior();
+        behavior.setTargetOriginId(origin.getId());
+        behavior.setViewerProtocolPolicy("allow-all");
+        Map<String, Object> spaFallback = new LinkedHashMap<>();
+        spaFallback.put("ErrorCode", "404");
+        spaFallback.put("ResponseCode", "200");
+        spaFallback.put("ResponsePagePath", "/index.html");
+        DistributionConfig config = new DistributionConfig();
+        config.setEnabled(true);
+        config.setOrigins(List.of(origin));
+        config.setDefaultCacheBehavior(behavior);
+        config.setCustomErrorResponses(List.of(spaFallback));
+        Distribution distribution = new Distribution();
+        distribution.setConfig(config);
+        return cloudFrontService.createDistribution(distribution, Map.of());
+    }
+
+    public static final class S3OnDiskProfile implements QuarkusTestProfile {
+        @Override
+        public Map<String, String> getConfigOverrides() {
+            return Map.of(
+                    "floci.storage.services.s3.mode", "persistent",
+                    "floci.storage.persistent-path", "target/cloudfront-s3-disk-origin-it");
+        }
+    }
+}

src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontViewerBodyTest.java

@@ -0,0 +1,62 @@
+package io.github.hectorvent.floci.services.cloudfront;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** How the length of a viewer's request body is taken before it is forwarded to a custom origin. */
+class CloudFrontViewerBodyTest {
+
+    private static final byte[] DATA = "a viewer upload".getBytes(StandardCharsets.UTF_8);
+
+    @Test
+    void aChunkedBodyIsForwardedChunkedWhateverItsContentLength() throws IOException {
+        CloudFrontServingController.ViewerBody body =
+                CloudFrontServingController.viewerBody(new ByteArrayInputStream(DATA), "chunked", "3");
+
+        assertEquals(-1, body.length());
+        assertArrayEquals(DATA, readAll(body));
+    }
+
+    @Test
+    void aBodyWithAContentLengthKeepsIt() throws IOException {
+        CloudFrontServingController.ViewerBody body = CloudFrontServingController.viewerBody(
+                new ByteArrayInputStream(DATA), null, Integer.toString(DATA.length));
+
+        assertEquals(DATA.length, body.length());
+        assertArrayEquals(DATA, readAll(body));
+    }
+
+    @Test
+    void anInvalidContentLengthIsForwardedChunked() {
+        assertEquals(-1, CloudFrontServingController.viewerBody(
+                new ByteArrayInputStream(DATA), null, "not-a-length").length());
+    }
+
+    @Test
+    void aRequestWithNeitherHeaderAndNoBodyIsEmpty() {
+        assertEquals(0, CloudFrontServingController.viewerBody(InputStream.nullInputStream(), null, null).length());
+        assertEquals(0, CloudFrontServingController.viewerBody(null, null, null).length());
+    }
+
+    @Test
+    void aBodyWithNeitherHeaderAsHttp2AllowsIsForwardedChunkedAndWhole() throws IOException {
+        CloudFrontServingController.ViewerBody body =
+                CloudFrontServingController.viewerBody(new ByteArrayInputStream(DATA), null, null);
+
+        assertEquals(-1, body.length());
+        assertArrayEquals(DATA, readAll(body), "the byte read to look at the body was lost");
+    }
+
+    private static byte[] readAll(CloudFrontServingController.ViewerBody body) throws IOException {
+        try (InputStream stream = body.stream()) {
+            return stream.readAllBytes();
+        }
+    }
+}