Reproducible request bodies

core/src/main/java/com/linecorp/armeria/client/RedirectingClient.java

@@ -185,9 +185,12 @@ private void execute0(ClientRequestContext ctx, RedirectContext redirectCtx,
             return;
         }
 
-        final HttpRequest duplicateReq = reqDuplicator.duplicate();
+        final HttpRequest duplicateReq;
         final ClientRequestContext derivedCtx;
         try {
+            // duplicate() may throw if the request body cannot be reproduced (see
+            // HttpRequest.reproducible); fail the request instead of following the redirect.
+            duplicateReq = reqDuplicator.duplicate();
             derivedCtx = ClientUtil.newDerivedContext(ctx, duplicateReq, ctx.rpcRequest(), initialAttempt);
         } catch (Throwable t) {
             handleException(ctx, reqDuplicator, responseFuture, t, initialAttempt);

core/src/main/java/com/linecorp/armeria/client/retry/RetryingClient.java

@@ -296,16 +296,17 @@ private void doExecute0(ClientRequestContext ctx, HttpRequestDuplicator rootReqD
         }
 
         final HttpRequest duplicateReq;
-        if (initialAttempt) {
-            duplicateReq = rootReqDuplicator.duplicate();
-        } else {
-            final RequestHeadersBuilder newHeaders = originalReq.headers().toBuilder();
-            newHeaders.setInt(ARMERIA_RETRY_COUNT, totalAttempts - 1);
-            duplicateReq = rootReqDuplicator.duplicate(newHeaders.build());
-        }
-
         final ClientRequestContext derivedCtx;
         try {
+            // duplicate() may throw if the request body cannot be reproduced (see
+            // HttpRequest.reproducible); fail the request instead of retrying an unreproducible body.
+            if (initialAttempt) {
+                duplicateReq = rootReqDuplicator.duplicate();
+            } else {
+                final RequestHeadersBuilder newHeaders = originalReq.headers().toBuilder();
+                newHeaders.setInt(ARMERIA_RETRY_COUNT, totalAttempts - 1);
+                duplicateReq = rootReqDuplicator.duplicate(newHeaders.build());
+            }
             derivedCtx = newDerivedContext(ctx, duplicateReq, ctx.rpcRequest(), initialAttempt);
         } catch (Throwable t) {
             handleException(ctx, rootReqDuplicator, future, t, initialAttempt);

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

@@ -30,6 +30,7 @@
 import java.util.concurrent.CompletionStage;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
@@ -357,6 +358,54 @@ static HttpRequest of(RequestHeaders headers,
         return of(headers, StreamMessage.of(stage, subscriberExecutor));
     }
 
+    /**
+     * Creates a new {@link HttpRequest} whose body can be reproduced on demand, so that
+     * {@code RetryingClient} and {@code RedirectingClient} can resend it without buffering the whole
+     * body in memory.
+     *
+     * <p>For streaming requests larger than about 2 GiB, buffering the body for replay is neither
+     * possible (an {@code int} size limit is exceeded) nor desirable. Instead, supply a factory that
+     * opens a fresh body {@link StreamMessage} on demand:
+     *
+     * <pre>{@code
+     * final RequestHeaders headers =
+     *         RequestHeaders.of(HttpMethod.POST, "/upload",
+     *                           HttpHeaderNames.CONTENT_TYPE, "application/octet-stream");
+     * final HttpRequest req = HttpRequest.reproducible(headers, () -> StreamMessage.of(path));
+     * final RequestOptions options =
+     *         RequestOptions.builder()
+     *                       .exchangeType(ExchangeType.REQUEST_STREAMING)
+     *                       .build();
+     * client.execute(req, options);
+     * }</pre>
+     *
+     * <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 to the outermost duplicating decorator (typically
+     * {@code RetryingClient}). Each attempt handed downstream is an ordinary {@link HttpRequest};
+     * an inner decorator (e.g. a redirect within a single retry attempt) treats it as a normal
+     * request and may buffer it. It 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}
+     */
+    @UnstableApi
+    static HttpRequest reproducible(
+            RequestHeaders headers,
+            Supplier<? extends StreamMessage<? extends HttpObject>> bodyFactory) {
+        requireNonNull(headers, "headers");
+        requireNonNull(bodyFactory, "bodyFactory");
+        return new ReproducibleHttpRequest(headers, bodyFactory);
+    }
+
     /**
      * Returns a new {@link HttpRequestBuilder}.
      */

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

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2026 LY Corporation
+ *
+ * LY Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.linecorp.armeria.common;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
+
+import org.reactivestreams.Publisher;
+
+import com.linecorp.armeria.common.stream.StreamMessage;
+import com.linecorp.armeria.internal.common.ReproducibleHttpRequestDuplicator;
+import com.linecorp.armeria.internal.common.stream.NonOverridableStreamMessageWrapper;
+
+import io.netty.util.concurrent.EventExecutor;
+
+/**
+ * An {@link HttpRequest} whose body can be reproduced on demand, so {@code RetryingClient} and
+ * {@code RedirectingClient} can resend it without buffering the whole body in memory.
+ *
+ * <p>See {@link HttpRequest#reproducible(RequestHeaders, Supplier)} for details and usage.
+ *
+ * <p>The body factory is invoked lazily: when this request is duplicated (the retry/redirect path),
+ * {@link #toDuplicator(EventExecutor, long)} returns a {@link ReproducibleHttpRequestDuplicator} that
+ * calls the factory once per attempt and this request's own delegate is never subscribed. When this
+ * request is subscribed directly (no retry/redirect decorator), the factory is invoked exactly once,
+ * on subscription, to produce the single body. Either way the factory is never called eagerly.
+ */
+final class ReproducibleHttpRequest
+        extends NonOverridableStreamMessageWrapper<HttpObject, HttpRequestDuplicator> implements HttpRequest {
+
+    private final RequestHeaders headers;
+    private final Supplier<? extends StreamMessage<? extends HttpObject>> bodyFactory;
+
+    ReproducibleHttpRequest(RequestHeaders headers,
+                            Supplier<? extends StreamMessage<? extends HttpObject>> bodyFactory) {
+        super(lazyBody(bodyFactory));
+        this.headers = headers;
+        this.bodyFactory = bodyFactory;
+    }
+
+    /**
+     * Returns a {@link StreamMessage} that invokes {@code bodyFactory} on its first (and only)
+     * subscription, so the factory is not called until this request is actually consumed directly.
+     */
+    private static StreamMessage<HttpObject> lazyBody(
+            Supplier<? extends StreamMessage<? extends HttpObject>> bodyFactory) {
+        return StreamMessage.of((Publisher<HttpObject>) subscriber -> {
+            final StreamMessage<? extends HttpObject> body;
+            try {
+                body = bodyFactory.get();
+            } catch (Throwable t) {
+                StreamMessage.<HttpObject>aborted(t).subscribe(subscriber);
+                return;
+            }
+            if (body == null) {
+                StreamMessage.<HttpObject>aborted(
+                        new NullPointerException("bodyFactory.get() returned null.")).subscribe(subscriber);
+                return;
+            }
+            @SuppressWarnings("unchecked")
+            final StreamMessage<HttpObject> cast = (StreamMessage<HttpObject>) body;
+            cast.subscribe(subscriber);
+        });
+    }
+
+    @Override
+    public RequestHeaders headers() {
+        return headers;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public CompletableFuture<AggregatedHttpRequest> aggregate(AggregationOptions options) {
+        return super.aggregate(options);
+    }
+
+    @Override
+    public HttpRequestDuplicator toDuplicator(EventExecutor executor) {
+        // Ignore the executor: the reproducible duplicator never buffers, so it needs no subscriber
+        // executor. Every duplicate() obtains a fresh body from the factory.
+        return new ReproducibleHttpRequestDuplicator(headers, bodyFactory);
+    }
+
+    @Override
+    public HttpRequestDuplicator toDuplicator(EventExecutor executor, long maxRequestLength) {
+        // maxRequestLength does not apply: this duplicator accumulates nothing, so there is no
+        // buffered length to cap. Each attempt streams a fresh body straight from the factory.
+        return new ReproducibleHttpRequestDuplicator(headers, bodyFactory);
+    }
+}

core/src/main/java/com/linecorp/armeria/internal/common/ReproducibleHttpRequestDuplicator.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright 2026 LY Corporation
+ *
+ * LY Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.linecorp.armeria.internal.common;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.function.Supplier;
+
+import com.linecorp.armeria.common.HttpObject;
+import com.linecorp.armeria.common.HttpRequest;
+import com.linecorp.armeria.common.HttpRequestDuplicator;
+import com.linecorp.armeria.common.RequestHeaders;
+import com.linecorp.armeria.common.annotation.Nullable;
+import com.linecorp.armeria.common.stream.StreamMessage;
+
+/**
+ * An {@link HttpRequestDuplicator} that reproduces the request body without buffering it. Every
+ * {@link #duplicate()} — including the first — obtains a fresh body from the supplied factory, so no
+ * attempt reuses another attempt's stream and the original request handed to the client is never put
+ * on the wire. This avoids the ~2 GiB {@code int} size limit and the memory cost of
+ * {@code DefaultStreamMessageDuplicator}, which buffers the whole body for replay.
+ *
+ * <p>{@code RetryingClient} and {@code RedirectingClient} drive this duplicator strictly
+ * sequentially: at most one produced request is outstanding at a time, and each access
+ * happens-after the previous one via the async response future. The duplicator tracks the single
+ * last-produced request so that {@link #abort(Throwable)} can tear it down if it was produced but
+ * never subscribed (e.g. endpoint selection threw before the wire took ownership). The fields are
+ * {@code volatile} for safe publication across the caller-thread → event-loop hand-off, matching how
+ * the clients drive it.
+ */
+public final class ReproducibleHttpRequestDuplicator implements HttpRequestDuplicator {
+
+    private final RequestHeaders headers;
+    private final Supplier<? extends StreamMessage<? extends HttpObject>> bodyFactory;
+
+    @Nullable
+    private volatile HttpRequest lastProduced;
+    private volatile boolean closed;
+
+    public ReproducibleHttpRequestDuplicator(
+            RequestHeaders headers,
+            Supplier<? extends StreamMessage<? extends HttpObject>> bodyFactory) {
+        this.headers = requireNonNull(headers, "headers");
+        this.bodyFactory = requireNonNull(bodyFactory, "bodyFactory");
+    }
+
+    @Override
+    public RequestHeaders headers() {
+        return headers;
+    }
+
+    @Override
+    public HttpRequest duplicate() {
+        return duplicate(headers);
+    }
+
+    @Override
+    public HttpRequest duplicate(RequestHeaders newHeaders) {
+        requireNonNull(newHeaders, "newHeaders");
+        if (closed) {
+            throw new IllegalStateException("duplicator is closed or aborted.");
+        }
+
+        // A produced-but-unsubscribed request from a previous attempt (e.g. endpoint selection threw
+        // before the wire subscribed) would otherwise be lost when we overwrite lastProduced; abort it
+        // so its body is released. Aborting an already-subscribed stream is a no-op.
+        final HttpRequest previous = lastProduced;
+        if (previous != null) {
+            previous.abort();
+        }
+
+        // Fail fast on a broken factory: propagate to the caller (RetryingClient / RedirectingClient),
+        // which completes the response exceptionally instead of re-judging an aborted request and
+        // wasting the retry budget.
+        final StreamMessage<? extends HttpObject> body =
+                requireNonNull(bodyFactory.get(), "bodyFactory.get() returned null.");
+        final HttpRequest produced = HttpRequest.of(newHeaders, body);
+        lastProduced = produced;
+        return produced;
+    }
+
+    @Override
+    public void close() {
+        // Let the last-produced request finish: on the success path the wire owns it and is still
+        // streaming, and the StreamMessageDuplicator contract says close() must not abort issued
+        // duplicates. If it was never subscribed, it is completed/aborted by its own consumer or by a
+        // subsequent abort(); we intentionally do not abort it here.
+        closed = true;
+    }
+
+    @Override
+    public void abort() {
+        abortLastProduced(null);
+    }
+
+    @Override
+    public void abort(Throwable cause) {
+        abortLastProduced(requireNonNull(cause, "cause"));
+    }
+
+    private void abortLastProduced(@Nullable Throwable cause) {
+        closed = true;
+        final HttpRequest lastProduced = this.lastProduced;
+        if (lastProduced == null) {
+            return;
+        }
+        this.lastProduced = null;
+        // If the wire already owns the subscription, abort() on an already-subscribed stream is a
+        // no-op. Otherwise this releases the produced-but-unsubscribed body (e.g. an open file).
+        if (cause != null) {
+            lastProduced.abort(cause);
+        } else {
+            lastProduced.abort();
+        }
+    }
+}