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}