praxis_filter/pipeline/
tcp.rs

1// SPDX-License-Identifier: MIT
2// Copyright (c) 2024 Praxis Contributors
3
4//! TCP pipeline execution: connect and disconnect filter phases.
5
6use tracing::trace;
7
8use super::{FilterPipeline, check_failure_mode};
9use crate::{FilterError, actions::FilterAction, any_filter::AnyFilter, tcp_filter::TcpFilterContext};
10
11// -----------------------------------------------------------------------------
12// FilterPipeline TCP
13// -----------------------------------------------------------------------------
14
15#[expect(
16    clippy::multiple_inherent_impl,
17    reason = "pipeline concerns are split across modules"
18)]
19impl FilterPipeline {
20    /// Run all TCP connect filters in order.
21    ///
22    /// TCP filters do not participate in branch chain
23    /// evaluation currently. TCP pipelines execute filters
24    /// sequentially without conditional branching or rejoin logic.
25    ///
26    /// # Errors
27    ///
28    /// Returns [`FilterError`] if any filter rejects or fails.
29    pub async fn execute_tcp_connect(&self, ctx: &mut TcpFilterContext<'_>) -> Result<FilterAction, FilterError> {
30        for pf in &self.filters {
31            let tcp_filter = match &pf.filter {
32                AnyFilter::Tcp(f) => f.as_ref(),
33                AnyFilter::Http(_) => continue,
34            };
35            trace!(filter = tcp_filter.name(), "on_connect");
36            match tcp_filter.on_connect(ctx).await {
37                Ok(FilterAction::Continue | FilterAction::Release | FilterAction::BodyDone) => {},
38                Ok(FilterAction::Reject(r)) => return Ok(FilterAction::Reject(r)),
39                Err(e) => {
40                    check_failure_mode(tcp_filter.name(), e, "tcp connect", pf.failure_mode)?;
41                },
42            }
43        }
44
45        Ok(FilterAction::Continue)
46    }
47
48    /// Run all TCP disconnect filters in reverse order.
49    ///
50    /// # Errors
51    ///
52    /// Returns [`FilterError`] if any filter fails.
53    pub async fn execute_tcp_disconnect(&self, ctx: &mut TcpFilterContext<'_>) -> Result<(), FilterError> {
54        for pf in self.filters.iter().rev() {
55            let tcp_filter = match &pf.filter {
56                AnyFilter::Tcp(f) => f.as_ref(),
57                AnyFilter::Http(_) => continue,
58            };
59            trace!(filter = tcp_filter.name(), "on_disconnect");
60            if let Err(e) = tcp_filter.on_disconnect(ctx).await {
61                check_failure_mode(tcp_filter.name(), e, "tcp disconnect", pf.failure_mode)?;
62            }
63        }
64
65        Ok(())
66    }
67}
68
69// -----------------------------------------------------------------------------
70// Tests
71// -----------------------------------------------------------------------------
72
73#[cfg(test)]
74#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
75#[allow(
76    clippy::unwrap_used,
77    clippy::expect_used,
78    clippy::indexing_slicing,
79    clippy::panic,
80    reason = "tests"
81)]
82mod tests {
83    use std::sync::{
84        Arc,
85        atomic::{AtomicUsize, Ordering},
86    };
87
88    use async_trait::async_trait;
89    use praxis_core::config::FailureMode;
90
91    use super::*;
92    use crate::{FilterError, FilterRegistry, Rejection, body::BodyCapabilities, tcp_filter::TcpFilter};
93    #[tokio::test]
94    async fn empty_pipeline_connect_continues() {
95        let registry = FilterRegistry::with_builtins();
96        let pipeline = FilterPipeline::build(&mut [], &registry).unwrap();
97        let mut ctx = make_ctx();
98        let action = pipeline.execute_tcp_connect(&mut ctx).await.unwrap();
99        assert!(matches!(action, FilterAction::Continue));
100    }
101
102    #[tokio::test]
103    async fn empty_pipeline_disconnect_succeeds() {
104        let registry = FilterRegistry::with_builtins();
105        let pipeline = FilterPipeline::build(&mut [], &registry).unwrap();
106        let mut ctx = make_ctx();
107        pipeline.execute_tcp_disconnect(&mut ctx).await.unwrap();
108    }
109
110    #[tokio::test]
111    async fn connect_runs_all_filters() {
112        let connects = Arc::new(AtomicUsize::new(0));
113        let disconnects = Arc::new(AtomicUsize::new(0));
114        let pipeline = make_tcp_pipeline(vec![
115            Box::new(CountingTcpFilter {
116                connects: Arc::clone(&connects),
117                disconnects: Arc::clone(&disconnects),
118            }),
119            Box::new(CountingTcpFilter {
120                connects: Arc::clone(&connects),
121                disconnects: Arc::clone(&disconnects),
122            }),
123        ]);
124        let mut ctx = make_ctx();
125        drop(pipeline.execute_tcp_connect(&mut ctx).await.unwrap());
126        assert_eq!(connects.load(Ordering::SeqCst), 2);
127    }
128
129    #[tokio::test]
130    async fn connect_stops_on_reject() {
131        let connects = Arc::new(AtomicUsize::new(0));
132        let disconnects = Arc::new(AtomicUsize::new(0));
133        let pipeline = make_tcp_pipeline(vec![
134            Box::new(RejectTcpFilter { status: 403 }),
135            Box::new(CountingTcpFilter {
136                connects: Arc::clone(&connects),
137                disconnects: Arc::clone(&disconnects),
138            }),
139        ]);
140        let mut ctx = make_ctx();
141        let action = pipeline.execute_tcp_connect(&mut ctx).await.unwrap();
142        assert!(matches!(action, FilterAction::Reject(r) if r.status == 403));
143        assert_eq!(connects.load(Ordering::SeqCst), 0);
144    }
145
146    #[tokio::test]
147    async fn connect_propagates_error() {
148        let pipeline = make_tcp_pipeline(vec![Box::new(ErrorTcpFilter)]);
149        let mut ctx = make_ctx();
150        let result = pipeline.execute_tcp_connect(&mut ctx).await;
151        assert!(result.is_err(), "error filter should propagate error");
152        assert!(
153            result.unwrap_err().to_string().contains("tcp filter error"),
154            "error message should contain filter error text"
155        );
156    }
157
158    #[tokio::test]
159    async fn connect_runs_in_forward_order() {
160        let log = Arc::new(std::sync::Mutex::new(Vec::new()));
161        let pipeline = make_tcp_pipeline(vec![
162            Box::new(OrderTcpFilter {
163                label: "first",
164                log: Arc::clone(&log),
165            }),
166            Box::new(OrderTcpFilter {
167                label: "second",
168                log: Arc::clone(&log),
169            }),
170            Box::new(OrderTcpFilter {
171                label: "third",
172                log: Arc::clone(&log),
173            }),
174        ]);
175        let mut ctx = make_ctx();
176        drop(pipeline.execute_tcp_connect(&mut ctx).await.unwrap());
177        let recorded = log.lock().unwrap().clone();
178        assert_eq!(recorded, vec!["first", "second", "third"]);
179    }
180
181    #[tokio::test]
182    async fn disconnect_runs_in_reverse_order() {
183        let log = Arc::new(std::sync::Mutex::new(Vec::new()));
184        let pipeline = make_tcp_pipeline(vec![
185            Box::new(OrderTcpFilter {
186                label: "first",
187                log: Arc::clone(&log),
188            }),
189            Box::new(OrderTcpFilter {
190                label: "second",
191                log: Arc::clone(&log),
192            }),
193            Box::new(OrderTcpFilter {
194                label: "third",
195                log: Arc::clone(&log),
196            }),
197        ]);
198        let mut ctx = make_ctx();
199        pipeline.execute_tcp_disconnect(&mut ctx).await.unwrap();
200        let recorded = log.lock().unwrap().clone();
201        assert_eq!(recorded, vec!["third", "second", "first"]);
202    }
203
204    #[tokio::test]
205    async fn disconnect_all_filters_called() {
206        let connects = Arc::new(AtomicUsize::new(0));
207        let disconnects = Arc::new(AtomicUsize::new(0));
208        let pipeline = make_tcp_pipeline(vec![
209            Box::new(CountingTcpFilter {
210                connects: Arc::clone(&connects),
211                disconnects: Arc::clone(&disconnects),
212            }),
213            Box::new(CountingTcpFilter {
214                connects: Arc::clone(&connects),
215                disconnects: Arc::clone(&disconnects),
216            }),
217        ]);
218        let mut ctx = make_ctx();
219        pipeline.execute_tcp_disconnect(&mut ctx).await.unwrap();
220        assert_eq!(disconnects.load(Ordering::SeqCst), 2);
221    }
222
223    #[tokio::test]
224    async fn connect_failure_mode_open_continues() {
225        let connects = Arc::new(AtomicUsize::new(0));
226        let disconnects = Arc::new(AtomicUsize::new(0));
227        let mut pipeline = make_tcp_pipeline(vec![
228            Box::new(ErrorTcpFilter),
229            Box::new(CountingTcpFilter {
230                connects: Arc::clone(&connects),
231                disconnects: Arc::clone(&disconnects),
232            }),
233        ]);
234        pipeline.filters[0].failure_mode = FailureMode::Open;
235        let mut ctx = make_ctx();
236
237        let action = pipeline.execute_tcp_connect(&mut ctx).await.unwrap();
238
239        assert!(
240            matches!(action, FilterAction::Continue),
241            "open failure_mode should continue past error"
242        );
243        assert_eq!(
244            connects.load(Ordering::SeqCst),
245            1,
246            "second filter should still run after open error"
247        );
248    }
249
250    #[tokio::test]
251    async fn connect_failure_mode_closed_propagates() {
252        let pipeline = make_tcp_pipeline(vec![Box::new(ErrorTcpFilter)]);
253        let mut ctx = make_ctx();
254
255        let result = pipeline.execute_tcp_connect(&mut ctx).await;
256
257        assert!(result.is_err(), "closed (default) failure_mode should propagate error");
258    }
259
260    #[tokio::test]
261    async fn disconnect_failure_mode_open_continues() {
262        let connects = Arc::new(AtomicUsize::new(0));
263        let disconnects = Arc::new(AtomicUsize::new(0));
264        let mut pipeline = make_tcp_pipeline(vec![
265            Box::new(CountingTcpFilter {
266                connects: Arc::clone(&connects),
267                disconnects: Arc::clone(&disconnects),
268            }),
269            Box::new(ErrorDisconnectTcpFilter),
270        ]);
271        pipeline.filters[1].failure_mode = FailureMode::Open;
272        let mut ctx = make_ctx();
273
274        pipeline.execute_tcp_disconnect(&mut ctx).await.unwrap();
275
276        assert_eq!(
277            disconnects.load(Ordering::SeqCst),
278            1,
279            "first filter disconnect should still run"
280        );
281    }
282
283    #[tokio::test]
284    async fn disconnect_failure_mode_closed_propagates() {
285        let pipeline = make_tcp_pipeline(vec![Box::new(ErrorDisconnectTcpFilter)]);
286        let mut ctx = make_ctx();
287
288        let result = pipeline.execute_tcp_disconnect(&mut ctx).await;
289
290        assert!(
291            result.is_err(),
292            "closed (default) failure_mode should propagate disconnect error"
293        );
294    }
295
296    #[tokio::test]
297    async fn connect_mixed_chain_open_skipped_closed_blocks() {
298        let connects = Arc::new(AtomicUsize::new(0));
299        let disconnects = Arc::new(AtomicUsize::new(0));
300        let mut pipeline = make_tcp_pipeline(vec![
301            Box::new(ErrorTcpFilter),
302            Box::new(ErrorTcpFilter),
303            Box::new(CountingTcpFilter {
304                connects: Arc::clone(&connects),
305                disconnects: Arc::clone(&disconnects),
306            }),
307        ]);
308        pipeline.filters[0].failure_mode = FailureMode::Open;
309        let mut ctx = make_ctx();
310
311        let result = pipeline.execute_tcp_connect(&mut ctx).await;
312
313        assert!(
314            result.is_err(),
315            "open filter should be skipped but closed filter should block"
316        );
317        assert_eq!(
318            connects.load(Ordering::SeqCst),
319            0,
320            "third filter should not run after closed error"
321        );
322    }
323
324    #[tokio::test]
325    async fn http_filters_skipped_in_tcp_execution() {
326        let registry = FilterRegistry::with_builtins();
327        let mut entries = vec![crate::FilterEntry {
328            branch_chains: None,
329            filter_type: "router".into(),
330            config: serde_yaml::from_str("routes: []").unwrap(),
331            conditions: vec![],
332            name: None,
333            response_conditions: vec![],
334            failure_mode: FailureMode::default(),
335        }];
336        let pipeline = FilterPipeline::build(&mut entries, &registry).unwrap();
337        let mut ctx = make_ctx();
338
339        let action = pipeline.execute_tcp_connect(&mut ctx).await.unwrap();
340        assert!(
341            matches!(action, FilterAction::Continue),
342            "TCP connect should skip HTTP filters"
343        );
344        pipeline.execute_tcp_disconnect(&mut ctx).await.unwrap();
345    }
346
347    // -------------------------------------------------------------------------
348    // Test Utilities
349    // -------------------------------------------------------------------------
350
351    /// TCP filter that counts connect and disconnect calls.
352    struct CountingTcpFilter {
353        connects: Arc<AtomicUsize>,
354        disconnects: Arc<AtomicUsize>,
355    }
356
357    #[async_trait]
358    impl TcpFilter for CountingTcpFilter {
359        fn name(&self) -> &'static str {
360            "counting_tcp"
361        }
362
363        async fn on_connect(&self, _ctx: &mut TcpFilterContext<'_>) -> Result<FilterAction, FilterError> {
364            self.connects.fetch_add(1, Ordering::SeqCst);
365            Ok(FilterAction::Continue)
366        }
367
368        async fn on_disconnect(&self, _ctx: &mut TcpFilterContext<'_>) -> Result<(), FilterError> {
369            self.disconnects.fetch_add(1, Ordering::SeqCst);
370            Ok(())
371        }
372    }
373
374    /// TCP filter that rejects connections with a given status.
375    struct RejectTcpFilter {
376        status: u16,
377    }
378
379    #[async_trait]
380    impl TcpFilter for RejectTcpFilter {
381        fn name(&self) -> &'static str {
382            "reject_tcp"
383        }
384
385        async fn on_connect(&self, _ctx: &mut TcpFilterContext<'_>) -> Result<FilterAction, FilterError> {
386            Ok(FilterAction::Reject(Rejection::status(self.status)))
387        }
388    }
389
390    /// TCP filter that always returns an error on connect.
391    struct ErrorTcpFilter;
392
393    #[async_trait]
394    impl TcpFilter for ErrorTcpFilter {
395        fn name(&self) -> &'static str {
396            "error_tcp"
397        }
398
399        async fn on_connect(&self, _ctx: &mut TcpFilterContext<'_>) -> Result<FilterAction, FilterError> {
400            Err("tcp filter error".into())
401        }
402    }
403
404    /// TCP filter that always errors on disconnect.
405    struct ErrorDisconnectTcpFilter;
406
407    #[async_trait]
408    impl TcpFilter for ErrorDisconnectTcpFilter {
409        fn name(&self) -> &'static str {
410            "error_disconnect_tcp"
411        }
412
413        async fn on_disconnect(&self, _ctx: &mut TcpFilterContext<'_>) -> Result<(), FilterError> {
414            Err("tcp disconnect error".into())
415        }
416    }
417
418    /// TCP filter that records its label to a shared log.
419    struct OrderTcpFilter {
420        label: &'static str,
421        log: Arc<std::sync::Mutex<Vec<&'static str>>>,
422    }
423
424    #[async_trait]
425    impl TcpFilter for OrderTcpFilter {
426        fn name(&self) -> &'static str {
427            self.label
428        }
429
430        async fn on_connect(&self, _ctx: &mut TcpFilterContext<'_>) -> Result<FilterAction, FilterError> {
431            self.log.lock().unwrap().push(self.label);
432            Ok(FilterAction::Continue)
433        }
434
435        async fn on_disconnect(&self, _ctx: &mut TcpFilterContext<'_>) -> Result<(), FilterError> {
436            self.log.lock().unwrap().push(self.label);
437            Ok(())
438        }
439    }
440
441    /// Build a [`FilterPipeline`] wrapping the given TCP filters.
442    fn make_tcp_pipeline(filters: Vec<Box<dyn TcpFilter>>) -> FilterPipeline {
443        let filters: Vec<_> = filters
444            .into_iter()
445            .enumerate()
446            .map(|(i, f)| crate::pipeline::filter::PipelineFilter::new(i, AnyFilter::Tcp(f), vec![], vec![]))
447            .collect();
448        FilterPipeline {
449            body_capabilities: BodyCapabilities::default(),
450            compression: None,
451            filters,
452            health_registry: None,
453            id_generator: Arc::new(praxis_core::id::IdGenerator::with_seed(0)),
454            kv_stores: None,
455            pipeline_extensions: Vec::new(),
456            time_source: Arc::new(praxis_core::time::SystemTimeSource),
457        }
458    }
459
460    /// Build a default [`TcpFilterContext`] for testing.
461    fn make_ctx() -> TcpFilterContext<'static> {
462        TcpFilterContext {
463            remote_addr: "127.0.0.1:12345",
464            local_addr: "0.0.0.0:8080",
465            sni: None,
466            upstream_addr: Some(std::borrow::Cow::Borrowed("10.0.0.1:80")),
467            cluster: None,
468            health_registry: None,
469            kv_stores: None,
470            connect_time: std::time::Instant::now(),
471            bytes_in: 0,
472            bytes_out: 0,
473        }
474    }
475}
praxis_filter/pipeline/tcp.rs

praxis_filter/pipeline/
tcp.rs
