Support early request body buffering before upstream peer selection by CodyPubNub · Pull Request #816 · cloudflare/pingora

CodyPubNub · 2026-02-16T19:42:59Z

Resolves #780

Adds opt-in early body buffering to ProxyHttp behind the early_body_buffer Cargo feature. When early_request_body_buffer_limit() returns Some(max_size), the full request body is read before request_filter runs. The buffered body is available via Session::get_buffered_body() for inspection and Session::set_buffered_body() for mutation, and is automatically forwarded to HTTP/1.x and HTTP/2 upstreams during the proxy phase. proxy_custom support is intentionally left for a Cloudflare-side follow-up.

New trait methods:

early_request_body_buffer_limit() — opt in to buffering with a size limit (default None)
early_request_body_buffer_timeout() — optionally bound the complete buffering phase, including body-filter callbacks (default None)
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 on request_filter state. The normal request_body_filter() still runs during upstream forwarding.

Use cases:

Auth signature verification that needs the full body before making an auth decision
Content-based routing (e.g., routing GraphQL queries by operation name)
Body transformation before upstream selection
Streaming decompression during buffering via early_request_body_filter

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 a downstream read timeout. With early_body_buffer disabled, no API or runtime path is added; when enabled, default None values preserve existing streaming behavior.

Expect: 100-continue: automatically buffered HTTP/1.1 requests are acknowledged locally before the body is read. The original header remains visible to application filters, then is removed after upstream_request_filter() so it is not forwarded to either HTTP/1.x or HTTP/2 upstreams.

HTTP/2 body handling: requests without Content-Length (valid in HTTP/2) are handled correctly — only Content-Length: 0 skips 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, including None for a confirmed empty body. Applications using this path are responsible for fully consuming the downstream body, handling Expect: 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, local 100 Continue, total buffering timeout, and empty HTTP/2 bodies.

Includes a body_routing example demonstrating all three patterns — stream, peek, and mutate:

RUST_LOG=INFO cargo run --features openssl,early_body_buffer --example body_routing

# Peek + mutate — body is inspected for routing then wrapped in an envelope:
curl -X POST 127.0.0.1:6193/post -H "Host: httpbin.org" -H "Content-Type: application/json" -d '{"route": "beta"}'

# Multi-chunk — early_request_body_filter fires per-chunk:
printf 'POST /post HTTP/1.1\r\nHost: httpbin.org\r\nTransfer-Encoding: chunked\r\n\r\na\r\n{"part":1}\r\na\r\n{"part":2}\r\n0\r\n\r\n' | nc 127.0.0.1 6193

# No buffering — GET requests pass through unchanged:
curl 127.0.0.1:6193/get -H "Host: httpbin.org"

Phase docs and mermaid charts updated to include the new phase.

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

CodyPubNub · 2026-03-12T17:00:30Z

Hi @johnhurt 👋 friendly ping on this. This implements the feature requested in #780 (which has the help wanted label). Happy to adjust the approach if the team has a different direction in mind. Would appreciate any initial feedback on whether this is something you'd consider merging.

johnhurt · 2026-03-13T17:31:29Z

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.

CodyPubNub · 2026-03-13T17:36:37Z

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

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.

PiotrSikora · 2026-03-17T03:59:00Z

+            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;
+                }


Both of those branches contain virtually the same code.

Could you de-duplicate this and use if let Some(data) = body_chunk { ... } where needed?

PiotrSikora · 2026-03-17T04:03:05Z

+                    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?;


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.

PiotrSikora · 2026-03-17T04:22:23Z

+        // 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(());
+        }


This doesn't work with HTTP/2 requests that don't contain the Content-Length header.

PiotrSikora · 2026-03-17T04:55:38Z

+    /// 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)


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.

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.

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)

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).

@PiotrSikora

- 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.

@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.

CodyPubNub · 2026-03-17T07:05:40Z

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.

Thanks for the thorough review, @PiotrSikora. I really appreciate you taking the time.

I've addressed comments 1-3:

