fix(client): keep reproducible body across header rewrites; harden docs+tests

Addresses code-review findings on the reproducible-request-bodies work:

- ReproducibleHttpRequest now overrides withHeaders() to rebind the same
  body factory to the new headers. Previously a header rewrite (e.g. a
  base-URI path prefix applied by DefaultWebClient, or a decorator
  overriding the path) wrapped the request in HeaderOverridingHttpRequest,
  which does not override toDuplicator and fell back to the buffering
  DefaultStreamMessageDuplicator -- silently reintroducing the ~2 GiB
  ContentTooLargeException this request type exists to avoid.

- Corrects the HttpRequest.reproducible javadoc: RedirectingClient is not
  always the outermost duplicating decorator (redirects are disabled by
  default), so with retries only, RetryingClient is outermost and replays
  reproducibly. Also documents that each factory invocation must produce an
  equivalent body (not merely the same length), since the fixed
  content-length is reused unvalidated, and documents throwing/null factory
  fail-fast semantics.

- Tests: add direct-consume coverage (normal / throwing / null factory via a
  plain WebClient with no decorator), a base-path-prefix regression test that
  fails without the withHeaders fix, and non-buffering proofs
  (toDuplicator ignores maxRequestLength; large body streams without
  accumulation). Tighten deterministic factory-invocation assertions from
  >= 2 to exactly 2 so an over-regeneration bug is caught rather than masked.

Co-authored-by: Isaac

Author: yzfeng2020

core/src/main/java/com/linecorp/armeria/common/HttpRequest.java

