sim_expr_tree_calc/calc/
refresh.rs

1use std::{
2    collections::{BTreeMap, BTreeSet},
3    error::Error,
4    fmt,
5    sync::Arc,
6};
7
8use sim_expr_tree_core::MountEpoch;
9use sim_table_core::TablePath;
10
11use super::*;
12
13/// One explicit observation of a mounted backend that lacks a watch contract.
14///
15/// Listing and stamp keys are canonical absolute expression-tree paths. Their
16/// numeric values are backend-owned monotone or content-derived stamps; the
17/// calculator compares them for equality and never interprets their magnitude.
18#[derive(Clone, Debug, Eq, PartialEq)]
19pub struct BackendRefreshSample {
20    /// Backend-wide generation.
21    pub epoch: MountEpoch,
22    /// Directory listing stamps by absolute directory path.
23    pub listings: BTreeMap<String, u64>,
24    /// Entry stamps by absolute entry path.
25    pub stamps: BTreeMap<String, u64>,
26}
27
28impl BackendRefreshSample {
29    /// Creates an epoch-only backend observation.
30    #[must_use]
31    pub fn new(epoch: MountEpoch) -> Self {
32        Self {
33            epoch,
34            listings: BTreeMap::new(),
35            stamps: BTreeMap::new(),
36        }
37    }
38
39    /// Adds a directory listing stamp.
40    #[must_use]
41    pub fn with_listing(mut self, path: impl Into<String>, stamp: u64) -> Self {
42        self.listings.insert(path.into(), stamp);
43        self
44    }
45
46    /// Adds an entry stamp.
47    #[must_use]
48    pub fn with_stamp(mut self, path: impl Into<String>, stamp: u64) -> Self {
49        self.stamps.insert(path.into(), stamp);
50        self
51    }
52}
53
54/// Explicit sampling contract for one mounted backend.
55///
56/// Backends with a native watch contract return `true` from
57/// [`Self::has_watch_contract`] and are skipped by [`ExprTreeCalc::refresh`].
58/// `sample` is called only by that explicit refresh operation.
59pub trait MountRefreshSource: Send + Sync {
60    /// Whether this backend already reports mutations through a watch contract.
61    fn has_watch_contract(&self) -> bool {
62        false
63    }
64
65    /// Samples the backend's epoch, relevant listings, and entry stamps.
66    fn sample(&self) -> Result<BackendRefreshSample, String>;
67}
68
69/// Evidence from one explicit refresh pass.
70#[derive(Clone, Debug, Default, Eq, PartialEq)]
71pub struct RefreshReport {
72    /// Mount paths actually sampled.
73    pub sampled_mounts: Vec<String>,
74    /// Watch-managed mount paths deliberately not polled.
75    pub watch_managed_mounts: Vec<String>,
76    /// Mount paths whose backend-wide epoch changed.
77    pub changed_epochs: Vec<String>,
78    /// Absolute directory paths whose listing stamp changed.
79    pub changed_listings: Vec<String>,
80    /// Absolute entry paths whose stamp changed.
81    pub changed_stamps: Vec<String>,
82    /// Number of distinct incremental observation keys invalidated.
83    pub invalidated_observations: usize,
84}
85
86/// A refresh registration, sample, or validation failure.
87#[derive(Clone, Debug, Eq, PartialEq)]
88pub enum RefreshError {
89    /// A sampler was attached to a path that is not mounted.
90    UnknownMount {
91        /// Canonical supplied mount path.
92        path: String,
93    },
94    /// A backend failed while explicitly sampled.
95    Sample {
96        /// Canonical mount path.
97        path: String,
98        /// Stable backend diagnostic.
99        message: String,
100    },
101    /// A backend returned a listing or stamp outside its mount.
102    InvalidSamplePath {
103        /// Canonical mount path.
104        mount: String,
105        /// Rejected sample path.
106        path: String,
107    },
108}
109
110impl fmt::Display for RefreshError {
111    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
112        match self {
113            Self::UnknownMount { path } => write!(f, "no mounted backend at {path}"),
114            Self::Sample { path, message } => {
115                write!(f, "cannot refresh mounted backend {path}: {message}")
116            }
117            Self::InvalidSamplePath { mount, path } => {
118                write!(f, "refresh sample path {path:?} is outside mount {mount}")
119            }
120        }
121    }
122}
123
124impl Error for RefreshError {}
125
126impl ExprTreeCalc {
127    /// Attaches an explicit refresh sampler to an existing mount.
128    ///
129    /// Registration does not sample the backend. The source remains idle until
130    /// [`Self::refresh`] is called.
131    pub fn attach_refresh_source(
132        &mut self,
133        path: &TablePath,
134        source: Arc<dyn MountRefreshSource>,
135    ) -> Result<(), RefreshError> {
136        let key = path_key(path);
137        if !self
138            .state
139            .read()
140            .expect("calc state poisoned")
141            .mounts
142            .contains_key(&key)
143        {
144            return Err(RefreshError::UnknownMount { path: key });
145        }
146        self.refresh_sources.insert(key, source);
147        Ok(())
148    }
149
150    /// Detaches and returns an explicit refresh sampler.
151    pub fn detach_refresh_source(
152        &mut self,
153        path: &TablePath,
154    ) -> Option<Arc<dyn MountRefreshSource>> {
155        self.refresh_sources.remove(&path_key(path))
156    }
157
158    /// Explicitly samples every non-watch mount and invalidates changed inputs.
159    ///
160    /// Sampling completes before graph or control state is mutated. If any
161    /// backend fails, the refresh is rejected atomically and no observation is
162    /// advanced.
163    pub fn refresh(&mut self) -> Result<RefreshReport, RefreshError> {
164        let mut report = RefreshReport::default();
165        let mut sampled = Vec::new();
166        for (mount, source) in &self.refresh_sources {
167            if source.has_watch_contract() {
168                report.watch_managed_mounts.push(mount.clone());
169                continue;
170            }
171            let sample = source.sample().map_err(|message| RefreshError::Sample {
172                path: mount.clone(),
173                message,
174            })?;
175            validate_sample_paths(mount, &sample)?;
176            report.sampled_mounts.push(mount.clone());
177            sampled.push((mount.clone(), sample));
178        }
179
180        let mut invalidated = BTreeSet::new();
181        let mut control_changed = false;
182        for (mount, sample) in sampled {
183            let previous = self
184                .refresh_samples
185                .get(&mount)
186                .cloned()
187                .unwrap_or_else(|| BackendRefreshSample::new(MountEpoch::default()));
188            if previous.epoch != sample.epoch {
189                report.changed_epochs.push(mount.clone());
190                invalidated.insert(CalcQuery::MountEpoch(mount.clone()));
191            }
192            for path in changed_paths(&previous.listings, &sample.listings) {
193                report.changed_listings.push(path.clone());
194                invalidated.insert(CalcQuery::Listing(path));
195            }
196            for path in changed_paths(&previous.stamps, &sample.stamps) {
197                report.changed_stamps.push(path.clone());
198                invalidated.insert(CalcQuery::LookupStep(path.clone()));
199                invalidated.insert(CalcQuery::NameSlot(path.clone()));
200                invalidated.insert(CalcQuery::Cell(path));
201            }
202            if previous != sample {
203                control_changed = true;
204                if let Some(state) = self
205                    .state
206                    .write()
207                    .expect("calc state poisoned")
208                    .mounts
209                    .get_mut(&mount)
210                {
211                    state.epoch = sample.epoch;
212                }
213                self.refresh_samples.insert(mount, sample);
214            }
215        }
216        if control_changed {
217            let mut state = self.state.write().expect("calc state poisoned");
218            bump_generation(&mut state.control_generation);
219        }
220        for query in &invalidated {
221            self.engine.invalidate(query);
222        }
223        report.invalidated_observations = invalidated.len();
224        if !invalidated.is_empty() {
225            self.schedule_dirty_automatic();
226        }
227        Ok(report)
228    }
229}
230
231fn changed_paths(before: &BTreeMap<String, u64>, after: &BTreeMap<String, u64>) -> Vec<String> {
232    before
233        .keys()
234        .chain(after.keys())
235        .collect::<BTreeSet<_>>()
236        .into_iter()
237        .filter(|path| before.get(*path) != after.get(*path))
238        .cloned()
239        .collect()
240}
241
242fn validate_sample_paths(mount: &str, sample: &BackendRefreshSample) -> Result<(), RefreshError> {
243    let mount_path =
244        TablePath::parse_absolute(mount).map_err(|_| RefreshError::InvalidSamplePath {
245            mount: mount.to_owned(),
246            path: mount.to_owned(),
247        })?;
248    for path in sample.listings.keys().chain(sample.stamps.keys()) {
249        let parsed =
250            TablePath::parse_absolute(path).map_err(|_| RefreshError::InvalidSamplePath {
251                mount: mount.to_owned(),
252                path: path.clone(),
253            })?;
254        if !is_path_prefix(&mount_path, &parsed) {
255            return Err(RefreshError::InvalidSamplePath {
256                mount: mount.to_owned(),
257                path: path.clone(),
258            });
259        }
260    }
261    Ok(())
262}
263
264fn is_path_prefix(candidate: &TablePath, path: &TablePath) -> bool {
265    candidate.segments().len() <= path.segments().len()
266        && candidate
267            .segments()
268            .iter()
269            .zip(path.segments())
270            .all(|(left, right)| left == right)
271}
sim_expr_tree_calc/calc/refresh.rs

sim_expr_tree_calc/calc/
refresh.rs