Comment 1 (dedup read loop): collapsed the two Some/None match arms into a unified flow with an end_of_body flag.
Comment 2 (callback ordering): added early_request_body_filter() as a dedicated trait method. The early buffering loop calls this instead of request_body_filter, so the existing callback contract is preserved and request_body_filter runs normally during upstream forwarding with no skip guards. Updated phase docs and mermaid charts.
Comment 3 (HTTP/2 body detection): replaced the has_body heuristic with content_length == Some(0). Only skips the read when body is explicitly zero-length.

PiotrSikora

Thanks! This looks much better now.

PiotrSikora · 2026-03-17T16:14:45Z

+        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()
+        };


Nit: You could return (downstream_state, buffer) tuple here (same in H1 proxy).

PiotrSikora · 2026-03-17T16:16:37Z

+    /// - 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> {


Nit: early_request_body_buffer_limit

PiotrSikora · 2026-03-17T16:37:06Z

+    /// 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)


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).

@PiotrSikora

- 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.

QiiHeng3 · 2026-03-31T14:59:34Z

@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.

CodyPubNub · 2026-05-04T16:19:02Z

Hi @johnhurt, I was wondering if there was any updates or feedback from your team to share about this change. Thank you!

johnhurt · 2026-05-04T16:29:45Z

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.

CodyPubNub · 2026-05-04T17:18:18Z

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 early_body_buffer feature to gate the behavior introduced in this PR. Please let me know if there's anything else. Thank you!

CodyPubNub · 2026-06-17T16:58:23Z

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 🙏

jeremyhamning · 2026-07-20T12:51:06Z

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.

CodyPubNub · 2026-07-27T18:32:50Z

Hi @drcaramelsyrup I was wondering if you or others on the team have had a chance to think about this PR. Thank you 🙏

drcaramelsyrup

Directionally good with changes requested, I'm ok with deferred the "stream-if-exceeds-buffer" behavior for a followup.

I noticed we'll need to change proxy_custom as well but it's a patch we can also apply internally since I suspect you don't have a great way of testing that.

drcaramelsyrup · 2026-08-22T22:06:20Z

+
+    /// Handle each chunk of request body during early buffering.
+    ///
+    /// This is called during [`buffer_request_body_early()`] for each body chunk, **before**


Suggested change

/// This is called during [`buffer_request_body_early()`] for each body chunk, **before**

/// This is called while the body is buffered early (enabled by

/// [`Self::early_request_body_buffer_limit()`]) for each body chunk, **before**

buffer_request_body_early() is a private method on HttpProxy, so we get a doc warning.

drcaramelsyrup · 2026-08-22T22:07:17Z

    ReadingFinished,
+    /// body was pre-buffered before upstream connection, skip all downstream polling
+    #[cfg(feature = "early_body_buffer")]
+    PreBuffered,


IMO the variant is unnecessary and actually undesirable when we already have ReadingFinished. It results in an odd behavior diff for abort on close: i.e., we do want to continue polling after the body is finished to be able to detect EOF or error.

drcaramelsyrup · 2026-08-22T22:08:50Z

+        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());


Suggested change

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());

let content_length = header_value_content_length(

session

.downstream_session

.req_header()

.headers

.get(header::CONTENT_LENGTH),

);

drcaramelsyrup · 2026-08-22T22:17:25Z

