memstead_base/engine/
roster.rs

1//! Mem membership follows reload-before-operation.
2//!
3//! Content drift is probed per operation ([`super::Engine::reload_if_stale`]
4//! compares each mount's branch tip with the head it last loaded). Membership
5//! used to be fixed at boot: a mem registered or unregistered by another
6//! process (the CLI, a sibling server) stayed invisible, or kept being served,
7//! until a restart — on 2026-09-02 a ui-api served a retired mem for hours.
8//! This module is the membership half of the same discipline: before each
9//! operation the engine compares a fingerprint of the mount roster
10//! (`.memstead/state/mounts.json`) with the one it booted or last reconciled
11//! on, and on a change mounts the new entries cold under the boot quarantine
12//! rules, unmounts the gone entries atomically, re-scans the schema sources,
13//! and reports the change (`MEM_ROSTER_CHANGED`) so an agent drops the cached
14//! hashes of the mems that left.
15//!
16//! The probe is a `stat` plus, only when size or mtime moved, a hash of the
17//! file — the same cost band as the branch-tip probe, never a boot.
18
19use std::collections::{BTreeSet, HashSet};
20use std::path::PathBuf;
21use std::sync::Arc;
22
23use serde::{Deserialize, Serialize};
24
25use super::{Engine, EngineError};
26
27/// What the engine last saw of the roster file: size and mtime for the
28/// cheap comparison, the content hash for the decision.
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub struct RosterFingerprint {
31    len: u64,
32    modified: Option<std::time::SystemTime>,
33    hash: u64,
34}
35
36/// One applied roster reconciliation.
37#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
38pub struct RosterChange {
39    /// Mems the roster gained and the engine mounted cold, ready to serve.
40    pub added: Vec<String>,
41    /// Mems the roster lost and the engine unmounted: their entities,
42    /// search index, community partition and derived views are gone.
43    pub removed: Vec<String>,
44    /// Mems the roster gained that failed to mount and are quarantined
45    /// under the boot rules (each with its reason on the quarantine
46    /// roster); the other mems keep serving.
47    pub quarantined: Vec<String>,
48    /// Per-item failures that were not a quarantine: the schema-source
49    /// re-scan, or an unmount that could not complete (the roster change
50    /// for that mem is not applied and it stays fully served).
51    pub failures: Vec<crate::ops::RefreshFailure>,
52}
53
54impl RosterChange {
55    /// Whether anything at all changed or failed.
56    pub fn is_empty(&self) -> bool {
57        self.added.is_empty()
58            && self.removed.is_empty()
59            && self.quarantined.is_empty()
60            && self.failures.is_empty()
61    }
62}
63
64/// The event a roster subscriber receives — the applied change.
65pub type RosterChangedEvent = RosterChange;
66
67/// Callback shape for roster subscribers, mirroring
68/// [`super::events::EventCallback`].
69pub type RosterCallback = Arc<dyn Fn(&RosterChangedEvent) + Send + Sync + 'static>;
70
71/// The roster subscriber registry: the next id, then `(id, callback)`.
72pub(crate) type RosterSubscribers = std::sync::Mutex<(u64, Vec<(u64, RosterCallback)>)>;
73
74fn hash_bytes(bytes: &[u8]) -> u64 {
75    use std::hash::{Hash, Hasher};
76    let mut h = std::collections::hash_map::DefaultHasher::new();
77    bytes.hash(&mut h);
78    h.finish()
79}
80
81impl Engine {
82    /// The roster file this engine's membership is read from, when the
83    /// engine knows its workspace root.
84    fn roster_path(&self) -> Option<PathBuf> {
85        self.workspace_root.as_ref().map(|root| {
86            root.join(crate::workspace_store::WORKSPACE_STORE_DIR)
87                .join("state")
88                .join("mounts.json")
89        })
90    }
91
92    /// The roster's current fingerprint. `Ok(None)` when the engine has no
93    /// workspace root or the file does not exist (an ad-hoc mount list, a
94    /// standalone folder mem). Size and mtime unchanged from the cached
95    /// fingerprint short-circuits without reading the file.
96    fn roster_fingerprint_now(&self) -> Result<Option<RosterFingerprint>, std::io::Error> {
97        let Some(path) = self.roster_path() else {
98            return Ok(None);
99        };
100        let meta = match std::fs::metadata(&path) {
101            Ok(m) => m,
102            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
103            Err(e) => return Err(e),
104        };
105        let len = meta.len();
106        let modified = meta.modified().ok();
107        if let Some(cached) = &self.roster_fingerprint
108            && cached.len == len
109            && cached.modified == modified
110        {
111            return Ok(Some(cached.clone()));
112        }
113        let bytes = std::fs::read(&path)?;
114        Ok(Some(RosterFingerprint {
115            len,
116            modified,
117            hash: hash_bytes(&bytes),
118        }))
119    }
120
121    /// Capture the roster as it stands as the reconciliation baseline —
122    /// what boot and every applied reconcile leave behind.
123    pub(crate) fn capture_roster_fingerprint(&mut self) {
124        self.roster_fingerprint = self.roster_fingerprint_now().ok().flatten();
125    }
126
127    /// Reconcile membership with the roster file: `Ok(None)` when the roster
128    /// did not change since the last reconcile (or the first observation,
129    /// captured silently as the baseline); `Ok(Some(change))` after applying
130    /// a change; `Err` only when the roster could not be read or parsed at
131    /// all (the baseline is kept, so the next operation retries).
132    ///
133    /// Per mem the change is atomic: an unmount that cannot complete leaves
134    /// that mem fully served, is reported under `failures`, and keeps the
135    /// baseline where it was so it is retried next time; a cold mount that
136    /// fails quarantines the mem exactly as boot would. Read-only
137    /// attachments (installed archives) are not roster entries and never
138    /// count as removals.
139    pub fn reconcile_roster(&mut self) -> Result<Option<RosterChange>, EngineError> {
140        let now = self.roster_fingerprint_now().map_err(|e| {
141            EngineError::Backend(crate::backend::BackendError::Other(format!(
142                "roster unreadable: {e}"
143            )))
144        })?;
145        let Some(now) = now else {
146            return Ok(None);
147        };
148        let Some(cached) = self.roster_fingerprint.clone() else {
149            self.roster_fingerprint = Some(now);
150            return Ok(None);
151        };
152        if cached.hash == now.hash {
153            self.roster_fingerprint = Some(now);
154            return Ok(None);
155        }
156        self.apply_roster(now).map(Some)
157    }
158
159    /// The reconciliation without the fingerprint gate — what
160    /// `full_refresh` runs so its report is authoritative even when the
161    /// roster file did not move (a first observation, a probe that was
162    /// skipped). An engine without a roster file yields an empty change.
163    pub(crate) fn reconcile_roster_forced(&mut self) -> Result<RosterChange, EngineError> {
164        let now = self.roster_fingerprint_now().map_err(|e| {
165            EngineError::Backend(crate::backend::BackendError::Other(format!(
166                "roster unreadable: {e}"
167            )))
168        })?;
169        match now {
170            Some(now) => self.apply_roster(now),
171            None => Ok(RosterChange::default()),
172        }
173    }
174
175    fn apply_roster(&mut self, now: RosterFingerprint) -> Result<RosterChange, EngineError> {
176        let root = self
177            .workspace_root
178            .clone()
179            .expect("a roster fingerprint implies a workspace root");
180        let workspace = crate::workspace_store::WorkspaceStoreAdapter::load(
181            &crate::workspace_store::FileWorkspaceStore::new(),
182            &root,
183        )
184        .map_err(|e| {
185            EngineError::Backend(crate::backend::BackendError::Other(format!(
186                "roster under {} unreadable: {e}",
187                root.display()
188            )))
189        })?;
190
191        let manifest: BTreeSet<String> = workspace
192            .mounts
193            .iter()
194            .filter(|m| m.capability == crate::workspace::MountCapability::Write)
195            .map(|m| m.mem.clone())
196            .collect();
197        let mounted: BTreeSet<String> = self
198            .mounts
199            .iter()
200            .filter(|m| m.mount.capability == crate::workspace::MountCapability::Write)
201            .map(|m| m.mount.mem.clone())
202            .collect();
203        let quarantined_now: BTreeSet<String> = self
204            .quarantined
205            .iter()
206            .map(|q| q.mount.mem.clone())
207            .collect();
208
209        let mut change = RosterChange::default();
210        let mut all_applied = true;
211
212        // The schema catalogue first: a mem that arrived may pin a schema
213        // installed since boot, and a cold mount resolves against the
214        // catalogue as it stands.
215        let mut schema_report = crate::ops::FullRefreshReport::default();
216        self.refresh_schema_sources(&mut schema_report);
217        change.failures.extend(schema_report.failures);
218
219        // Removals first, each atomic. A gone quarantined entry just leaves
220        // the quarantine roster: it served nothing.
221        for name in mounted.difference(&manifest) {
222            match self.unmount_mem(name) {
223                Ok(()) => change.removed.push(name.clone()),
224                Err(e) => {
225                    all_applied = false;
226                    change.failures.push(crate::ops::RefreshFailure {
227                        item: format!("unmount:{name}"),
228                        error: e.to_string(),
229                    });
230                }
231            }
232        }
233        let gone_quarantined: Vec<String> =
234            quarantined_now.difference(&manifest).cloned().collect();
235        if !gone_quarantined.is_empty() {
236            self.quarantined
237                .retain(|q| !gone_quarantined.contains(&q.mount.mem));
238            change.removed.extend(gone_quarantined);
239        }
240
241        // Additions: cold mount under the boot quarantine rules.
242        let mut any_mounted = false;
243        for mount in workspace.mounts {
244            if mount.capability != crate::workspace::MountCapability::Write
245                || mounted.contains(&mount.mem)
246                || quarantined_now.contains(&mount.mem)
247            {
248                continue;
249            }
250            let name = mount.mem.clone();
251            let backend = match (self.backend_factory)(&mount) {
252                Ok(b) => b,
253                Err(e) => {
254                    self.quarantine_mount(mount, e.code(), e.to_string());
255                    change.quarantined.push(name);
256                    continue;
257                }
258            };
259            // Boot's rule for storage that is gone (a branch the mem-repo
260            // lacks, a folder that does not exist): quarantine under
261            // MOUNT_UNBACKED rather than serving an empty graph.
262            if let Some(crate::ops::WarningHint::MountUnbacked { reason, .. }) =
263                super::boot::unbacked_mount_warning(&mount, backend.as_ref(), None)
264                && reason != crate::ops::MountUnbackedReason::Empty
265            {
266                let location = match &mount.storage {
267                    crate::workspace::MountStorage::GitBranch { branch, .. } => branch.clone(),
268                    crate::workspace::MountStorage::Folder { path }
269                    | crate::workspace::MountStorage::Archive { path } => {
270                        path.display().to_string()
271                    }
272                    crate::workspace::MountStorage::InMemory => String::new(),
273                };
274                self.quarantine_mount(
275                    mount,
276                    "MOUNT_UNBACKED",
277                    format!(
278                        "the mount's storage is gone ({location}); it is configured but cannot \
279                         serve, so it is held out of the roster rather than answering reads \
280                         with an empty graph"
281                    ),
282                );
283                change.quarantined.push(name);
284                continue;
285            }
286            match self.register_writable_mem_batched(
287                mount.clone(),
288                backend,
289                crate::mem::MemOrigin::ExplicitToml,
290            ) {
291                Ok(()) => {
292                    any_mounted = true;
293                    self.recently_unmounted.remove(&name);
294                    change.added.push(name);
295                }
296                Err(e) => {
297                    self.quarantine_mount(mount, e.code(), e.to_string());
298                    change.quarantined.push(name);
299                }
300            }
301        }
302        if any_mounted {
303            self.finish_batched_registrations();
304        }
305
306        if all_applied {
307            self.roster_fingerprint = Some(now);
308        }
309        self.invalidate_communities();
310        self.invalidate_search_indexes();
311        self.emit_roster_changed(&change);
312        Ok(change)
313    }
314
315    fn quarantine_mount(&mut self, mount: crate::workspace::Mount, code: &str, message: String) {
316        self.quarantined.push(super::QuarantinedMem {
317            mount,
318            reason_code: code.to_string(),
319            reason_message: message,
320        });
321    }
322
323    /// Unmount one writable mem atomically: every derived structure for it
324    /// (store slice, schema entry, router slot, load warnings, pending
325    /// change notices, community and search memos, its quarantine entry)
326    /// is gone afterwards, or nothing is touched and the error names why.
327    /// The mem is remembered as recently unmounted so an operation naming
328    /// it refuses with `MEM_UNMOUNTED` rather than a bare unknown-mem.
329    pub(crate) fn unmount_mem(&mut self, mem: &str) -> Result<(), EngineError> {
330        #[cfg(test)]
331        if self.inject_unmount_failure.as_deref() == Some(mem) {
332            return Err(EngineError::Backend(crate::backend::BackendError::Other(
333                format!("injected unmount failure for mem `{mem}`"),
334            )));
335        }
336        let removed = self.unregister_writable_mem(mem)?;
337        if removed.is_none() {
338            // Not mounted: a quarantined entry leaving the roster.
339            self.quarantined.retain(|q| q.mount.mem != mem);
340        }
341        self.pending_mem_changed.retain(|n| n.mem != mem);
342        self.labelling_memo = std::cell::OnceCell::new();
343        self.recently_unmounted.insert(mem.to_string());
344        Ok(())
345    }
346
347    /// Whether `mem` left the roster during this engine's lifetime and has
348    /// not returned — the typed-refusal memory behind `MEM_UNMOUNTED`.
349    pub fn recently_unmounted(&self, mem: &str) -> bool {
350        self.recently_unmounted.contains(mem)
351    }
352
353    /// Subscribe to applied roster changes. Returns the subscription id;
354    /// pass it to [`Self::unsubscribe_roster_changes`] to stop.
355    pub fn subscribe_roster_changes(&self, callback: RosterCallback) -> u64 {
356        let mut subs = self
357            .roster_subscribers
358            .lock()
359            .expect("roster subscriber registry mutex must not be poisoned");
360        let id = subs.0 + 1;
361        subs.0 = id;
362        subs.1.push((id, callback));
363        id
364    }
365
366    /// Drop a roster subscription; a no-op for an unknown id.
367    pub fn unsubscribe_roster_changes(&self, id: u64) {
368        let mut subs = self
369            .roster_subscribers
370            .lock()
371            .expect("roster subscriber registry mutex must not be poisoned");
372        subs.1.retain(|(slot, _)| *slot != id);
373    }
374
375    fn emit_roster_changed(&self, change: &RosterChange) {
376        if change.is_empty() {
377            return;
378        }
379        let callbacks: Vec<RosterCallback> = self
380            .roster_subscribers
381            .lock()
382            .expect("roster subscriber registry mutex must not be poisoned")
383            .1
384            .iter()
385            .map(|(_, cb)| cb.clone())
386            .collect();
387        for cb in callbacks {
388            cb(change);
389        }
390    }
391
392    /// The set of mems this engine serves as writable, for tests and
393    /// consumers that compare rosters.
394    pub fn writable_mem_set(&self) -> HashSet<String> {
395        self.mounts
396            .iter()
397            .filter(|m| m.mount.capability == crate::workspace::MountCapability::Write)
398            .map(|m| m.mount.mem.clone())
399            .collect()
400    }
401}
memstead_base/engine/roster.rs

memstead_base/engine/
roster.rs
