Support HTTP/1.1 request pipelining on the downstream session

[CodyPubNub]

Resolves #377, #673

What

Adds an opt-in HTTP/1.1 pipelining mode on the downstream session that serves pipelined requests sequentially in request order per RFC 9112 §9.3.2. Pipelining remains disabled by default.

• Adopters opt in per-session with HttpSession::set_pipelining_enabled(true) (mirrors the field+accessor shape of existing setters like keepalive_timeout, abort_on_close, proxy_tasks_enabled). No new ProxyHttp trait method.
• ServerSession::finish() and H1 HttpSession::reuse() are pipelining-aware and return a ReusableHttpStream when the H1 connection can be reused. The wrapper carries the underlying Stream plus any bytes already read for the next pipelined request; into_parts() remains the escape hatch.
• The proxy/server app path converts ReusableHttpStream into ReusedHttpStream with ReusedHttpStream::from_reusable_stream(), preserving the pipelined prefix in HttpPersistentSettings for the next session.

Wire shapes covered

Both observed pipelining wire shapes are handled:

1. Same-segment overread — request 1 and request 2 arrive in one read. HttpSession::reuse() hands the body-reader overread back as the pipelined prefix for the next session.
2. Two-segment overread — request 2 arrives while request 1's response is still being written. read_body_or_idle()'s idle branch stashes non-zero idle reads onto the body reader's overread surface instead of raising ConnectError("Sent data after end of body"). The read == 0 FIN path remains unchanged; this branch only handles non-zero pipelining bytes and leaves half_closed / abort_on_close untouched.

HttpPersistentSettings carries pipelining_enabled + the prefix across keep-alive reuses; read_request() consumes the prefix first before pulling more bytes from the stream.

Positioning

Restores nginx parity for the traffic shape in #377 without inheriting nginx's upstream re-pipelining behavior. Each pipelined request gets its own independent upstream selection and its response is fully written before the next request begins — matching Envoy, AWS CloudFront, Google Classic LB, Akamai, and IIS.

Tests

cargo test -p pingora-core test_pipelining — 12 unit tests covering:

• default-off behavior
• toggle round-trip
• reuse rejects overread when pipelining is disabled (both explicit-off and never-set)
• reuse extracts overread when pipelining is enabled (with and without prior body poll, including Content-Length: 0)
• read_request() consumes a complete prefix without touching the stream
• read_request() falls through to the stream for partial prefixes
• idle-branch stashes bytes when pipelining is enabled (two-segment shape)
• idle-branch still raises ConnectError when pipelining is disabled
• end-to-end session A → session B chain via take_body_overread + set_pipelined_prefix

Plus pingora-proxy/examples/pipelining.rs that opts in from early_request_filter. Minimal reproducer:

RUST_LOG=INFO cargo run --example pipelining

printf 'GET /get HTTP/1.1\r\nHost: httpbin.org\r\n\r\nGET /get HTTP/1.1\r\nHost: httpbin.org\r\n\r\n' \
  | ncat --no-shutdown localhost 6191 \
  | grep -oE 'HTTP/1.1 [0-9]{3}'
HTTP/1.1 200
HTTP/1.1 200

Before this PR the same command returns a single HTTP/1.1 200 and the connection is marked un-reusable.

Runtime cost for non-adopters

• Two extra bool and one Option<BytesMut> on HttpSession (comparable scale to existing fields like abort_on_close, proxy_tasks_enabled).
• The idle-branch check is self.pipelining_enabled && self.pipelined_idle_bytes_stashed (two bool reads) inside an already-async idle poll.
• Two extra fields on HttpPersistentSettings, constructed once per keep-alive boundary.
• No changes to the ProxyHttp trait.

If a cargo feature gate is preferred (e.g. matching the connection_filter pattern), happy to add one in a follow-up commit on this PR.

[drcaramelsyrup]

This implementation is similar to something we were working on internally as well, I will pass it in for internal review so we can layer additional features on top.

[drcaramelsyrup]

I think I'm fine with the overall direction here, but left some comments. I will see if others internally have opinions.

[drcaramelsyrup]

Note that this comment seems slightly misleading, as the abort won't happen if any pipelined bytes are stashed (no 0 byte read can occur).

[drcaramelsyrup]

The if seems unnecessary, the reserve will allocate if the additional bytes requested exceed capacity. I think a simple buf.reserve(prefix.len()) is what we want here.

[drcaramelsyrup]

I think I would almost prefer removing finish_reuse and just making finish() return ReusedHttpConnection as a breaking API change as it seems like an easy footgun to drop the pipelined bytes without handling them otherwise.

I do not expect it to be a commonly used API either.

[drcaramelsyrup]

nit: I slightly prefer something like ReusableHttpStream even though it's very similar. I wonder if the current naming makes some arbitrary distinction between Stream and Connection that doesn't really hold here; Reusable implies that we will transform it into Reused.

[CodyPubNub]

Thank you for the feedback. I've addressed everything you've flagged. I'm looking forward to hearing any internal concerns, considerations, nitpicks, or otherwise that come up!