feat: streambuffer

Signed-off-by: Shane Utt <shaneutt@linux.com>

Author: shaneutt

benchmarks/microbenchmarks/common.rs

@@ -42,8 +42,10 @@ pub(crate) fn make_ctx(req: &Request) -> HttpFilterContext<'_> {
         health_registry: None,
         request: req,
         request_body_bytes: 0,
+        request_body_mode: praxis_filter::BodyMode::Stream,
         request_start: std::time::Instant::now(),
         response_body_bytes: 0,
+        response_body_mode: praxis_filter::BodyMode::Stream,
         response_header: None,
         response_headers_modified: false,
         rewritten_path: None,

docs/architecture.md

@@ -250,13 +250,10 @@ flowchart TD
     Proto --> |"on Release or EOS: forward buffer"| Upstream
 ```
 
-Three delivery modes:
+Two delivery modes:
 
 - **Stream**: chunks flow through filters as they arrive.
   Low latency, low memory.
-- **Buffer**: the protocol layer accumulates chunks into a
-  `BodyBuffer`, then delivers the full body in a single
-  call. Required when any filter needs the complete body.
 - **StreamBuffer**: chunks are delivered to filters
   incrementally (like Stream) but accumulated in a buffer
   and not forwarded to upstream until a filter returns
@@ -275,10 +272,9 @@ routing decisions. The pre-read body is stored and
 forwarded to the upstream after the connection is
 established.
 
-Precedence: `Buffer` > `StreamBuffer` > `Stream`. If any
-filter requests `Buffer`, the entire pipeline buffers. If
-any filter requests `StreamBuffer` (and none requests
-`Buffer`), the pipeline uses stream-buffered mode.
+Precedence: `StreamBuffer` > `SizeLimit` > `Stream`. If
+any filter requests `StreamBuffer`, the pipeline uses
+stream-buffered mode.
 Global `body_limits.max_request_bytes` / `body_limits.max_response_bytes`
 config limits force buffer mode for size enforcement even
 when no filter requests body access.
@@ -411,7 +407,7 @@ praxis-filter                   Filter pipeline engine
 │   ├── access                  BodyAccess enum
 │   ├── buffer                  BodyBuffer and overflow handling
 │   ├── builder                 Pre-computed BodyCapabilities
-│   └── mode                    BodyMode enum (Stream, Buffer, StreamBuffer)
+│   └── mode                    BodyMode enum (Stream, StreamBuffer, SizeLimit)
 ├── condition/                  Condition evaluation for filter gating
 │   ├── request                 Request condition evaluation
 │   └── response                Response condition evaluation

docs/configuration.md

@@ -545,8 +545,8 @@ Each rule has:
 | `negate` | bool | no | Invert match (default: false) |
 
 Each rule must have either `contains` or `pattern`, not
-both. Body rules use Buffer mode (up to 1 MiB by default)
-to inspect the full request body.
+both. Body rules use StreamBuffer mode (up to 1 MiB by
+default) to inspect the full request body.
 
 ### CORS
 
@@ -649,7 +649,7 @@ body hooks. A filter can have both `conditions` and
 ## Payload Size Limits
 
 Global hard ceilings on request and response payload
-size. These apply across all body modes (Stream, Buffer,
+size. These apply across all body modes (Stream,
 StreamBuffer). When a filter also declares a per-filter
 `max_bytes`, the smaller of the two limits is enforced.
 Requests exceeding the limit receive 413 (Payload Too

docs/extensions.md

@@ -326,15 +326,40 @@ chunks in place.
 - `Stream`: lowest latency; chunks flow through as they
   arrive. Best for filters that inspect headers only or
   process chunks independently.
-- `Buffer`: accumulates the entire body before
-  delivering it. Use when your filter needs the complete
-  body (e.g. signature verification). Set `max_bytes` to
-  avoid unbounded memory growth.
 - `StreamBuffer`: chunks flow through filters
   incrementally but forwarding to upstream is deferred
   until `Release` or end-of-stream. Use when body
-  content influences routing or when you need to inspect
-  the full body before upstream selection.
+  content influences routing, when you need the complete
+  body (e.g. signature verification), or when you need
+  to inspect the full body before upstream selection.
+  Set `max_bytes` to avoid unbounded memory growth.
+
+Two patterns for declaring `StreamBuffer`:
+
+**Static declaration** (filter always needs the body):
+
+```rust
+fn request_body_mode(&self) -> BodyMode {
+    BodyMode::StreamBuffer { max_bytes: Some(1_048_576) }
+}
+```
+
+**Per-request upgrade** (conditional buffering):
+
+```rust
+async fn on_request(
+    &self, ctx: &mut HttpFilterContext<'_>,
+) -> Result<FilterAction, FilterError> {
+    if needs_body_inspection(ctx) {
+        ctx.set_request_body_mode(
+            BodyMode::StreamBuffer {
+                max_bytes: Some(1_048_576),
+            },
+        );
+    }
+    Ok(FilterAction::Continue)
+}
+```
 
 ### `on_response_body` is synchronous
 

docs/features.md

@@ -44,8 +44,8 @@
   by default, opt-in buffered or stream-buffered payload
   access with configurable size limits.
   Stream mode passes chunks through as they arrive
-  (lowest latency). Buffer mode accumulates the full
-  body. StreamBuffer delivers chunks to filters
+  (lowest latency). StreamBuffer delivers chunks to
+  filters
   incrementally but defers upstream forwarding until
   release. See [Payload Processing][payload-processing]
   in the architecture docs.

docs/filters.md

@@ -336,10 +336,10 @@ fn request_body_access(&self) -> BodyAccess {
 | Mode                          | Behavior        | Use case                  |
 | ----------------------------- | --------------- | ------------------------- |
 | `Stream` (default)            | Per chunk       | Logging, transforms       |
-| `Buffer { max_bytes }`        | Full body       | JSON, payload routing     |
 | `StreamBuffer { max_bytes }`  | Deferred stream | Inspection before forward |
 
-If any filter requests `Buffer`, the pipeline buffers.
+If any filter requests `StreamBuffer`, the pipeline
+defers upstream forwarding until release.
 
 ### StreamBuffer Mode
 

filter/src/actions.rs

@@ -90,17 +90,17 @@ impl Rejection {
         }
     }
 
-    /// Add a header to the rejection response.
-    pub fn with_header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
-        self.headers.push((name.into(), value.into()));
-        self
-    }
-
     /// Set the body of the rejection response.
     pub fn with_body(mut self, body: impl Into<Bytes>) -> Self {
         self.body = Some(body.into());
         self
     }
+
+    /// Add a header to the rejection response.
+    pub fn with_header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
+        self.headers.push((name.into(), value.into()));
+        self
+    }
 }
 
 // -----------------------------------------------------------------------------

filter/src/body/builder.rs

@@ -22,13 +22,17 @@ use super::BodyMode;
 ///
 /// let caps = BodyCapabilities {
 ///     needs_request_body: true,
-///     request_body_mode: BodyMode::Buffer { max_bytes: 4096 },
+///     request_body_mode: BodyMode::StreamBuffer {
+///         max_bytes: Some(4096),
+///     },
 ///     ..Default::default()
 /// };
 /// assert!(caps.needs_request_body);
 /// assert!(matches!(
 ///     caps.request_body_mode,
-///     BodyMode::Buffer { max_bytes: 4096 }
+///     BodyMode::StreamBuffer {
+///         max_bytes: Some(4096)
+///     }
 /// ));
 /// assert!(!caps.needs_response_body, "unset fields stay at default");
 /// ```