+        // read body chunks until end of stream
+        loop {
+            let body_chunk: Option<Bytes> =
+                match session.downstream_session.read_body_or_idle(false).await {


Suggested change

match session.downstream_session.read_body_or_idle(false).await {

match session.downstream_session.read_request_body().await {

There's no real point in choosing read_body_or_idle here since this isn't part of a select!. Also, you'd just hang if the body is done.

drcaramelsyrup · 2026-08-22T22:28:11Z

+        if total_size > 0 {
+            let mut combined = bytes::BytesMut::with_capacity(total_size);
+            for part in body_parts {
+                combined.extend_from_slice(&part);
+            }
+            session.set_buffered_body(Some(combined.freeze()));
+        } else {
+            session.mark_body_buffered();
+        }


Suggested change

if total_size > 0 {

let mut combined = bytes::BytesMut::with_capacity(total_size);

for part in body_parts {

combined.extend_from_slice(&part);

}

session.set_buffered_body(Some(combined.freeze()));

} else {

session.mark_body_buffered();

}

if total_size == 0 {

session.mark_body_buffered();

} else if body_parts.len() == 1 {

// 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);

for part in body_parts {

combined.extend_from_slice(&part);

}

session.set_buffered_body(Some(combined.freeze()));

}

optimization when you have a single chunk as the whole body

…buffering

- reuse Session initialization to prevent test-only drift - preserve downstream close detection after buffering - use shared Content-Length parsing and avoid empty-body hangs - avoid copying single-chunk bodies and fix public API docs

CodyPubNub · 2026-08-26T17:22:19Z

Directionally good with changes requested, I'm ok with deferred the "stream-if-exceeds-buffer" behavior for a followup.

I noticed we'll need to change proxy_custom as well but it's a patch we can also apply internally since I suspect you don't have a great way of testing that.

Thank you very much for taking the time to review this and provide feedback. I've addressed all of the inline comments and would be happy to take you up on your offer to implement and test proxy_custom support internally. I've left the "stream-if-exceeds-buffer" behavior for a follow-up.

stevehu · 2026-08-27T18:44:23Z

We are implementing raw-body HMAC webhook authentication in networknt/light-fabric and independently hit the lifecycle gap addressed by this PR.

Our ordering requirement is slightly different from the automatic early-buffering use case: handler-chain rate limiting and JWT/API-key factors must be able to reject a request before its body is buffered, while the application also enforces a profile-specific read timeout and an aggregate memory budget. The integration we need is therefore:

leave early_request_body_buffer_limit() at its default None;
run header-phase policy in request_filter();
consume and verify the bounded body there;
call Session::set_buffered_body(Some(verified_body)); and
rely on the normal request_body_filter(), H1/H2 upstream forwarding, and retry paths to use that application-supplied body.

The current implementation appears to support this cleanly: both H1 and H2 select the same retained buffered-body source, and the normal request-body filter is applied while forwarding it. If this is intended as a supported contract, it would let us remove our small Pingora fork patch without moving security checks ahead of the handler chain or duplicating Pingora's forwarding machinery.

I added an end-to-end regression for the application-managed path in CodyPubNub/pingora#1. It disables automatic buffering, reads in request_filter(), calls set_buffered_body(), forces an upstream retry, verifies exact bytes at the origin, and asserts that request_body_filter() runs for every forwarding attempt. The complete early-body/retry test file passes 5/5 locally.

Could the maintainers confirm that application-supplied buffering through set_buffered_body() is intended to remain supported alongside automatic early buffering?

drcaramelsyrup · 2026-09-03T07:07:46Z

@stevehu Thanks for adding that test, I think we can PR it directly after this change (though it is also fine to include here). I don't think we can guarantee long-term API stability beyond our normal policy but I do think that set_buffered_body call makes sense to work across retries and h1/h2. Some of my comments hint toward making that more explicit in the comments.

I do think your use case might benefit from another convenience primitive to explore in a follow up like an async buffer_request_body that could handle the fiddly bits like 100-continue handling, draining/storing the buffered body, read timeouts, etc.

drcaramelsyrup · 2026-09-03T06:58:54Z

+        // read body chunks until end of stream
+        loop {
+            let body_chunk: Option<Bytes> =
+                match session.downstream_session.read_request_body().await {


We noticed this can end up hanging with an Expect: 100-continue client.

The client may wait for 100 before sending the body, but the early buffering feature will wait for the body before upstream_peer() runs.

To mirror nginx's behavior in this case we could send 100 Continue locally e.g. with write_continue_response. We should keep the Expect header across application filters, but if session.is_body_buffered() is true, remove the Expect header from the cloned header after upstream_request_filter (right before it's sent upstream).

drcaramelsyrup · 2026-09-03T06:58:54Z

+
        // retry, send buffer if it exists
-        if let Some(buffer) = session.as_mut().get_retry_buffer() {
+        if let Some(buffer) = buffer {


If you have an H2 origin and an application's early_request_body_filter removes the body and updates framing, then the session is buffered but its buffer is None.

e.g. something like

async fn early_request_body_filter( &self, session: &mut Session, body: &mut Option<Bytes>, _end_of_stream: bool, _ctx: &mut Self::CTX, ) -> Result<()> { *body = None; let req = session.downstream_session.req_header_mut(); req.remove_header(&header::CONTENT_LENGTH); req.remove_header(&header::TRANSFER_ENCODING); Ok(()) }

Then HEADERS can get sent without END_STREAM and, because of this guard, we may also skip send_body_to2(), so the upstream may never see END_STREAM.

I think we can send_body_to2() even when the session is buffered and its buffer is None to fix that.

drcaramelsyrup · 2026-09-03T06:58:54Z

+    /// Sets the buffered request body.
+    ///
+    /// This is called by `buffer_request_body_early()` after reading the full body.
+    /// Also useful for app code that wants to replace the body (e.g., decompression).


It would be good to note that we'll retain this body across retries.

Also, I believe the documentation should tell the caller that with using this API, they will need to: fully consume the downstream body, handle Expect: 100-continue before reading, and update request framing for their buffered body.

drcaramelsyrup · 2026-09-03T06:58:54Z

+    ///
+    /// Requires the `early_body_buffer` feature.
+    #[cfg(feature = "early_body_buffer")]
+    fn early_request_body_buffer_limit(


Could we also make a total buffering timeout optionally configurable alongside max_size? A total buffering deadline would give this phase a consistent bound across protocols and while there is no upstream peer.

drcaramelsyrup · 2026-09-03T07:10:45Z

@CodyPubNub Thanks for addressing the previous round of comments! We found some more issues after internal reviews, so I left follow-up comments on Expect: 100-continue handling and empty buffered H2 bodies, plus total buffering timeout and API documentation. As mentioned, we’ll track proxy_custom changes on our end.

- prove callers can buffer after header-phase policy checks - verify request body filters run for every forwarding attempt - preserve application-supplied bodies across upstream retries Signed-off-by: Steve Hu <stevehu@gmail.com>

- answer Expect: 100-continue locally while preserving application filter visibility - close empty buffered HTTP/2 request streams and retain explicit empty bodies - add an optional total buffering deadline with H1/H2 protocol regressions - document application-managed framing and retry responsibilities

CodyPubNub · 2026-09-03T16:52:13Z

@CodyPubNub Thanks for addressing the previous round of comments! We found some more issues after internal reviews, so I left follow-up comments on Expect: 100-continue handling and empty buffered H2 bodies, plus total buffering timeout and API documentation. As mentioned, we’ll track proxy_custom changes on our end.

Thank you very much for another thoughtful review. I’ve addressed all follow-up items:

local 100 Continue handling without forwarding Expect upstream
END_STREAM for empty buffered H2 bodies
an optional total buffering timeout
clearer retry and application-managed buffering contracts

I also incorporated @stevehu's regression test as b67b904

Please let me know if I missed anything else. 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

rxdiscovery · 2026-09-10T13:17:28Z

Hello,
Thank you very much @CodyPubNub for your contribution.

I noticed that session.set_buffered_body() has no effect when called inside early_request_body_filter.
In this example, the buffer remains full even though we assigned an empty Bytes to it.

async fn early_request_body_filter(
    &self,
    session: &mut Session,
    body: &mut Option<Bytes>,
    end_of_stream: bool,
    ctx: &mut Self::CTX,
) -> Result<()>
where
    Self::CTX: Send + Sync,
{

    if end_of_stream {
         session.set_buffered_body(Some(Bytes::new()));
    }
  
    Ok(())
}

Do you have any idea what might be causing this?

CodyPubNub · 2026-09-10T15:50:57Z

Hello, Thank you very much @CodyPubNub for your contribution.

I noticed that session.set_buffered_body() has no effect when called inside early_request_body_filter. In this example, the buffer remains full even though we assigned an empty Bytes to it.
async fn early_request_body_filter(
    &self,
    session: &mut Session,
    body: &mut Option<Bytes>,
    end_of_stream: bool,
    ctx: &mut Self::CTX,
) -> Result<()>
where
    Self::CTX: Send + Sync,
{

    if end_of_stream {
         session.set_buffered_body(Some(Bytes::new()));
    }
  
    Ok(())
}
Do you have any idea what might be causing this?

Hi @rxdiscovery, set_buffered_body() should not be called from early_request_body_filter(). That callback operates on each chunk through its body argument. Once the buffering loop finishes, Pingora assembles the filtered chunks and stores the result, overwriting anything assigned during the loop. If the filtered chunks total zero bytes, Pingora only marks the body as buffered and does not replace the assignment.

To discard a chunk, set *body = None. To drop the entire body, do so on every invocation. To replace the fully assembled body, call session.set_buffered_body() later from request_filter().

Either way, update the request framing to match the resulting body. For example, leaving Content-Length unchanged after removing the body can leave an HTTP/1.1 upstream waiting for bytes that will never arrive.

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

rxdiscovery · 2026-09-11T00:19:23Z

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]

CodyPubNub · 2026-09-11T03:01:00Z

@rxdiscovery I may be misunderstanding the proposal, but a few premises here don't match the implementation.

early_request_body_buffer_limit() does not behave like request_filter(). It selects whether automatic buffering is enabled and supplies the maximum permitted size. The decision can use both the Session and context populated by early_request_filter(); it is not limited to inspecting headers.

Content-Length is only used for an optional fail-fast rejection. Pingora also checks the accumulated body size as each chunk is read and filtered, so missing, chunked, or incorrect Content-Length values do not control or bypass enforcement.

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 request_filter() before buffering would remove the primary capability introduced by this PR: making the complete body available inside request_filter() for authentication, mutation, and routing before upstream_peer().

Finally, request_body_filter() remains a forwarding-time callback and runs for every upstream attempt, including retries. early_request_body_filter() instead runs once while consuming the downstream body, so the two callbacks intentionally have different lifecycles.

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.

rxdiscovery · 2026-09-11T09:12:50Z

@CodyPubNub

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 early_request_filter, such as enable_request_body_buffer(max_buffer_size: usize)?
When called, it would trigger internal processing within Pingora right before request_filter is invoked. This would allow us to retrieve the buffer via get_request_body_buffer or modify it using set_request_body_buffer.

I hope this makes things clearer.

On another note, I have a question: in your PR, does early_buffer share the same buffer as Pingora's internal one, so we avoid duplicating data in memory if retry is enabled?

CodyPubNub · 2026-09-11T14:21:09Z

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.

zaidoon1 · 2026-09-15T17:33:49Z

-        {
+        #[cfg(feature = "early_body_buffer")]
+        let body_replayable =
+            session.req_header().method.is_idempotent() || session.is_body_buffered();


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.

zaidoon1 · 2026-09-15T17:33:49Z

+        };
+
        // retry, send buffer if it exists or body empty
        if buffer.is_some() || session.as_mut().is_body_empty() {


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.

zaidoon1 · 2026-09-15T17:33:49Z

+
+            // accumulate the (possibly filtered) data
+            if let Some(filtered) = filter_data {
+                total_size += filtered.len();


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.

zaidoon1 · 2026-09-15T17:33:49Z

+        // 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) {


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.

zaidoon1 · 2026-09-15T17:33:49Z

+            // 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);


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.

zaidoon1 · 2026-09-15T17:33:49Z

+    /// - 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.


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.

zaidoon1 · 2026-09-15T17:33:49Z

+//! retry replay, protocol completion, local `100 Continue` handling, and total buffering
+//! deadlines.
+
+#![cfg(feature = "early_body_buffer")]


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.

rxdiscovery · 2026-09-16T14:30:26Z

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

CodyPubNub · 2026-09-17T20:07:33Z

@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.

CodyPubNub added 4 commits February 16, 2026 11:21

Add body_routing example for early request body buffering

b50ad3f

Mark body read errors as downstream in buffer_request_body_early

42e909d

Simplify early body buffering read errors to use into_down()

56f9065

davidthing mentioned this pull request Feb 23, 2026

Support Early Request Body Access for Dynamic Upstream Peer Selection #780

Open

Merge remote-tracking branch 'upstream/main' into early-request-body-…

82642a1

…buffering # Conflicts: # pingora-proxy/src/lib.rs

johnhurt added the enhancement New feature or request label Mar 13, 2026

PiotrSikora reviewed Mar 17, 2026

View reviewed changes

CodyPubNub added 2 commits March 16, 2026 23:15

PiotrSikora reviewed Mar 17, 2026

View reviewed changes

CodyPubNub added 2 commits May 4, 2026 09:58

Merge branch 'main' into early-request-body-buffering

ba51f61

Gate early_body_buffer behind cargo feature

b2524b9

CodyPubNub mentioned this pull request Jun 8, 2026

Enhanced LLM reverse proxy based on Model Name routing #907

Closed

Merge branch 'main' into early-request-body-buffering

ae6d2d9

Merge branch 'main' into early-request-body-buffering

09c4360

CodyPubNub added 2 commits July 27, 2026 10:04

drcaramelsyrup requested changes Aug 26, 2026

View reviewed changes

CodyPubNub added 2 commits August 26, 2026 09:58

Merge remote-tracking branch 'upstream/main' into early-request-body-…

5916496

…buffering

Address early body buffering review feedback

1dd0166

- reuse Session initialization to prevent test-only drift - preserve downstream close detection after buffering - use shared Content-Length parsing and avoid empty-body hangs - avoid copying single-chunk bodies and fix public API docs

CodyPubNub requested a review from drcaramelsyrup August 26, 2026 17:22

stevehu mentioned this pull request Aug 27, 2026

test(proxy): cover application-managed body buffering CodyPubNub/pingora#1

Closed

stevehu mentioned this pull request Aug 27, 2026

Upgrade Pingora after #816 release and remove local proxy patch networknt/light-fabric#355

Open

11 tasks

drcaramelsyrup reviewed Sep 3, 2026

View reviewed changes

stevehu and others added 2 commits September 3, 2026 09:07

test(proxy): cover application-managed body buffering

b67b904

- prove callers can buffer after header-phase policy checks - verify request body filters run for every forwarding attempt - preserve application-supplied bodies across upstream retries Signed-off-by: Steve Hu <stevehu@gmail.com>

CodyPubNub requested a review from drcaramelsyrup September 3, 2026 16:57

CodyPubNub added 2 commits September 4, 2026 10:55

Fix zero-length buffered requests to H2 upstreams

6ba8a67

- 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

Merge branch 'main' into early-request-body-buffering

a35f2cd

CodyPubNub added 2 commits September 10, 2026 09:07

Merge branch 'main' into early-request-body-buffering

898d4b3

zaidoon1 reviewed Sep 15, 2026

View reviewed changes

	/// This is called during [`buffer_request_body_early()`] for each body chunk, before
	/// This is called while the body is buffered early (enabled by
	/// [`Self::early_request_body_buffer_limit()`]) for each body chunk, before

	match session.downstream_session.read_body_or_idle(false).await {
	match session.downstream_session.read_request_body().await {

Conversation

CodyPubNub commented Feb 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

CodyPubNub commented Mar 12, 2026

Uh oh!

johnhurt commented Mar 13, 2026

Uh oh!

CodyPubNub commented Mar 13, 2026

Uh oh!

PiotrSikora left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

CodyPubNub commented Mar 17, 2026

Uh oh!

PiotrSikora left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

QiiHeng3 commented Mar 31, 2026

Uh oh!

CodyPubNub commented May 4, 2026

Uh oh!

johnhurt commented May 4, 2026

Uh oh!

CodyPubNub commented May 4, 2026

Uh oh!

CodyPubNub commented Jun 17, 2026

Uh oh!

jeremyhamning commented Jul 20, 2026

Uh oh!

CodyPubNub commented Jul 27, 2026

Uh oh!

drcaramelsyrup left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

CodyPubNub commented Aug 26, 2026

Uh oh!

stevehu commented Aug 27, 2026

Uh oh!

drcaramelsyrup commented Sep 3, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

drcaramelsyrup commented Sep 3, 2026

CodyPubNub commented Feb 16, 2026 •

edited

Loading

rxdiscovery commented Sep 10, 2026 •

edited

Loading