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

praxis_protocol/http/pingora/
context.rs
