amethystate_core/primitives/
intercept.rs

1use crate::path::StorePath;
2use std::cell::RefCell;
3use std::collections::HashMap;
4use std::sync::Arc;
5
6pub(crate) const MAX_INTERCEPT_DEPTH: usize = 10;
7
8/// How deep the interceptors of one field or map are running inside each
9/// other, counted on each thread by itself.
10///
11/// The limit is there for an interceptor that writes back into what it
12/// guards, which is one thread calling itself. Writers on two threads at once
13/// are not nested, and counting them together refuses the eleventh writer of a
14/// busy map for a recursion nobody made.
15#[derive(Clone, Default)]
16pub struct InterceptDepth(Arc<()>);
17
18thread_local! {
19    static NESTED: RefCell<HashMap<usize, usize>> = RefCell::new(HashMap::new());
20}
21
22impl InterceptDepth {
23    /// Which field or map this is, for as long as a clone of it is alive -
24    /// and a guard holds one.
25    fn key(&self) -> usize {
26        Arc::as_ptr(&self.0) as usize
27    }
28
29    fn deepen(&self, levels: usize) -> usize {
30        NESTED.with(|nested| {
31            let mut nested = nested.borrow_mut();
32            let at = nested.entry(self.key()).or_insert(0);
33            let was = *at;
34            *at += levels;
35            was
36        })
37    }
38
39    fn rise(&self, levels: usize) {
40        let _ = NESTED.try_with(|nested| {
41            let mut nested = nested.borrow_mut();
42            if let Some(at) = nested.get_mut(&self.key()) {
43                *at = at.saturating_sub(levels);
44                if *at == 0 {
45                    nested.remove(&self.key());
46                }
47            }
48        });
49    }
50
51    /// Stands this thread `levels` deep until the guard goes, as if that many
52    /// interceptors were running inside each other.
53    #[doc(hidden)]
54    pub fn nested_on_this_thread(&self, levels: usize) -> InterceptGuard {
55        self.deepen(levels);
56        InterceptGuard {
57            depth: self.clone(),
58            levels,
59        }
60    }
61}
62
63impl std::fmt::Debug for InterceptDepth {
64    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
65        f.write_str("InterceptDepth")
66    }
67}
68
69/// Why a change did not get past the interceptors.
70#[derive(Debug, Clone, PartialEq, Eq)]
71pub enum Refusal {
72    /// One of them turned it down, in these words.
73    Said(String),
74
75    /// They wrote back into what they guard deeper than a write may nest.
76    Recursed,
77
78    /// A rule the field was declared with turned it down, in these words.
79    Ruled(String),
80}
81
82pub struct InterceptGuard {
83    depth: InterceptDepth,
84    levels: usize,
85}
86
87impl InterceptGuard {
88    pub(crate) fn enter(depth: &InterceptDepth, path: StorePath) -> Option<Self> {
89        let prev = depth.deepen(1);
90        if prev >= MAX_INTERCEPT_DEPTH {
91            depth.rise(1);
92            tracing::warn!(
93                target: "amethystate::intercept",
94                path = %path,
95                depth = prev + 1,
96                "maximum intercept depth reached, skipping execution"
97            );
98            None
99        } else {
100            Some(Self {
101                depth: depth.clone(),
102                levels: 1,
103            })
104        }
105    }
106}
107
108impl Drop for InterceptGuard {
109    fn drop(&mut self) {
110        self.depth.rise(self.levels);
111    }
112}
113
114pub struct InterceptDisposer {
115    pub id: u64,
116    pub path: StorePath,
117    pub(crate) cleanup: Arc<dyn Fn(u64) + Send + Sync + 'static>,
118}
119
120impl InterceptDisposer {
121    pub fn remove(self) {
122        (self.cleanup)(self.id);
123    }
124}
125
126impl std::fmt::Debug for InterceptDisposer {
127    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
128        f.debug_struct("InterceptDisposer")
129            .field("id", &self.id)
130            .field("path", &self.path)
131            .finish()
132    }
133}
134
135#[cfg(all(test, not(target_arch = "wasm32")))]
136mod tests {
137    use super::*;
138    use crate::primitives::field_core::FieldCore;
139    use std::time::Duration;
140
141    #[test]
142    fn writers_on_many_threads_at_once_are_not_a_recursion() {
143        let core = FieldCore::new(0u32);
144        let _slow = core.intercept(StorePath::segment("busy"), |change| {
145            std::thread::sleep(Duration::from_millis(300));
146            Some(change)
147        });
148
149        let writers: Vec<_> = (0..MAX_INTERCEPT_DEPTH as u32 + 2)
150            .map(|n| {
151                let core = core.clone();
152                std::thread::spawn(move || {
153                    core.run_interceptors(StorePath::segment("busy"), n, None)
154                        .map(|_| ())
155                })
156            })
157            .collect();
158
159        let refused: Vec<_> = writers
160            .into_iter()
161            .map(|writer| writer.join().unwrap())
162            .filter(Result::is_err)
163            .collect();
164        assert_eq!(refused, Vec::new());
165    }
166}
amethystate_core/primitives/intercept.rs

amethystate_core/primitives/
intercept.rs
