Reproducible request bodies - #6841
Conversation
… request bodies Co-authored-by: Isaac
Co-authored-by: Isaac
…bute present Co-authored-by: Isaac
…tribute present Co-authored-by: Isaac
…ssued volatile Follow-ups from the whole-branch review: - Mark RequestFactoryHttpRequestDuplicator.firstDuplicateIssued volatile for safe publication across the caller-thread -> event-loop hand-off (accesses are ordered via the async response future, but not confined to one thread). - Rename the factory-duplicator size test to reflect what it actually asserts. - Add end-to-end tests: factory throwing on a retry fails the request; a non-streaming request ignores the factory; a 303 See Other redirect drops the body (method -> GET) without invoking the factory again. - Reset the server-hit counter via @beforeeach. Co-authored-by: Isaac
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds ChangesReproducible body regeneration for streaming requests
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant RetryingClient
participant RedirectingClient
participant ReproducibleHttpRequestDuplicator
participant BodyFactory
participant Server
Client->>RetryingClient: execute streaming request
RetryingClient->>ReproducibleHttpRequestDuplicator: duplicate()
ReproducibleHttpRequestDuplicator->>BodyFactory: create fresh StreamMessage
BodyFactory-->>ReproducibleHttpRequestDuplicator: new body
RetryingClient->>RedirectingClient: execute request
RedirectingClient->>ReproducibleHttpRequestDuplicator: duplicate()
ReproducibleHttpRequestDuplicator->>BodyFactory: create fresh StreamMessage
RedirectingClient->>Server: send retry or redirect attempt
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
core/src/test/java/com/linecorp/armeria/internal/client/RequestFactoryHttpRequestDuplicatorTest.java (1)
110-126: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueTest doesn't actually exercise "no size cap" — only asserts non-null over 6 tiny requests.
The comment claims to verify behavior across "large-'reported'-length duplications," but the test body never uses large content; it just checks
dup.duplicate()returns non-null a handful of times. Consider renaming/trimming the comment to reflect what's actually tested (repeated duplication succeeds), or add a request with a genuinely large/streaming body if the size-cap-avoidance claim is meant to be verified here.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/src/test/java/com/linecorp/armeria/internal/client/RequestFactoryHttpRequestDuplicatorTest.java` around lines 110 - 126, The test in RequestFactoryHttpRequestDuplicatorTest#factoryDuplicatorHasNoSizeCapAcrossManyDuplicates currently claims to verify no size cap for large reported lengths, but it only checks a few non-null duplicates of tiny requests. Either update the test/comment to match the actual behavior being exercised in RequestFactoryHttpRequestDuplicator.duplicate() (repeated duplication succeeds), or change the setup to use a genuinely large or streaming HttpRequest body if you intend to prove the size-cap path is bypassed.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@core/src/test/java/com/linecorp/armeria/client/RedirectingClientReproducibleBodyTest.java`:
- Around line 95-124: The 303 redirect test in
RedirectingClientReproducibleBodyTest verifies the response but never checks
that factoryCalls stays at the expected count. Update
seeOtherRedirectDropsBodyAndDoesNotThrow() to assert the factory is only used
for the initial request and not invoked again for the SEE_OTHER hop, using the
existing factoryCalls AtomicInteger so the redirect body-drop behavior is
actually enforced.
---
Nitpick comments:
In
`@core/src/test/java/com/linecorp/armeria/internal/client/RequestFactoryHttpRequestDuplicatorTest.java`:
- Around line 110-126: The test in
RequestFactoryHttpRequestDuplicatorTest#factoryDuplicatorHasNoSizeCapAcrossManyDuplicates
currently claims to verify no size cap for large reported lengths, but it only
checks a few non-null duplicates of tiny requests. Either update the
test/comment to match the actual behavior being exercised in
RequestFactoryHttpRequestDuplicator.duplicate() (repeated duplication succeeds),
or change the setup to use a genuinely large or streaming HttpRequest body if
you intend to prove the size-cap path is bypassed.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 716d108e-71a8-4c6c-92a0-ad132a0b79b9
📒 Files selected for processing (8)
core/src/main/java/com/linecorp/armeria/client/ClientRequestBodyFactory.javacore/src/main/java/com/linecorp/armeria/client/RedirectingClient.javacore/src/main/java/com/linecorp/armeria/client/retry/RetryingClient.javacore/src/main/java/com/linecorp/armeria/internal/client/RequestFactoryHttpRequestDuplicator.javacore/src/test/java/com/linecorp/armeria/client/ClientRequestBodyFactoryTest.javacore/src/test/java/com/linecorp/armeria/client/RedirectingClientReproducibleBodyTest.javacore/src/test/java/com/linecorp/armeria/client/retry/RetryingClientReproducibleBodyTest.javacore/src/test/java/com/linecorp/armeria/internal/client/RequestFactoryHttpRequestDuplicatorTest.java
| @Test | ||
| void seeOtherRedirectDropsBodyAndDoesNotThrow() { | ||
| final AtomicInteger factoryCalls = new AtomicInteger(); | ||
| final Supplier<HttpRequest> factory = () -> { | ||
| factoryCalls.incrementAndGet(); | ||
| return HttpRequest.of(RequestHeaders.of(HttpMethod.POST, "/see-other", | ||
| HttpHeaderNames.CONTENT_TYPE, | ||
| "text/plain; charset=utf-8"), | ||
| StreamMessage.of(HttpData.ofUtf8("see-other-body"))); | ||
| }; | ||
|
|
||
| final WebClient client = | ||
| WebClient.builder(server.httpUri()) | ||
| .followRedirects() | ||
| .build(); | ||
|
|
||
| final RequestOptions options = | ||
| RequestOptions.builder() | ||
| .exchangeType(ExchangeType.REQUEST_STREAMING) | ||
| .attr(ClientRequestBodyFactory.REQUEST_BODY_FACTORY, factory) | ||
| .build(); | ||
|
|
||
| final AggregatedHttpResponse res = | ||
| client.execute(factory.get(), options).aggregate().join(); | ||
|
|
||
| // On a 303 the method is rewritten to GET and the body is dropped; the redirected GET has | ||
| // an empty body. The factory duplicator is aborted and never invoked again for the hop. | ||
| assertThat(res.status()).isEqualTo(HttpStatus.OK); | ||
| assertThat(res.contentUtf8()).isEqualTo("GET:"); | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Missing assertion on factoryCalls leaves the 303 behavior unverified.
The test declares and increments factoryCalls but never asserts on it, so the claim in the comment ("aborted and never invoked again for the hop") is not actually verified. A regression that invokes the factory on the SEE_OTHER hop would pass silently.
🧪 Proposed fix
assertThat(res.status()).isEqualTo(HttpStatus.OK);
assertThat(res.contentUtf8()).isEqualTo("GET:");
+ // The factory is invoked once for the initial request; the SEE_OTHER hop drops the body
+ // via reqDuplicator.abort() instead of calling the factory again.
+ assertThat(factoryCalls).hasValue(1);
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| @Test | |
| void seeOtherRedirectDropsBodyAndDoesNotThrow() { | |
| final AtomicInteger factoryCalls = new AtomicInteger(); | |
| final Supplier<HttpRequest> factory = () -> { | |
| factoryCalls.incrementAndGet(); | |
| return HttpRequest.of(RequestHeaders.of(HttpMethod.POST, "/see-other", | |
| HttpHeaderNames.CONTENT_TYPE, | |
| "text/plain; charset=utf-8"), | |
| StreamMessage.of(HttpData.ofUtf8("see-other-body"))); | |
| }; | |
| final WebClient client = | |
| WebClient.builder(server.httpUri()) | |
| .followRedirects() | |
| .build(); | |
| final RequestOptions options = | |
| RequestOptions.builder() | |
| .exchangeType(ExchangeType.REQUEST_STREAMING) | |
| .attr(ClientRequestBodyFactory.REQUEST_BODY_FACTORY, factory) | |
| .build(); | |
| final AggregatedHttpResponse res = | |
| client.execute(factory.get(), options).aggregate().join(); | |
| // On a 303 the method is rewritten to GET and the body is dropped; the redirected GET has | |
| // an empty body. The factory duplicator is aborted and never invoked again for the hop. | |
| assertThat(res.status()).isEqualTo(HttpStatus.OK); | |
| assertThat(res.contentUtf8()).isEqualTo("GET:"); | |
| } | |
| `@Test` | |
| void seeOtherRedirectDropsBodyAndDoesNotThrow() { | |
| final AtomicInteger factoryCalls = new AtomicInteger(); | |
| final Supplier<HttpRequest> factory = () -> { | |
| factoryCalls.incrementAndGet(); | |
| return HttpRequest.of(RequestHeaders.of(HttpMethod.POST, "/see-other", | |
| HttpHeaderNames.CONTENT_TYPE, | |
| "text/plain; charset=utf-8"), | |
| StreamMessage.of(HttpData.ofUtf8("see-other-body"))); | |
| }; | |
| final WebClient client = | |
| WebClient.builder(server.httpUri()) | |
| .followRedirects() | |
| .build(); | |
| final RequestOptions options = | |
| RequestOptions.builder() | |
| .exchangeType(ExchangeType.REQUEST_STREAMING) | |
| .attr(ClientRequestBodyFactory.REQUEST_BODY_FACTORY, factory) | |
| .build(); | |
| final AggregatedHttpResponse res = | |
| client.execute(factory.get(), options).aggregate().join(); | |
| // On a 303 the method is rewritten to GET and the body is dropped; the redirected GET has | |
| // an empty body. The factory duplicator is aborted and never invoked again for the hop. | |
| assertThat(res.status()).isEqualTo(HttpStatus.OK); | |
| assertThat(res.contentUtf8()).isEqualTo("GET:"); | |
| // The factory is invoked once for the initial request; the SEE_OTHER hop drops the body | |
| // via reqDuplicator.abort() instead of calling the factory again. | |
| assertThat(factoryCalls).hasValue(1); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@core/src/test/java/com/linecorp/armeria/client/RedirectingClientReproducibleBodyTest.java`
around lines 95 - 124, The 303 redirect test in
RedirectingClientReproducibleBodyTest verifies the response but never checks
that factoryCalls stays at the expected count. Update
seeOtherRedirectDropsBodyAndDoesNotThrow() to assert the factory is only used
for the initial request and not invoked again for the SEE_OTHER hop, using the
existing factoryCalls AtomicInteger so the redirect body-drop behavior is
actually enforced.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #6841 +/- ##
============================================
+ Coverage 74.46% 75.19% +0.73%
- Complexity 22234 25536 +3302
============================================
Files 1963 2269 +306
Lines 82437 94636 +12199
Branches 10764 12382 +1618
============================================
+ Hits 61385 71164 +9779
- Misses 15918 17600 +1682
- Partials 5134 5872 +738 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
…ss request type Replace the attribute-based `ClientRequestBodyFactory.REQUEST_BODY_FACTORY` (a `Supplier<HttpRequest>` read out of band by RetryingClient and RedirectingClient) with a first-class `HttpRequest.reproducible(RequestHeaders, Supplier<StreamMessage<HttpObject>>)`. The returned `ReproducibleHttpRequest` overrides `toDuplicator()` to return a non-buffering `ReproducibleHttpRequestDuplicator`, so both clients drop their special-case selection block and call `req.toDuplicator(...)` unconditionally. Every attempt — including the first — is regenerated from the factory, so the caller's request is never itself put on the wire. This fixes the prior design's retry-after-mid-body-failure regression: because the first attempt used the live request, a mid-body transport failure completed its `whenComplete()` exceptionally and RetryingClient short-circuited all retries. The body factory is invoked lazily (never eagerly at construction): once per attempt on the retry/redirect path, or once on direct subscription when no duplicating decorator is present. Fixed headers are reused for every attempt so the request method and content-length cannot drift. The duplicator tracks the single last-produced request (volatile, for the caller-thread to event-loop hand-off) so abort() can release a produced-but-unsubscribed body; close() lets an in-flight subscribed request finish per the StreamMessageDuplicator contract. duplicate() fails fast (propagates) when the factory throws or returns null, so an unreproducible body terminates the request instead of burning the retry budget. Reproducibility applies at the outermost duplicating decorator only; inner hops see ordinary requests. Copyright headers on new files use "LY Corporation". Co-authored-by: Isaac
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
core/src/test/java/com/linecorp/armeria/client/ReproducibleHttpRequestClientTest.java (1)
158-161: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winTighten the exception assertion in
factoryThrowingOnRetryFailsFast.
isInstanceOf(Exception.class)matches almost any throwable and could mask an unrelated failure. Consider asserting the wrapped cause to verify the factory'sIllegalStateExceptionpropagates correctly.♻️ Suggested refinement
- assertThatThrownBy(() -> client.execute(HttpRequest.reproducible(headers, bodyFactory), - streamingOptions()) - .aggregate().join()) - .isInstanceOf(Exception.class); + assertThatThrownBy(() -> client.execute(HttpRequest.reproducible(headers, bodyFactory), + streamingOptions()) + .aggregate().join()) + .isInstanceOf(CompletionException.class) + .hasCauseInstanceOf(IllegalStateException.class) + .hasRootCauseMessage("cannot regenerate body");🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/src/test/java/com/linecorp/armeria/client/ReproducibleHttpRequestClientTest.java` around lines 158 - 161, The assertion in factoryThrowingOnRetryFailsFast is too broad and can hide unrelated failures. Update the thrown-exception check around client.execute(...).aggregate().join() to verify the wrapped cause from the retry factory path, using the existing ReproducibleHttpRequestClientTest and factoryThrowingOnRetryFailsFast identifiers, so it specifically confirms the IllegalStateException is propagated instead of matching any Exception.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@core/src/main/java/com/linecorp/armeria/common/HttpRequest.java`:
- Line 374: The Javadoc example in HttpRequest uses an undefined variable name,
so update the snippet referenced by HttpRequest.reproducible to use a declared
value or a clearer placeholder instead of path. Make the example self-contained
and compilable by ensuring the argument passed to StreamMessage.of comes from a
properly defined local variable in the documentation sample.
In
`@core/src/test/java/com/linecorp/armeria/client/ReproducibleHttpRequestClientTest.java`:
- Around line 218-242: The stacked retry/redirect test in
ReproducibleHttpRequestClientTest leaves bodyCalls unused, so the body
regeneration count is not verified. Update stackedRetryAndRedirect to assert the
expected number of bodyFactory invocations via bodyCalls, or remove the
AtomicInteger tracking entirely if it is not needed. Use the existing
bodyFactory, bodyCalls, and client.execute(HttpRequest.reproducible(...)) flow
to locate the test and keep the redirect/retry coverage aligned with the
intended behavior.
---
Nitpick comments:
In
`@core/src/test/java/com/linecorp/armeria/client/ReproducibleHttpRequestClientTest.java`:
- Around line 158-161: The assertion in factoryThrowingOnRetryFailsFast is too
broad and can hide unrelated failures. Update the thrown-exception check around
client.execute(...).aggregate().join() to verify the wrapped cause from the
retry factory path, using the existing ReproducibleHttpRequestClientTest and
factoryThrowingOnRetryFailsFast identifiers, so it specifically confirms the
IllegalStateException is propagated instead of matching any Exception.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 03e8158e-56dd-4cea-99d7-11cd5a0014c4
📒 Files selected for processing (8)
core/src/main/java/com/linecorp/armeria/client/RedirectingClient.javacore/src/main/java/com/linecorp/armeria/client/retry/RetryingClient.javacore/src/main/java/com/linecorp/armeria/common/HttpRequest.javacore/src/main/java/com/linecorp/armeria/common/ReproducibleHttpRequest.javacore/src/main/java/com/linecorp/armeria/internal/common/ReproducibleHttpRequestDuplicator.javacore/src/test/java/com/linecorp/armeria/client/ReproducibleHttpRequestClientTest.javacore/src/test/java/com/linecorp/armeria/client/retry/ReproducibleHttpRequestRetryTest.javacore/src/test/java/com/linecorp/armeria/internal/common/ReproducibleHttpRequestDuplicatorTest.java
✅ Files skipped from review due to trivial changes (1)
- core/src/main/java/com/linecorp/armeria/common/ReproducibleHttpRequest.java
…ucibleHttpRequest The single-arg toDuplicator now delegates to the two-arg sink; both returned an identical duplicator. Merges the explanatory comments into one. No behavioral change. Co-authored-by: Isaac
- HttpRequest.reproducible javadoc: declare the `path` local used in the example so the snippet is self-contained and compilable. - ReproducibleHttpRequestClientTest.stackedRetryAndRedirect: assert bodyCalls so the body-regeneration count through the retry+redirect chain is actually verified. Co-authored-by: Isaac
|
Tick the box to add this pull request to the merge queue (same as
|
jrhee17
left a comment
jrhee17
left a comment
There was a problem hiding this comment.
Looks good overall, left some thoughts I had while looking through the PR
| // 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; |
There was a problem hiding this comment.
I think I prefer that the semantics of duplicate match that of the existing DefaultStreamMessageDuplicator in that duplicated streams are left to be active as long as the originating StreamMessage is not closed. (So a collection of child streams would be maintained, and closing the upstream will also close all child streams)
This would also help with supporting hedging more easily in the future.
Alternatively, I'm fine with keeping the current implementation and improving in the future as well as I do think complexity of the impl would go up slightly.
There was a problem hiding this comment.
duplicated streams are left to be active as long as the originating StreamMessage is not closed.
Yeah, I think it's the contract of a duplicated request.
There was a problem hiding this comment.
and closing the upstream will also close all child streams
If a factory creates a new child stream, it creates an independent stream, so each child may publish different elements. I don't think it will be used that way in most cases.
That said, given the limitation of the current approach - that the upstream and downstream data may differ - I'm not sure it makes sense to close the downstream when the upstream is closed.
There was a problem hiding this comment.
If a factory creates a new child stream, it creates an independent stream, so each child may publish different elements.
The contract guarantees that it's the same stream:
* 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
Unlike DefaultStreamMessageDuplicator, there is no single upstream. So children never share buffered signals and — as you note — could in principle publish different elements. We treat that as a caller contract (each factory invocation must produce an equivalent body matching the fixed content-length) rather than enforcing it, since the intended use is re-opening the same source, e.g. a file.
There was a problem hiding this comment.
Summarizing where this landed: there's no shared upstream by design — each duplicate() produces an independent, equivalent body from the factory, which is exactly what avoids the 2 GiB buffer. So "closing the upstream closes children" doesn't map; there's nothing shared to propagate. abort() tears down all outstanding children to release unsubscribed bodies (e.g. open files), and close() leaves in-flight children streaming per the StreamMessageDuplicator contract, matching DefaultHttpRequestDuplicator. Concurrent children are already supported, so hedging is accommodated without the shared-upstream model. Keeping the current semantics.
| * <p>{@code RetryingClient} and {@code RedirectingClient} drive this duplicator strictly | ||
| * sequentially: at most one produced request is outstanding at a time, and each access |
There was a problem hiding this comment.
I do think it is possible that abort on the original req can be called from any thread, which may result in a leak where the duplicated request isn't closed
There was a problem hiding this comment.
This is handled by the instance lock, and I've now added tests that pin it. The race — abort(cause) firing after the factory produced a body but before duplicate() registers it — is closed under the lock: if abort wins, duplicate observes closed, tears its just-produced body down with the remembered cause, and throws; if duplicate wins, the child is in children and abort's snapshot catches it. New coverage: duplicateAfterAbortTearsDownProducedBodyWithCause (sequential) and concurrentAbortDuringDuplicateTearsDownProducedBody (a barrier-gated, genuinely multi-threaded interleave). I mutation-verified the latter is non-vacuous: removing the synchronized guards makes it fail (leaked body, no throw).
| * return {@code null} | ||
| */ | ||
| @UnstableApi | ||
| static HttpRequest reproducible( |
There was a problem hiding this comment.
So I understood this is combining two concerns:
- Adding a static
HttpRequest of(RequestHeaders, Supplier<StreamMessage>)variant - Adding a
HttpRequest#withDuplicatorBodyFactory
e.g.
static HttpRequest of(RequestHeaders headers,
Supplier<? extends StreamMessage<? extends HttpObject>> bodyFactory) {
requireNonNull(headers, "headers");
requireNonNull(bodyFactory, "bodyFactory");
return of(headers, bodyFactory.get());
}
...
static HttpRequest reproducible(
RequestHeaders headers,
Supplier<? extends StreamMessage<? extends HttpObject>> bodyFactory) {
requireNonNull(headers, "headers");
requireNonNull(bodyFactory, "bodyFactory");
return of(headers, bodyFactory).withDuplicatorFactory(bodyFactory);
}For your use-case, do you see most users will prefer using reproducible method directly rather than separately? If so, I think it's fine to add a convenience method, but I just wanted to double-check.
I do think the single factory method has value in that users only need to specify a single Supplier.
There was a problem hiding this comment.
Thanks — I'd prefer to keep the single reproducible(headers, factory) method rather than the two-method decomposition. The single-Supplier form makes the key invariant — the body source and the replay source are the same factory — unbreakable by construction. The decomposed shape (of(headers, Supplier) plus a separate withDuplicatorFactory(Supplier)) invites a footgun: two supplier references that can drift, or a forgotten withDuplicatorFactory silently falling back to the buffering (2 GiB-capped) path. Since this is @UnstableApi, we can add the composable primitives later if a concrete need arises, without committing to the wider surface now.
Address maintainer review (jrhee17, minwoox) on PR line#6841: - Match DefaultStreamMessageDuplicator semantics: every request produced by duplicate() stays active until it completes on its own, for as long as the duplicator is not aborted. Replace single last-produced tracking with a Set<HttpRequest> of children. close() prevents further duplication but leaves outstanding requests streaming; abort()/abort(cause) tears down every outstanding request. This also supports multiple concurrent outstanding requests (e.g. future hedging), not just the most recent one. - Fix the abort-from-any-thread leak: abort() may run on the event loop while the caller thread is mid-duplicate(). Guard all state transitions with the instance lock so a concurrently produced request cannot be missed and leaked; a request produced after close/abort is aborted immediately and duplicate() then throws instead of returning an untearable request. Co-authored-by: Isaac
…ten assert Address review findings on reproducible request bodies: - HttpRequest.reproducible Javadoc had the retry/redirect nesting backwards. In a WebClient, RedirectingClient wraps the user decorators (incl. RetryingClient), so RedirectingClient is the outermost duplicating decorator and replays reproducibly across redirect hops; an inner RetryingClient re-buffers when retrying within a single hop. Correct the direction. - ReproducibleHttpRequest Javadoc claimed the factory is "never called eagerly / invoked exactly once on subscription". abort() before any real subscriber subscribes an aborting subscriber, which runs the factory once. Document that abort-before-subscribe counts as the first subscription (factory runs at most once on the direct path), and note the cold-path single-arg subscribe drops SubscriptionOptions / the caller executor. - Add retryReproducesMultiChunkBodyAndTrailers: a genuinely multi-chunk body terminated by a trailer, asserting order and trailer survive reproduction on the re-sent attempt (previous tests only used a single chunk, no trailers). - Tighten factoryThrowingOnRetryFailsFast to assert the surfaced root cause is the factory's IllegalStateException, not merely some Exception. Co-authored-by: Isaac
There was a problem hiding this comment.
🧹 Nitpick comments (1)
core/src/test/java/com/linecorp/armeria/internal/common/ReproducibleHttpRequestDuplicatorTest.java (1)
104-151: 🩺 Stability & Availability | 🔵 Trivial | 🏗️ Heavy liftConsider a dedicated test for the duplicate()/abort() race window.
These tests cover sequential abort-after-produce, close-leaves-active, and completed-untracked cases well, but none exercises the actual concurrent race the class Javadoc (lines 49-54 in
ReproducibleHttpRequestDuplicator.java) is designed to close:abort()firing from another thread whileduplicate()is mid-execution, before the produced request is added tochildren. This is precisely the scenariojrhee17flagged in a past review as a potential leak. A test using a latch to pause a thread insideduplicate()right before/after the synchronized block while a second thread callsabort()would give regression protection for this specific fix.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/src/test/java/com/linecorp/armeria/internal/common/ReproducibleHttpRequestDuplicatorTest.java` around lines 104 - 151, Add a dedicated concurrency test for the duplicate()/abort() race in ReproducibleHttpRequestDuplicator: use latches to pause duplicate() around its synchronized child-registration boundary, invoke abort() from another thread, then release duplication and assert the produced request completes exceptionally. Ensure the test specifically verifies a request created during the race cannot remain active or leak after abort().
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In
`@core/src/test/java/com/linecorp/armeria/internal/common/ReproducibleHttpRequestDuplicatorTest.java`:
- Around line 104-151: Add a dedicated concurrency test for the
duplicate()/abort() race in ReproducibleHttpRequestDuplicator: use latches to
pause duplicate() around its synchronized child-registration boundary, invoke
abort() from another thread, then release duplication and assert the produced
request completes exceptionally. Ensure the test specifically verifies a request
created during the race cannot remain active or leak after abort().
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 62c775f5-6224-47eb-88b2-6d22f1c8a14c
📒 Files selected for processing (5)
core/src/main/java/com/linecorp/armeria/common/HttpRequest.javacore/src/main/java/com/linecorp/armeria/common/ReproducibleHttpRequest.javacore/src/main/java/com/linecorp/armeria/internal/common/ReproducibleHttpRequestDuplicator.javacore/src/test/java/com/linecorp/armeria/client/ReproducibleHttpRequestClientTest.javacore/src/test/java/com/linecorp/armeria/internal/common/ReproducibleHttpRequestDuplicatorTest.java
🚧 Files skipped from review as they are similar to previous changes (3)
- core/src/main/java/com/linecorp/armeria/common/ReproducibleHttpRequest.java
- core/src/main/java/com/linecorp/armeria/common/HttpRequest.java
- core/src/test/java/com/linecorp/armeria/client/ReproducibleHttpRequestClientTest.java
The private static abort(HttpRequest, Throwable) helper was an overload of abort()/abort(Throwable) with abortAll() declared between them, tripping checkstyle's OverloadMethodsDeclarationOrder. Rename the helper to abortQuietly so it is no longer an overload; no behavior change. Co-authored-by: Isaac
jrhee17
left a comment
jrhee17
left a comment
There was a problem hiding this comment.
Mostly looks good, only nits and a question for me
| * {@code duplicate}, and {@code duplicate} then throws instead of returning a request that would never | ||
| * be torn down. | ||
| */ | ||
| public final class ReproducibleHttpRequestDuplicator implements HttpRequestDuplicator { |
There was a problem hiding this comment.
Question) Does this have to be in the internal package? Can this class be in the common pkg with ReproducibleHttpRequest?
There was a problem hiding this comment.
Good call — moved it to com.linecorp.armeria.common and made both the class and constructor package-private. It was only public to be reachable across the package boundary from ReproducibleHttpRequest#toDuplicator, and it has no external consumers, so this matches the sibling DefaultHttpRequestDuplicator exactly. Done in the latest push.
| final HttpRequest produced = HttpRequest.of(newHeaders, body); | ||
|
|
||
| final Throwable abortCause; | ||
| synchronized (this) { |
There was a problem hiding this comment.
note) synchronized seems to no longer have the thread pinning issue since Java 24
There was a problem hiding this comment.
Good note. The lock here never wraps a blocking call — the factory (bodyFactory.get(), the only potentially-blocking/file-opening call) runs outside it, and the critical sections only mutate an in-memory IdentityHashMap set. So pinning isn't a concern even pre-Java-24. No change needed.
| */ | ||
| private static StreamMessage<HttpObject> lazyBody( | ||
| Supplier<? extends StreamMessage<? extends HttpObject>> bodyFactory) { | ||
| return StreamMessage.of((Publisher<HttpObject>) subscriber -> { |
There was a problem hiding this comment.
Question) Is it necessary that the supplier isn't called eagerly? Asking because from what I understand, StreamMessage usually don't open resources until a downstream subscription is done.
If it is necessary for your use-case, I think it's fine to go ahead with this impl. I think it may be possible to generalize DeferredStreamMessage later on s.t. we can create a lazy supplier StreamMessage variant.
There was a problem hiding this comment.
Laziness is necessary here: the caller's factory may have subscribe-time side effects (opening a file, acquiring a handle), and on the retry/redirect path the delegate is never subscribed — toDuplicator(...) drives the factory directly. An eager bodyFactory.get() at construction would materialize and leak a body on every reproducible request. The lazy wrapper confines materialization to the direct-consume path. Agreed that generalizing DeferredStreamMessage into a lazy-supplier StreamMessage variant would be cleaner and remove the dropped-SubscriptionOptions caveat — I'd like to do that as a follow-up rather than expand this PR.
There was a problem hiding this comment.
Thanks
I'd like to do that as a follow-up rather than expand this PR.
Feel free to go ahead, or we can handle this as well if you are busy 🙇 Let us know
There was a problem hiding this comment.
That'd be great — please go ahead with the DeferredStreamMessage generalization on your side. I'll keep the hand-rolled lazy body in this PR (its one caveat, dropped SubscriptionOptions, only affects the cold direct-consume path), and it can be swapped over to the generalized variant once yours lands. Thanks for offering to take it!
…cs+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
|
I'm not sure you are ready for the next round of reviews - let me know if you are 🙇 |
|
thanks for the comments so far, I will re-request when ready! |
… package-private The duplicator was public only to be reachable across the internal.common -> common package boundary from ReproducibleHttpRequest#toDuplicator. It has no external consumers, so co-locating it with its sole caller in com.linecorp.armeria.common lets it (and its constructor) become package-private, matching the sibling DefaultHttpRequestDuplicator. No behavior change. Co-authored-by: Isaac
…d 303 Lock in the reproducibility contracts that were previously only asserted in prose: - factoryInvokedOnceOnDirectConsumption: direct consumption (no retry/redirect decorator) subscribes the lazy delegate and runs the factory exactly once. - abortBeforeSubscriptionInvokesFactoryOnce: aborting a never-sent request runs the factory exactly once to release the produced body, and never regenerates. - seeOtherRedirectDropsBody now asserts the factory is called exactly once: the 303 hop drops the body (aborting the duplicator) instead of regenerating it, addressing the missing-assertion review note. Co-authored-by: Isaac
f8e4295 to
1286461
Compare
…hten retry assert Address code-review findings (all test-only, no production change): - duplicateAfterAbortTearsDownProducedBodyWithCause: exercises the abortCause propagation branch of the duplicate()/abort(cause) race — the just-produced body is torn down with the remembered cause and duplicate() throws. Previously only the close() variant (abortCause == null) was covered. - factoryThrowingOnRedirectHopFailsFast: a factory that throws while regenerating the body for a redirect hop fails the request fast via RedirectingClient's try/catch, mirroring the existing RetryingClient coverage. - retriesAfterFirstAttemptBodyFailsMidStream: assert exactly 2 factory calls instead of >= 2, so an over-regeneration bug is caught rather than masked. Co-authored-by: Isaac
…dundant test Second-round review follow-up (test-only): - concurrentAbortDuringDuplicateTearsDownProducedBody: a barrier-gated test that actually runs duplicate() on one thread while abort(cause) runs on another, forcing the interleave the instance lock exists for. Verified non-vacuous by mutation: with the synchronized guards removed it fails (the produced body leaks and duplicate() does not throw). The prior sequential test only covered the non-interleaved branch. - Remove streamsBodyWithoutAccumulatingIt: a 64 KiB aggregate is delivered identically by a buffering duplicator, so it could not detect a regression to buffering. The companion toDuplicatorIgnoresMaxRequestLength (cap=8) is the real non-buffering proof, and body integrity is already covered by everyDuplicateProducesAFreshBody. Co-authored-by: Isaac
|
@jrhee17 this is ready for another round — I've replied to all the open threads and pushed follow-up changes:
Thanks for the thorough review! |
jrhee17
left a comment
jrhee17
left a comment
There was a problem hiding this comment.
Looks good to me overall 👍
| // 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); |
There was a problem hiding this comment.
Question) Is there a reason we aren't creating another reproducible request here?
| final HttpRequest produced = HttpRequest.of(newHeaders, body); | |
| final HttpRequest produced = HttpRequest.reproducible(newHeaders, bodyFactory); |
There was a problem hiding this comment.
I prototyped this — returning HttpRequest.defer(newHeaders, bodyFactory) from duplicate() so each child is itself lazy/deferred. It's appealing (it would delete the child-tracking + lock + abort-teardown, and all 16 end-to-end retry/redirect tests still passed), but it regresses the fail-fast guarantee, so I kept the eager HttpRequest.of(newHeaders, body).
The reason: eager duplicate() distinguishes "the factory is broken" (throw now, fail the request) from "the wire faulted" (retryable). With a lazy child the factory only runs at subscribe time, so a broken factory becomes indistinguishable from a transport fault. I measured it — a factory that always throws, under RetryRule.onException():
- eager (current): factory called once, fails fast;
- lazy child: factory called 7×, burning the entire retry budget on a factory that can never succeed.
So the child-tracking/lock machinery is load-bearing: it's what buys fail-fast. Happy to revisit if you'd like, but that's why I've left it eager.
There was a problem hiding this comment.
Flagging in case it crossed with your note: I replied above with the outcome of prototyping this. I did build the "each duplicate() returns a reproducible child" version — it deletes the child-tracking/lock/teardown and all 16 end-to-end tests pass — but it regresses fail-fast: a broken factory under RetryRule.onException() burns the whole retry budget (measured 7× vs 1×) instead of failing immediately, because a lazy child can't distinguish "factory broken" from "wire faulted". So I've kept the eager duplicate(). Happy to revisit if you'd still prefer the lazy version knowing that trade-off.
There was a problem hiding this comment.
So there's a trade-off. I understand it. So we can probably leave it as is and revisit this if we need to make the child reproducible also. @jrhee17 What do you think of it?
| static HttpRequest reproducible( | ||
| RequestHeaders headers, | ||
| Supplier<? extends StreamMessage<? extends HttpObject>> bodyFactory) { |
There was a problem hiding this comment.
Question) I'm unsure about the naming. What do you think of defer? It seems like the API looks similar to Flux#defer, and we'll likely be adding a StreamMessage#defer as well.
There was a problem hiding this comment.
@minwoox suggested that we keep the name reproducible since users may mistakenly think it may be safe to call subscribe HttpRequest#reproducible multiple times
There was a problem hiding this comment.
Good idea — renamed to defer. I've taken it all the way through: the public @UnstableApi method is now HttpRequest.defer(...), and the internal classes are DeferredHttpRequest / DeferredHttpRequestDuplicator (plus the test classes) so the vocabulary is consistent end to end. "Regenerate" stays as the verb for producing an equivalent body per attempt, since that describes the actual behavior where "defer" wouldn't fit. Pushed in 8a66f3e0e.
There was a problem hiding this comment.
Sorry but, could we revert to HttpRequest.reproducible(...)?
StreamMessage is single-subscription by contract, so the Flux.defer mental model (re-subscribe → regenerate) doesn't apply here — you can't re-subscribe. So defer names a property that isn't the point.
What actually defines this type is the opposite constraint: each call must supply an
equivalent StreamMessage (same bytes and trailers). reproducible names
that contract directly; defer only says "later" and says nothing about equivalence.
There was a problem hiding this comment.
Agreed, and reverted to HttpRequest.reproducible(...) (pushed in 0cd4d64e8). You and @minwoox are right — since StreamMessage is single-subscription, the Flux#defer mental model doesn't carry over, and reproducible names the equivalent-body contract directly rather than just "later". The internal ReproducibleHttpRequest* classes and tests are back too. I kept the content-length doc scoping from the reverted commit (the framing warning now applies only when the caller sets content-length explicitly).
| * {@code duplicate}, and {@code duplicate} then throws instead of returning a request that would never | ||
| * be torn down. | ||
| */ | ||
| final class ReproducibleHttpRequestDuplicator implements HttpRequestDuplicator { |
There was a problem hiding this comment.
Note) My understanding is that the purpose of this class is to sync the lifecycle of HttpRequestDuplicator with the lifecycle of the upstream+child streams.
e.g. HttpRequestDuplicator#abort or HttpRequestDuplicator#close cleans up upstream+child streams.
This usage pattern is documented in the javadocs for HttpRequestDuplicator, and RetryingClient uses HttpRequestDuplicator#abort to close all child/upstream streams at once.
Also, there is a small nuance that DefaultHttpRequestDuplicator assumes the upstream lifecycle is synced with the duplicator. (since the upstream subscribes to the duplicator internally)
For the reproducible case, this won't be true - calling toDuplicator won't subscribe to the upstream.
However, since we never call bodyFactory#get in the first place, there likely isn't anything to leak so there should be no issue.
There was a problem hiding this comment.
One clarification on the premise: in the current (eager) design duplicate() does call bodyFactory.get() eagerly, so a produced-but-unsubscribed child does hold a resource — which is exactly why the child-set + abort-teardown exists. Your conclusion (no leak) is right, but it holds because abort()/close() tear those children down, not because the factory is never called.
I looked at making children lazy (so nothing is produced until subscribed, matching your mental model) — see my reply on the duplicate() thread — but it regresses fail-fast: a broken factory under onException() burns the full retry budget (measured 7× vs 1×). So I've kept eager production + teardown. And you're right that the upstream-lifecycle assumption DefaultHttpRequestDuplicator makes doesn't apply here (toDuplicator never subscribes an upstream); the teardown covers the eagerly-produced children instead.
| */ | ||
| private static StreamMessage<HttpObject> lazyBody( | ||
| Supplier<? extends StreamMessage<? extends HttpObject>> bodyFactory) { | ||
| return StreamMessage.of((Publisher<HttpObject>) subscriber -> { |
There was a problem hiding this comment.
Thanks
I'd like to do that as a follow-up rather than expand this PR.
Feel free to go ahead, or we can handle this as well if you are busy 🙇 Let us know
| * 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 |
There was a problem hiding this comment.
Question) Did you actually encounter any content-length related issues? Asking bc I think we set content-length right before a req goes through the wire (unless a user explicitly set it)
There was a problem hiding this comment.
No, I didn't hit it in practice — it was a precautionary note, and you're right that it overstated the risk. A streaming request without an explicit content-length goes out chunked (self-delimiting), so a body-length difference between attempts is harmless to framing. I've scoped the caveat down to the one case where it actually applies: when the caller sets content-length explicitly in the fixed headers, that value is reused verbatim and not re-validated, so a length mismatch corrupts framing. Updated in 8a66f3e0e.
…th caveat Address jrhee17's review round: - Rename the public @UnstableApi factory HttpRequest.reproducible(...) to HttpRequest.defer(...), aligning with Reactor's Flux#defer and a future StreamMessage#defer. Rename the internal ReproducibleHttpRequest -> DeferredHttpRequest and ReproducibleHttpRequestDuplicator -> DeferredHttpRequestDuplicator (plus the three test classes) so the vocabulary is consistent end to end. "Regenerate" remains the verb for producing an equivalent body per attempt, where "defer" (laziness) would not fit. - Scope the content-length caveat: a streaming request without an explicit content-length is sent chunked (self-delimiting), so a body-length difference is harmless to framing. The framing-corruption warning now applies only to the case where the caller sets content-length explicitly. No behavioral change; the eager per-attempt duplicator design is unchanged. Co-authored-by: Isaac
|
Sorry @yzfeng2020 but it seems like there are some nuances where the behavior is slightly different from
Once these two are addressed, I think this PR looks done to me |
…ntent-length fix jrhee17 and minwoox asked to revert the `defer` rename: StreamMessage is single-subscription by contract, so the Flux#defer mental model (re-subscribe -> regenerate) does not apply, and `defer` says nothing about the equivalent-body contract that actually defines this type. `reproducible` names that contract directly. This fully reverts commit 8a66f3e's rename — HttpRequest.defer(...) is back to HttpRequest.reproducible(...), and the DeferredHttpRequest* classes/tests are back to ReproducibleHttpRequest*. The content-length caveat scoping from that commit is retained (the framing warning applies only when the caller sets content-length explicitly; otherwise the request is chunked and self-delimiting). Co-authored-by: Isaac
…ore asserting The BlockHound CI profile hit a latent race: after `aggregate().join()` returns, the request's own `whenComplete()` callback can fire slightly later on the event loop, so `assertThat(whenComplete()).isCompleted()` occasionally saw an incomplete future. Join `whenComplete()` first to make the assertion deterministic, matching the join-based pattern used elsewhere in the file. Co-authored-by: Isaac
4 participants
Motivation:
When a streaming request body larger than ~2 GiB is sent through
RetryingClientorRedirectingClient, the request fails withContentTooLargeExceptioneven though theapplication configured no content-length limit.
Both clients must be able to replay the request body across retry attempts / redirect hops, so
for streaming requests they wrap it in a duplicator:
The
0argument means "no limit", butDefaultStreamMessageDuplicatormaps0(and any value> Integer.MAX_VALUE) toInteger.MAX_VALUEand then buffers the entire body in memory,accumulating the total into an
int signalLengthfield. Once the accumulated body crossesInteger.MAX_VALUE(~2 GiB), the duplicator throwsContentTooLargeExceptionand aborts therequest.
There are two coupled problems:
int, so replay-buffering is capped at ~2 GiBeven when the caller explicitly asked for no limit (
0).replay, which is undesirable for large uploads.
Modifications:
HttpRequest.reproducible(RequestHeaders, Supplier<StreamMessage<? extends HttpObject>>),a first-class request type (
@UnstableApi). The caller supplies fixed headers and a factorythat opens a fresh body
StreamMessageon demand, declaring the body reproducible insteadof relying on the client to buffer it for replay.
ReproducibleHttpRequestDuplicator, a non-bufferingHttpRequestDuplicator. Everyduplicate()— including the first — obtains a fresh body from the factory, so no attemptreuses another attempt's stream and the caller's request is never itself put on the wire. This
avoids both the ~2 GiB
intcap and the memory cost ofDefaultStreamMessageDuplicator.ReproducibleHttpRequestoverridestoDuplicator(...)to return that non-buffering duplicator,so
RetryingClientandRedirectingClientkeep their single unconditionalreq.toDuplicator(...)call — no attribute read, no special-case selection block. Othercallers of
toDuplicator(circuit-breaker / rule utilities) benefit automatically.Design notes:
HttpRequestis supplied; only arequest built via
HttpRequest.reproducible(...)takes the non-buffering path.retry/redirect path, or exactly once on direct subscription when no duplicating decorator is
present.
request is never the live wire stream, a mid-body transport failure on the first attempt no
longer completes its
whenComplete()exceptionally and short-circuits all retries — the priorattribute-based design's retry regression.
content-lengthcannot drift between attempts. The body each factory invocation produces mustmatch the declared
content-length, exactly as for any streamingHttpRequest.duplicate()fails fast: a factory that throws or returnsnullpropagates to the client,which completes the response exceptionally instead of re-judging an aborted request and burning
the retry budget.
volatile, for the caller-thread →event-loop hand-off) so
abort()releases a produced-but-unsubscribed body (e.g. endpointselection threw before the wire subscribed);
close()lets an in-flight subscribed requestfinish, per the
StreamMessageDuplicatorcontract.downstream is an ordinary
HttpRequest; an inner hop (e.g. a redirect within a single retryattempt) treats it as a normal request and may buffer it.
ExchangeType#isRequestStreaming()); an aggregatedexchange type buffers the body as usual.
calls the factory again.
Result:
ContentTooLargeExceptionwhen usingRetryingClientorRedirectingClient#6835.with
HttpRequest.reproducible(...), without theContentTooLargeExceptioncap and withoutbuffering the whole body in memory. Callers who use a plain
HttpRequestsee no behavioralchange.