praxis_protocol/http/pingora/
context.rs

1// SPDX-License-Identifier: MIT
2// Copyright (c) 2024 Praxis Contributors
3
4//! Per-request context that carries filter pipeline results through Pingora's request/response lifecycle hooks.
5
6use std::{collections::VecDeque, net::IpAddr, sync::Arc, time::Instant};
7
8use bytes::Bytes;
9use praxis_core::connectivity::Upstream;
10use praxis_filter::{BodyBuffer, BodyMode, FilterPipeline, Request, Response, TrustedHeaderMutation};
11use tokio::sync::OwnedSemaphorePermit;
12
13// -----------------------------------------------------------------------------
14// PingoraRequestCtx
15// -----------------------------------------------------------------------------
16
17/// Per-request context carrying filter pipeline results through Pingora hooks.
18///
19/// ```
20/// use std::sync::Arc;
21///
22/// use praxis_protocol::http::pingora::context::PingoraRequestCtx;
23///
24/// let mut ctx = PingoraRequestCtx::default();
25/// ctx.cluster = Some(Arc::from("api-cluster"));
26/// assert_eq!(ctx.cluster.as_deref(), Some("api-cluster"));
27/// ```
28#[expect(clippy::struct_excessive_bools, reason = "lifecycle flags")]
29pub struct PingoraRequestCtx {
30    /// Connection permit from the per-listener semaphore.
31    ///
32    /// Held for the lifetime of the request. RAII drop
33    /// releases the permit when the context is dropped,
34    /// including error and timeout paths.
35    pub _connection_permit: Option<OwnedSemaphorePermit>,
36
37    /// Permit from the process-wide connection semaphore.
38    ///
39    /// Present only when `runtime.max_connections` is configured.
40    pub _global_connection_permit: Option<OwnedSemaphorePermit>,
41
42    /// Downstream client IP address.
43    pub client_addr: Option<IpAddr>,
44
45    /// HTTP version of the downstream client request.
46    ///
47    /// Captured during `request_filter` so the response-phase Via
48    /// header can reflect the protocol the client used.
49    pub client_http_version: Option<http::Version>,
50
51    /// Name of the cluster selected by a cluster-selecting filter.
52    pub cluster: Option<Arc<str>>,
53
54    /// Cached per-filter body-done indices. Swapped into each
55    /// [`HttpFilterContext`] and written back after execution so
56    /// that the heap allocation is reused across pipeline phases.
57    ///
58    /// [`HttpFilterContext`]: praxis_filter::HttpFilterContext
59    pub cached_body_done_indices: Vec<bool>,
60
61    /// Cached per-filter execution indices. Same lifecycle as
62    /// [`cached_body_done_indices`].
63    ///
64    /// [`cached_body_done_indices`]: Self::cached_body_done_indices
65    pub cached_executed_filter_indices: Vec<bool>,
66
67    /// Whether the downstream connection uses TLS.
68    ///
69    /// Derived from the Pingora session's SSL digest during
70    /// `request_filter`. Used by the forwarded headers filter
71    /// to set `X-Forwarded-Proto` correctly for HTTP/1.1
72    /// connections where the URI lacks a scheme.
73    pub downstream_tls: bool,
74
75    /// Verified downstream TLS peer identity.
76    ///
77    /// Set once from the SSL digest in `request_filter` before
78    /// the first filter runs.  Cloned (not moved) into each
79    /// `HttpFilterContext` so it is available in both pre-read
80    /// body phases and the main filter pipeline.  `None` for
81    /// non-mTLS or no-client-cert connections.
82    pub peer_identity: Option<praxis_tls::TlsPeerIdentity>,
83
84    /// Whether the connection was upgraded via 101 Switching Protocols.
85    ///
86    /// Set during `response_filter` when the upstream returns 101.
87    /// Body filter hooks skip processing when true, since post-upgrade
88    /// bytes are raw protocol frames (e.g. `WebSocket`), not HTTP bodies.
89    pub connection_upgraded: bool,
90
91    /// Type-safe request-scoped extension container. Swapped into each
92    /// [`HttpFilterContext`] and written back after filter execution,
93    /// following the same lifecycle as [`filter_metadata`].
94    ///
95    /// [`HttpFilterContext`]: praxis_filter::HttpFilterContext
96    /// [`filter_metadata`]: PingoraRequestCtx::filter_metadata
97    pub extensions: praxis_filter::RequestExtensions,
98
99    /// Durable per-request metadata that persists across all lifecycle
100    /// phases. Swapped into each [`HttpFilterContext`] and written back
101    /// after filter execution.
102    ///
103    /// [`HttpFilterContext`]: praxis_filter::HttpFilterContext
104    pub filter_metadata: std::collections::HashMap<String, String>,
105
106    /// Ordered log of trusted header mutations from pre-read body
107    /// processing. Swapped into each [`HttpFilterContext`] and written
108    /// back after filter execution.
109    ///
110    /// [`HttpFilterContext`]: praxis_filter::HttpFilterContext
111    pub pre_read_mutations: Vec<TrustedHeaderMutation>,
112
113    /// Structured per-request metadata keyed by namespace.
114    /// Swapped into each [`HttpFilterContext`] and written back
115    /// after filter execution.
116    ///
117    /// [`HttpFilterContext`]: praxis_filter::HttpFilterContext
118    pub structured_metadata: std::collections::HashMap<String, serde_json::Value>,
119
120    /// Post-mutation request body length produced during `StreamBuffer`
121    /// pre-read.
122    ///
123    /// Stored so `upstream_request_filter` can repair request framing
124    /// before Pingora sends headers to the backend.
125    pub mutated_request_body_len: Option<usize>,
126
127    /// Pipeline pinned for this request's entire lifecycle.
128    ///
129    /// Set once during `request_filter` by cloning the [`Arc`] from the
130    /// listener's [`ArcSwap`]. All subsequent hooks (request body, response,
131    /// response body, logging) use this reference instead of re-loading
132    /// from the [`ArcSwap`], ensuring that a hot configuration reload
133    /// cannot change the pipeline mid-request.
134    ///
135    /// [`ArcSwap`]: arc_swap::ArcSwap
136    pub pinned_pipeline: Option<Arc<FilterPipeline>>,
137
138    /// Filter results from body pre-read. Carried into the next
139    /// [`HttpFilterContext`] so that branch chains attached to the
140    /// first `on_request` filter can evaluate body-derived results.
141    ///
142    /// [`HttpFilterContext`]: praxis_filter::HttpFilterContext
143    pub filter_results: std::collections::HashMap<&'static str, praxis_filter::FilterResultSet>,
144
145    /// Typed per-filter state that persists across all lifecycle
146    /// phases. Keyed by stable filter invocation ID, unique within
147    /// the request's pinned [`FilterPipeline`]. Swapped into each
148    /// [`HttpFilterContext`] and written back after filter execution,
149    /// following the same pattern as [`filter_metadata`].
150    ///
151    /// [`FilterPipeline`]: praxis_filter::FilterPipeline
152    /// [`HttpFilterContext`]: praxis_filter::HttpFilterContext
153    /// [`filter_metadata`]: Self::filter_metadata
154    pub filter_state: std::collections::HashMap<usize, Box<dyn std::any::Any + Send + Sync>>,
155
156    /// Cluster name snapshot retained for metrics emission in the
157    /// `logging()` hook, after `cluster` has been consumed by filter
158    /// context construction.
159    pub metrics_cluster: Option<Arc<str>>,
160
161    /// Pre-built [`SharedString`] for the metrics cluster label.
162    ///
163    /// Cached when `metrics_cluster` is set so that
164    /// `emit_request_metrics` avoids an `Arc` clone per request.
165    ///
166    /// [`SharedString`]: ::metrics::SharedString
167    pub metrics_cluster_shared: Option<::metrics::SharedString>,
168
169    /// Matched route path-match pattern for the `route` metric label.
170    pub metrics_route: Option<::metrics::SharedString>,
171
172    /// RAII guard that decrements `praxis_connections_active` on drop.
173    pub(crate) _active_connection: Option<crate::http::pingora::metrics::ActiveConnectionGuard>,
174
175    /// When the current upstream connect attempt started.
176    pub upstream_connect_start: Option<Instant>,
177
178    /// Pre-read body chunks (`StreamBuffer` mode). When `StreamBuffer` is
179    /// active, the body is read during `request_filter` (before upstream
180    /// selection) so that body-based routing can influence `upstream_peer`.
181    /// The `request_body_filter` hook then forwards these stored chunks
182    /// instead of reading from the session.
183    ///
184    /// Uses `VecDeque` so that draining from the front is O(1).
185    pub pre_read_body: Option<VecDeque<Bytes>>,
186
187    /// Buffer for request body accumulation in [`StreamBuffer`] mode.
188    ///
189    /// [`StreamBuffer`]: praxis_filter::BodyMode::StreamBuffer
190    pub request_body_buffer: Option<BodyBuffer>,
191
192    /// Accumulated request body bytes seen so far.
193    pub request_body_bytes: u64,
194
195    /// Per-request body delivery mode for the request direction.
196    /// Seeded from static pipeline capabilities, then potentially
197    /// upgraded by filters during `on_request`.
198    pub request_body_mode: BodyMode,
199
200    /// Whether the request body has been released (`StreamBuffer` mode).
201    /// Once true, remaining chunks bypass buffering and stream through.
202    pub request_body_released: bool,
203
204    /// Whether the request method is idempotent (GET, HEAD, OPTIONS).
205    pub request_is_idempotent: bool,
206
207    /// Snapshot of the original request for body/response body phases.
208    pub request_snapshot: Option<Request>,
209
210    /// When this request was received.
211    pub request_start: Instant,
212
213    /// Buffer for response body accumulation in [`StreamBuffer`] mode.
214    ///
215    /// [`StreamBuffer`]: praxis_filter::BodyMode::StreamBuffer
216    pub response_body_buffer: Option<BodyBuffer>,
217
218    /// Accumulated response body bytes seen so far.
219    pub response_body_bytes: u64,
220
221    /// Per-request body delivery mode for the response direction.
222    /// Seeded from static pipeline capabilities, then potentially
223    /// upgraded by filters during `on_response`.
224    pub response_body_mode: BodyMode,
225
226    /// Whether the response body has been released (`StreamBuffer` mode).
227    pub response_body_released: bool,
228
229    /// Snapshot of response headers after response-phase filters.
230    ///
231    /// Used to evaluate `response_conditions` consistently during
232    /// response-body hooks, where Pingora no longer exposes mutable
233    /// response headers.
234    pub response_header_snapshot: Option<Response>,
235
236    /// Upstream response status code, captured during `response_filter`
237    /// for passive health recording in the `logging` hook.
238    pub upstream_response_status: Option<u16>,
239
240    /// Whether the response phase has been executed. Used to ensure
241    /// cleanup (e.g. least-connections counter release) in the
242    /// `logging()` hook when errors bypass `response_filter`.
243    pub response_phase_done: bool,
244
245    /// Number of upstream connection retries attempted.
246    pub retries: u32,
247
248    /// Index of the selected endpoint in the cluster's
249    /// endpoint list. Set during load balancing; used
250    /// for passive health recording in the logging hook.
251    pub selected_endpoint_index: Option<usize>,
252
253    /// Rewritten URI path for the upstream request.
254    ///
255    /// Set by the `path_rewrite` filter via [`HttpFilterContext`] and
256    /// applied in `upstream_request_filter`.
257    ///
258    /// [`HttpFilterContext`]: praxis_filter::HttpFilterContext
259    pub rewritten_path: Option<String>,
260
261    /// Upstream endpoint selected by the load balancer filter.
262    pub upstream: Option<Upstream>,
263
264    /// Saved upstream for retry (cloned before first use).
265    pub upstream_for_retry: Option<Upstream>,
266}
267
268/// Build an [`HttpFilterContext`] from a `PingoraRequestCtx`.
269///
270/// Macro (not a function) so Rust's disjoint field borrowing
271/// works: `filter_context_for` borrows `self.request_snapshot`
272/// immutably while `cluster`, `upstream`, and `rewritten_path`
273/// are taken mutably. A function call with `&mut self` would
274/// collapse these into a single mutable borrow.
275///
276/// [`HttpFilterContext`]: praxis_filter::HttpFilterContext
277macro_rules! filter_context {
278    ($ctx:expr, $pipeline:expr, $request:expr, $response_header:expr) => {{
279        $pipeline.prepare_extensions(&mut $ctx.extensions);
280        praxis_filter::HttpFilterContext {
281            buffered_request_body: $ctx
282                .pre_read_body
283                .as_ref()
284                .map(|chunks| chunks.front().cloned().unwrap_or_default()),
285            body_done_indices: std::mem::take(&mut $ctx.cached_body_done_indices),
286            branch_iterations: std::collections::HashMap::new(),
287            client_addr: $ctx.client_addr,
288            cluster: $ctx.cluster.take(),
289            current_filter_id: None,
290            downstream_tls: $ctx.downstream_tls,
291            metrics_route: $ctx.metrics_route.clone(),
292            peer_identity: $ctx.peer_identity.clone(),
293            extensions: std::mem::take(&mut $ctx.extensions),
294            executed_filter_indices: std::mem::take(&mut $ctx.cached_executed_filter_indices),
295            extra_request_headers: Vec::new(),
296            request_headers_to_remove: Vec::new(),
297            request_headers_to_set: Vec::new(),
298            filter_metadata: std::mem::take(&mut $ctx.filter_metadata),
299            pre_read_mutations: std::mem::take(&mut $ctx.pre_read_mutations),
300            structured_metadata: std::mem::take(&mut $ctx.structured_metadata),
301            filter_results: std::mem::take(&mut $ctx.filter_results),
302            filter_state: std::mem::take(&mut $ctx.filter_state),
303            health_registry: $pipeline.health_registry(),
304            id_generator: $pipeline.id_generator(),
305            kv_stores: $pipeline.kv_stores(),
306            subrequest_client: $pipeline.subrequest_client(),
307            request: $request,
308            request_body_bytes: $ctx.request_body_bytes,
309            request_body_mode: $ctx.request_body_mode,
310            request_start: $ctx.request_start,
311            response_body_bytes: $ctx.response_body_bytes,
312            response_body_mode: $ctx.response_body_mode,
313            response_header: $response_header,
314            response_headers_modified: false,
315            rewritten_path: $ctx.rewritten_path.take(),
316            selected_endpoint_index: $ctx.selected_endpoint_index,
317            time_source: $pipeline.time_source(),
318            upstream: $ctx.upstream.take(),
319        }
320    }};
321}
322
323impl PingoraRequestCtx {
324    /// Build an [`HttpFilterContext`] using an external request reference.
325    ///
326    /// Takes `cluster` and `upstream` from `self` (leaving `None`
327    /// behind) so that filters can reassign them. The caller must
328    /// write those fields back after filter execution.
329    ///
330    /// ```
331    /// use praxis_filter::{FilterPipeline, FilterRegistry, Request};
332    /// use praxis_protocol::http::pingora::context::PingoraRequestCtx;
333    ///
334    /// let registry = FilterRegistry::with_builtins();
335    /// let pipeline = FilterPipeline::build(&mut [], &registry).unwrap();
336    /// let request = Request {
337    ///     method: http::Method::GET,
338    ///     uri: http::Uri::from_static("/"),
339    ///     headers: http::HeaderMap::new(),
340    /// };
341    /// let mut ctx = PingoraRequestCtx::default();
342    /// let filter_ctx = ctx.build_filter_context(&pipeline, &request, None);
343    /// assert!(filter_ctx.cluster.is_none());
344    /// ```
345    ///
346    /// [`HttpFilterContext`]: praxis_filter::HttpFilterContext
347    pub fn build_filter_context<'a>(
348        &mut self,
349        pipeline: &'a FilterPipeline,
350        request: &'a Request,
351        response_header: Option<&'a mut Response>,
352    ) -> praxis_filter::HttpFilterContext<'a> {
353        filter_context!(self, pipeline, request, response_header)
354    }
355
356    /// Build an [`HttpFilterContext`] from the stored [`request_snapshot`].
357    ///
358    /// Uses disjoint field borrowing so that `request_snapshot` is
359    /// borrowed immutably while `cluster` and `upstream` are taken
360    /// mutably.
361    ///
362    /// Returns `None` when `request_snapshot` is not set.
363    ///
364    /// ```
365    /// use praxis_filter::{FilterPipeline, FilterRegistry, Request};
366    /// use praxis_protocol::http::pingora::context::PingoraRequestCtx;
367    ///
368    /// let registry = FilterRegistry::with_builtins();
369    /// let pipeline = FilterPipeline::build(&mut [], &registry).unwrap();
370    /// let mut ctx = PingoraRequestCtx::default();
371    /// ctx.request_snapshot = Some(Request {
372    ///     method: http::Method::GET,
373    ///     uri: http::Uri::from_static("/"),
374    ///     headers: http::HeaderMap::new(),
375    /// });
376    /// let filter_ctx = ctx.filter_context_for(&pipeline, None);
377    /// assert!(filter_ctx.is_some());
378    /// ```
379    ///
380    /// [`HttpFilterContext`]: praxis_filter::HttpFilterContext
381    /// [`request_snapshot`]: PingoraRequestCtx::request_snapshot
382    pub fn filter_context_for<'a>(
383        &'a mut self,
384        pipeline: &'a FilterPipeline,
385        response_header: Option<&'a mut Response>,
386    ) -> Option<praxis_filter::HttpFilterContext<'a>> {
387        let request = self.request_snapshot.as_ref()?;
388        Some(filter_context!(self, pipeline, request, response_header))
389    }
390
391    /// Build an [`HttpFilterContext`] plus the saved response header for body conditions.
392    ///
393    /// Response body hooks do not receive mutable headers, but their
394    /// `response_conditions` still need the response status and headers
395    /// captured during the response phase.
396    ///
397    /// [`HttpFilterContext`]: praxis_filter::HttpFilterContext
398    pub fn response_body_context_for<'a>(
399        &'a mut self,
400        pipeline: &'a FilterPipeline,
401    ) -> Option<(praxis_filter::HttpFilterContext<'a>, Option<&'a Response>)> {
402        let request = self.request_snapshot.as_ref()?;
403        let response_header = self.response_header_snapshot.as_ref();
404        Some((filter_context!(self, pipeline, request, None), response_header))
405    }
406
407    /// Pin the current pipeline for this request's entire lifecycle.
408    ///
409    /// Clones the [`Arc`] from the [`ArcSwap`] and stores it in
410    /// [`pinned_pipeline`]. All subsequent hooks should call
411    /// [`pipeline`] instead of re-loading from the [`ArcSwap`].
412    ///
413    /// Called once by `request_filter` in both body-capable and
414    /// no-body handlers.
415    ///
416    /// [`ArcSwap`]: arc_swap::ArcSwap
417    /// [`pinned_pipeline`]: Self::pinned_pipeline
418    /// [`pipeline`]: Self::pipeline
419    pub fn pin_pipeline(&mut self, swap: &arc_swap::ArcSwap<FilterPipeline>) -> Arc<FilterPipeline> {
420        if let Some(existing) = &self.pinned_pipeline {
421            return Arc::clone(existing);
422        }
423        let pipeline = swap.load_full();
424        self.pinned_pipeline = Some(Arc::clone(&pipeline));
425        pipeline
426    }
427
428    /// Return the pinned pipeline, falling back to a fresh
429    /// [`ArcSwap`] load when no pipeline was pinned.
430    ///
431    /// The fallback covers early-failure paths where a lifecycle
432    /// hook runs before `request_filter` (e.g. after
433    /// `early_request_filter` rejection triggers `logging`).
434    ///
435    /// Per-body-chunk hooks (`request_body_filter`,
436    /// `response_body_filter`) call this on every chunk,
437    /// incurring one [`Arc::clone`] per invocation.
438    ///
439    /// [`ArcSwap`]: arc_swap::ArcSwap
440    pub fn pipeline(&self, swap: &arc_swap::ArcSwap<FilterPipeline>) -> Arc<FilterPipeline> {
441        self.pinned_pipeline
442            .as_ref()
443            .map_or_else(|| swap.load_full(), Arc::clone)
444    }
445}
446
447impl Default for PingoraRequestCtx {
448    #[expect(
449        clippy::too_many_lines,
450        reason = "context default enumerates all lifecycle fields explicitly"
451    )]
452    fn default() -> Self {
453        Self {
454            _connection_permit: None,
455            _global_connection_permit: None,
456            cached_body_done_indices: Vec::new(),
457            cached_executed_filter_indices: Vec::new(),
458            client_addr: None,
459            client_http_version: None,
460            cluster: None,
461            connection_upgraded: false,
462            downstream_tls: false,
463            peer_identity: None,
464            extensions: praxis_filter::RequestExtensions::new(),
465            filter_metadata: std::collections::HashMap::new(),
466            pre_read_mutations: Vec::new(),
467            structured_metadata: std::collections::HashMap::new(),
468            mutated_request_body_len: None,
469            pinned_pipeline: None,
470            filter_results: std::collections::HashMap::new(),
471            filter_state: std::collections::HashMap::new(),
472            metrics_cluster: None,
473            metrics_cluster_shared: None,
474            metrics_route: None,
475            _active_connection: None,
476            upstream_connect_start: None,
477            pre_read_body: None,
478            request_body_buffer: None,
479            request_body_bytes: 0,
480            request_body_mode: BodyMode::Stream,
481            request_body_released: false,
482            request_is_idempotent: false,
483            request_snapshot: None,
484            request_start: Instant::now(),
485            response_body_buffer: None,
486            response_body_bytes: 0,
487            response_body_mode: BodyMode::Stream,
488            response_body_released: false,
489            response_header_snapshot: None,
490            upstream_response_status: None,
491            response_phase_done: false,
492            retries: 0,
493            rewritten_path: None,
494            selected_endpoint_index: None,
495            upstream: None,
496            upstream_for_retry: None,
497        }
498    }
499}
500
501// -----------------------------------------------------------------------------
502// Tests
503// -----------------------------------------------------------------------------
504
505#[cfg(test)]
506#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
507#[allow(
508    clippy::unwrap_used,
509    clippy::expect_used,
510    clippy::indexing_slicing,
511    clippy::significant_drop_tightening,
512    clippy::too_many_lines,
513    reason = "tests"
514)]
515mod tests {
516    use std::{
517        collections::VecDeque,
518        net::{IpAddr, Ipv4Addr},
519        sync::Arc,
520    };
521
522    use bytes::Bytes;
523    use http::{HeaderMap, Method, Uri};
524    use praxis_core::connectivity::Upstream;
525    use praxis_filter::{BodyBuffer, BodyMode, FilterPipeline, FilterRegistry};
526
527    use super::*;
528
529    #[test]
530    fn default_state_has_no_client_addr() {
531        let ctx = default_ctx();
532        assert!(ctx.client_addr.is_none(), "default client_addr should be None");
533    }
534
535    #[test]
536    fn default_state_has_no_cluster() {
537        let ctx = default_ctx();
538        assert!(ctx.cluster.is_none(), "default cluster should be None");
539    }
540
541    #[test]
542    fn default_state_has_zero_retries() {
543        let ctx = default_ctx();
544        assert_eq!(ctx.retries, 0, "default retries should be zero");
545    }
546
547    #[test]
548    fn default_state_flags_are_false() {
549        let ctx = default_ctx();
550        assert!(
551            !ctx.request_body_released,
552            "default request_body_released should be false"
553        );
554        assert!(
555            !ctx.response_body_released,
556            "default response_body_released should be false"
557        );
558        assert!(
559            !ctx.request_is_idempotent,
560            "default request_is_idempotent should be false"
561        );
562        assert!(!ctx.response_phase_done, "default response_phase_done should be false");
563    }
564
565    #[test]
566    fn default_state_buffers_are_none() {
567        let ctx = default_ctx();
568        assert!(
569            ctx.request_body_buffer.is_none(),
570            "default request_body_buffer should be None"
571        );
572        assert!(
573            ctx.response_body_buffer.is_none(),
574            "default response_body_buffer should be None"
575        );
576        assert!(ctx.pre_read_body.is_none(), "default pre_read_body should be None");
577    }
578
579    #[test]
580    fn default_state_snapshots_are_none() {
581        let ctx = default_ctx();
582        assert!(
583            ctx.request_snapshot.is_none(),
584            "default request_snapshot should be None"
585        );
586        assert!(ctx.upstream.is_none(), "default upstream should be None");
587        assert!(
588            ctx.upstream_for_retry.is_none(),
589            "default upstream_for_retry should be None"
590        );
591    }
592
593    #[test]
594    fn set_client_addr() {
595        let mut ctx = default_ctx();
596        let addr = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100));
597        ctx.client_addr = Some(addr);
598        assert_eq!(
599            ctx.client_addr.unwrap(),
600            addr,
601            "client_addr should match assigned value"
602        );
603    }
604
605    #[test]
606    fn set_cluster() {
607        let mut ctx = default_ctx();
608        ctx.cluster = Some(Arc::from("api-cluster"));
609        assert_eq!(
610            ctx.cluster.as_deref(),
611            Some("api-cluster"),
612            "cluster should match assigned value"
613        );
614    }
615
616    #[test]
617    fn set_upstream() {
618        let mut ctx = default_ctx();
619        let upstream = Upstream {
620            address: Arc::from("10.0.0.1:80"),
621            tls: None,
622            connection: Arc::new(praxis_core::connectivity::ConnectionOptions::default()),
623        };
624        ctx.upstream = Some(upstream.clone());
625        assert_eq!(
626            &*ctx.upstream.as_ref().unwrap().address,
627            "10.0.0.1:80",
628            "upstream address should match assigned value"
629        );
630    }
631
632    #[test]
633    fn increment_retries() {
634        let mut ctx = default_ctx();
635        ctx.retries += 1;
636        ctx.retries += 1;
637        assert_eq!(ctx.retries, 2, "retries should be 2 after two increments");
638    }
639
640    #[test]
641    fn release_request_body_flag() {
642        let mut ctx = default_ctx();
643        assert!(!ctx.request_body_released, "request_body_released should start false");
644        ctx.request_body_released = true;
645        assert!(
646            ctx.request_body_released,
647            "request_body_released should be true after setting"
648        );
649    }
650
651    #[test]
652    fn release_response_body_flag() {
653        let mut ctx = default_ctx();
654        assert!(!ctx.response_body_released, "response_body_released should start false");
655        ctx.response_body_released = true;
656        assert!(
657            ctx.response_body_released,
658            "response_body_released should be true after setting"
659        );
660    }
661
662    #[test]
663    fn response_phase_done_flag() {
664        let mut ctx = default_ctx();
665        assert!(!ctx.response_phase_done, "response_phase_done should start false");
666        ctx.response_phase_done = true;
667        assert!(
668            ctx.response_phase_done,
669            "response_phase_done should be true after setting"
670        );
671    }
672
673    #[test]
674    fn set_pre_read_body() {
675        let mut ctx = default_ctx();
676        let chunks = VecDeque::from([Bytes::from_static(b"chunk1"), Bytes::from_static(b"chunk2")]);
677        ctx.pre_read_body = Some(chunks);
678        let body = ctx.pre_read_body.as_ref().unwrap();
679        assert_eq!(body.len(), 2, "pre_read_body should contain 2 chunks");
680        assert_eq!(body[0], Bytes::from_static(b"chunk1"), "first chunk should be 'chunk1'");
681        assert_eq!(
682            body[1],
683            Bytes::from_static(b"chunk2"),
684            "second chunk should be 'chunk2'"
685        );
686    }
687
688    #[test]
689    fn set_request_snapshot() {
690        let mut ctx = default_ctx();
691        let snapshot = Request {
692            method: Method::POST,
693            uri: "/api/data".parse::<Uri>().unwrap(),
694            headers: HeaderMap::new(),
695        };
696        ctx.request_snapshot = Some(snapshot);
697        let snap = ctx.request_snapshot.as_ref().unwrap();
698        assert_eq!(snap.method, Method::POST, "snapshot method should be POST");
699        assert_eq!(snap.uri.path(), "/api/data", "snapshot URI path should be /api/data");
700    }
701
702    #[test]
703    fn request_body_buffer_lifecycle() {
704        let mut ctx = default_ctx();
705        let mut buf = BodyBuffer::new(100);
706        buf.push(Bytes::from_static(b"data")).unwrap();
707        ctx.request_body_buffer = Some(buf);
708
709        assert!(
710            ctx.request_body_buffer.is_some(),
711            "buffer should be present after assignment"
712        );
713        let taken = ctx.request_body_buffer.take().unwrap();
714        assert_eq!(
715            taken.freeze(),
716            Bytes::from_static(b"data"),
717            "frozen buffer should contain pushed data"
718        );
719        assert!(ctx.request_body_buffer.is_none(), "buffer should be None after take");
720    }
721
722    #[test]
723    fn default_request_body_mode_is_stream() {
724        let ctx = default_ctx();
725        assert_eq!(
726            ctx.request_body_mode,
727            BodyMode::Stream,
728            "default request_body_mode should be Stream"
729        );
730    }
731
732    #[test]
733    fn default_response_body_mode_is_stream() {
734        let ctx = default_ctx();
735        assert_eq!(
736            ctx.response_body_mode,
737            BodyMode::Stream,
738            "default response_body_mode should be Stream"
739        );
740    }
741
742    #[test]
743    fn set_request_body_mode() {
744        let mut ctx = default_ctx();
745        ctx.request_body_mode = BodyMode::StreamBuffer { max_bytes: Some(4096) };
746        assert_eq!(
747            ctx.request_body_mode,
748            BodyMode::StreamBuffer { max_bytes: Some(4096) },
749            "request_body_mode should match assigned value"
750        );
751    }
752
753    #[test]
754    fn set_response_body_mode() {
755        let mut ctx = default_ctx();
756        ctx.response_body_mode = BodyMode::StreamBuffer { max_bytes: Some(8192) };
757        assert_eq!(
758            ctx.response_body_mode,
759            BodyMode::StreamBuffer { max_bytes: Some(8192) },
760            "response_body_mode should match assigned value"
761        );
762    }
763
764    // -------------------------------------------------------------------------
765    // Metadata Roundtrip Tests
766    // -------------------------------------------------------------------------
767
768    #[test]
769    fn metadata_roundtrip_through_filter_context() {
770        let registry = FilterRegistry::with_builtins();
771        let pipeline = FilterPipeline::build(&mut [], &registry).unwrap();
772        let request = Request {
773            method: Method::GET,
774            uri: "/".parse::<Uri>().unwrap(),
775            headers: HeaderMap::new(),
776        };
777
778        let mut ctx = default_ctx();
779        ctx.filter_metadata.insert("rpc.method".to_owned(), "echo".to_owned());
780        ctx.filter_metadata.insert("rpc.status".to_owned(), "ok".to_owned());
781
782        let fctx = ctx.build_filter_context(&pipeline, &request, None);
783        assert_eq!(
784            fctx.get_metadata("rpc.method"),
785            Some("echo"),
786            "metadata written before build should survive into filter context"
787        );
788        assert_eq!(
789            fctx.get_metadata("rpc.status"),
790            Some("ok"),
791            "multiple metadata keys should round-trip"
792        );
793    }
794
795    #[test]
796    fn metadata_written_in_filter_context_persists_back() {
797        let registry = FilterRegistry::with_builtins();
798        let pipeline = FilterPipeline::build(&mut [], &registry).unwrap();
799        let request = Request {
800            method: Method::GET,
801            uri: "/".parse::<Uri>().unwrap(),
802            headers: HeaderMap::new(),
803        };
804
805        let mut ctx = default_ctx();
806        let mut fctx = ctx.build_filter_context(&pipeline, &request, None);
807        fctx.set_metadata("trace.id", "abc-123");
808        fctx.set_metadata("trace.span", "42");
809
810        ctx.filter_metadata = fctx.filter_metadata;
811        assert_eq!(
812            ctx.filter_metadata.get("trace.id").map(String::as_str),
813            Some("abc-123"),
814            "metadata set in filter context should persist back to protocol context"
815        );
816        assert_eq!(
817            ctx.filter_metadata.get("trace.span").map(String::as_str),
818            Some("42"),
819            "multiple metadata keys should persist back"
820        );
821    }
822
823    // -------------------------------------------------------------------------
824    // Hot-Reload Pipeline Pinning (via production helpers)
825    // -------------------------------------------------------------------------
826
827    #[test]
828    fn pin_pipeline_captures_current_arc() {
829        let registry = FilterRegistry::with_builtins();
830        let pipeline_a = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
831        let swap = arc_swap::ArcSwap::from(Arc::clone(&pipeline_a));
832
833        let mut ctx = default_ctx();
834        let pinned = ctx.pin_pipeline(&swap);
835        assert!(
836            Arc::ptr_eq(&pinned, &pipeline_a),
837            "pin_pipeline should return the current pipeline"
838        );
839        assert!(
840            Arc::ptr_eq(ctx.pinned_pipeline.as_ref().unwrap(), &pipeline_a),
841            "pinned_pipeline should be stored in ctx"
842        );
843    }
844
845    #[test]
846    fn pipeline_returns_pinned_after_reload() {
847        let registry = FilterRegistry::with_builtins();
848        let pipeline_a = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
849        let swap = arc_swap::ArcSwap::from(Arc::clone(&pipeline_a));
850
851        let mut ctx = default_ctx();
852        ctx.pin_pipeline(&swap);
853
854        let pipeline_b = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
855        swap.store(pipeline_b);
856
857        let later = ctx.pipeline(&swap);
858        assert!(
859            Arc::ptr_eq(&later, &pipeline_a),
860            "later hooks should still return pipeline A after reload"
861        );
862    }
863
864    #[test]
865    fn new_request_after_reload_pins_new_pipeline() {
866        let registry = FilterRegistry::with_builtins();
867        let pipeline_a = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
868        let swap = arc_swap::ArcSwap::from(Arc::clone(&pipeline_a));
869
870        let mut ctx_a = default_ctx();
871        ctx_a.pin_pipeline(&swap);
872
873        let pipeline_b = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
874        swap.store(Arc::clone(&pipeline_b));
875
876        let mut ctx_b = default_ctx();
877        ctx_b.pin_pipeline(&swap);
878
879        assert!(
880            Arc::ptr_eq(&ctx_a.pipeline(&swap), &pipeline_a),
881            "request A should use old pipeline"
882        );
883        assert!(
884            Arc::ptr_eq(&ctx_b.pipeline(&swap), &pipeline_b),
885            "request B should use new pipeline"
886        );
887    }
888
889    #[test]
890    fn old_pipeline_drops_after_ctx_drops() {
891        let registry = FilterRegistry::with_builtins();
892        let pipeline_a = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
893        let weak_a = Arc::downgrade(&pipeline_a);
894        let swap = arc_swap::ArcSwap::from(pipeline_a);
895
896        let mut ctx = default_ctx();
897        ctx.pin_pipeline(&swap);
898
899        swap.store(Arc::new(FilterPipeline::build(&mut [], &registry).unwrap()));
900
901        assert!(weak_a.upgrade().is_some(), "old pipeline alive while ctx exists");
902
903        drop(ctx);
904
905        assert!(weak_a.upgrade().is_none(), "old pipeline drops with ctx");
906    }
907
908    #[test]
909    fn pipeline_helper_returns_pinned_for_every_phase() {
910        let registry = FilterRegistry::with_builtins();
911        let pipeline_a = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
912        let swap = arc_swap::ArcSwap::from(Arc::clone(&pipeline_a));
913
914        let mut ctx = default_ctx();
915        ctx.pin_pipeline(&swap);
916
917        swap.store(Arc::new(FilterPipeline::build(&mut [], &registry).unwrap()));
918
919        for phase in ["request_body", "response", "response_body", "logging"] {
920            let p = ctx.pipeline(&swap);
921            assert!(
922                Arc::ptr_eq(&p, &pipeline_a),
923                "{phase}: should still return pinned pipeline A"
924            );
925        }
926    }
927
928    #[test]
929    fn pipeline_fallback_when_not_pinned() {
930        let registry = FilterRegistry::with_builtins();
931        let pipeline_a = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
932        let swap = arc_swap::ArcSwap::from(Arc::clone(&pipeline_a));
933
934        let ctx = default_ctx();
935
936        let loaded = ctx.pipeline(&swap);
937        assert!(
938            Arc::ptr_eq(&loaded, &pipeline_a),
939            "unpinned ctx should fall back to current ArcSwap value"
940        );
941    }
942
943    #[test]
944    fn pin_pipeline_is_idempotent_after_reload() {
945        let registry = FilterRegistry::with_builtins();
946        let pipeline_a = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
947        let swap = arc_swap::ArcSwap::from(Arc::clone(&pipeline_a));
948
949        let mut ctx = default_ctx();
950        ctx.pin_pipeline(&swap);
951
952        let pipeline_b = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
953        swap.store(pipeline_b);
954
955        let second_pin = ctx.pin_pipeline(&swap);
956        assert!(
957            Arc::ptr_eq(&second_pin, &pipeline_a),
958            "repeated pin_pipeline after reload should return the original pin"
959        );
960    }
961
962    #[test]
963    fn filter_state_isolated_across_pipelines_with_same_ids() {
964        let registry = FilterRegistry::with_builtins();
965        let pipeline_a = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
966        let pipeline_b = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
967
968        let request = Request {
969            method: Method::GET,
970            uri: "/".parse::<Uri>().unwrap(),
971            headers: HeaderMap::new(),
972        };
973
974        let mut ctx_a = default_ctx();
975        ctx_a.pinned_pipeline = Some(Arc::clone(&pipeline_a));
976        ctx_a.request_snapshot = Some(request.clone());
977        let mut fctx_a = ctx_a.build_filter_context(&pipeline_a, &request, None);
978        fctx_a.current_filter_id = Some(0);
979        fctx_a.insert_filter_state(String::from("from_pipeline_a"));
980        ctx_a.filter_state = fctx_a.filter_state;
981
982        let mut ctx_b = default_ctx();
983        ctx_b.pinned_pipeline = Some(Arc::clone(&pipeline_b));
984        ctx_b.request_snapshot = Some(request.clone());
985        let fctx_b = ctx_b.build_filter_context(&pipeline_b, &request, None);
986
987        assert!(
988            fctx_b.filter_state.is_empty(),
989            "request B should have its own empty state map"
990        );
991
992        let fctx_a2 = ctx_a.filter_context_for(&pipeline_a, None).unwrap();
993        assert_eq!(
994            fctx_a2.filter_state.get(&0).and_then(|v| v.downcast_ref::<String>()),
995            Some(&String::from("from_pipeline_a")),
996            "request A should still see its own state in a later phase"
997        );
998    }
999
1000    // -------------------------------------------------------------------------
1001    // Test Utilities
1002    // -------------------------------------------------------------------------
1003
1004    /// Create a default request context for tests.
1005    fn default_ctx() -> PingoraRequestCtx {
1006        PingoraRequestCtx::default()
1007    }
1008}
praxis_protocol/http/pingora/context.rs

praxis_protocol/http/pingora/
context.rs
