praxis_protocol/
pipelines.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2024 Praxis Contributors
3
4//! Hot-swappable pipeline storage for protocol adapters.
5//!
6//! [`ListenerPipelines`] lives in the protocol crate because it is
7//! the interface between protocol adapters (which invoke filter
8//! execution) and the filter engine (which owns [`FilterPipeline`]).
9//!
10//! Each pipeline is wrapped in [`ArcSwap`] for lock-free atomic
11//! replacement during hot reloads. In-flight requests hold an
12//! [`Arc`] guard to the old pipeline, so they drain safely while
13//! new requests pick up the replacement.
14//!
15//! [`FilterPipeline`]: praxis_filter::FilterPipeline
16//! [`ArcSwap`]: arc_swap::ArcSwap
17
18use std::{collections::HashMap, sync::Arc};
19
20use arc_swap::ArcSwap;
21use praxis_filter::FilterPipeline;
22
23// -----------------------------------------------------------------------------
24// ListenerPipelines
25// -----------------------------------------------------------------------------
26
27/// Maps listener names to their resolved [`FilterPipeline`]s.
28///
29/// Each pipeline is wrapped in [`ArcSwap`] so it can be atomically
30/// replaced at runtime without blocking in-flight requests.
31///
32/// ```
33/// use std::{collections::HashMap, sync::Arc};
34///
35/// use praxis_filter::{FilterPipeline, FilterRegistry};
36/// use praxis_protocol::ListenerPipelines;
37///
38/// let registry = FilterRegistry::with_builtins();
39/// let pipeline = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
40///
41/// let mut map = HashMap::new();
42/// map.insert("web".to_owned(), pipeline);
43/// let pipelines = ListenerPipelines::new(map);
44///
45/// assert!(pipelines.get("web").is_some());
46/// assert!(pipelines.get("missing").is_none());
47/// ```
48///
49/// [`ArcSwap`]: arc_swap::ArcSwap
50pub struct ListenerPipelines {
51    /// Maps listener names to their swappable filter pipelines.
52    pipelines: HashMap<String, Arc<ArcSwap<FilterPipeline>>>,
53}
54
55impl ListenerPipelines {
56    /// Create from a map of listener name to pipeline.
57    pub fn new(pipelines: HashMap<String, Arc<FilterPipeline>>) -> Self {
58        let swappable = pipelines
59            .into_iter()
60            .map(|(name, p)| (name, Arc::new(ArcSwap::from(p))))
61            .collect();
62        Self { pipelines: swappable }
63    }
64
65    /// Get the swappable pipeline for a listener by name.
66    pub fn get(&self, listener_name: &str) -> Option<&Arc<ArcSwap<FilterPipeline>>> {
67        self.pipelines.get(listener_name)
68    }
69
70    /// Atomically replace the pipeline for a listener.
71    ///
72    /// No-op if the listener name is not present.
73    ///
74    /// ```
75    /// use std::{collections::HashMap, sync::Arc};
76    ///
77    /// use praxis_filter::{FilterPipeline, FilterRegistry};
78    /// use praxis_protocol::ListenerPipelines;
79    ///
80    /// let registry = FilterRegistry::with_builtins();
81    /// let old = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
82    /// let new = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
83    ///
84    /// let mut map = HashMap::new();
85    /// map.insert("web".to_owned(), old);
86    /// let pipelines = ListenerPipelines::new(map);
87    ///
88    /// pipelines.swap("web", new);
89    /// pipelines.swap(
90    ///     "nonexistent",
91    ///     Arc::new(FilterPipeline::build(&mut [], &registry).unwrap()),
92    /// );
93    /// ```
94    pub fn swap(&self, listener_name: &str, new_pipeline: Arc<FilterPipeline>) {
95        if let Some(slot) = self.pipelines.get(listener_name) {
96            slot.store(new_pipeline);
97        }
98    }
99
100    /// Every filesystem path any filter in any listener's pipeline reads
101    /// configuration from, de-duplicated.
102    ///
103    /// Two listeners can share a filter chain, so the same document would
104    /// otherwise appear more than once and be watched and hashed repeatedly.
105    pub fn referenced_files(&self) -> Vec<std::path::PathBuf> {
106        let mut seen = std::collections::BTreeSet::new();
107        for name in self.listener_names() {
108            if let Some(slot) = self.get(name) {
109                for path in slot.load().referenced_files() {
110                    seen.insert(path);
111                }
112            }
113        }
114        seen.into_iter().collect()
115    }
116
117    /// Returns an iterator over listener names.
118    pub fn listener_names(&self) -> impl Iterator<Item = &str> {
119        self.pipelines.keys().map(String::as_str)
120    }
121}
122
123// -----------------------------------------------------------------------------
124// Tests
125// -----------------------------------------------------------------------------
126
127#[cfg(test)]
128#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
129#[allow(
130    clippy::unwrap_used,
131    clippy::expect_used,
132    clippy::indexing_slicing,
133    clippy::too_many_lines,
134    reason = "tests"
135)]
136mod tests {
137    use praxis_filter::FilterRegistry;
138
139    use super::*;
140
141    #[test]
142    fn get_returns_pipeline() {
143        let pipelines = make_pipelines(&["web"]);
144        assert!(pipelines.get("web").is_some(), "should find 'web' pipeline");
145    }
146
147    #[test]
148    fn get_returns_none_for_missing() {
149        let pipelines = make_pipelines(&["web"]);
150        assert!(pipelines.get("missing").is_none(), "should return None for missing");
151    }
152
153    #[test]
154    fn swap_replaces_pipeline_pointer() {
155        let pipelines = make_pipelines(&["web"]);
156        let old_ptr = Arc::as_ptr(&pipelines.get("web").unwrap().load());
157
158        let registry = FilterRegistry::with_builtins();
159        let new_pipeline = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
160        pipelines.swap("web", Arc::clone(&new_pipeline));
161
162        let new_ptr = Arc::as_ptr(&pipelines.get("web").unwrap().load());
163        assert_ne!(old_ptr, new_ptr, "swap should replace the pipeline pointer");
164    }
165
166    #[test]
167    fn old_guard_remains_valid_after_swap() {
168        let pipelines = make_pipelines(&["web"]);
169        let old_guard = pipelines.get("web").unwrap().load();
170        let old_ptr = Arc::as_ptr(&old_guard);
171
172        let registry = FilterRegistry::with_builtins();
173        let new_pipeline = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
174        pipelines.swap("web", new_pipeline);
175
176        let still_old_ptr = Arc::as_ptr(&old_guard);
177        assert_eq!(
178            old_ptr, still_old_ptr,
179            "old guard should still point to the original pipeline"
180        );
181    }
182
183    #[test]
184    fn swap_nonexistent_is_noop() {
185        let pipelines = make_pipelines(&["web"]);
186        let registry = FilterRegistry::with_builtins();
187        let new_pipeline = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
188        pipelines.swap("nonexistent", new_pipeline);
189        assert!(pipelines.get("web").is_some(), "existing pipeline should be unaffected");
190    }
191
192    #[test]
193    fn get_returns_arcswap_reference() {
194        let pipelines = make_pipelines(&["web"]);
195        let slot: &Arc<ArcSwap<FilterPipeline>> = pipelines.get("web").unwrap();
196        let _loaded: arc_swap::Guard<Arc<FilterPipeline>> = slot.load();
197    }
198
199    #[test]
200    fn referenced_files_empty_without_listeners() {
201        let pipelines = make_pipelines(&[]);
202        assert!(
203            pipelines.referenced_files().is_empty(),
204            "no listeners means no referenced documents"
205        );
206    }
207
208    #[test]
209    fn referenced_files_empty_when_no_filter_declares_one() {
210        let pipelines = make_pipelines(&["web"]);
211        assert!(
212            pipelines.referenced_files().is_empty(),
213            "a pipeline of non-declaring filters contributes nothing"
214        );
215    }
216
217    #[test]
218    fn referenced_files_collects_across_listeners() {
219        let pipelines =
220            make_pipelines_with_documents(&[("web", "/etc/praxis/web.yaml"), ("api", "/etc/praxis/api.yaml")]);
221        assert_eq!(
222            pipelines.referenced_files(),
223            vec![
224                std::path::PathBuf::from("/etc/praxis/api.yaml"),
225                std::path::PathBuf::from("/etc/praxis/web.yaml"),
226            ],
227            "every listener's documents must be collected, sorted by the BTreeSet"
228        );
229    }
230
231    #[test]
232    fn referenced_files_dedupes_a_document_shared_by_two_listeners() {
233        let shared = "/etc/praxis/shared.yaml";
234        let pipelines = make_pipelines_with_documents(&[("web", shared), ("api", shared)]);
235        assert_eq!(
236            pipelines.referenced_files(),
237            vec![std::path::PathBuf::from(shared)],
238            "a shared document must appear once"
239        );
240    }
241
242    // -------------------------------------------------------------------------
243    // Test Utilities
244    // -------------------------------------------------------------------------
245
246    /// A filter that declares the document named by its `document:` config key.
247    struct DocumentReaderFilter {
248        document: std::path::PathBuf,
249    }
250
251    #[async_trait::async_trait]
252    impl praxis_filter::HttpFilter for DocumentReaderFilter {
253        fn name(&self) -> &'static str {
254            "document_reader"
255        }
256
257        fn referenced_files(&self) -> Vec<std::path::PathBuf> {
258            vec![self.document.clone()]
259        }
260
261        async fn on_request(
262            &self,
263            _ctx: &mut praxis_filter::HttpFilterContext<'_>,
264        ) -> Result<praxis_filter::FilterAction, praxis_filter::FilterError> {
265            Ok(praxis_filter::FilterAction::Continue)
266        }
267    }
268
269    /// Build [`ListenerPipelines`] where each named listener runs a single filter
270    /// declaring the given document.
271    fn make_pipelines_with_documents(listeners: &[(&str, &str)]) -> ListenerPipelines {
272        let mut registry = FilterRegistry::with_builtins();
273        registry
274            .register(
275                "document_reader",
276                praxis_filter::FilterFactory::Http(Arc::new(|cfg: &serde_yaml::Value| {
277                    let document = cfg
278                        .get("document")
279                        .and_then(serde_yaml::Value::as_str)
280                        .ok_or_else(|| praxis_filter::FilterError::from("document_reader: missing document"))?;
281                    let filter: Box<dyn praxis_filter::HttpFilter> = Box::new(DocumentReaderFilter {
282                        document: std::path::PathBuf::from(document),
283                    });
284                    Ok(filter)
285                })),
286            )
287            .unwrap();
288
289        let mut map = HashMap::new();
290        for (listener, document) in listeners {
291            let yaml = format!("- filter: document_reader\n  document: {document}\n");
292            let mut entries: Vec<praxis_core::config::FilterEntry> = serde_yaml::from_str(&yaml).unwrap();
293            let pipeline = Arc::new(FilterPipeline::build(&mut entries, &registry).unwrap());
294            map.insert((*listener).to_owned(), pipeline);
295        }
296        ListenerPipelines::new(map)
297    }
298
299    /// Build [`ListenerPipelines`] with empty pipelines for the given names.
300    fn make_pipelines(names: &[&str]) -> ListenerPipelines {
301        let registry = FilterRegistry::with_builtins();
302        let mut map = HashMap::new();
303        for name in names {
304            let pipeline = Arc::new(FilterPipeline::build(&mut [], &registry).unwrap());
305            map.insert((*name).to_owned(), pipeline);
306        }
307        ListenerPipelines::new(map)
308    }
309}
praxis_protocol/pipelines.rs

praxis_protocol/
pipelines.rs
