Skip to main content

plugmem_host/
workspace.rs

1//! A workspace: many databases under one directory, addressed by name.
2//!
3//! **The default is one database.** Nothing here switches on by itself — a
4//! caller that opens a [`crate::Database`] at a path behaves exactly as it did
5//! before this module existed. A workspace is what you reach for when one
6//! process serves many independent memories (a chat per conversation, a
7//! database per tenant) and wants them to stay independent.
8//!
9//! The layout is deliberately mechanical:
10//!
11//! ```text
12//! <root>/registry.plugmem      the registry — an ordinary plugmem database
13//! <root>/db/<name>.plugmem     the databases themselves
14//! ```
15//!
16//! Two properties fall out of it, and both are the point:
17//!
18//! - **a name is not a path, and cannot become one.** [`DbName`] admits only
19//!   `[a-z0-9][a-z0-9_-]*`, so `..`, `/`, a drive letter or an absolute path
20//!   are not filtered out — they are unrepresentable. Resolution is then a
21//!   join, with nothing to get wrong;
22//! - **the directory is the truth.** [`WorkspaceLayout::list`] reads the
23//!   filesystem, never the registry. The registry (see the `registry` module)
24//!   is a searchable index over descriptions and can be rebuilt from the
25//!   databases themselves; losing it costs search, not data.
26//!
27//! The registry lives in the root while the databases live one level down, so
28//! no name can ever collide with it.
29
30mod registry;
31
32use std::fmt;
33use std::path::{Path, PathBuf};
34use std::sync::{Mutex, MutexGuard};
35
36use crate::{Database, HostError};
37
38pub use registry::{
39    ARCHIVED_TAG, DbEntry, Description, ENTRY_TAG, ReindexReport, SELF_ENTITY, WorkspaceIssue,
40};
41
42/// Longest database name a workspace accepts, in bytes.
43///
44/// Names are ASCII, so this is also the character count. The limit exists so a
45/// name plus the extension stays comfortably inside the shortest filename limit
46/// worth caring about (255 bytes on ext4/APFS/NTFS) with room for the sidecar
47/// suffixes (`.lock`, `.jrnl`, `.snap.N`) the storage layer appends.
48pub const MAX_DB_NAME: usize = 64;
49
50/// Directory, below the workspace root, holding the databases.
51const DB_DIR: &str = "db";
52
53/// The registry's file name, directly in the workspace root.
54const REGISTRY_FILE: &str = "registry.plugmem";
55
56/// Extension of a database file. The storage layer appends its sidecar
57/// suffixes *after* this (`chat-42.plugmem.lock`), so matching on the
58/// extension picks out base files and nothing else.
59const DB_EXT: &str = "plugmem";
60
61/// Names Windows resolves to devices rather than files, in every directory and
62/// **whatever extension is appended** — `con.plugmem` opens the console, not a
63/// file called `con.plugmem`.
64///
65/// Refused on every platform, not only on Windows. A workspace is a directory
66/// someone may copy between machines, and a memory that exists on Linux and
67/// silently becomes the printer port on Windows is the worst kind of portability
68/// bug: it appears at someone else's desk, on data that was already fine.
69///
70/// Written out rather than taken from a crate (`sanitize-filename` and friends
71/// exist) because those *rewrite* an arbitrary string into a safe filename,
72/// which is a different job: here the alphabet has already refused separators,
73/// dots, colons, `$`, control bytes, non-ASCII, uppercase and trailing spaces,
74/// and this list is the entire remainder. It is 22 strings, frozen by Windows
75/// for thirty years — `CONIN$` and the superscript `COM¹` forms cannot be
76/// spelled in the alphabet at all. A dependency would be carrying a sanitizer
77/// to do the part we already did.
78const RESERVED_DEVICE_NAMES: &[&str] = &[
79    "con", "prn", "aux", "nul", "com1", "com2", "com3", "com4", "com5", "com6", "com7", "com8",
80    "com9", "lpt1", "lpt2", "lpt3", "lpt4", "lpt5", "lpt6", "lpt7", "lpt8", "lpt9",
81];
82
83/// Why a string is not a usable database name.
84///
85/// A typed reason rather than a message, so a caller (and a test) can react to
86/// the specific problem instead of matching on prose.
87#[derive(Clone, Copy, Debug, PartialEq, Eq)]
88#[non_exhaustive]
89pub enum NameProblem {
90    /// The name was empty.
91    Empty,
92    /// The name was longer than [`MAX_DB_NAME`].
93    TooLong,
94    /// The first character was neither a lowercase ASCII letter nor a digit.
95    /// Leading `-`, `_` and `.` are refused so a name can never be read as a
96    /// flag or as a relative path component.
97    LeadingChar,
98    /// Some character was outside `[a-z0-9_-]`.
99    Character,
100    /// The name is a Windows device name (`con`, `nul`, `com1`, …). Refused
101    /// everywhere so a workspace stays portable between machines.
102    ReservedDevice,
103}
104
105impl fmt::Display for NameProblem {
106    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
107        match self {
108            Self::Empty => f.write_str("it is empty"),
109            Self::TooLong => write!(f, "it is longer than {MAX_DB_NAME} bytes"),
110            Self::LeadingChar => f.write_str("it must start with a lowercase letter or a digit"),
111            Self::Character => {
112                f.write_str("it may hold only lowercase letters, digits, '-' and '_'")
113            }
114            Self::ReservedDevice => f.write_str(
115                "it is a Windows device name, which would open a device rather than a file there",
116            ),
117        }
118    }
119}
120
121/// A validated database name — the only thing a workspace resolves.
122///
123/// Construct it with [`DbName::parse`]; there is no other way in, which is what
124/// makes "a name is not a path" a property of the type rather than a rule
125/// somebody has to remember at every call site.
126#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
127pub struct DbName(String);
128
129impl DbName {
130    /// Validates `s` as a database name.
131    ///
132    /// The rule: ASCII, first character `[a-z0-9]`, the rest `[a-z0-9_-]`,
133    /// length `1..=`[`MAX_DB_NAME`].
134    ///
135    /// # Errors
136    ///
137    /// [`WorkspaceError::BadName`] carrying the specific [`NameProblem`].
138    pub fn parse(s: &str) -> Result<Self, WorkspaceError> {
139        let bad = |why| {
140            Err(WorkspaceError::BadName {
141                name: s.to_string(),
142                why,
143            })
144        };
145        let Some(&first) = s.as_bytes().first() else {
146            return bad(NameProblem::Empty);
147        };
148        if !first.is_ascii_lowercase() && !first.is_ascii_digit() {
149            return bad(NameProblem::LeadingChar);
150        }
151        if !s.bytes().all(is_name_byte) {
152            return bad(NameProblem::Character);
153        }
154        if s.len() > MAX_DB_NAME {
155            return bad(NameProblem::TooLong);
156        }
157        if RESERVED_DEVICE_NAMES.contains(&s) {
158            return bad(NameProblem::ReservedDevice);
159        }
160        Ok(DbName(s.to_string()))
161    }
162
163    /// The name as a string slice.
164    pub fn as_str(&self) -> &str {
165        &self.0
166    }
167}
168
169impl fmt::Display for DbName {
170    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
171        f.write_str(&self.0)
172    }
173}
174
175/// True for a byte allowed anywhere in a name.
176fn is_name_byte(b: u8) -> bool {
177    b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-' || b == b'_'
178}
179
180/// Every way a workspace operation can fail.
181///
182/// Opening one database still fails as a [`HostError`]; this type adds the
183/// failures that only exist once databases are addressed by name.
184#[derive(Debug, thiserror::Error)]
185#[non_exhaustive]
186pub enum WorkspaceError {
187    /// The string is not a usable database name.
188    #[error("{name:?} is not a usable database name: {why}")]
189    BadName {
190        /// The rejected string, as given.
191        name: String,
192        /// What specifically was wrong with it.
193        why: NameProblem,
194    },
195
196    /// No database of that name exists, and the caller asked not to create one.
197    #[error("no database named {name} in this workspace (looked for {})", path.display())]
198    NoSuchDatabase {
199        /// The name that did not resolve.
200        name: DbName,
201        /// Where it would have been.
202        path: PathBuf,
203    },
204
205    /// The database is open for writing elsewhere.
206    ///
207    /// Distinct from [`HostError::Locked`] so the message can name the database
208    /// rather than a path the caller never typed. One local database has one writer, and
209    /// in a workspace the other writer is usually a long-running sidecar that
210    /// will release it once the handle goes idle.
211    #[error(
212        "database {name} is in use by another process; it is released once that process closes it (a pooled handle does so after its idle timeout)"
213    )]
214    Busy {
215        /// The database that is held elsewhere.
216        name: DbName,
217    },
218
219    /// A filesystem operation on the workspace itself failed.
220    #[error("i/o on {}: {source}", path.display())]
221    Io {
222        /// The path the operation touched.
223        path: PathBuf,
224        /// The underlying error.
225        #[source]
226        source: std::io::Error,
227    },
228
229    /// Opening or using one of the databases failed.
230    #[error(transparent)]
231    Host(#[from] HostError),
232}
233
234impl WorkspaceError {
235    /// Shorthand for wrapping an I/O error with its path.
236    pub(crate) fn io(path: &Path, source: std::io::Error) -> Self {
237        Self::Io {
238            path: path.to_path_buf(),
239            source,
240        }
241    }
242}
243
244/// Where a workspace keeps its files.
245///
246/// Pure path arithmetic and one directory listing — it opens nothing and locks
247/// nothing. A caller that only needs "which file is `work`?" (the CLI resolving
248/// `--db work`) wants this and not the handle pool.
249#[derive(Clone, Debug, PartialEq, Eq)]
250pub struct WorkspaceLayout {
251    root: PathBuf,
252}
253
254impl WorkspaceLayout {
255    /// A layout rooted at `root`. Creates nothing; the directories appear when
256    /// a database is first written.
257    pub fn new(root: impl Into<PathBuf>) -> Self {
258        Self { root: root.into() }
259    }
260
261    /// The workspace root.
262    pub fn root(&self) -> &Path {
263        &self.root
264    }
265
266    /// The directory holding the databases, `<root>/db`.
267    pub fn db_dir(&self) -> PathBuf {
268        self.root.join(DB_DIR)
269    }
270
271    /// The file backing `name`, `<root>/db/<name>.plugmem`.
272    ///
273    /// A join of validated components, so the result is always inside
274    /// [`WorkspaceLayout::db_dir`].
275    pub fn path_of(&self, name: &DbName) -> PathBuf {
276        self.db_dir().join(format!("{}.{DB_EXT}", name.0))
277    }
278
279    /// The registry's file, `<root>/registry.plugmem`. It sits in the root
280    /// rather than beside the databases, so no name can collide with it.
281    pub fn registry_path(&self) -> PathBuf {
282        self.root.join(REGISTRY_FILE)
283    }
284
285    /// Whether `name` is a database on disk.
286    ///
287    /// Asks the storage layer rather than stat-ing the path: a database that
288    /// has been written to but not yet checkpointed has no file at its base
289    /// path, and treating it as absent would mean creating over it.
290    pub fn exists(&self, name: &DbName) -> bool {
291        crate::storage::database_exists(&self.path_of(name))
292    }
293
294    /// Every database in the workspace, sorted by name.
295    ///
296    /// Reads the directory — the filesystem is the truth, the registry is only
297    /// an index over it. A missing `db/` is an empty workspace, not an error.
298    /// Files whose name is not a database's are skipped here and reported by
299    /// the registry's `verify`, which is where a person is asking about
300    /// consistency rather than about what they can open.
301    ///
302    /// One database is several files (`chat-42.plugmem`, `.journal`, `.lock`,
303    /// `.snap.N`), so the listing folds them back to one name each: a name
304    /// cannot contain a dot, so everything up to the first dot is the candidate,
305    /// and the storage layer confirms whether a database is really there.
306    ///
307    /// # Errors
308    ///
309    /// [`WorkspaceError::Io`] if the directory exists but cannot be read.
310    pub fn list(&self) -> Result<Vec<DbName>, WorkspaceError> {
311        let dir = self.db_dir();
312        let entries = match std::fs::read_dir(&dir) {
313            Ok(entries) => entries,
314            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
315            Err(e) => return Err(WorkspaceError::io(&dir, e)),
316        };
317        let mut names = Vec::new();
318        for entry in entries {
319            let entry = entry.map_err(|e| WorkspaceError::io(&dir, e))?;
320            let file_name = entry.file_name();
321            let Some(candidate) = file_name.to_str().and_then(|n| n.split('.').next()) else {
322                continue;
323            };
324            if let Ok(name) = DbName::parse(candidate)
325                && !names.contains(&name)
326                && self.exists(&name)
327            {
328                names.push(name);
329            }
330        }
331        names.sort_unstable();
332        Ok(names)
333    }
334}
335
336/// How many databases a workspace keeps open, and for how long.
337#[derive(Clone, Copy, Debug, PartialEq, Eq)]
338pub struct WorkspaceLimits {
339    /// Most databases held open at once. The least recently used is closed to
340    /// make room. `0` is read as `1` — a pool that holds nothing would reopen
341    /// on every call.
342    pub max_open: usize,
343    /// How long a database may sit unused before [`Workspace::close_idle`]
344    /// closes it. `0` disables the sweep, so handles stay until evicted.
345    ///
346    /// This is not a memory knob, it is a *liveness* knob: an open writer holds
347    /// the file's exclusive lock, so nothing else on the machine can touch that
348    /// database until the handle goes. A long-running server that never let go
349    /// would make its databases permanently unreachable from the CLI.
350    pub idle_timeout_ms: u64,
351}
352
353/// Handles kept open by default. Small on purpose: reopening a database the
354/// size of a chat is milliseconds, so caching is a latency win, not a
355/// requirement, and every extra handle is a lock somebody else cannot take.
356pub const DEFAULT_MAX_OPEN: usize = 16;
357
358/// Default idle window before a pooled handle is closed.
359pub const DEFAULT_IDLE_TIMEOUT_MS: u64 = 60_000;
360
361/// Descriptors one open database occupies: its lock, its journal, the mapped
362/// generation, and room for the staging file a checkpoint writes.
363const FILES_PER_OPEN_DATABASE: usize = 4;
364
365/// The lowest per-process descriptor limit worth planning for — the usual POSIX
366/// soft limit.
367const ASSUMED_FD_LIMIT: usize = 1024;
368
369/// Descriptors left for the rest of the process: stdio, the config file, the
370/// registry, and the embedder's sockets.
371const RESERVED_FDS: usize = 64;
372
373/// Most databases a pool will hold open, whatever it is configured with.
374///
375/// Derived rather than picked: `max_open` can come from a config file, and a
376/// file must not be able to talk a process into running itself out of file
377/// descriptors. Exceeding this would not fail at the pool — it would fail at
378/// some unrelated `open` elsewhere in the program, which is the worst place to
379/// find out.
380pub const MAX_OPEN_CEILING: usize = (ASSUMED_FD_LIMIT - RESERVED_FDS) / FILES_PER_OPEN_DATABASE;
381
382const _: () = {
383    assert!(MAX_OPEN_CEILING > DEFAULT_MAX_OPEN);
384};
385
386impl Default for WorkspaceLimits {
387    fn default() -> Self {
388        Self {
389            max_open: DEFAULT_MAX_OPEN,
390            idle_timeout_ms: DEFAULT_IDLE_TIMEOUT_MS,
391        }
392    }
393}
394
395impl WorkspaceLimits {
396    /// The effective ceiling: at least one, never more than
397    /// [`MAX_OPEN_CEILING`]. Clamped here, in one place, because `max_open` can
398    /// arrive from a config file and neither end of the range is safe to trust.
399    pub fn ceiling(&self) -> usize {
400        self.max_open.clamp(1, MAX_OPEN_CEILING)
401    }
402}
403
404/// Whether resolving a name that has no file yet creates one.
405#[derive(Clone, Copy, Debug, PartialEq, Eq)]
406pub enum IfMissing {
407    /// Create the database. What makes a workspace usable without a
408    /// registration step: a new chat writes, and its memory exists.
409    Create,
410    /// Fail with [`WorkspaceError::NoSuchDatabase`]. What reads should do, so a
411    /// misspelled name is diagnosed instead of silently answering nothing.
412    Fail,
413}
414
415/// Opens one database at a path. Supplied by the caller so this module stays
416/// out of the business of configuration — [`crate::Settings`] builds one that
417/// applies the engine config, the maintenance policy and a shared embedder.
418pub type Opener = Box<dyn Fn(&Path) -> Result<Database, HostError> + Send + Sync>;
419
420/// One open database and when it was last handed out.
421struct Pooled {
422    name: DbName,
423    db: Database,
424    last_used_ms: u64,
425}
426
427/// Many databases in one directory, opened on demand and kept open for a while.
428///
429/// The pool is a flat `Vec` rather than a map: `max_open` is a handful, and a
430/// linear scan of a handful is cheaper than hashing — the same reason the
431/// engine's own structures stay flat.
432///
433/// **A returned [`Database`] outlives its pool entry.** The handle is an `Arc`,
434/// so eviction drops the pool's copy and nothing else; the file lock is
435/// released when the *last* clone goes. A caller that parks a handle for hours
436/// keeps the database locked for hours, whatever the idle timeout says. Hold
437/// one for the length of a request, as the MCP server does, and this never
438/// comes up.
439///
440/// # What the pool lock covers
441///
442/// One `Mutex` guards the pool, and every path mutates it — a *hit* writes
443/// `last_used_ms`, so an `RwLock` would buy nothing and cost more. It is not
444/// the engine lock: a handle is cloned out and the verb runs with the pool
445/// released, so two threads working in two databases never meet here.
446///
447/// Two things are worth knowing about how long it is held.
448///
449/// [`Workspace::get`] holds it **across the open** — creating the file,
450/// taking the lock, mapping the snapshot, replaying the journal. That is
451/// deliberate (see the comment in the body: dropping it first lets two threads
452/// of one process race for a file and hand the loser a `Busy` it caused
453/// itself), and the cost is that a cold open of one database queues a hit on
454/// an unrelated one. With a pool that warms in a few requests and a `max_open`
455/// in the tens, that queue is short.
456///
457/// The closing paths do the opposite: they take the evicted entries out under
458/// the lock and drop them **after** releasing it. Today that only defers
459/// closing a few file descriptors, since nothing in the handle has a `Drop`.
460/// It is written that way so it stays true if one ever gains one — a
461/// checkpoint-on-close would otherwise turn a timer tick into disk I/O under a
462/// lock every worker wants, which is a stall with no visible cause.
463pub struct Workspace {
464    layout: WorkspaceLayout,
465    open: Opener,
466    limits: WorkspaceLimits,
467    pool: Mutex<Vec<Pooled>>,
468    registry: Mutex<Option<Database>>,
469}
470
471impl Workspace {
472    /// A workspace over `layout`, opening databases with `open`.
473    pub fn new(layout: WorkspaceLayout, open: Opener, limits: WorkspaceLimits) -> Self {
474        Self {
475            layout,
476            open,
477            limits,
478            pool: Mutex::new(Vec::new()),
479            registry: Mutex::new(None),
480        }
481    }
482
483    /// The registry database, opened on first use.
484    ///
485    /// Lazily, and that matters: a process that only ever resolves names it was
486    /// given never opens the registry, so it neither creates the file nor holds
487    /// a lock on it. The registry is a search index — a caller that is not
488    /// searching should not pay for it, and two processes that never search can
489    /// share one workspace without contending over it.
490    ///
491    /// It lives outside the handle pool because it is not one of the databases:
492    /// it has no [`DbName`], it is never evicted, and it is never handed out by
493    /// [`Workspace::get`].
494    ///
495    /// # Errors
496    ///
497    /// [`WorkspaceError::Io`] if the root cannot be created, or whatever the
498    /// open reports — including [`HostError::Locked`] if another process holds
499    /// the registry.
500    pub fn registry(&self) -> Result<Database, WorkspaceError> {
501        let mut slot = self.registry.lock().unwrap_or_else(|e| e.into_inner());
502        if let Some(db) = slot.as_ref() {
503            return Ok(db.clone());
504        }
505        let root = self.layout.root();
506        std::fs::create_dir_all(root).map_err(|e| WorkspaceError::io(root, e))?;
507        let db = (self.open)(&self.layout.registry_path())?;
508        *slot = Some(db.clone());
509        Ok(db)
510    }
511
512    /// Closes the registry handle, if one is open. Returns whether there was
513    /// one. The same liveness concern as [`Workspace::close_idle`]: a held
514    /// registry is a registry no other process can write.
515    pub fn close_registry(&self) -> bool {
516        self.registry
517            .lock()
518            .unwrap_or_else(|e| e.into_inner())
519            .take()
520            .is_some()
521    }
522
523    /// Where the files are.
524    pub fn layout(&self) -> &WorkspaceLayout {
525        &self.layout
526    }
527
528    /// The limits in force.
529    pub fn limits(&self) -> WorkspaceLimits {
530        self.limits
531    }
532
533    /// How many databases are open right now. Observability for tests and
534    /// `stats`; not a number to make decisions on, since it moves.
535    pub fn open_count(&self) -> usize {
536        self.pooled().len()
537    }
538
539    /// Resolves `name` to an open database, opening it if it is not pooled.
540    ///
541    /// `now_ms` is the host clock (unix milliseconds), used only for the idle
542    /// bookkeeping — it is passed in rather than read here for the same reason
543    /// every verb takes `now`: the host owns time.
544    ///
545    /// # Errors
546    ///
547    /// [`WorkspaceError::NoSuchDatabase`] when the file is absent and `missing`
548    /// is [`IfMissing::Fail`]; [`WorkspaceError::Busy`] when another process
549    /// holds the writer; [`WorkspaceError::Io`] if the directory cannot be
550    /// created; [`WorkspaceError::Host`] for anything the open itself rejects.
551    pub fn get(
552        &self,
553        name: &DbName,
554        now_ms: u64,
555        missing: IfMissing,
556    ) -> Result<Database, WorkspaceError> {
557        // The lock is held across the open, deliberately. Releasing it first
558        // would let two callers race to open the same file, and the loser would
559        // see a spurious `Busy` — from its own process. An open is bounded
560        // (the lock is tried, never waited on), so the cost is a short queue.
561        let mut pool = self.pooled();
562
563        if let Some(slot) = pool.iter_mut().find(|p| &p.name == name) {
564            slot.last_used_ms = now_ms;
565            return Ok(slot.db.clone());
566        }
567
568        let path = self.layout.path_of(name);
569        if !self.layout.exists(name) {
570            if missing == IfMissing::Fail {
571                return Err(WorkspaceError::NoSuchDatabase {
572                    name: name.clone(),
573                    path,
574                });
575            }
576            let dir = self.layout.db_dir();
577            std::fs::create_dir_all(&dir).map_err(|e| WorkspaceError::io(&dir, e))?;
578        }
579
580        // Make room *before* opening, so the ceiling counts this database too.
581        let ceiling = self.limits.ceiling();
582        while pool.len() >= ceiling {
583            let lru = Self::least_recently_used(&pool);
584            pool.remove(lru);
585        }
586
587        let db = (self.open)(&path).map_err(|e| match e {
588            HostError::Locked { .. } => WorkspaceError::Busy { name: name.clone() },
589            other => WorkspaceError::Host(other),
590        })?;
591        pool.push(Pooled {
592            name: name.clone(),
593            db: db.clone(),
594            last_used_ms: now_ms,
595        });
596        Ok(db)
597    }
598
599    /// Closes every database unused for longer than the idle timeout, returning
600    /// how many were closed. A no-op when the timeout is `0`.
601    ///
602    /// Call it on a timer. Nothing else releases a file lock a server is
603    /// holding on a database nobody is asking about.
604    pub fn close_idle(&self, now_ms: u64) -> usize {
605        let timeout = self.limits.idle_timeout_ms;
606        if timeout == 0 {
607            return 0;
608        }
609        // `extract_if` rather than `retain` so the entries come *out* instead of
610        // being dropped in place: the guard is a temporary, so it is released at
611        // the semicolon and the handles die on the next line, unlocked.
612        //
613        // `saturating_sub`: a clock that stepped backwards leaves handles open
614        // rather than closing all of them at once.
615        let closing: Vec<Pooled> = self
616            .pooled()
617            .extract_if(.., |p| now_ms.saturating_sub(p.last_used_ms) >= timeout)
618            .collect();
619        closing.len()
620    }
621
622    /// Closes every open handle. The pool's copies, that is — see the note on
623    /// [`Workspace`] about clones the caller still holds.
624    pub fn close_all(&self) -> usize {
625        // Same shape as `close_idle`: the handles leave the pool under the lock
626        // and are dropped once it is released.
627        let closing = std::mem::take(&mut *self.pooled());
628        closing.len()
629    }
630
631    /// The pool guard. A panic in a verb cannot leave the pool half-updated
632    /// (every mutation here is a single push, remove or retain), so a poisoned
633    /// lock is recovered — the same rule the engine lock follows.
634    fn pooled(&self) -> MutexGuard<'_, Vec<Pooled>> {
635        self.pool.lock().unwrap_or_else(|e| e.into_inner())
636    }
637
638    /// Index of the least recently used entry. The pool is never empty here.
639    fn least_recently_used(pool: &[Pooled]) -> usize {
640        let mut oldest = 0;
641        for (i, p) in pool.iter().enumerate() {
642            if p.last_used_ms < pool[oldest].last_used_ms {
643                oldest = i;
644            }
645        }
646        oldest
647    }
648}
649
650impl fmt::Debug for Workspace {
651    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
652        f.debug_struct("Workspace")
653            .field("root", &self.layout.root())
654            .field("limits", &self.limits)
655            .field("open", &self.open_count())
656            .finish()
657    }
658}
659
660/// Fixtures shared by this module's tests and the registry's.
661#[cfg(test)]
662pub(crate) mod testkit {
663    use std::path::PathBuf;
664    use std::sync::Arc;
665    use std::sync::atomic::{AtomicUsize, Ordering};
666
667    use super::{DbName, Opener, Workspace, WorkspaceLayout, WorkspaceLimits};
668    use crate::Database;
669
670    /// A unique temp directory; removed on drop.
671    pub(crate) struct TempDir(pub PathBuf);
672
673    impl TempDir {
674        pub(crate) fn new(tag: &str) -> Self {
675            let dir = std::env::temp_dir().join(format!(
676                "plugmem-workspace-{tag}-{}-{}",
677                std::process::id(),
678                std::time::SystemTime::now()
679                    .duration_since(std::time::UNIX_EPOCH)
680                    .unwrap()
681                    .as_nanos()
682            ));
683            std::fs::create_dir_all(&dir).unwrap();
684            TempDir(dir)
685        }
686    }
687
688    impl Drop for TempDir {
689        fn drop(&mut self) {
690            let _ = std::fs::remove_dir_all(&self.0);
691        }
692    }
693
694    /// A workspace of plain default databases, plus a count of how many times
695    /// the opener actually ran — the only way to tell a pool hit from a reopen.
696    pub(crate) fn workspace(
697        tmp: &TempDir,
698        limits: WorkspaceLimits,
699    ) -> (Workspace, Arc<AtomicUsize>) {
700        let opens = Arc::new(AtomicUsize::new(0));
701        let counted = Arc::clone(&opens);
702        let open: Opener = Box::new(move |path: &std::path::Path| {
703            counted.fetch_add(1, Ordering::SeqCst);
704            Ok(Database::open(path, crate::Config::default())?.0)
705        });
706        (
707            Workspace::new(WorkspaceLayout::new(&tmp.0), open, limits),
708            opens,
709        )
710    }
711
712    /// A name that must parse.
713    pub(crate) fn name(s: &str) -> DbName {
714        DbName::parse(s).unwrap()
715    }
716}
717
718#[cfg(test)]
719mod tests {
720    use std::sync::atomic::Ordering;
721
722    use super::testkit::{TempDir, name, workspace};
723    use super::*;
724
725    fn problem(s: &str) -> NameProblem {
726        match DbName::parse(s) {
727            Err(WorkspaceError::BadName { why, .. }) => why,
728            other => panic!("expected {s:?} to be refused, got {other:?}"),
729        }
730    }
731
732    #[test]
733    fn a_name_admits_only_the_safe_alphabet() {
734        for ok in [
735            "a",
736            "0",
737            "chat-42",
738            "common",
739            "x_y-9",
740            &"a".repeat(MAX_DB_NAME),
741        ] {
742            assert_eq!(DbName::parse(ok).unwrap().as_str(), ok);
743        }
744
745        assert_eq!(problem(""), NameProblem::Empty);
746        assert_eq!(problem(&"a".repeat(MAX_DB_NAME + 1)), NameProblem::TooLong);
747
748        // A leading character that could be read as a flag, a path component or
749        // a hidden file is refused before anything else looks at the string.
750        // A non-ASCII name trips this same check, because its first *byte* is
751        // already outside the alphabet.
752        for bad in ["-x", "_x", ".x", "..", "/x", "Ab", "чат"] {
753            assert_eq!(problem(bad), NameProblem::LeadingChar, "{bad:?}");
754        }
755
756        // Path separators, dots, spaces, uppercase and non-ASCII are not
757        // filtered late — they simply are not names.
758        for bad in ["a/b", "a\\b", "a.b", "a b", "aB", "a:b", "aчат", "a\0b"] {
759            assert_eq!(problem(bad), NameProblem::Character, "{bad:?}");
760        }
761
762        // Windows resolves these to devices in every directory, extension or
763        // no extension, so `con.plugmem` is the console. Refused everywhere,
764        // because a workspace directory is a thing people copy between
765        // machines.
766        for bad in ["con", "nul", "prn", "aux", "com1", "lpt9"] {
767            assert_eq!(problem(bad), NameProblem::ReservedDevice, "{bad:?}");
768        }
769        // Only the exact names — a memory called `console` is fine.
770        for ok in ["console", "con1", "com0", "com10", "nula"] {
771            assert!(DbName::parse(ok).is_ok(), "{ok:?}");
772        }
773    }
774
775    #[test]
776    fn a_name_prints_as_itself() {
777        assert_eq!(DbName::parse("chat-42").unwrap().to_string(), "chat-42");
778        assert_eq!(NameProblem::Empty.to_string(), "it is empty");
779        assert_eq!(
780            NameProblem::TooLong.to_string(),
781            format!("it is longer than {MAX_DB_NAME} bytes")
782        );
783        assert!(NameProblem::LeadingChar.to_string().contains("start with"));
784        assert!(NameProblem::Character.to_string().contains("lowercase"));
785        assert!(
786            NameProblem::ReservedDevice
787                .to_string()
788                .contains("Windows device name")
789        );
790    }
791
792    #[test]
793    fn the_layout_puts_the_registry_out_of_reach_of_names() {
794        let layout = WorkspaceLayout::new("/ws");
795        let name = DbName::parse("chat-42").unwrap();
796
797        assert_eq!(layout.root(), Path::new("/ws"));
798        assert_eq!(layout.db_dir(), Path::new("/ws/db"));
799        assert_eq!(layout.path_of(&name), Path::new("/ws/db/chat-42.plugmem"));
800        assert_eq!(layout.registry_path(), Path::new("/ws/registry.plugmem"));
801
802        // The registry is one level above the databases, so no name — not even
803        // one spelled like the registry file — can resolve onto it.
804        let lookalike = DbName::parse("registry").unwrap();
805        assert_ne!(layout.path_of(&lookalike), layout.registry_path());
806    }
807
808    #[test]
809    fn listing_reads_the_directory_and_ignores_what_is_not_a_database() {
810        let tmp = TempDir::new("list");
811        let layout = WorkspaceLayout::new(&tmp.0);
812
813        // A workspace nobody has written to yet lists nothing rather than failing.
814        assert!(layout.list().unwrap().is_empty());
815
816        std::fs::create_dir_all(layout.db_dir()).unwrap();
817        for file in [
818            "chat-42.plugmem",
819            "common.plugmem",
820            // Sidecars of a database already counted: folded back to its name,
821            // not listed again.
822            "chat-42.plugmem.lock",
823            "chat-42.plugmem.journal",
824            "chat-42.plugmem.snap.3",
825            // Neither is an unrelated file, nor one whose stem is not a name.
826            "notes.txt",
827            "Chat-43.plugmem",
828        ] {
829            std::fs::write(layout.db_dir().join(file), b"").unwrap();
830        }
831
832        let names: Vec<String> = layout
833            .list()
834            .unwrap()
835            .iter()
836            .map(DbName::to_string)
837            .collect();
838        assert_eq!(names, ["chat-42", "common"]);
839
840        assert!(layout.exists(&DbName::parse("chat-42").unwrap()));
841        assert!(!layout.exists(&DbName::parse("nope").unwrap()));
842    }
843
844    #[test]
845    fn a_database_exists_before_its_first_checkpoint() {
846        let tmp = TempDir::new("list-uncheckpointed");
847        let layout = WorkspaceLayout::new(&tmp.0);
848        let fresh = DbName::parse("fresh").unwrap();
849        std::fs::create_dir_all(layout.db_dir()).unwrap();
850
851        // The base path holds the published snapshot, and that is written by
852        // the first checkpoint — so a brand-new database has a journal and no
853        // base file. Reporting it as absent would mean creating over live data.
854        let (db, _) = Database::open(layout.path_of(&fresh), crate::Config::default()).unwrap();
855        db.remember(crate::RememberInput::text(1_000, "not yet checkpointed"))
856            .unwrap();
857        assert!(!layout.path_of(&fresh).exists());
858        assert!(layout.exists(&fresh));
859        assert_eq!(layout.list().unwrap(), [fresh]);
860    }
861
862    #[test]
863    fn an_unreadable_directory_is_an_error_not_an_empty_workspace() {
864        let tmp = TempDir::new("list-io");
865        let layout = WorkspaceLayout::new(&tmp.0);
866        // `db` is a *file*, so reading it as a directory fails with something
867        // other than NotFound — the caller must hear about it.
868        std::fs::write(layout.db_dir(), b"not a directory").unwrap();
869        assert!(matches!(layout.list(), Err(WorkspaceError::Io { .. })));
870    }
871
872    #[test]
873    fn every_failure_names_what_the_caller_typed() {
874        let busy = WorkspaceError::Busy {
875            name: DbName::parse("chat-42").unwrap(),
876        };
877        assert!(busy.to_string().contains("chat-42"));
878
879        let host = WorkspaceError::from(HostError::Embed("no".into()));
880        assert!(matches!(host, WorkspaceError::Host(HostError::Embed(_))));
881
882        let io = WorkspaceError::io(
883            Path::new("/ws"),
884            std::io::Error::new(std::io::ErrorKind::PermissionDenied, "denied"),
885        );
886        assert!(io.to_string().contains("/ws"));
887
888        let missing = WorkspaceError::NoSuchDatabase {
889            name: DbName::parse("gone").unwrap(),
890            path: PathBuf::from("/ws/db/gone.plugmem"),
891        };
892        assert!(missing.to_string().contains("gone"));
893    }
894
895    #[test]
896    fn a_pooled_database_is_reused_and_a_missing_one_is_created_only_on_request() {
897        let tmp = TempDir::new("pool-reuse");
898        let (ws, opens) = workspace(&tmp, WorkspaceLimits::default());
899        let chat = name("chat-42");
900
901        // Reading a name nobody has written names the name, not a path the
902        // caller never typed.
903        let missed = ws.get(&chat, 1_000, IfMissing::Fail).unwrap_err();
904        assert!(
905            matches!(&missed, WorkspaceError::NoSuchDatabase { name, .. } if name == &chat),
906            "{missed}"
907        );
908        assert_eq!(opens.load(Ordering::SeqCst), 0);
909        assert!(!ws.layout().db_dir().exists());
910
911        // Writing creates it, directory and all.
912        let db = ws.get(&chat, 1_000, IfMissing::Create).unwrap();
913        db.remember(crate::RememberInput::text(1_000, "prefers tokio"))
914            .unwrap();
915        assert!(ws.layout().exists(&chat));
916        assert_eq!(ws.open_count(), 1);
917
918        // The second call is a pool hit: same state, no reopen.
919        let again = ws.get(&chat, 2_000, IfMissing::Fail).unwrap();
920        assert_eq!(again.stats().facts, 1);
921        assert_eq!(opens.load(Ordering::SeqCst), 1);
922
923        assert!(format!("{ws:?}").contains("open: 1"));
924    }
925
926    #[test]
927    fn databases_in_one_workspace_do_not_see_each_other() {
928        let tmp = TempDir::new("isolation");
929        let (ws, _) = workspace(&tmp, WorkspaceLimits::default());
930
931        for (db, text) in [
932            ("chat-42", "the sky is blue"),
933            ("chat-43", "the sky is red"),
934        ] {
935            ws.get(&name(db), 1_000, IfMissing::Create)
936                .unwrap()
937                .remember(crate::RememberInput::text(1_000, text))
938                .unwrap();
939        }
940
941        // The same query, the same fact id in each — and each answers only for
942        // itself. Ids are per database, which is why they can be plain `[f1]`.
943        for (db, expected) in [("chat-42", "blue"), ("chat-43", "red")] {
944            let out = ws
945                .get(&name(db), 2_000, IfMissing::Fail)
946                .unwrap()
947                .recall(crate::RecallQuery::text(2_000, "sky"))
948                .unwrap();
949            assert_eq!(out.facts.len(), 1, "{db}");
950            assert!(out.rendered.contains(expected), "{db}: {}", out.rendered);
951        }
952    }
953
954    #[test]
955    fn the_ceiling_evicts_the_least_recently_used() {
956        let tmp = TempDir::new("pool-evict");
957        let (ws, opens) = workspace(
958            &tmp,
959            WorkspaceLimits {
960                max_open: 2,
961                ..WorkspaceLimits::default()
962            },
963        );
964
965        // Three databases through a pool of two, touching `a` in between so it
966        // is `b` that is coldest when `c` arrives.
967        ws.get(&name("a"), 1_000, IfMissing::Create).unwrap();
968        ws.get(&name("b"), 2_000, IfMissing::Create).unwrap();
969        ws.get(&name("a"), 3_000, IfMissing::Fail).unwrap();
970        ws.get(&name("c"), 4_000, IfMissing::Create).unwrap();
971        assert_eq!(ws.open_count(), 2);
972        assert_eq!(opens.load(Ordering::SeqCst), 3);
973
974        // `a` is still pooled; `b` was evicted and has to be reopened.
975        ws.get(&name("a"), 5_000, IfMissing::Fail).unwrap();
976        assert_eq!(opens.load(Ordering::SeqCst), 3);
977        ws.get(&name("b"), 6_000, IfMissing::Fail).unwrap();
978        assert_eq!(opens.load(Ordering::SeqCst), 4);
979    }
980
981    #[test]
982    fn a_ceiling_of_zero_still_serves_one_database() {
983        let tmp = TempDir::new("pool-zero");
984        let (ws, opens) = workspace(
985            &tmp,
986            WorkspaceLimits {
987                max_open: 0,
988                idle_timeout_ms: 0,
989            },
990        );
991        ws.get(&name("a"), 1_000, IfMissing::Create).unwrap();
992        ws.get(&name("b"), 2_000, IfMissing::Create).unwrap();
993        assert_eq!(ws.open_count(), 1);
994        assert_eq!(opens.load(Ordering::SeqCst), 2);
995
996        // A zero timeout disables the sweep rather than closing everything.
997        assert_eq!(ws.close_idle(u64::MAX), 0);
998        assert_eq!(ws.open_count(), 1);
999    }
1000
1001    #[test]
1002    fn an_idle_database_is_closed_and_its_lock_released() {
1003        let tmp = TempDir::new("pool-idle");
1004        let (ws, _) = workspace(
1005            &tmp,
1006            WorkspaceLimits {
1007                max_open: 8,
1008                idle_timeout_ms: 1_000,
1009            },
1010        );
1011        let chat = name("chat-42");
1012        let path = ws.layout().path_of(&chat);
1013        drop(ws.get(&chat, 1_000, IfMissing::Create).unwrap());
1014
1015        // Inside the window nothing moves, and the file stays locked.
1016        assert_eq!(ws.close_idle(1_500), 0);
1017        assert!(matches!(
1018            Database::open(&path, crate::Config::default()),
1019            Err(HostError::Locked { .. })
1020        ));
1021
1022        // A clock that stepped backwards must not close everything.
1023        assert_eq!(ws.close_idle(500), 0);
1024
1025        // Past it, the handle goes and the database is reachable again — this
1026        // is the whole point of the timeout, not memory.
1027        assert_eq!(ws.close_idle(2_000), 1);
1028        assert_eq!(ws.open_count(), 0);
1029        assert!(Database::open(&path, crate::Config::default()).is_ok());
1030    }
1031
1032    #[test]
1033    fn a_handle_held_by_a_caller_outlives_its_pool_entry() {
1034        let tmp = TempDir::new("pool-outlive");
1035        let (ws, _) = workspace(&tmp, WorkspaceLimits::default());
1036        let chat = name("chat-42");
1037        let held = ws.get(&chat, 1_000, IfMissing::Create).unwrap();
1038        let path = ws.layout().path_of(&chat);
1039
1040        assert_eq!(ws.close_all(), 1);
1041        assert_eq!(ws.open_count(), 0);
1042
1043        // The pool let go; the caller did not, so the lock is still taken and
1044        // the handle still works. Documented on `Workspace`, checked here.
1045        held.remember(crate::RememberInput::text(2_000, "still mine"))
1046            .unwrap();
1047        assert!(matches!(
1048            Database::open(&path, crate::Config::default()),
1049            Err(HostError::Locked { .. })
1050        ));
1051        drop(held);
1052        assert!(Database::open(&path, crate::Config::default()).is_ok());
1053    }
1054
1055    #[test]
1056    fn a_database_held_by_another_process_is_reported_by_name() {
1057        let tmp = TempDir::new("pool-busy");
1058        let (ws, _) = workspace(&tmp, WorkspaceLimits::default());
1059        let chat = name("chat-42");
1060        let path = ws.layout().path_of(&chat);
1061        std::fs::create_dir_all(ws.layout().db_dir()).unwrap();
1062
1063        // Stand in for the other process with a handle the pool knows nothing
1064        // about: the lock is what matters, not who holds it.
1065        let outsider = Database::open(&path, crate::Config::default()).unwrap().0;
1066        let e = ws.get(&chat, 1_000, IfMissing::Create).unwrap_err();
1067        assert!(matches!(&e, WorkspaceError::Busy { name } if name == &chat));
1068        assert!(e.to_string().contains("chat-42"), "{e}");
1069
1070        drop(outsider);
1071        assert!(ws.get(&chat, 2_000, IfMissing::Fail).is_ok());
1072    }
1073
1074    #[test]
1075    fn an_open_that_fails_for_another_reason_keeps_its_own_error() {
1076        let tmp = TempDir::new("pool-open-err");
1077        let open: Opener = Box::new(|_| Err(HostError::Embed("no provider".into())));
1078        let ws = Workspace::new(
1079            WorkspaceLayout::new(&tmp.0),
1080            open,
1081            WorkspaceLimits::default(),
1082        );
1083        let e = ws.get(&name("a"), 1_000, IfMissing::Create).unwrap_err();
1084        assert!(
1085            matches!(e, WorkspaceError::Host(HostError::Embed(_))),
1086            "{e}"
1087        );
1088    }
1089
1090    #[test]
1091    fn a_directory_that_cannot_be_created_is_an_error_not_a_panic() {
1092        let tmp = TempDir::new("pool-mkdir");
1093        // `db` is a file, so `create_dir_all` cannot make it a directory.
1094        std::fs::write(tmp.0.join(DB_DIR), b"in the way").unwrap();
1095        let (ws, _) = workspace(&tmp, WorkspaceLimits::default());
1096        assert!(matches!(
1097            ws.get(&name("a"), 1_000, IfMissing::Create),
1098            Err(WorkspaceError::Io { .. })
1099        ));
1100    }
1101
1102    proptest::proptest! {
1103        /// The property the whole design rests on: whatever a caller sends,
1104        /// either it is refused, or the file it names sits directly inside
1105        /// `<root>/db` — one component, no traversal, no absolute path, no
1106        /// device name. Checked over arbitrary strings rather than a list of
1107        /// attacks somebody thought of.
1108        #[test]
1109        fn a_name_that_parses_can_only_resolve_inside_the_workspace(s in ".*") {
1110            let Ok(name) = DbName::parse(&s) else { return Ok(()) };
1111            let layout = WorkspaceLayout::new("/ws");
1112            let path = layout.path_of(&name);
1113
1114            let rest: Vec<_> = path
1115                .strip_prefix(layout.db_dir())
1116                .expect("resolved outside the workspace")
1117                .components()
1118                .collect();
1119            let expected = format!("{s}.{DB_EXT}");
1120            proptest::prop_assert_eq!(rest.len(), 1);
1121            proptest::prop_assert_eq!(
1122                path.file_name().and_then(|n| n.to_str()),
1123                Some(expected.as_str())
1124            );
1125        }
1126    }
1127}