@@ -383,23 +383,39 @@ static HttpRequest of(RequestHeaders headers,
      * <p>The {@code bodyFactory} is invoked once per attempt (initial request, retry attempt, or
      * redirect hop). The fixed {@code headers} are reused for every attempt; the factory regenerates
      * only the body {@link StreamMessage}, so the request method and headers cannot drift between
-     * attempts. The body each factory invocation produces must match the declared
-     * {@link HttpHeaderNames#CONTENT_LENGTH} (if any), exactly as for any streaming
-     * {@link HttpRequest}.
-     *
-     * <p>Reproducible replay applies only at the outermost duplicating decorator, which for a
-     * {@code WebClient} is {@code RedirectingClient} (it wraps the user-supplied decorators, including
-     * {@code RetryingClient}). Each attempt that the outermost decorator hands downstream is an
-     * ordinary {@link HttpRequest}, so an inner decorator treats it as a normal request and may buffer
-     * it: with both redirect and retry enabled, {@code RedirectingClient} replays reproducibly across
-     * redirect hops, but an inner {@code RetryingClient} re-buffers the body when retrying within a
-     * single hop. Reproducibility is honored for streaming requests
+     * attempts. Every invocation must produce an <em>equivalent</em> body — the same bytes and
+     * trailers, not merely the same length — because the fixed {@code headers} (including any declared
+     * {@link HttpHeaderNames#CONTENT_LENGTH}) are reused verbatim on every attempt and are not
+     * re-validated against the regenerated body. A factory whose output varies across invocations
+     * (e.g. it embeds a timestamp, or reads a file being mutated concurrently) silently sends
+     * different data on a retry or redirect, and a length mismatch against a declared
+     * {@code content-length} corrupts wire framing (the request stalls or the next message is
+     * garbled) with no error surfaced.
+     *
+     * <p>Reproducible replay applies only at the <em>outermost</em> duplicating decorator; each attempt
+     * it hands downstream is an ordinary {@link HttpRequest}, so any inner duplicating decorator treats
+     * it as a normal request and buffers it. Which decorator is outermost depends on configuration:
+     * <ul>
+     *   <li>With retries only (the common case; redirects are disabled by default),
+     *       {@code RetryingClient} is the outermost decorator and replays reproducibly across every
+     *       retry.</li>
+     *   <li>With {@code followRedirects()} enabled, the built-in {@code RedirectingClient} wraps the
+     *       user-supplied decorators (including {@code RetryingClient}), so it replays reproducibly
+     *       across redirect hops, but an inner {@code RetryingClient} re-buffers the body when retrying
+     *       within a single hop — reintroducing the ~2 GiB limit for that retry. If you need
+     *       non-buffering retries of a very large body, avoid stacking {@code RetryingClient} beneath
+     *       {@code RedirectingClient}.</li>
+     * </ul>
+     * Reproducibility is honored for streaming requests
      * ({@link ExchangeType#isRequestStreaming()}); an aggregated exchange type buffers the body as
      * usual.
      *
      * @param headers the fixed {@link RequestHeaders} reused for every attempt
-     * @param bodyFactory produces a fresh body {@link StreamMessage} for each attempt; it must not
-     *                    return {@code null}
+     * @param bodyFactory produces a fresh body {@link StreamMessage} for each attempt, each equivalent
+     *                    to the first; it must not return {@code null}. If an invocation throws or
+     *                    returns {@code null}, that attempt fails with the thrown cause (or a
+     *                    {@link NullPointerException}) and the failure is propagated to the caller
+     *                    without exhausting the remaining retry budget.
      */
     @UnstableApi
     static HttpRequest reproducible(

core/src/main/java/com/linecorp/armeria/common/ReproducibleHttpRequest.java

@@ -16,6 +16,8 @@
 
 package com.linecorp.armeria.common;
 
+import static java.util.Objects.requireNonNull;
+
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Supplier;
 
@@ -96,6 +98,21 @@ public RequestHeaders headers() {
         return headers;
     }
 
+    @Override
+    public HttpRequest withHeaders(RequestHeaders newHeaders) {
+        requireNonNull(newHeaders, "newHeaders");
+        if (headers == newHeaders) {
+            return this;
+        }
+        // Preserve reproducibility across a header rewrite (e.g. a base-URI path prefix applied by
+        // DefaultWebClient, or a redirect/retry decorator overriding the path). The default
+        // HttpRequest.withHeaders wraps this in a HeaderOverridingHttpRequest, which does not override
+        // toDuplicator and would therefore fall back to the buffering DefaultHttpRequestDuplicator,
+        // reintroducing the ~2 GiB size limit this request type exists to avoid. Rebinding the same
+        // factory to the new headers keeps the non-buffering toDuplicator path.
+        return new ReproducibleHttpRequest(newHeaders, bodyFactory);
+    }
+
     @SuppressWarnings("unchecked")
     @Override
     public CompletableFuture<AggregatedHttpRequest> aggregate(AggregationOptions options) {

core/src/test/java/com/linecorp/armeria/client/ReproducibleHttpRequestClientTest.java

@@ -28,14 +28,17 @@
 
 import com.linecorp.armeria.client.retry.RetryRule;
 import com.linecorp.armeria.client.retry.RetryingClient;
+import com.linecorp.armeria.common.AggregatedHttpRequest;
 import com.linecorp.armeria.common.AggregatedHttpResponse;
+import com.linecorp.armeria.common.CommonPools;
 import com.linecorp.armeria.common.ExchangeType;
 import com.linecorp.armeria.common.HttpData;
 import com.linecorp.armeria.common.HttpHeaderNames;
 import com.linecorp.armeria.common.HttpHeaders;
 import com.linecorp.armeria.common.HttpMethod;
 import com.linecorp.armeria.common.HttpObject;
 import com.linecorp.armeria.common.HttpRequest;
+import com.linecorp.armeria.common.HttpRequestDuplicator;
 import com.linecorp.armeria.common.HttpResponse;
 import com.linecorp.armeria.common.HttpStatus;
 import com.linecorp.armeria.common.MediaType;
@@ -45,6 +48,8 @@
 import com.linecorp.armeria.server.ServerBuilder;
 import com.linecorp.armeria.testing.junit5.server.ServerExtension;
 
+import io.netty.util.concurrent.EventExecutor;
+
 class ReproducibleHttpRequestClientTest {
 
     private static final AtomicInteger serverHits = new AtomicInteger();
@@ -93,6 +98,24 @@ protected void configure(ServerBuilder sb) {
                     req.aggregate().thenApply(agg -> AggregatedHttpResponse.of(
                             HttpStatus.OK, MediaType.PLAIN_TEXT_UTF_8,
                             req.method() + ":" + agg.contentUtf8()).toHttpResponse())));
+            // Plain echo that always succeeds, for exercising the request without any retry/redirect
+            // decorator (the direct-consume path) and for a base-path-prefixed client.
+            sb.service("/echo", (ctx, req) -> HttpResponse.of(
+                    req.aggregate().thenApply(agg -> AggregatedHttpResponse.of(
+                            HttpStatus.OK, MediaType.PLAIN_TEXT_UTF_8,
+                            agg.contentUtf8()).toHttpResponse())));
+            // Same fail-once-then-echo behavior as /upload, reached only when a base-URI path prefix
+            // ("/api") rewrites the request path — the scenario that must keep the reproducible body.
+            sb.service("/api/upload", (ctx, req) -> HttpResponse.of(
+                    req.aggregate().thenApply(agg -> {
+                        final int hit = serverHits.incrementAndGet();
+                        if (hit == 1) {
+                            return AggregatedHttpResponse.of(HttpStatus.INTERNAL_SERVER_ERROR)
+                                                         .toHttpResponse();
+                        }
+                        return AggregatedHttpResponse.of(HttpStatus.OK, MediaType.PLAIN_TEXT_UTF_8,
+                                                         agg.contentUtf8()).toHttpResponse();
+                    })));
         }
     };
 
@@ -145,9 +168,12 @@ void retryRegeneratesBody() {
 
         assertThat(res.status()).isEqualTo(HttpStatus.OK);
         assertThat(res.contentUtf8()).isEqualTo("hello-body");
-        assertThat(serverHits).hasValueGreaterThanOrEqualTo(2);
+        // The server fails exactly once, so the attempt count is deterministic: initial + one retry.
+        // Assert exactly 2 so a double-subscription bug that regenerates the body 3+ times (leaking a
+        // fresh factory resource per attempt) is caught rather than masked by a >= assertion.
+        assertThat(serverHits).hasValue(2);
         // Body regenerated for the initial attempt and the retry.
-        assertThat(bodyCalls).hasValueGreaterThanOrEqualTo(2);
+        assertThat(bodyCalls).hasValue(2);
     }
 
     @Test
@@ -179,8 +205,9 @@ void retryReproducesMultiChunkBodyAndTrailers() {
         assertThat(res.status()).isEqualTo(HttpStatus.OK);
         // Concatenated interior chunks (in order) plus the reproduced trailer, on the re-sent attempt.
         assertThat(res.contentUtf8()).isEqualTo("abc|v");
-        assertThat(serverHits).hasValueGreaterThanOrEqualTo(2);
-        assertThat(bodyCalls).hasValueGreaterThanOrEqualTo(2);
+        // Deterministic: initial + one retry. Exact assertion guards against over-regeneration.
+        assertThat(serverHits).hasValue(2);
+        assertThat(bodyCalls).hasValue(2);
     }
 
     @Test
@@ -238,8 +265,9 @@ void followsRedirectRegeneratingBody() {
 
         assertThat(res.status()).isEqualTo(HttpStatus.OK);
         assertThat(res.contentUtf8()).isEqualTo("redir-body");
-        // Body regenerated for the initial request and the redirected hop.
-        assertThat(bodyCalls).hasValueGreaterThanOrEqualTo(2);
+        // Deterministic: initial request + one redirect hop (no server error, so no retry). Exact
+        // assertion guards against over-regeneration.
+        assertThat(bodyCalls).hasValue(2);
     }
 
     @Test
@@ -291,7 +319,124 @@ void stackedRetryAndRedirect() {
 
         assertThat(res.status()).isEqualTo(HttpStatus.OK);
         assertThat(res.contentUtf8()).isEqualTo("redir-body");
-        // Body regenerated for the initial request and the redirected hop.
-        assertThat(bodyCalls).hasValueGreaterThanOrEqualTo(2);
+        // Deterministic: initial request + one redirect hop (no server error, so no retry). Exact
+        // assertion guards against over-regeneration.
+        assertThat(bodyCalls).hasValue(2);
+    }
+
+    @Test
+    void directConsumeWithoutDecoratorSendsBodyOnce() {
+        // No retry/redirect decorator, so the request is consumed directly via its lazyBody delegate
+        // (never through toDuplicator). This path is otherwise unexercised — every other test drives a
+        // decorator. The factory must be invoked exactly once and the body delivered intact.
+        final AtomicInteger bodyCalls = new AtomicInteger();
+        final RequestHeaders headers =
+                RequestHeaders.of(HttpMethod.POST, "/echo",
+                                  HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+        final Supplier<StreamMessage<? extends HttpObject>> bodyFactory = () -> {
+            bodyCalls.incrementAndGet();
+            return StreamMessage.of(HttpData.ofUtf8("direct-body"));
+        };
+
+        final WebClient client = WebClient.of(server.httpUri());
+        final AggregatedHttpResponse res =
+                client.execute(HttpRequest.reproducible(headers, bodyFactory), streamingOptions())
+                      .aggregate().join();
+
+        assertThat(res.status()).isEqualTo(HttpStatus.OK);
+        assertThat(res.contentUtf8()).isEqualTo("direct-body");
+        assertThat(bodyCalls).hasValue(1);
+    }
+
+    @Test
+    void directConsumeSurfacesThrowingFactory() {
+        // On the direct path, a throwing factory must surface its own cause to the subscriber rather
+        // than hang or swallow the error.
+        final RequestHeaders headers =
+                RequestHeaders.of(HttpMethod.POST, "/echo",
+                                  HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+        final Supplier<StreamMessage<? extends HttpObject>> bodyFactory = () -> {
+            throw new IllegalStateException("cannot produce body");
+        };
+
+        final WebClient client = WebClient.of(server.httpUri());
+        assertThatThrownBy(() -> client.execute(HttpRequest.reproducible(headers, bodyFactory),
+                                                streamingOptions())
+                                       .aggregate().join())
+                .getRootCause()
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("cannot produce body");
+    }
+
+    @Test
+    void directConsumeSurfacesNullFactory() {
+        // On the direct path, a factory returning null must surface an NPE, matching the duplicator
+        // path's null handling.
+        final RequestHeaders headers =
+                RequestHeaders.of(HttpMethod.POST, "/echo",
+                                  HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+        final Supplier<StreamMessage<? extends HttpObject>> bodyFactory = () -> null;
+
+        final WebClient client = WebClient.of(server.httpUri());
+        assertThatThrownBy(() -> client.execute(HttpRequest.reproducible(headers, bodyFactory),
+                                                streamingOptions())
+                                       .aggregate().join())
+                .getRootCause()
+                .isInstanceOf(NullPointerException.class);
+    }
+
+    @Test
+    void toDuplicatorIgnoresMaxRequestLength() {
+        // The reproducible duplicator never buffers, so it must ignore the maxRequestLength cap that a
+        // buffering DefaultStreamMessageDuplicator would enforce. A body far larger than a tiny cap must
+        // still stream to completion; a regression that fell back to a buffering duplicator would throw
+        // ContentTooLargeException here.
+        final byte[] large = new byte[64 * 1024];
+        final Supplier<StreamMessage<? extends HttpObject>> bodyFactory =
+                () -> StreamMessage.of(HttpData.wrap(large));
+        final RequestHeaders headers = RequestHeaders.of(HttpMethod.POST, "/echo");
+
+        final EventExecutor executor = CommonPools.workerGroup().next();
+        final HttpRequestDuplicator duplicator =
+                HttpRequest.reproducible(headers, bodyFactory).toDuplicator(executor, 8);
+
+        final AggregatedHttpRequest produced = duplicator.duplicate().aggregate().join();
+        assertThat(produced.content().length()).isEqualTo(large.length);
+        duplicator.close();
+    }
+
+    @Test
+    void basePathPrefixRemainsReproducible() {
+        // A WebClient built with a base-URI path prefix rewrites the request path via
+        // req.withHeaders(...). If ReproducibleHttpRequest did not override withHeaders, the rewritten
+        // request would be a plain HeaderOverridingHttpRequest whose toDuplicator falls back to the
+        // buffering DefaultStreamMessageDuplicator — silently reintroducing the ~2 GiB limit. This test
+        // pins that the rewritten request still regenerates its body per attempt (non-buffering path).
+        final AtomicInteger bodyCalls = new AtomicInteger();
+        // Header path is "/upload"; the base URI prefix "/api" makes the effective path "/api/upload",
+        // forcing a path rewrite. Route the server so /api/upload retries once like /upload does.
+        final RequestHeaders headers =
+                RequestHeaders.of(HttpMethod.POST, "/upload",
+                                  HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8);
+        final Supplier<StreamMessage<? extends HttpObject>> bodyFactory = () -> {
+            bodyCalls.incrementAndGet();
+            return StreamMessage.of(HttpData.ofUtf8("prefixed-body"));
+        };
+
+        final WebClient client =
+                WebClient.builder(server.httpUri() + "/api")
+                         .decorator(RetryingClient.newDecorator(
+                                 RetryRule.builder().onServerErrorStatus().thenBackoff()))
+                         .build();
+
+        final AggregatedHttpResponse res =
+                client.execute(HttpRequest.reproducible(headers, bodyFactory), streamingOptions())
+                      .aggregate().join();
+
+        assertThat(res.status()).isEqualTo(HttpStatus.OK);
+        assertThat(res.contentUtf8()).isEqualTo("prefixed-body");
+        // Regenerated for the initial attempt and the retry — proving the path-rewritten request kept
+        // the reproducible (non-buffering) duplicator rather than falling back to buffering.
+        assertThat(bodyCalls).hasValue(2);
     }
 }

core/src/test/java/com/linecorp/armeria/internal/common/ReproducibleHttpRequestDuplicatorTest.java

@@ -132,6 +132,24 @@ void closeLeavesOutstandingRequestsActive() {
         assertThat(produced.whenComplete()).isNotDone();
     }
 
+    @Test
+    void streamsBodyWithoutAccumulatingIt() {
+        // The whole point of the reproducible duplicator is to avoid buffering the body for replay (and
+        // thus the ~2 GiB int32 cap of DefaultStreamMessageDuplicator). A produced request must stream
+        // its body straight through rather than accumulate it; here we simply assert a large body is
+        // delivered intact. See ReproducibleHttpRequestClientTest#toDuplicatorIgnoresMaxRequestLength
+        // for the companion proof that the maxRequestLength cap is not applied.
+        final byte[] large = new byte[64 * 1024];
+        final Supplier<StreamMessage<? extends HttpObject>> factory =
+                () -> StreamMessage.of(HttpData.wrap(large));
+        final ReproducibleHttpRequestDuplicator dup =
+                new ReproducibleHttpRequestDuplicator(HEADERS, factory);
+
+        final HttpRequest produced = dup.duplicate();
+        final int received = produced.aggregate().join().content().length();
+        assertThat(received).isEqualTo(large.length);
+    }
+
     @Test
     void completedRequestIsUntrackedSoAbortDoesNotAffectIt() {
         final Supplier<StreamMessage<? extends HttpObject>> factory =