@@ -38,12 +42,12 @@ pub struct BodyCapabilities {
     /// Whether any filter writes to the request body.
     pub any_request_body_writer: bool,
 
-    /// Whether any response condition references headers.
-    pub any_response_condition_uses_headers: bool,
-
     /// Whether any filter writes to the response body.
     pub any_response_body_writer: bool,
 
+    /// Whether any response condition references headers.
+    pub any_response_condition_uses_headers: bool,
+
     /// Whether any filter needs request body access.
     pub needs_request_body: bool,
 
@@ -53,10 +57,10 @@ pub struct BodyCapabilities {
     /// Whether any filter needs response body access.
     pub needs_response_body: bool,
 
-    /// Resolved request body mode (Buffer if any filter requires it).
+    /// Resolved request body mode (`StreamBuffer` if any filter requires it).
     pub request_body_mode: BodyMode,
 
-    /// Resolved response body mode (Buffer if any filter requires it).
+    /// Resolved response body mode (`StreamBuffer` if any filter requires it).
     pub response_body_mode: BodyMode,
 }
 

filter/src/body/mode.rs

@@ -15,8 +15,15 @@
 /// let mode = BodyMode::default();
 /// assert!(matches!(mode, BodyMode::Stream));
 ///
-/// let buffered = BodyMode::Buffer { max_bytes: 1024 };
-/// assert!(matches!(buffered, BodyMode::Buffer { max_bytes: 1024 }));
+/// let buffered = BodyMode::StreamBuffer {
+///     max_bytes: Some(1024),
+/// };
+/// assert!(matches!(
+///     buffered,
+///     BodyMode::StreamBuffer {
+///         max_bytes: Some(1024)
+///     }
+/// ));
 ///
 /// let stream_buf = BodyMode::StreamBuffer { max_bytes: None };
 /// assert!(matches!(
@@ -41,6 +48,7 @@
 /// ));
 /// ```
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+#[non_exhaustive]
 pub enum BodyMode {
     /// Deliver chunks as they arrive. Low latency, low memory.
     ///
@@ -53,19 +61,6 @@ pub enum BodyMode {
     #[default]
     Stream,
 
-    /// Buffer the entire body, then deliver it in a single call.
-    ///
-    /// ```
-    /// use praxis_filter::BodyMode;
-    ///
-    /// let mode = BodyMode::Buffer { max_bytes: 8192 };
-    /// assert!(matches!(mode, BodyMode::Buffer { max_bytes: 8192 }));
-    /// ```
-    Buffer {
-        /// Maximum body size in bytes.
-        max_bytes: usize,
-    },
-
     /// Deliver chunks incrementally (like [`Stream`]) but accumulate
     /// them and defer upstream forwarding until a filter returns
     /// [`FilterAction::Release`] or end-of-stream is reached.
@@ -141,16 +136,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn body_mode_buffer_carries_limit() {
-        let mode = BodyMode::Buffer { max_bytes: 4096 };
-
-        assert!(
-            matches!(mode, BodyMode::Buffer { max_bytes: 4096 }),
-            "Buffer variant should carry configured limit"
-        );
-    }
-
     #[test]
     fn body_mode_stream_buffer_unlimited() {
         let mode = BodyMode::StreamBuffer { max_bytes: None };
@@ -179,21 +164,16 @@ mod tests {
     }
 
     #[test]
-    fn body_mode_size_limit_is_distinct_from_buffer() {
+    fn body_mode_size_limit_is_distinct_from_stream_buffer() {
         assert_ne!(
             BodyMode::SizeLimit { max_bytes: 100 },
-            BodyMode::Buffer { max_bytes: 100 },
-            "SizeLimit and Buffer should be distinct even with same limit"
+            BodyMode::StreamBuffer { max_bytes: Some(100) },
+            "SizeLimit and StreamBuffer should be distinct even with same limit"
         );
     }
 
     #[test]
-    fn body_mode_stream_buffer_is_distinct() {
-        assert_ne!(
-            BodyMode::StreamBuffer { max_bytes: None },
-            BodyMode::Buffer { max_bytes: 100 },
-            "StreamBuffer and Buffer should be distinct variants"
-        );
+    fn body_mode_stream_buffer_is_distinct_from_stream() {
         assert_ne!(
             BodyMode::StreamBuffer { max_bytes: None },
             BodyMode::Stream,

filter/src/builtins/http/security/guardrails/filter.rs

@@ -199,8 +199,8 @@ impl HttpFilter for GuardrailsFilter {
 
     fn request_body_mode(&self) -> BodyMode {
         if self.needs_body {
-            BodyMode::Buffer {
-                max_bytes: DEFAULT_MAX_BODY_BYTES,
+            BodyMode::StreamBuffer {
+                max_bytes: Some(DEFAULT_MAX_BODY_BYTES),
             }
         } else {
             BodyMode::Stream
@@ -224,8 +224,12 @@ impl HttpFilter for GuardrailsFilter {
         &self,
         ctx: &mut HttpFilterContext<'_>,
         body: &mut Option<Bytes>,
-        _end_of_stream: bool,
+        end_of_stream: bool,
     ) -> Result<FilterAction, FilterError> {
+        if !end_of_stream {
+            return Ok(FilterAction::Continue);
+        }
+
         let Some(chunk) = body.as_ref() else {
             write_result(ctx, "passed");
             return Ok(FilterAction::Continue);