Repository navigation
Support early request body buffering before upstream peer selection - #816
CodyPubNub wants to merge 32 commits into
Conversation
Add opt-in request body buffering via request_body_buffer_limit() trait method. When implemented, the full request body is read and filtered before request_filter runs, making it available for auth signature verification and content-based routing decisions. Resolves cloudflare#780
…buffering # Conflicts: # pingora-proxy/src/lib.rs
|
Hey, thanks for your patience. I realize this ticket missed some of our triage steps. It's a big change, but it seems worthwhile. We will check it out to make sure the impact on our system wouldn't be too extreme. One thing that will make this easier to review is to make this fully configurable so that we can opt out of this change. |
Thanks for taking a look! The feature is fully opt-in, request_body_buffer_limit() returns None by default, and when it does, buffer_request_body_early() returns immediately with no side effects. No existing code paths are altered unless a user explicitly overrides that trait method to return Some(max_size). Happy to add additional gating if needed. |
PiotrSikora
left a comment
There was a problem hiding this comment.
Hi @CodyPubNub,
this is completely unsolicited drive-by review (I have no relationship with the project, so my feedback might be different than that from maintainers), but I was recently looking at Pingora and was surprised by the lack of request body buffering before establishing connection to the upstream, so I'm also interested in solving this, albeit for a more generic use case.
| match body_chunk { | ||
| Some(data) => { | ||
| let is_body_done = session.downstream_session.is_body_done(); | ||
|
|
||
| // Call request_body_filter for each chunk | ||
| let mut filter_data: Option<Bytes> = Some(data); | ||
| session | ||
| .downstream_modules_ctx | ||
| .request_body_filter(&mut filter_data, is_body_done) | ||
| .await?; | ||
| self.inner | ||
| .request_body_filter(session, &mut filter_data, is_body_done, ctx) | ||
| .await?; | ||
|
|
||
| // Accumulate the (possibly filtered) data | ||
| if let Some(filtered) = filter_data { | ||
| total_size += filtered.len(); | ||
|
|
||
| // Check size limit during accumulation (streaming protection) | ||
| if total_size > max_size { | ||
| return Error::e_explain( | ||
| HTTPStatus(413), | ||
| format!( | ||
| "Request body exceeded limit: {} > {} bytes", | ||
| total_size, max_size | ||
| ), | ||
| ); | ||
| } | ||
|
|
||
| body_parts.push(filtered); | ||
| } | ||
|
|
||
| if is_body_done { | ||
| break; | ||
| } | ||
| } | ||
| None => { | ||
| // End of body, call filter with end_of_stream=true | ||
| let mut filter_data: Option<Bytes> = None; | ||
| session | ||
| .downstream_modules_ctx | ||
| .request_body_filter(&mut filter_data, true) | ||
| .await?; | ||
| self.inner | ||
| .request_body_filter(session, &mut filter_data, true, ctx) | ||
| .await?; | ||
|
|
||
| // Collect any final data from the filter | ||
| if let Some(filtered) = filter_data { | ||
| total_size += filtered.len(); | ||
|
|
||
| // Final size check | ||
| if total_size > max_size { | ||
| return Error::e_explain( | ||
| HTTPStatus(413), | ||
| format!( | ||
| "Request body exceeded limit: {} > {} bytes", | ||
| total_size, max_size | ||
| ), | ||
| ); | ||
| } | ||
|
|
||
| body_parts.push(filtered); | ||
| } | ||
| break; | ||
| } |
There was a problem hiding this comment.
Both of those branches contain virtually the same code.
Could you de-duplicate this and use if let Some(data) = body_chunk { ... } where needed?
| session | ||
| .downstream_modules_ctx | ||
| .request_body_filter(&mut filter_data, is_body_done) | ||
| .await?; | ||
| self.inner | ||
| .request_body_filter(session, &mut filter_data, is_body_done, ctx) | ||
| .await?; |
There was a problem hiding this comment.
This results in wrong and error-prone ordering of callbacks, i.e. request body callbacks (HttpModule::request_body_filter and ProxyHttp::request_body_filter) are called before request headers callbacks (HttpModule::request_header_filter and ProxyHttp::request_filter).
The buffered request body is available in request_filter using get_buffered_body to perform any business logic based on the request body, so I'm not sure why you need to call those filters here. You should use this step only to pre-read and buffer the request body, and then push it through request_body_filter after request_filter is done.
Alternatively, you could add early_request_body_filter to avoid messing with the existing request flow.
| // Get Content-Length if present (for early size check) | ||
| let content_length = session | ||
| .downstream_session | ||
| .req_header() | ||
| .headers | ||
| .get(header::CONTENT_LENGTH) | ||
| .and_then(|v| v.to_str().ok()) | ||
| .and_then(|s| s.parse::<usize>().ok()); | ||
|
|
||
| // Fail fast: check Content-Length before reading | ||
| if let Some(cl) = content_length { | ||
| if cl > max_size { | ||
| return Error::e_explain( | ||
| HTTPStatus(413), | ||
| format!( | ||
| "Request body too large: Content-Length {} exceeds limit {} bytes", | ||
| cl, max_size | ||
| ), | ||
| ); | ||
| } | ||
| } | ||
|
|
||
| // Check if there's a body to read (Content-Length > 0 or Transfer-Encoding) | ||
| let has_body = content_length.is_some_and(|len| len > 0) | ||
| || session | ||
| .downstream_session | ||
| .req_header() | ||
| .headers | ||
| .get(header::TRANSFER_ENCODING) | ||
| .is_some(); | ||
|
|
||
| if !has_body { | ||
| // No body to buffer, mark as done | ||
| session.mark_body_buffered(); | ||
| return Ok(()); | ||
| } |
There was a problem hiding this comment.
This doesn't work with HTTP/2 requests that don't contain the Content-Length header.
| /// Determine whether to buffer the entire request body before connecting to upstream. | ||
| /// | ||
| /// This is called after [`Self::early_request_filter()`] but before [`Self::request_filter()`] | ||
| /// and [`Self::upstream_peer()`]. The body is buffered in `Session::buffered_request_body` | ||
| /// and can be accessed via [`Session::get_buffered_body()`]. | ||
| /// | ||
| /// # Returns | ||
| /// - `None`: Don't buffer, stream body to upstream (default) | ||
| /// - `Some(max_size)`: Buffer body with size limit, return 413 error if exceeded | ||
| /// | ||
| /// # Use Cases | ||
| /// - Auth signature verification (need full body before auth decision) | ||
| /// - Content-based routing decisions | ||
| /// - Body transformation before upstream selection | ||
| /// | ||
| /// # Size Limit Enforcement | ||
| /// When returning `Some(max_size)`: | ||
| /// - Content-Length header is checked first (fail fast before reading) | ||
| /// - Body size is checked during accumulation (streaming protection) | ||
| /// - If exceeded, returns HTTP 413 (Payload Too Large) |
There was a problem hiding this comment.
This approach works well for a few specific use cases, but generic proxies should allow buffering up to the buffer limit, and then resume reading and forward remaining data once the upstream is connected, without rejecting the requests with request body larger than the buffer limit.
There was a problem hiding this comment.
This approach works well for a few specific use cases, but generic proxies should allow buffering up to the buffer limit, and then resume reading and forward remaining data once the upstream is connected, without rejecting the requests with request body larger than the buffer limit.
I agree buffer-then-stream is useful for generic proxies. Inspecting the head of a large upload without rejecting it has clear value.
The use case driving this PR is authorization: the auth decision depends on a signature computed over the complete body, so a partial buffer isn't sufficient. The full body has to be available before upstream_peer. Without a hard size limit that becomes an unbounded memory commitment per request, which is why request_body_buffer_limit returns a max size and rejects with 413 if exceeded.
I think buffer-then-stream is a different feature with different semantics (partial visibility, no 413, resume streaming after peer selection). I've scoped this PR to the full-buffer case, but buffer-then-stream would be a solid follow-up.
There was a problem hiding this comment.
FWIW, my use case is similar to @CodyPubNub's. We need the full request body in order to make decisions about whether to allow the request to proceed.
(also unaffiliated with the project, just 👀 this PR because I want this feature)
There was a problem hiding this comment.
I think buffer-then-stream is a different feature with different semantics (partial visibility, no 413, resume streaming after peer selection). I've scoped this PR to the full-buffer case, but buffer-then-stream would be a solid follow-up.
Right, the feature is different on a high-level, but both require buffering request body before it can be forwarded upstream, and once you have the generic capability, it's easy to support your use case (i.e. block forwarding until max_size bytes or complete request body).
- Fix HTTP/2 body detection: replace has_body heuristic (Content-Length or Transfer-Encoding) with explicit Content-Length == 0 skip. H2 POST without Content-Length was incorrectly treated as bodyless. - Collapse two near-identical match arms (Some/None) into a unified flow using end_of_body flag, removing ~40 lines of duplication. Addresses review comments 1 and 3 from @PiotrSikora.
- New trait method runs per-chunk during buffer_request_body_early(), before request_header_filter — avoids calling request_body_filter out of phase order. - Remove is_body_buffered() skip guards from proxy_h1/h2 — normal request_body_filter runs unguarded during upstream forwarding. - Update body_routing example to demonstrate the streaming callback. - Add early_request_body_filter to phase docs and mermaid charts. Addresses review comment 2 from @PiotrSikora.
Thanks for the thorough review, @PiotrSikora. I really appreciate you taking the time. I've addressed comments 1-3:
|
PiotrSikora
left a comment
There was a problem hiding this comment.
Thanks! This looks much better now.
| let mut downstream_state = if body_was_buffered { | ||
| DownstreamStateMachine::PreBuffered | ||
| } else { | ||
| DownstreamStateMachine::new(session.as_mut().is_body_done()) | ||
| }; | ||
|
|
||
| // Use pre-buffered body if available, otherwise check for retry buffer | ||
| let buffer = if body_was_buffered { | ||
| pre_buffered_body | ||
| } else { | ||
| session.as_mut().get_retry_buffer() | ||
| }; |
There was a problem hiding this comment.
Nit: You could return (downstream_state, buffer) tuple here (same in H1 proxy).
| /// - Content-Length header is checked first (fail fast before reading) | ||
| /// - Body size is checked during accumulation (streaming protection) | ||
| /// - If exceeded, returns HTTP 413 (Payload Too Large) | ||
| fn request_body_buffer_limit(&self, _session: &Session, _ctx: &Self::CTX) -> Option<usize> { |
There was a problem hiding this comment.
Nit: early_request_body_buffer_limit
| /// Determine whether to buffer the entire request body before connecting to upstream. | ||
| /// | ||
| /// This is called after [`Self::early_request_filter()`] but before [`Self::request_filter()`] | ||
| /// and [`Self::upstream_peer()`]. The body is buffered in `Session::buffered_request_body` | ||
| /// and can be accessed via [`Session::get_buffered_body()`]. | ||
| /// | ||
| /// # Returns | ||
| /// - `None`: Don't buffer, stream body to upstream (default) | ||
| /// - `Some(max_size)`: Buffer body with size limit, return 413 error if exceeded | ||
| /// | ||
| /// # Use Cases | ||
| /// - Auth signature verification (need full body before auth decision) | ||
| /// - Content-based routing decisions | ||
| /// - Body transformation before upstream selection | ||
| /// | ||
| /// # Size Limit Enforcement | ||
| /// When returning `Some(max_size)`: | ||
| /// - Content-Length header is checked first (fail fast before reading) | ||
| /// - Body size is checked during accumulation (streaming protection) | ||
| /// - If exceeded, returns HTTP 413 (Payload Too Large) |
There was a problem hiding this comment.
I think buffer-then-stream is a different feature with different semantics (partial visibility, no 413, resume streaming after peer selection). I've scoped this PR to the full-buffer case, but buffer-then-stream would be a solid follow-up.
Right, the feature is different on a high-level, but both require buffering request body before it can be forwarded upstream, and once you have the generic capability, it's easy to support your use case (i.e. block forwarding until max_size bytes or complete request body).
- Rename request_body_buffer_limit → early_request_body_buffer_limit for consistency with early_request_body_filter (comment 6) - Collapse downstream_state + buffer into tuple return in proxy_h1 and proxy_h2 (comment 5) - Align inline comments with existing Cloudflare style - Update body_routing example and phase docs Addresses review comments 5 and 6 from @PiotrSikora.
|
@CodyPubNub This is a great feature I've been eagerly waiting for!!!I'm building a Rust-based Kubernetes ingress/gateway on top of Pingora, and the lack of early request body access has been a major pain point for us. |
|
Hi @johnhurt, I was wondering if there was any updates or feedback from your team to share about this change. Thank you! |
|
Hey, yeah. Sorry for the delay. We have been discussing this internally, so we would like to pull this in. I should have come back and explained what I meant by "configurable". These kinds of features even if they are off by default still incur a cost in runtime (even if it's a minimal branch) and risk. That's why we ask contributors to make changes that touch the main proxy trait configurable by cargo feature to avoid the. Checkout the connection filter feature for an example. |
I appreciate the feedback. I've added an |
|
Hi @johnhurt I was wondering if there were any updates to share on the team's thoughts toward this PR. It would be very helpful for this, or something very much like this to be available in the main branch. Thank you 🙏 |
|
Hello @johnhurt, I'm also wondering if this could be reviewed and merged soon? |
Resolve conflict in pingora-proxy/src/lib.rs: - downstream_custom_message() now returns the DownstreamCustomMessageReader type alias introduced by main's custom-message retry fix (7c04f54), rather than the inlined boxed-Stream signature the branch forked from. - Add upstream_h1_upgrade_status_mismatch to the test-only new_h1_with_http_session() constructor. Main added this Session field, and because the constructor is gated behind early_body_buffer it compiled fine on default features while breaking the feature build — a silent conflict git could not flag.
- Early buffering drains downstream before retry buffering starts, so taking the session buffer leaves retries with no body to replay. - Clone the bounded buffer per attempt, share lazy selection across H1/H2, and remove the destructive take_buffered_body API. - Add self-contained regression coverage for fixed-length and chunked bodies.
|
Hi @drcaramelsyrup I was wondering if you or others on the team have had a chance to think about this PR. Thank you 🙏 |
- avoid sending a second END_STREAM after empty requests complete on HEADERS - preserve explicit END_STREAM forwarding when filters remove a non-empty body - add regression coverage for Content-Length: 0 requests
|
Hello, I noticed that session.set_buffered_body() has no effect when called inside early_request_body_filter. Do you have any idea what might be causing this? |
Hi @rxdiscovery, To discard a chunk, set Either way, update the request framing to match the resulting body. For example, leaving This is expected lifecycle behavior, though the API documentation could make the restriction clearer. |
Reported as a bug: set_buffered_body() appeared to have no effect from early_request_body_filter(), because the buffering loop assembles the filtered chunks after the callback returns and overwrites the assignment. - explain why buffered-body replacement belongs in request_filter - document chunk mutation and request framing responsibilities - note the current limitation when adding bodies to empty H2 requests
Replacing an empty H2 request body failed because stream completion was derived from the consumed downstream body instead of the replacement. - use buffered body state for H2 stream completion - cover body replacement across protocol boundaries - remove the obsolete documented limitation
|
Thank you for your responsiveness @CodyPubNub ; (@CodyPubNub + @drcaramelsyrup) In my view, reading the request body before request_filter feels somewhat counterintuitive in the overall request lifecycle. Furthermore, early_request_body_buffer_limit effectively behaves like a request_filter, since the buffer size is determined solely by inspecting the headers. Wouldn't it be cleaner to introduce an enable_early_request_body_filter() method that explicitly instructs Pingora whether or not to defer sending the headers upstream? This would yield two clear, distinct scenarios: Standard streaming / non-buffered:request_filter -> [enable_early_request_body_filter() = false] -> [send headers] -> request_body_filter() -> [send body] Buffered / deferred headers:request_filter -> [enable_early_request_body_filter() = true] -> request_body_filter() -> [send headers + body] |
|
@rxdiscovery I may be misunderstanding the proposal, but a few premises here don't match the implementation.
This hook also does not decide whether to defer upstream headers. At this phase no upstream peer has been selected and no upstream connection or header transmission exists yet. Running Finally, Could you clarify the concrete operation you need to perform, and which information must be available before and after it? I'm not yet seeing what behavior the proposed ordering enables that the current lifecycle does not. |
|
Sorry if I wasn't very clear earlier, let me explain the idea : Instead of adding several additional phases, which could impact overall performance and increase the complexity of the Pingora workflow. Why not just expose a method that can be called, for example, from I hope this makes things clearer. On another note, I have a question: in your PR, does |
|
Thanks @rxdiscovery, I understand the proposal. It is an alternative API design rather than a correctness issue with this implementation. This PR implements the behavior requested in #780 and refined through maintainer review. The limit and timeout methods are configuration hooks, not additional phases, and the per-chunk callback is intentional. I don't plan to redesign the API around an imperative Session setter unless the maintainers request it. If you would like that alternative considered, please open a separate issue. Regarding memory, the early-buffered body becomes the canonical replay source. Pingora's ordinary retry buffer does not retain another payload copy, and retries use reference-counted Bytes clones sharing the same allocation. To keep this PR focused, I'm going to leave further exploration of alternative API designs to a separate discussion. |
| { | ||
| #[cfg(feature = "early_body_buffer")] | ||
| let body_replayable = | ||
| session.req_header().method.is_idempotent() || session.is_body_buffered(); |
There was a problem hiding this comment.
The session.is_body_buffered() alternative lets the default policy retry POSTs that have already reached the origin. In a local test, the origin read the full POST and closed before replying on a reused connection, then received the same POST again.
Please keep the default restriction to idempotent methods. Applications can opt in through error_while_proxy() when they know the operation can be repeated. The replay tests can use PUT or an explicit policy override.
| }; | ||
|
|
||
| // retry, send buffer if it exists or body empty | ||
| if buffer.is_some() || session.as_mut().is_body_empty() { |
There was a problem hiding this comment.
When an early filter removes a non-empty body and clears its framing headers, buffer is None but is_body_empty() still reflects the original downstream body. We skip the body task here, even though the upstream request was given Transfer-Encoding: chunked, so the origin never receives the terminating chunk.
A strict H1 origin times out in this case. Please send the final body task for a buffered request even when its buffer is None, and add an H1 counterpart to the filtered-empty H2 test.
|
|
||
| // accumulate the (possibly filtered) data | ||
| if let Some(filtered) = filter_data { | ||
| total_size += filtered.len(); |
There was a problem hiding this comment.
total_size only counts bytes left after early_request_body_filter(). With a limit of 8 bytes and a filter that discards the body, the same 17-byte POST returns 413 with Content-Length but 200 with chunked encoding. A shrinking filter can therefore keep this loop reading past the advertised limit.
Please enforce the limit on input bytes before running the filter as well as checking the retained bytes.
| // Content-Length: 0 means no body; for all other cases (no Content-Length, | ||
| // Transfer-Encoding, HTTP/2) attempt to read. read_request_body returns | ||
| // None immediately if there's nothing. | ||
| if content_length == Some(0) { |
There was a problem hiding this comment.
This returns before initializing the H1 body reader. If request_filter() then replaces the empty body and updates Content-Length, the reader initializes later from the replacement length and thinks the client still owes those bytes. The response gets Connection: close, even though the original request body is complete.
I reproduced this with an empty request followed by body replacement. Content-Length: 0 loses keepalive, while the same request without that header keeps it. Initialize the downstream reader from the original framing before marking the body buffered, and check connection reuse in the replacement test.
| // common case: a single chunk can be moved out without a second copy | ||
| session.set_buffered_body(body_parts.pop()); | ||
| } else { | ||
| let mut combined = bytes::BytesMut::with_capacity(total_size); |
There was a problem hiding this comment.
This allocates total_size while all the original chunks are still live. For a body arriving in several chunks, that adds another allocation as large as the whole body, on top of the retained chunks and the Vec<Bytes>.
Please accumulate into a BytesMut once the second chunk arrives, using the checked Content-Length when available, while keeping the single-chunk fast path.
| /// - Body size is checked during accumulation (streaming protection) | ||
| /// - If exceeded, returns HTTP 413 (Payload Too Large) | ||
| /// | ||
| /// Use [`Self::early_request_body_buffer_timeout()`] to apply a total deadline to this phase. |
There was a problem hiding this comment.
Please mention the retry implications here too. Automatically buffered bodies bypass the usual 64 KiB retry buffer and replay the complete retained body on each eligible retry. For example, Some(64 << 20) allows replaying a 64 MiB body on every attempt. retry_buffer_truncated() doesn't reflect the size of this separate buffer.
| //! retry replay, protocol completion, local `100 Continue` handling, and total buffering | ||
| //! deadlines. | ||
|
|
||
| #![cfg(feature = "early_body_buffer")] |
There was a problem hiding this comment.
early_body_buffer isn't enabled by any build or test command in .github/workflows/build.yml. Since it isn't a default feature either, the early buffering code and this entire test file are compiled out of those build and test runs.
Please add a CI run with --features early_body_buffer and a supported TLS backend, while keeping the existing run with the feature disabled.
|
Hi @CodyPubNub , could you please take into account the optimizations mentioned by @zaidoon1 (...whom I’d like to thank, by the way, for his very detailed feedback.) ? This will prevent high memory usage and improve performance. |
- keep non-idempotent retries an explicit application policy to avoid duplicate POSTs by default - enforce limits before and after filtering while reducing multi-chunk peak memory - complete empty H1 bodies and preserve keepalive when replacing originally empty requests - add deterministic regression coverage and a feature-enabled CI lane
|
@zaidoon1 Thank you for the thorough review. I've addressed all the issues you flagged and added regression coverage for the retained-size limit and default non-idempotent retry behavior. I'd appreciate another look when you have a chance. |
|
Hello @zaidoon1 and @drcaramelsyrup, could please review again so we can potentially get this merged soon? 🙏 |
zaidoon1
left a comment
There was a problem hiding this comment.
Thanks, my earlier comments all look addressed. Found a few more things on another pass.
| /// | ||
| /// The fallback must remain lazy because `is_body_done()` can initialize the H1 body reader. | ||
| #[cfg(feature = "early_body_buffer")] | ||
| pub(crate) fn select_upstream_body_source( |
There was a problem hiding this comment.
proxy_custom.rs doesn't use this, so a buffered body never makes it to a custom upstream. The header goes out as non-empty, is_body_done() is already true and the retry buffer is empty, so the upstream gets no body and no end of stream. I know custom support is a follow-up, but until then can we return an error in the custom path when is_body_buffered() is set? Moving this onto Session would also let h1, h2 and custom share it instead of copying the same block into each.
| /// | ||
| /// Requires the `early_body_buffer` feature. | ||
| #[cfg(feature = "early_body_buffer")] | ||
| fn early_request_body_buffer_timeout( |
There was a problem hiding this comment.
With this defaulting to None, nothing bounds an H2 upload during buffering. H2 body reads have no per-read timeout (read_body_bytes in v2/server.rs still has a TODO), so a client can send HEADERS without END_STREAM and park the request forever. H1 isn't much better since the 60s read timeout resets on every read. In streaming mode the upstream's timeouts eventually kick in, but there's no upstream yet here. Can we default to a finite deadline when buffering is on, or at least call this out in the docs?
| let capacity = content_length.unwrap_or(retained_size).max(retained_size); | ||
| let mut body = bytes::BytesMut::with_capacity(capacity); |
There was a problem hiding this comment.
This sizes the allocation from the client's Content-Length, not from what we've actually received. A client can claim max_size, send two 1-byte chunks and get a max_size allocation. And if someone uses Some(usize::MAX) to mean no limit, Content-Length: 9223372036854775807 makes the allocation fail, which aborts the process. My earlier suggestion to pre-size from Content-Length didn't account for this, so it needs a cap, something like cl.min(1 << 20), and let BytesMut grow past that.
| return Ok(()); | ||
| } | ||
|
|
||
| if is_expect_continue_req(session.downstream_session.req_header()) { |
There was a problem hiding this comment.
is_expect_continue_req only matches HTTP/1.1, but the Expect header gets stripped for every buffered request in both proxy_h1 and proxy_h2. So an H2 client sending expect: 100-continue never gets a 100 from us or the upstream, and it stalls until its own expect timeout (or forever if it doesn't have one). write_continue_response already handles H2, so I think H2 just needs to be checked here too.
| { | ||
| Ok(result) => result, | ||
| Err(_) => Error::e_explain( | ||
| ReadTimedout, |
There was a problem hiding this comment.
With the default fail_to_proxy this turns into a 400, since a downstream ReadTimedout falls through to the 400 case. The 408 in total_buffering_timeout_bounds_slow_request_body only comes from the test's own fail_to_proxy override. Should this be HTTPStatus(408)? Either way I'd have the test use the default so it shows what users actually get.
| /// handle `Expect: 100-continue` before reading. It must also update `Content-Length` and | ||
| /// `Transfer-Encoding` to describe the supplied body. | ||
| #[cfg(feature = "early_body_buffer")] | ||
| pub fn set_buffered_body(&mut self, body: Option<Bytes>) { |
There was a problem hiding this comment.
Nothing checks that the caller actually read the downstream body before calling this. If they didn't and then update Content-Length like the docs say, the H1 body reader initializes from the new length (it reads the live header), so the original body gets framed with the wrong length. Can we require is_body_done() here and return an error otherwise? That also pins the reader to the original framing.
I'd also make mark_body_buffered() crate-private. It overlaps with set_buffered_body(None) but keeps an existing body instead of clearing it, which is confusing as public API.
| /// before forwarding. Pingora emits HTTP/2 stream termination itself based on the downstream | ||
| /// request rather than the chunks retained by this filter. |
There was a problem hiding this comment.
This isn't true anymore after 2247998. H2 completion now comes from the buffered body (get_buffered_body().is_none_or(Bytes::is_empty) in proxy_h2), not the downstream request. The same sentence is in docs/user_guide/phase.md too.
| let mut e = e.more_context(format!("Peer: {}", peer)); | ||
| if !session.req_header().method.is_idempotent() || session.as_ref().retry_buffer_truncated() | ||
| { | ||
| let body_replayable = session.req_header().method.is_idempotent(); |
There was a problem hiding this comment.
nit: this is method idempotency, not whether the body can be replayed, so the name is misleading. Since behavior didn't change I'd just revert this and the doc reflow above.
| } | ||
| } | ||
|
|
||
| struct NonIdempotentRetryProxy; |
There was a problem hiding this comment.
This only exercises RetryType::decide_reuse through a test double, so it isn't really testing the proxy, and it's unrelated to this feature. I'd drop it along with NonIdempotentRetryProxy.
| run: | | ||
| cargo check --package pingora-proxy --all-targets --features openssl,early_body_buffer | ||
| cargo test --package pingora-proxy --features openssl,early_body_buffer \ | ||
| --lib --test test_early_body_buffer_retry --no-fail-fast -- --test-threads=1 |
There was a problem hiding this comment.
These pass in parallel for me, so --test-threads=1 doesn't look necessary. If it's there because of flakes, it's probably spawn_proxy_service binding port 0, dropping the listener and rebinding later, which can race. Can we also add a clippy run with early_body_buffer on? Right now none of this code gets linted in CI.
- fail custom upstreams on early-buffered bodies instead of sending a header with no body or end of stream - share upstream body selection through Session so h1, h2, and custom stop duplicating cfg blocks - send H2 100 Continue via h2 send_informational; the existing H2 path silently dropped 1xx headers - cap the Content-Length pre-size hint so a client cannot force a max_size (or aborting) allocation - require set_buffered_body callers to have consumed the downstream body, pinning H1 framing; make mark_body_buffered crate-private - return 408 from the total buffering deadline so the default fail_to_proxy does not map it to 400 - document that no upstream timeout backstops buffering; keep the opt-in deadline default - fix stale H2 termination docs, revert the error_while_proxy rename, drop the unrelated retry test double - add deterministic regressions for the allocation cap, custom rejection, and H2 Expect handling - lint the early_body_buffer lane in CI and run its tests in parallel
|
@zaidoon1 Thanks again for another careful pass. I've pushed the fixes in 51d618b. A few things that aren't obvious from the diff:
Please let me know if I missed anything. Thank you! |
# Conflicts: # docs/user_guide/phase.md # pingora-proxy/src/proxy_custom.rs # pingora-proxy/src/proxy_h1.rs # pingora-proxy/src/proxy_h2.rs
- buffer after allow_spawning_subrequest and should_handle_downstream_abandonment so main's documented contract holds: the opt-in is decided right after early_request_filter succeeds - document the hook order on early_request_body_buffer_limit, and that buffering hooks must not read the downstream body themselves, including through a spawned subrequest - add a regression that observes handles_downstream_abandonment() from the buffering hook
- SaveBody(limit) assumed the downstream body was still unread; after early buffering it saw a finished body, cleared the subrequest's body headers, and sent nothing - keep the SaveBody path and send a retained non-empty body through send_body_to_pipe once the subrequest asks for input, so downstream modules filter it and the saved body captures their output with SaveBody's limit and truncation semantics - an empty buffered body leaves the downstream body done, as for a request without a body, so a subrequest that asks for input later is still read and watched normally - a subrequest that never reads its body gets nothing and leaves the saved body empty, as with a streamed body - add regressions for forwarding, the save limit, an empty body, module filtering with and without the abandonment opt-in, and a subrequest that never reads
- replace the custom-upstream rejection with the same forwarding contract as h1 and h2 - strip Expect after upstream_request_filter, since 100 Continue was answered locally - derive the header's end flag from the buffered body, not the original framing, so a filtered or replaced empty body ends the request on its header - skip an explicitly empty buffer, so the body writer is not finished after the header already ended the request - retries replay the retained body through upstream_body_source as on h1 and h2 - add custom-connector regressions for forwarding, retry replay, Expect, and both empty-body paths, and run them in the early_body_buffer CI lane
- the guard checked is_body_done(), which an H1 or subrequest read error also sets before the declared framing ends - an application that swallowed that error could store a partial body as fully consumed and forward it upstream - check is_body_complete() instead; automatic buffering already propagates the read error - add a regression for a Content-Length body cut short by EOF
…tlives - an attempt's initial body send raced the upstream write against downstream stream closure; a large replay blocked on upstream flow control lost to an already-reset client - for a complete body whose partial-write cache fill could still finish, including retries after recorded abandonment, the initial send then lost H2BodySendError.downstream_closed and failed the attempt instead of applying the loop's cache-fill policy - skip the race only for the initial send of a complete body while the cache supports streaming partial writes, matching the loop's wait_for_cache_fill at attempt start; the loop's next downstream read then applies upstream's policy without cutting a write short - initial-send error handling is otherwise unchanged: an incomplete retry buffer or an uncached replay still stops as soon as the client resets - decide after both request body filters, which may disable caching; streamed writes in the loop, including the final chunk, always race - this is upstream's replay path in every build: ordinary retry buffers take it with or without early_body_buffer or the abandonment opt-in - regressions cover retry buffers in both feature modes, early-buffered bodies, an incomplete retry buffer, and the loop's final streamed chunk; fail-fast cases use a stalled upstream and assert the closure watch's own error
|
Merged
CI is red for an unrelated reason: tokio 1.54 panics on
I didn't run the OpenResty-based suites. |
Resolves #780
Adds opt-in early body buffering to
ProxyHttpbehind theearly_body_bufferCargo feature. Whenearly_request_body_buffer_limit()returnsSome(max_size), the full request body is read beforerequest_filterruns. The buffered body is available viaSession::get_buffered_body()for inspection andSession::set_buffered_body()for mutation, and is automatically forwarded to HTTP/1.x, HTTP/2, and custom upstreams during the proxy phase, and throughpipe_subrequest(InputBodyType::SaveBody).New trait methods:
early_request_body_buffer_limit()— opt in to buffering with a size limit (defaultNone)early_request_body_buffer_timeout()— optionally bound the complete buffering phase, including body-filter callbacks (defaultNone)early_request_body_filter()— per-chunk callback during early buffering, before any header-phase filters run. Use for streaming processing (e.g., decompression) that doesn't depend onrequest_filterstate. The normalrequest_body_filter()still runs during upstream forwarding.Use cases:
early_request_body_filterHook order: buffering runs after
early_request_filter(),allow_spawning_subrequest(), andshould_handle_downstream_abandonment(), and beforerequest_filter()andupstream_peer(), so the abandonment opt-in is still decided right afterearly_request_filter()as documented onmain. Buffering hooks must not read the downstream body themselves, including through a spawned subrequest.Size limits and timeouts are enforced independently: Content-Length is checked before reading (fail fast), accumulated size is checked during streaming, and applications may configure a total buffering deadline. Exceeding the size limit returns HTTP 413; an expired deadline returns HTTP 408. The buffering hooks and the
Sessionbuffer APIs are gated behindearly_body_buffer; when enabled, defaultNonevalues preserve existing streaming behavior. Two shared H2 changes also apply with the feature disabled:write_continue_response()now sends the H2100 Continue(previously dropped, since the H2 server ignores 1xx response headers) through a newv2::server::HttpSession::write_continue_response(), and the H2 replay adjustment described below.Expect: 100-continue: automatically buffered HTTP/1.1 and HTTP/2 requests are acknowledged locally before the body is read. The original header remains visible to application filters, then is removed afterupstream_request_filter()so it is not forwarded to HTTP/1.x, HTTP/2, or custom upstreams.HTTP/2 body handling: requests without
Content-Length(valid in HTTP/2) are handled correctly — onlyContent-Length: 0skips the body read. A confirmed empty buffered body still sends END_STREAM upstream.Retries and application-managed buffering: buffered bodies are retained and replayed across upstream retry attempts. Calling
Session::set_buffered_body()explicitly marks the body as buffered, includingNonefor a confirmed empty body; it fails unless the downstream body completed its declared framing. Applications using this path are responsible for fully consuming the downstream body, handlingExpect: 100-continue, and updating framing headers after body mutation. Regression tests cover fixed-length and chunked retries, application-managed buffering, oversize rejection, single-attempt forwarding, local100 Continue, total buffering timeout, empty HTTP/2 bodies, custom-upstream forwarding and replay, subrequest pipes, and truncated application-supplied bodies.Custom upstreams: the buffered body is forwarded and replayed as on HTTP/1.x and HTTP/2. The request header's end flag comes from the buffered body, so a filtered or replaced empty body ends the request on its header.
Subrequest pipes:
SaveBody(limit)on a buffered session forwards the buffered body once the subrequest asks for input. Downstream module body filters run on it and the saved body captures their output, with the usual limit and truncation. A subrequest that never reads gets nothing.H2 replay (all builds): an attempt's initial send of a complete request body no longer races the upstream write against downstream stream closure while the cache supports streaming partial writes. Previously a replay blocked on upstream flow control lost
downstream_closedand failed the attempt as soon as an already-reset client was noticed, instead of applying the loop's cache-fill policy. The H2 replay adjustment also changes behavior withearly_body_bufferdisabled: complete initial replays from the ordinary retry buffer finish without racing downstream closure while partial-write caching remains enabled after body filters. Other request-body writes keep the closure watch, including uncached replays, incomplete retry buffers, and streamed writes.Includes a
body_routingexample demonstrating all three patterns — stream, peek, and mutate:Phase docs and mermaid charts updated to include the new phase.