sim_expr_tree_core/
store.rs

1use std::collections::BTreeMap;
2
3use sim_table_core::TablePath;
4
5use crate::{CellId, DirId, EffectiveCodecPolicy};
6
7/// Backend family named by a mounted expression-tree store descriptor.
8#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
9pub enum BackendKind {
10    /// Process-local memory backend.
11    Memory,
12    /// Filesystem-backed Table/Dir backend.
13    Filesystem,
14    /// Database-backed Table/Dir backend.
15    Database,
16    /// Read-only backend wrapper.
17    ReadOnly,
18    /// Already-composed mounted namespace backend.
19    MountedNamespace,
20}
21
22/// Monotonic observation of a mounted backend generation.
23#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
24pub struct MountEpoch(u64);
25
26impl MountEpoch {
27    /// Create an epoch from a backend supplied generation.
28    pub fn new(value: u64) -> Self {
29        Self(value)
30    }
31
32    /// Borrow the raw generation value.
33    pub fn value(self) -> u64 {
34        self.0
35    }
36
37    /// Return the next observed generation.
38    pub fn next_after(self) -> Self {
39        Self(self.0.saturating_add(1))
40    }
41}
42
43/// Mounted target shape. Table mounts are leaves; Dir mounts can be traversed.
44#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
45pub enum MountResource {
46    /// A mounted Table leaf.
47    Table,
48    /// A mounted Dir subtree.
49    Dir,
50}
51
52/// Explicit mount descriptor stored outside authored source cells.
53#[derive(Clone, Debug, PartialEq, Eq)]
54pub struct MountDescriptor {
55    path: TablePath,
56    resource: MountResource,
57    backend: BackendKind,
58    epoch: MountEpoch,
59}
60
61impl MountDescriptor {
62    /// Create an explicit Table mount descriptor.
63    pub fn table(path: TablePath, backend: BackendKind, epoch: MountEpoch) -> Self {
64        Self {
65            path,
66            resource: MountResource::Table,
67            backend,
68            epoch,
69        }
70    }
71
72    /// Create an explicit Dir mount descriptor.
73    pub fn dir(path: TablePath, backend: BackendKind, epoch: MountEpoch) -> Self {
74        Self {
75            path,
76            resource: MountResource::Dir,
77            backend,
78            epoch,
79        }
80    }
81
82    /// Absolute mount path.
83    pub fn path(&self) -> &TablePath {
84        &self.path
85    }
86
87    /// Mounted target shape.
88    pub fn resource(&self) -> MountResource {
89        self.resource
90    }
91
92    /// Mounted backend family.
93    pub fn backend(&self) -> BackendKind {
94        self.backend
95    }
96
97    /// Last observed backend epoch.
98    pub fn epoch(&self) -> MountEpoch {
99        self.epoch
100    }
101
102    fn set_epoch(&mut self, epoch: MountEpoch) {
103        self.epoch = epoch;
104    }
105}
106
107/// Authored source-store entry. Operational and derived values have no variants here.
108#[derive(Clone, Debug, PartialEq, Eq)]
109pub struct SourceEntry {
110    expr: String,
111    codec: Option<String>,
112}
113
114impl SourceEntry {
115    /// Create an authored expression entry.
116    pub fn new(expr: impl Into<String>) -> Self {
117        Self {
118            expr: expr.into(),
119            codec: None,
120        }
121    }
122
123    /// Attach the source codec used to parse the authored expression.
124    pub fn with_codec(mut self, codec: impl Into<String>) -> Self {
125        self.codec = Some(codec.into());
126        self
127    }
128
129    /// Authored expression text.
130    pub fn expr(&self) -> &str {
131        &self.expr
132    }
133
134    /// Optional source codec.
135    pub fn codec(&self) -> Option<&str> {
136        self.codec.as_deref()
137    }
138}
139
140/// Control-store entry for operational state.
141#[derive(Clone, Debug, PartialEq, Eq)]
142pub enum ControlEntry {
143    /// Durable generated-name or scheduler counter.
144    Counter(u64),
145    /// Effective policy snapshot.
146    Policy(EffectiveCodecPolicy),
147    /// UI preference kept out of authored source.
148    UiPreference(String),
149    /// Last observed mount backend epoch.
150    MountEpoch(MountEpoch),
151}
152
153/// Derived-store entry for rebuildable calculation artifacts.
154#[derive(Clone, Debug, PartialEq, Eq)]
155pub enum DerivedEntry {
156    /// Dependency graph materialization.
157    Graph(String),
158    /// Cached value for a calculated cell.
159    CachedValue(String),
160    /// Calculation receipt or scheduler evidence.
161    Receipt(String),
162}
163
164/// Recoverable source/control commit staged across separate backends.
165#[derive(Clone, Debug, PartialEq, Eq)]
166pub struct PendingCommit {
167    source_writes: BTreeMap<CellId, SourceEntry>,
168    control_writes: BTreeMap<String, ControlEntry>,
169    phase: CommitPhase,
170}
171
172impl PendingCommit {
173    fn new(
174        source_writes: BTreeMap<CellId, SourceEntry>,
175        control_writes: BTreeMap<String, ControlEntry>,
176    ) -> Self {
177        Self {
178            source_writes,
179            control_writes,
180            phase: CommitPhase::Prepared,
181        }
182    }
183
184    /// Whether the source side of the transaction boundary was persisted.
185    pub fn source_committed(&self) -> bool {
186        self.phase >= CommitPhase::SourceCommitted
187    }
188
189    /// Whether the control side of the transaction boundary was persisted.
190    pub fn control_committed(&self) -> bool {
191        self.phase >= CommitPhase::ControlCommitted
192    }
193}
194
195#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
196enum CommitPhase {
197    Prepared,
198    SourceCommitted,
199    ControlCommitted,
200}
201
202/// Store composition failures.
203#[derive(Clone, Debug, PartialEq, Eq)]
204pub enum StoreError {
205    /// A composed expression tree must have a root Dir.
206    MissingRootDir,
207    /// A mount path cannot be root and must not duplicate or conflict with existing mounts.
208    InvalidMount(String),
209    /// A table mount was treated as a directory.
210    TableMountIsLeaf(TablePath),
211    /// A persisted mount descriptor is corrupt.
212    CorruptMount(String),
213}
214
215/// Typed source/control/derived stores plus explicit Table/Dir mount descriptors.
216#[derive(Clone, Debug, PartialEq, Eq)]
217pub struct ExprTreeStores {
218    root_dir: DirId,
219    source: BTreeMap<CellId, SourceEntry>,
220    control: BTreeMap<String, ControlEntry>,
221    derived: BTreeMap<CellId, DerivedEntry>,
222    mounts: BTreeMap<String, MountDescriptor>,
223}
224
225impl ExprTreeStores {
226    /// Compose a tree over an existing root directory.
227    pub fn new(root_dir: DirId) -> Result<Self, StoreError> {
228        if root_dir.as_str().is_empty() {
229            return Err(StoreError::MissingRootDir);
230        }
231        Ok(Self {
232            root_dir,
233            source: BTreeMap::new(),
234            control: BTreeMap::new(),
235            derived: BTreeMap::new(),
236            mounts: BTreeMap::new(),
237        })
238    }
239
240    /// Reopen persisted stores, validating the mount table before accepting it.
241    pub fn reopen(
242        root_dir: DirId,
243        source: BTreeMap<CellId, SourceEntry>,
244        control: BTreeMap<String, ControlEntry>,
245        derived: BTreeMap<CellId, DerivedEntry>,
246        mounts: Vec<MountDescriptor>,
247    ) -> Result<Self, StoreError> {
248        let mut stores = Self::new(root_dir)?;
249        stores.source = source;
250        stores.control = control;
251        stores.derived = derived;
252        for descriptor in mounts {
253            stores.mount(descriptor)?;
254        }
255        Ok(stores)
256    }
257
258    /// Root directory required by the mounted namespace owner.
259    pub fn root_dir(&self) -> &DirId {
260        &self.root_dir
261    }
262
263    /// Authored source entries only.
264    pub fn source_entry(&self, id: &CellId) -> Option<&SourceEntry> {
265        self.source.get(id)
266    }
267
268    /// Operational control entries only.
269    pub fn control_entry(&self, key: &str) -> Option<&ControlEntry> {
270        self.control.get(key)
271    }
272
273    /// Rebuildable derived entries only.
274    pub fn derived_entry(&self, id: &CellId) -> Option<&DerivedEntry> {
275        self.derived.get(id)
276    }
277
278    /// All current explicit mounts.
279    pub fn mounts(&self) -> impl Iterator<Item = &MountDescriptor> {
280        self.mounts.values()
281    }
282
283    /// Return a Table or Dir value to the caller without mutating the namespace.
284    pub fn return_value_without_mounting(&self, _resource: MountResource) -> usize {
285        self.mounts.len()
286    }
287
288    /// Explicitly mount a Table or Dir descriptor.
289    pub fn mount(&mut self, descriptor: MountDescriptor) -> Result<(), StoreError> {
290        validate_mount(&descriptor)?;
291        let key = mount_key(descriptor.path());
292        if self.mounts.contains_key(&key) {
293            return Err(StoreError::InvalidMount(format!(
294                "duplicate mount point {}",
295                descriptor.path()
296            )));
297        }
298        for existing in self.mounts.values() {
299            if is_prefix(existing.path(), descriptor.path())
300                && existing.resource() == MountResource::Table
301                && existing.path() != descriptor.path()
302            {
303                return Err(StoreError::TableMountIsLeaf(existing.path().clone()));
304            }
305            if is_prefix(descriptor.path(), existing.path())
306                && descriptor.resource() == MountResource::Table
307            {
308                return Err(StoreError::InvalidMount(format!(
309                    "table mount {} would parent existing mount {}",
310                    descriptor.path(),
311                    existing.path()
312                )));
313            }
314        }
315        self.mounts.insert(key, descriptor);
316        Ok(())
317    }
318
319    /// Removes and returns one explicit mount descriptor.
320    pub fn unmount(&mut self, path: &TablePath) -> Result<MountDescriptor, StoreError> {
321        let key = mount_key(path);
322        let descriptor = self
323            .mounts
324            .remove(&key)
325            .ok_or_else(|| StoreError::InvalidMount(format!("missing mount point {path}")))?;
326        self.control.remove(&format!("mount-epoch:{path}"));
327        Ok(descriptor)
328    }
329
330    /// Record a mounted backend epoch in the control store.
331    pub fn observe_mount_epoch(
332        &mut self,
333        path: &TablePath,
334        epoch: MountEpoch,
335    ) -> Result<(), StoreError> {
336        let mount = self
337            .mounts
338            .get_mut(&mount_key(path))
339            .ok_or_else(|| StoreError::CorruptMount(format!("missing mount {}", path)))?;
340        mount.set_epoch(epoch);
341        self.control.insert(
342            format!("mount-epoch:{}", path),
343            ControlEntry::MountEpoch(epoch),
344        );
345        Ok(())
346    }
347
348    /// Prepare a recoverable source/control commit.
349    ///
350    /// The transaction boundary is exactly source plus control. Recovery replays the
351    /// same pending record until both sides are durable. Derived entries are
352    /// rebuildable and are not part of this boundary.
353    pub fn prepare_source_control_commit(
354        source_writes: BTreeMap<CellId, SourceEntry>,
355        control_writes: BTreeMap<String, ControlEntry>,
356    ) -> PendingCommit {
357        PendingCommit::new(source_writes, control_writes)
358    }
359
360    /// Persist the source side of a prepared commit.
361    pub fn commit_source(&mut self, pending: &mut PendingCommit) {
362        self.source.extend(pending.source_writes.clone());
363        pending.phase = CommitPhase::SourceCommitted;
364    }
365
366    /// Persist the control side of a source-committed commit.
367    pub fn commit_control(&mut self, pending: &mut PendingCommit) {
368        self.control.extend(pending.control_writes.clone());
369        pending.phase = CommitPhase::ControlCommitted;
370    }
371
372    /// Finish or replay a partially persisted source/control commit.
373    pub fn recover_commit(&mut self, pending: &mut PendingCommit) {
374        if !pending.source_committed() {
375            self.commit_source(pending);
376        }
377        if !pending.control_committed() {
378            self.commit_control(pending);
379        }
380    }
381
382    /// Store a rebuildable derived value outside source entries.
383    pub fn put_derived(&mut self, id: CellId, entry: DerivedEntry) {
384        self.derived.insert(id, entry);
385    }
386
387    /// Store an operational control value outside source entries.
388    pub fn put_control(&mut self, key: impl Into<String>, entry: ControlEntry) {
389        self.control.insert(key.into(), entry);
390    }
391
392    /// Removes authored source for a deleted cell.
393    pub fn remove_source(&mut self, id: &CellId) -> Option<SourceEntry> {
394        self.source.remove(id)
395    }
396}
397
398fn validate_mount(descriptor: &MountDescriptor) -> Result<(), StoreError> {
399    if descriptor.path().is_root() {
400        return Err(StoreError::InvalidMount(
401            "root is supplied as the required root Dir, not as a mount".to_owned(),
402        ));
403    }
404    if descriptor.backend() == BackendKind::ReadOnly && descriptor.resource() == MountResource::Dir
405    {
406        return Ok(());
407    }
408    Ok(())
409}
410
411fn is_prefix(candidate: &TablePath, path: &TablePath) -> bool {
412    let candidate_segments = segments(candidate);
413    let path_segments = segments(path);
414    candidate_segments.len() <= path_segments.len()
415        && candidate_segments
416            .iter()
417            .zip(path_segments.iter())
418            .all(|(left, right)| left == right)
419}
420
421fn segments(path: &TablePath) -> Vec<&str> {
422    path.segments().iter().map(String::as_str).collect()
423}
424
425fn mount_key(path: &TablePath) -> String {
426    path.to_absolute_reference()
427}
428
429#[cfg(test)]
430pub(crate) fn source_keys_for_test(stores: &ExprTreeStores) -> std::collections::BTreeSet<CellId> {
431    stores.source.keys().cloned().collect()
432}
sim_expr_tree_core/store.rs

sim_expr_tree_core/
store.rs
