mnemo/store.rs
1//! The Mnemo database — ties the storage engine and the memory model together.
2//!
3//! Layout of a populated file:
4//!
5//! ```text
6//! page 0 : Header (unencrypted)
7//! pages 1..W : write-ahead log region
8//! pages W.. : encrypted record runs + catalog runs + index +
9//! snapshot manifest (append-only)
10//! ```
11//!
12//! Durability is provided by a **write-ahead log** ([`crate::wal`]). A `flush`
13//! is one transaction: record data pages are written copy-on-write to fresh
14//! pages, then the new catalog, ANN index, and header are logged to the WAL
15//! and committed with a single fsync. A checkpoint then folds the WAL into the
16//! home pages. A crash before the commit leaves the previous state intact; a
17//! crash after it is repaired by replaying the WAL on open.
18//!
19//! Because pages are only ever appended, every past transaction's runs survive
20//! on disk. Each flush also appends an entry to a **snapshot manifest**, so
21//! [`Mnemo::restore_to`] can reinstate any past state. Stale pages — and the
22//! history along with them — are reclaimed by [`Mnemo::compact_file`].
23
24use std::collections::HashMap;
25use std::fs::{File, OpenOptions};
26use std::io::{Read, Seek, SeekFrom, Write};
27use std::path::{Path, PathBuf};
28
29use fs4::fs_std::FileExt;
30use serde::{Deserialize, Serialize};
31use ulid::Ulid;
32
33use crate::crypto::{self, KdfParams};
34use crate::error::{MnemoError, Result};
35use crate::format::{Header, FLAG_ENCRYPTED, PAGE_SIZE, PAYLOAD, VERSION, WRAPPED_DEK_LEN};
36use crate::index::{IndexConfig, IndexInfo, IvfPqIndex};
37use crate::memory::{self, Memory, MemoryType, Metric, Scope, ScoreWeights};
38use crate::pager::Pager;
39use crate::result_cache::{
40 self, BatchState, CacheBudget, CacheDirectoryEntry, CacheEntry, CacheFlushPolicy, CachePutOpts,
41 CacheStats, CachedValue, CounterState, DirectoryIndex, SemanticCachePutOpts,
42};
43use crate::wal;
44
45/// Default initial size of the write-ahead log region, in pages (64 KiB).
46/// The region grows automatically when a transaction's control plane outgrows
47/// it, so this is a *floor* on the fresh-file footprint, not a cap on
48/// transaction size. Was 64 pages (512 KiB) in v0.1.0 — lowered because that
49/// reservation dominated small-file size (~62% of a 31-memory dogfood file).
50const DEFAULT_WAL_PAGES: u64 = 8;
51
52/// Hard floor on the WAL reservation. A single header-page flush already
53/// needs one full DATA frame plus a COMMIT frame; two pages are the bare
54/// minimum that lets the first transaction commit without immediate growth.
55const MIN_WAL_PAGES: u64 = 2;
56
57/// Default cap on retained snapshot manifest entries. Each `flush` appends
58/// one entry; without a cap, long-running processes accumulate entries
59/// forever, growing per-flush manifest-serialize cost as O(total flushes)
60/// and the manifest run size on disk in lockstep. 256 keeps roughly a
61/// week of hourly snapshots while staying tiny (256 × ~82 bytes ≈ 21 KiB
62/// of manifest). Set `MnemoConfig::max_snapshots = 0` to disable the cap.
63pub const DEFAULT_MAX_SNAPSHOTS: usize = 256;
64
65/// Upper bound for the WAL scan when recovering a torn-header file. The
66/// real WAL region size lives in the header, but a torn header is exactly
67/// the case where we can't read it — so we scan a generous fixed window
68/// from the conventional WAL start, relying on `wal::recover` stopping at
69/// the first non-magic frame. 64 covers the historical pre-0.2 default
70/// of a 64-page reservation; users who configure larger WALs and then
71/// suffer a torn header at the same time will need offline recovery.
72const TORN_HEADER_SCAN_PAGES: u64 = 64;
73
74/// Configuration for creating a new database.
75#[derive(Clone, Copy, Debug)]
76pub struct MnemoConfig {
77 /// Embedding dimensionality. Every stored vector must match this.
78 pub dimensions: usize,
79 /// Key-derivation parameters.
80 pub kdf: KdfParams,
81 /// Initial size of the WAL region in 8 KiB pages. Default is
82 /// [`DEFAULT_WAL_PAGES`] (8 pages = 64 KiB), which keeps fresh files
83 /// small; the region auto-grows when a transaction's control plane needs
84 /// more space, so raising this is only a hint for the expected steady-state
85 /// transaction size. Write-heavy databases that routinely flush large
86 /// catalogs or ANN indexes can set this higher (e.g. 64 = 512 KiB) to
87 /// avoid early grow events; tiny embedded uses can leave it at the
88 /// default. Clamped to a minimum of [`MIN_WAL_PAGES`] (2 pages).
89 pub wal_pages_initial: u64,
90 /// Maximum number of snapshot manifest entries to retain. Each `flush`
91 /// appends one entry; once the cap is hit, the *oldest* entry is dropped
92 /// from the in-memory manifest before the new one is added, so the
93 /// retained set is always the most-recent N. Defaults to
94 /// [`DEFAULT_MAX_SNAPSHOTS`] (256). Set to `0` to disable the cap and
95 /// retain every snapshot forever (the pre-v0.3 behavior).
96 ///
97 /// Pages referenced *only* by pruned snapshots stay on disk until the
98 /// next [`Mnemo::compact_file`], which reclaims them. Point-in-time
99 /// recovery via [`Mnemo::restore_to`] into a pruned `txn_id` returns
100 /// [`MnemoError::NotFound`].
101 ///
102 /// Apply to an already-open database with [`Mnemo::set_max_snapshots`].
103 pub max_snapshots: usize,
104}
105
106impl Default for MnemoConfig {
107 fn default() -> Self {
108 Self {
109 dimensions: 768,
110 kdf: KdfParams::secure(),
111 wal_pages_initial: DEFAULT_WAL_PAGES,
112 max_snapshots: DEFAULT_MAX_SNAPSHOTS,
113 }
114 }
115}
116
117/// Catalog entry: maps a memory ID to its page run on disk.
118///
119/// **Schema (v5+):** carries `accessed_at` and `access_count` so `recall`
120/// can update access stats by mutating the catalog entry rather than
121/// rewriting the entire record (Phase 2.1 of the improvement plan). The
122/// values in a memory's serialized record body become a stale snapshot
123/// after the first recall; [`Mnemo::read_memory`] re-populates them from
124/// the catalog so consumers always see the live values.
125#[derive(Serialize, Deserialize, Clone, Debug)]
126struct CatalogEntry {
127 /// ULID as a raw `u128`.
128 id: u128,
129 start_page: u64,
130 page_count: u32,
131 /// Exact serialized byte length of the record.
132 len: u32,
133 deleted: bool,
134 /// Unix seconds of the most recent recall hit (or write, for fresh entries).
135 accessed_at: i64,
136 /// How many times `recall` has surfaced this memory.
137 access_count: u32,
138}
139
140/// Frozen v4 catalog entry shape, used only by the v4→v5 migration path in
141/// [`Mnemo::open`]. Its layout matches the catalog encoding written by
142/// pre-v5 builds (5 positional fields). Do not change this struct.
143#[derive(Serialize, Deserialize, Clone, Debug)]
144struct CatalogEntryV4 {
145 id: u128,
146 start_page: u64,
147 page_count: u32,
148 len: u32,
149 deleted: bool,
150}
151
152/// Returned by [`Mnemo::prepare_for_flush`] and consumed by [`Mnemo::flush`].
153/// Carries the already-serialized control-plane bytes through the prelude
154/// so we don't re-serialize after the lease has been persisted.
155struct FlushPrelude {
156 cat_bytes: Option<Vec<u8>>,
157 idx_bytes: Option<Vec<u8>>,
158 /// Serialized result-cache directory bytes when `dirty_cache` is
159 /// set (Phase 10.1). `None` if the cache didn't change since the
160 /// last flush — in that case the header keeps its current
161 /// `cache_start` / `cache_pages` / `cache_len` and no fresh cache
162 /// run is written.
163 cache_bytes: Option<Vec<u8>>,
164}
165
166/// One entry in the append-only snapshot manifest. Because record, catalog,
167/// and index pages are only ever *appended*, the runs a past flush wrote are
168/// still on disk; a `Snapshot` is the set of pointers needed to reconstruct
169/// the database exactly as that flush left it.
170#[derive(Serialize, Deserialize, Clone, Debug)]
171struct Snapshot {
172 txn_id: u64,
173 created_at: i64,
174 catalog_start: u64,
175 catalog_pages: u64,
176 catalog_len: u64,
177 index_start: u64,
178 index_pages: u64,
179 index_len: u64,
180 memory_count: u64,
181}
182
183/// A restorable point in a database's history — see [`Mnemo::snapshots`].
184#[derive(Clone, Copy, Debug)]
185pub struct SnapshotInfo {
186 /// Id of the transaction that produced this snapshot (monotonic from 1).
187 pub txn_id: u64,
188 /// When the snapshot was committed (unix seconds).
189 pub created_at: i64,
190 /// Live memory count captured in the snapshot.
191 pub memory_count: u64,
192}
193
194/// A retrieval request for [`Mnemo::recall`].
195#[derive(Clone, Debug)]
196pub struct RecallRequest {
197 /// Query embedding.
198 pub query: Vec<f32>,
199 /// Maximum results to return.
200 pub top_k: usize,
201 /// Restrict to these memory types (`None` = all types).
202 pub memory_types: Option<Vec<MemoryType>>,
203 /// Restrict to a single agent's view: its own memories plus shared ones.
204 pub agent_id: Option<String>,
205 /// Similarity metric.
206 pub metric: Metric,
207 /// Multi-signal score weights.
208 pub weights: ScoreWeights,
209 /// Index override: partitions to probe (`None` = index default). Ignored
210 /// when no ANN index is present.
211 pub n_probe: Option<usize>,
212 /// Index override: candidates to rerank (`None` = index default). Ignored
213 /// when no ANN index is present.
214 pub n_rerank: Option<usize>,
215 /// Whether to update each result's `accessed_at` and `access_count` in
216 /// the catalog. Defaults to `true`, matching pre-v5 behavior.
217 ///
218 /// Set to `false` for **fully read-only recall** — useful for batch
219 /// scoring, dry-run scoring, or recalls run by introspection tooling
220 /// (e.g. `mnemo about` consumers) where you don't want the score's own
221 /// observation to perturb the database. With `false`, recall does not
222 /// dirty the catalog and the next `flush` is a no-op.
223 pub track_access: bool,
224}
225
226impl RecallRequest {
227 /// A request with sensible defaults for a query embedding.
228 pub fn new(query: Vec<f32>) -> Self {
229 Self {
230 query,
231 top_k: 10,
232 memory_types: None,
233 agent_id: None,
234 metric: Metric::Cosine,
235 weights: ScoreWeights::default(),
236 n_probe: None,
237 n_rerank: None,
238 track_access: true,
239 }
240 }
241 /// Set the result cap.
242 pub fn top_k(mut self, k: usize) -> Self {
243 self.top_k = k;
244 self
245 }
246 /// Restrict to specific memory types.
247 pub fn types(mut self, t: Vec<MemoryType>) -> Self {
248 self.memory_types = Some(t);
249 self
250 }
251 /// Restrict to one agent's view.
252 pub fn agent(mut self, agent_id: impl Into<String>) -> Self {
253 self.agent_id = Some(agent_id.into());
254 self
255 }
256 /// Set the similarity metric (default: cosine).
257 pub fn metric(mut self, metric: Metric) -> Self {
258 self.metric = metric;
259 self
260 }
261 /// Replace the multi-signal score weights.
262 pub fn weights(mut self, weights: ScoreWeights) -> Self {
263 self.weights = weights;
264 self
265 }
266 /// Override the number of IVF partitions probed (accuracy/speed dial).
267 pub fn n_probe(mut self, n: usize) -> Self {
268 self.n_probe = Some(n);
269 self
270 }
271 /// Override the number of candidates reranked exactly (accuracy dial).
272 pub fn n_rerank(mut self, n: usize) -> Self {
273 self.n_rerank = Some(n);
274 self
275 }
276 /// Toggle whether recall updates `accessed_at` / `access_count` on
277 /// the returned memories' catalog entries (default `true`). Pass
278 /// `false` for a fully read-only recall — see the field doc.
279 pub fn track_access(mut self, track: bool) -> Self {
280 self.track_access = track;
281 self
282 }
283}
284
285/// One scored result from [`Mnemo::recall`].
286#[derive(Clone, Debug)]
287pub struct RecallResult {
288 /// The retrieved memory.
289 pub memory: Memory,
290 /// Combined multi-signal score.
291 pub score: f32,
292 /// The bare similarity component (before other signals).
293 pub similarity: f32,
294}
295
296/// Summary statistics for a database.
297#[derive(Clone, Debug)]
298pub struct Stats {
299 /// Live (non-deleted) memory count.
300 pub memories: usize,
301 /// Tombstoned entries awaiting compaction.
302 pub deleted: usize,
303 /// Embedding dimensionality.
304 pub dimensions: usize,
305 /// File size in bytes.
306 pub file_bytes: u64,
307 /// Distinct agent IDs present.
308 pub agents: Vec<String>,
309 /// Whether pages are encrypted (always true in v1).
310 pub encrypted: bool,
311 /// Creation time (unix seconds).
312 pub created_at: i64,
313 /// ANN index shape, if an index has been built.
314 pub index: Option<IndexInfo>,
315 /// Current size of the write-ahead log region, in 8 KiB pages.
316 pub wal_pages: u64,
317}
318
319/// Result of a [`Mnemo::compact_file`] run.
320#[derive(Clone, Copy, Debug)]
321pub struct CompactReport {
322 /// Live memories before compaction.
323 pub before: usize,
324 /// Live memories after compaction (expired ones dropped).
325 pub after: usize,
326}
327
328/// An encrypted, single-file agent memory database.
329pub struct Mnemo {
330 pager: Pager,
331 header: Header,
332 catalog: Vec<CatalogEntry>,
333 index: HashMap<u128, usize>,
334 #[allow(dead_code)]
335 path: PathBuf,
336 dimensions: usize,
337 kdf: KdfParams,
338 /// Set whenever the catalog changes; drives whether `flush` rewrites it.
339 dirty_catalog: bool,
340 /// Optional IVF+PQ approximate-nearest-neighbour index.
341 ann: Option<IvfPqIndex>,
342 /// Set whenever the ANN index changes; drives whether `flush` rewrites it.
343 dirty_index: bool,
344 /// Append-only manifest of committed snapshots, oldest first. Capped
345 /// at [`Mnemo::max_snapshots`] entries on every flush — the oldest
346 /// entries are pruned first.
347 manifest: Vec<Snapshot>,
348 /// Maximum manifest entries to retain across flushes. `0` disables
349 /// the cap (retain forever). Defaults to [`DEFAULT_MAX_SNAPSHOTS`].
350 max_snapshots: usize,
351 /// If `true`, this handle was opened via [`Mnemo::open_read_only`] and
352 /// every mutating API returns [`MnemoError::ReadOnly`]. The underlying
353 /// file was opened without write permission and holds an OS shared
354 /// lock; the OS would reject a write() anyway, but gating in Rust
355 /// gives callers a clean, typed error.
356 read_only: bool,
357 /// Result-cache directory (Phase 10.1). See [`crate::result_cache`].
358 /// Loaded from `header.cache_start` on open; persisted on `flush`
359 /// via the same seal_run pipeline as catalog/index/manifest.
360 cache_dir: Vec<CacheDirectoryEntry>,
361 /// In-memory index over `cache_dir` for O(1) `(namespace, key_hash)`
362 /// lookups. Rebuilt on open + kept in sync on `cache_put`/`delete`.
363 cache_idx: DirectoryIndex,
364 /// Set whenever `cache_dir` changes; drives whether `flush` rewrites
365 /// the cache directory run.
366 dirty_cache: bool,
367 /// Per-namespace eviction budgets. Missing entries fall back to
368 /// [`CacheBudget::default`]. Not persisted (per-handle policy).
369 cache_budgets: std::collections::HashMap<String, CacheBudget>,
370 /// Flush policy for cache mutations. Memory writes ignore this and
371 /// always behave as `Strict`.
372 cache_flush_policy: CacheFlushPolicy,
373 /// Bookkeeping for `CacheFlushPolicy::Batched` — pending mutation
374 /// count and first-dirty timestamp.
375 cache_batch: BatchState,
376 /// In-memory hit/miss/eviction counters (reset on reopen).
377 cache_counters: CounterState,
378}
379
380/// Open a file and take an OS advisory lock, returning `Locked { path }`
381/// if the requested lock cannot be acquired. On drop of the returned
382/// [`File`], the OS releases the lock. `flock`/`LockFileEx` semantics
383/// (per file-descriptor on Unix, per-open-file-handle on Windows) mean
384/// the lock is scoped to this handle: cloning the file (or the returned
385/// `File`) shares the lock, but a fresh `open()` from any process gets a
386/// distinct handle and will collide.
387fn open_with_lock(path: &Path, write: bool, create: bool) -> Result<File> {
388 let mut opts = OpenOptions::new();
389 opts.read(true);
390 if write {
391 opts.write(true);
392 }
393 if create {
394 opts.create(true).truncate(true);
395 }
396 let file = opts.open(path)?;
397 // On success the returned `Result<()>` is `Ok(())`. On conflict:
398 // * Unix `flock(2)` → `ErrorKind::WouldBlock` (EWOULDBLOCK / EAGAIN)
399 // * Windows `LockFileEx` → raw OS code 33 (ERROR_LOCK_VIOLATION)
400 // or 32 (ERROR_SHARING_VIOLATION), both surfaced by Rust as
401 // `ErrorKind::Uncategorized` because `std::io::ErrorKind`
402 // doesn't carry a portable "locked" variant.
403 // Anything else is a real I/O failure (permissions, invalid handle,
404 // ...) and gets bubbled up as `MnemoError::Io`.
405 let result = if write {
406 FileExt::try_lock_exclusive(&file)
407 } else {
408 FileExt::try_lock_shared(&file)
409 };
410 match result {
411 Ok(()) => Ok(file),
412 Err(e) if is_lock_conflict(&e) => Err(MnemoError::Locked {
413 path: path.to_path_buf(),
414 }),
415 Err(e) => Err(MnemoError::Io(e)),
416 }
417}
418
419/// Was this I/O error the OS reporting a conflicting file lock, or a
420/// genuine I/O failure? Platform-conditional because
421/// [`std::io::ErrorKind`] carries no portable "resource-locked" variant
422/// as of Rust 1.75 — Windows surfaces `ERROR_LOCK_VIOLATION` /
423/// `ERROR_SHARING_VIOLATION` as `Uncategorized`.
424fn is_lock_conflict(e: &std::io::Error) -> bool {
425 if e.kind() == std::io::ErrorKind::WouldBlock {
426 return true;
427 }
428 #[cfg(windows)]
429 if let Some(code) = e.raw_os_error() {
430 // 33 = ERROR_LOCK_VIOLATION, 32 = ERROR_SHARING_VIOLATION —
431 // both mean "someone else holds an incompatible lock on this
432 // region of the file". See
433 // https://learn.microsoft.com/en-us/windows/win32/debug/system-error-codes--0-499-
434 if code == 33 || code == 32 {
435 return true;
436 }
437 }
438 false
439}
440
441/// Read a run of consecutive encrypted pages and concatenate their plaintext.
442fn read_run_bytes(pager: &mut Pager, start: u64, pages: u64, len: u64) -> Result<Vec<u8>> {
443 let mut buf = Vec::with_capacity(len as usize);
444 for i in 0..pages {
445 buf.extend_from_slice(&pager.read_page(start + i)?);
446 }
447 buf.truncate(len as usize);
448 Ok(buf)
449}
450
451/// Load a v5+ catalog run (an empty `pages` yields an empty catalog).
452fn load_catalog(pager: &mut Pager, start: u64, pages: u64, len: u64) -> Result<Vec<CatalogEntry>> {
453 if pages == 0 {
454 return Ok(Vec::new());
455 }
456 let buf = read_run_bytes(pager, start, pages, len)?;
457 rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
458}
459
460/// Load a **v4** catalog run using the frozen [`CatalogEntryV4`] shape.
461/// Used only by the v4→v5 migration path in [`Mnemo::open`].
462fn load_catalog_v4(
463 pager: &mut Pager,
464 start: u64,
465 pages: u64,
466 len: u64,
467) -> Result<Vec<CatalogEntryV4>> {
468 if pages == 0 {
469 return Ok(Vec::new());
470 }
471 let buf = read_run_bytes(pager, start, pages, len)?;
472 rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
473}
474
475/// Walk a v4 catalog and synthesize the v5 shape. For each live entry,
476/// reads the memory's serialized body and copies its `accessed_at` and
477/// `access_count` into the catalog row (the v4 record body still carries
478/// them — they're moving *from* the body *to* the catalog). Deleted
479/// entries skip the read and zero the fields.
480fn migrate_v4_catalog(pager: &mut Pager, v4: Vec<CatalogEntryV4>) -> Result<Vec<CatalogEntry>> {
481 let mut out = Vec::with_capacity(v4.len());
482 for e in &v4 {
483 let (accessed_at, access_count) = if e.deleted {
484 (0i64, 0u32)
485 } else {
486 let mut buf = Vec::with_capacity(e.len as usize);
487 for i in 0..e.page_count as u64 {
488 buf.extend_from_slice(&pager.read_page(e.start_page + i)?);
489 }
490 buf.truncate(e.len as usize);
491 let mem: Memory = rmp_serde::from_slice(&buf)
492 .map_err(|err| MnemoError::Serialize(err.to_string()))?;
493 (mem.accessed_at, mem.access_count)
494 };
495 out.push(CatalogEntry {
496 id: e.id,
497 start_page: e.start_page,
498 page_count: e.page_count,
499 len: e.len,
500 deleted: e.deleted,
501 accessed_at,
502 access_count,
503 });
504 }
505 Ok(out)
506}
507
508/// Load the snapshot manifest (an empty `pages` yields an empty manifest).
509fn load_manifest(pager: &mut Pager, start: u64, pages: u64, len: u64) -> Result<Vec<Snapshot>> {
510 if pages == 0 {
511 return Ok(Vec::new());
512 }
513 let buf = read_run_bytes(pager, start, pages, len)?;
514 rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
515}
516
517/// Load the result-cache directory (v8+). An empty `pages` yields an
518/// empty directory, which is the natural state for v7 files migrated
519/// to v8 (no cache entries yet) and for freshly created v8 files.
520fn load_cache_directory(
521 pager: &mut Pager,
522 start: u64,
523 pages: u64,
524 len: u64,
525) -> Result<Vec<CacheDirectoryEntry>> {
526 if pages == 0 {
527 return Ok(Vec::new());
528 }
529 let buf = read_run_bytes(pager, start, pages, len)?;
530 rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
531}
532
533/// Load an ANN index run, validating its dimensionality.
534fn load_index(
535 pager: &mut Pager,
536 start: u64,
537 pages: u64,
538 len: u64,
539 dims: usize,
540) -> Result<Option<IvfPqIndex>> {
541 if pages == 0 {
542 return Ok(None);
543 }
544 let buf = read_run_bytes(pager, start, pages, len)?;
545 let mut idx: IvfPqIndex =
546 rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
547 if idx.dims() != dims {
548 return Err(MnemoError::Invalid(
549 "ANN index dimensionality does not match the database".into(),
550 ));
551 }
552 idx.rebuild_assignment();
553 Ok(Some(idx))
554}
555
556/// True if a memory's metadata marks it as the canonical onboarding manifest
557/// (`metadata.topic == "manifest"`). Used by [`Mnemo::about`] to hoist the
558/// manifest to the top of the briefing regardless of importance ordering.
559fn is_manifest(m: &Memory) -> bool {
560 m.metadata
561 .get("topic")
562 .and_then(|v| v.as_str())
563 .map(|s| s.eq_ignore_ascii_case("manifest"))
564 .unwrap_or(false)
565}
566
567/// Build the ULID → catalog-slot lookup map.
568fn build_id_index(catalog: &[CatalogEntry]) -> HashMap<u128, usize> {
569 let mut m = HashMap::with_capacity(catalog.len());
570 for (i, e) in catalog.iter().enumerate() {
571 m.insert(e.id, i);
572 }
573 m
574}
575
576impl Mnemo {
577 /// Create a brand-new encrypted database at `path`.
578 pub fn create(
579 path: impl Into<PathBuf>,
580 passphrase: &str,
581 config: MnemoConfig,
582 ) -> Result<Mnemo> {
583 let path: PathBuf = path.into();
584 if config.dimensions == 0 {
585 return Err(MnemoError::Invalid("dimensions must be > 0".into()));
586 }
587 // Exclusive lock covers the write path; second writer sees `Locked`.
588 let file = open_with_lock(&path, /*write=*/ true, /*create=*/ true)?;
589
590 let salt = crypto::random_salt();
591 let dek = crypto::random_dek();
592 let kek = crypto::derive_kek(passphrase.as_bytes(), &salt, config.kdf)?;
593 let dek_nonce = crypto::random_nonce();
594 let wrapped = crypto::wrap_dek(&kek, &dek_nonce, &dek)?;
595 if wrapped.len() != WRAPPED_DEK_LEN {
596 return Err(MnemoError::Crypto("unexpected wrapped DEK length".into()));
597 }
598 let mut wrapped_dek = [0u8; WRAPPED_DEK_LEN];
599 wrapped_dek.copy_from_slice(&wrapped);
600
601 let mut header = Header {
602 version: VERSION,
603 page_size: PAGE_SIZE as u32,
604 flags: FLAG_ENCRYPTED,
605 dimensions: config.dimensions as u32,
606 created_at: memory::now_secs(),
607 write_counter: 0,
608 // Page 0 is the header; pages 1..=wal_pages are the WAL;
609 // record/catalog/index pages start after it. Initial WAL size is
610 // chosen per-config (default 8 pages = 64 KiB) and clamped to a
611 // sane floor so the first transaction can always commit.
612 next_page: 1 + config.wal_pages_initial.max(MIN_WAL_PAGES),
613 catalog_start: 0,
614 catalog_pages: 0,
615 catalog_len: 0,
616 vector_count: 0,
617 m_cost: config.kdf.m_cost,
618 t_cost: config.kdf.t_cost,
619 p_cost: config.kdf.p_cost,
620 salt,
621 dek_nonce,
622 wrapped_dek,
623 index_start: 0,
624 index_pages: 0,
625 index_len: 0,
626 wal_start: 1,
627 wal_pages: config.wal_pages_initial.max(MIN_WAL_PAGES),
628 wal_seq: 0,
629 manifest_start: 0,
630 manifest_pages: 0,
631 manifest_len: 0,
632 // Regenerated per write by `apply_seal`; `None` until we seal.
633 seal_nonce: None,
634 seal_tag: None,
635 // v8+ result-cache pointer — zero-initialized on create,
636 // populated only when `cache_put` runs at least once and
637 // the next `flush` seals a directory run.
638 cache_start: 0,
639 cache_pages: 0,
640 cache_len: 0,
641 };
642
643 let mut pager = Pager::new(file, dek, 0, VERSION);
644 let mut page = header.to_page();
645 header.apply_seal(&mut page, pager.dek())?;
646 pager.write_raw(0, &page)?;
647 pager.sync()?;
648
649 Ok(Mnemo {
650 pager,
651 header,
652 catalog: Vec::new(),
653 index: HashMap::new(),
654 path,
655 dimensions: config.dimensions,
656 kdf: config.kdf,
657 dirty_catalog: false,
658 ann: None,
659 dirty_index: false,
660 manifest: Vec::new(),
661 max_snapshots: config.max_snapshots,
662 read_only: false,
663 cache_dir: Vec::new(),
664 cache_idx: DirectoryIndex::new(),
665 dirty_cache: false,
666 cache_budgets: std::collections::HashMap::new(),
667 cache_flush_policy: CacheFlushPolicy::default(),
668 cache_batch: BatchState::default(),
669 cache_counters: CounterState::default(),
670 })
671 }
672
673 /// Open an existing database. A wrong passphrase fails cleanly with
674 /// [`MnemoError::WrongPassphrase`]. Takes an OS exclusive lock for
675 /// the lifetime of the returned handle; a second concurrent open
676 /// (from any process) returns [`MnemoError::Locked`].
677 ///
678 /// For a caller that only needs to read (and can tolerate the
679 /// pending-recovery / pending-migration cases documented on
680 /// [`Mnemo::open_read_only`]), prefer that entry point — it takes a
681 /// shared lock and coexists with other readers.
682 pub fn open(path: impl Into<PathBuf>, passphrase: &str) -> Result<Mnemo> {
683 let path: PathBuf = path.into();
684 let mut file = open_with_lock(&path, /*write=*/ true, /*create=*/ false)?;
685
686 let mut hbuf = [0u8; PAGE_SIZE];
687 file.seek(SeekFrom::Start(0))?;
688 file.read_exact(&mut hbuf)?;
689 let mut header = match Header::from_page(&hbuf) {
690 Ok(h) => h,
691 Err(MnemoError::HeaderChecksum) => {
692 // Page 0 is torn — most likely a crash during a checkpoint's
693 // header write. Try to heal from the WAL at its default site
694 // (a never-grown WAL never moves from page 1). We don't know
695 // the WAL size (the header that holds it is what's torn), so
696 // scan a generous fixed window — `wal::recover` stops at the
697 // first non-magic frame, so reading extra is safe.
698 let healed = wal::recover(&mut file, 1, TORN_HEADER_SCAN_PAGES, 0)?;
699 let frames = healed.ok_or(MnemoError::HeaderChecksum)?;
700 for (page_no, bytes) in &frames {
701 if bytes.len() != PAGE_SIZE {
702 return Err(MnemoError::Invalid("WAL frame is not page-sized".into()));
703 }
704 file.seek(SeekFrom::Start(page_no * PAGE_SIZE as u64))?;
705 file.write_all(bytes)?;
706 }
707 file.sync_all()?;
708 file.seek(SeekFrom::Start(0))?;
709 file.read_exact(&mut hbuf)?;
710 Header::from_page(&hbuf)?
711 }
712 Err(e) => return Err(e),
713 };
714
715 // Crash recovery: replay a committed-but-uncheckpointed transaction.
716 // Each frame is a finished page image; the header frame, replayed to
717 // page 0, supersedes the header just read.
718 if let Some(frames) = wal::recover(
719 &mut file,
720 header.wal_start,
721 header.wal_pages,
722 header.wal_seq,
723 )? {
724 for (page_no, bytes) in &frames {
725 if bytes.len() != PAGE_SIZE {
726 return Err(MnemoError::Invalid("WAL frame is not page-sized".into()));
727 }
728 file.seek(SeekFrom::Start(page_no * PAGE_SIZE as u64))?;
729 file.write_all(bytes)?;
730 }
731 file.sync_all()?;
732 // Re-read the now-current header from the replayed page 0.
733 file.seek(SeekFrom::Start(0))?;
734 file.read_exact(&mut hbuf)?;
735 header = Header::from_page(&hbuf)?;
736 }
737
738 let kdf = KdfParams {
739 m_cost: header.m_cost,
740 t_cost: header.t_cost,
741 p_cost: header.p_cost,
742 };
743 let kek = crypto::derive_kek(passphrase.as_bytes(), &header.salt, kdf)?;
744 let dek = crypto::unwrap_dek(&kek, &header.dek_nonce, &header.wrapped_dek)?;
745
746 // Validate the v7 header seal now that we have the DEK. Returns
747 // `MnemoError::HeaderTampered` if any of the sealed mutable fields
748 // were rewritten after the last legitimate flush. No-op for v6 and
749 // older — pre-v7 files don't carry a seal.
750 header.validate_seal(&dek)?;
751
752 // Build the pager at the file's *current* version so initial reads
753 // use the matching AAD scheme — `Pager::page_aad` returns empty AAD
754 // for v4/v5 (no page binding) and `page_no.to_le_bytes()` for v6+.
755 let on_disk_version = header.version;
756 let mut pager = Pager::new(file, dek, header.write_counter, on_disk_version);
757 let dimensions = header.dimensions as usize;
758
759 // === v4→v5 catalog migration =====================================
760 // v6 reads directly; v5 reads directly; v4 replays the catalog into
761 // the v5 shape by reading each memory's body to populate the new
762 // `accessed_at` / `access_count` fields. All page reads here still
763 // use the v4/v5 AAD scheme (none).
764 let (catalog, migrated_from_v4) = if on_disk_version >= 5 {
765 (
766 load_catalog(
767 &mut pager,
768 header.catalog_start,
769 header.catalog_pages,
770 header.catalog_len,
771 )?,
772 false,
773 )
774 } else if on_disk_version == 4 {
775 let v4 = load_catalog_v4(
776 &mut pager,
777 header.catalog_start,
778 header.catalog_pages,
779 header.catalog_len,
780 )?;
781 (migrate_v4_catalog(&mut pager, v4)?, true)
782 } else {
783 return Err(MnemoError::UnsupportedVersion(on_disk_version));
784 };
785 let index = build_id_index(&catalog);
786 let ann = load_index(
787 &mut pager,
788 header.index_start,
789 header.index_pages,
790 header.index_len,
791 dimensions,
792 )?;
793 let mut manifest = load_manifest(
794 &mut pager,
795 header.manifest_start,
796 header.manifest_pages,
797 header.manifest_len,
798 )?;
799
800 // === v5→v6 page-encryption migration =============================
801 // Pre-v6 data pages were encrypted under no AAD; v6 binds the home
802 // `page_no` as AAD so a page-transplant attack fails authentication
803 // on read. Re-stage every live record page through `write_page` so
804 // the next flush re-encrypts it under v6 AAD at the same home slot.
805 // Tombstoned entries skip — they're reclaimed by `compact_file`.
806 // Old catalog / ANN / manifest pages stay as orphan ciphertext;
807 // the next flush writes the new runs at fresh slots.
808 let migrated_pages_to_v6 = on_disk_version < 6;
809 if migrated_pages_to_v6 {
810 for e in &catalog {
811 if e.deleted {
812 continue;
813 }
814 for i in 0..e.page_count as u64 {
815 let payload = pager.read_page(e.start_page + i)?;
816 pager.write_page(e.start_page + i, &payload)?;
817 }
818 }
819 pager.set_version(6);
820 }
821
822 // === v6→v7 header-seal migration =================================
823 // v7 adds an AES-GCM seal at the tail of the header page that
824 // authenticates every mutable header field. Pre-v7 files have no
825 // seal — we just bump the version in memory and mark `dirty_catalog`
826 // so the next flush writes a sealed v7 header. Pages are untouched
827 // (page crypto is unchanged from v6).
828 let migrated_header_to_v7 = on_disk_version < 7;
829 if migrated_header_to_v7 {
830 pager.set_version(7);
831 }
832
833 // === v7→v8 cache-directory migration =============================
834 // v8 adds three u64 header fields (cache_start/pages/len) for the
835 // result-cache directory (Phase 10.1). Pre-v8 files have those
836 // three bytes as zero (the on-disk region beyond the v7 seal is
837 // zero-initialized), which is the legal "empty cache directory"
838 // state — no data migration required. We bump the version in
839 // memory and mark dirty_catalog so the next flush writes a
840 // v8-sealed header with the new AAD shape covering the cache
841 // pointer. Pages are untouched.
842 let migrated_header_to_v8 = on_disk_version < 8;
843 if migrated_header_to_v8 {
844 pager.set_version(VERSION);
845 }
846
847 let migrated = migrated_from_v4
848 || migrated_pages_to_v6
849 || migrated_header_to_v7
850 || migrated_header_to_v8;
851 if migrated {
852 // Stamp the version forward in memory; the next flush persists
853 // it via the WAL-committed header frame (and the v7 seal).
854 header.version = VERSION;
855 // Snapshots written under an older format reference page runs
856 // encrypted under the old AAD scheme; this build can't decrypt
857 // them under the new pager. Drop the manifest so the next
858 // flush records a fresh snapshot of the migrated state.
859 // Point-in-time recovery into the pre-migration past is
860 // sacrificed for migration simplicity. Old pages remain on
861 // disk until `compact_file`. (For v6→v7 alone — which doesn't
862 // change page crypto — the old snapshots would technically
863 // still be readable, but dropping them uniformly keeps the
864 // migration policy simple.)
865 manifest.clear();
866 }
867
868 // The ANN index pages on disk were sealed under the *old* AAD scheme
869 // if we migrated pages to v6; force a rewrite at fresh v6-encrypted
870 // slots on the next flush so a future reopen doesn't try to read
871 // them under the new AAD and fail. (Catalog and manifest already
872 // get rewritten via dirty_catalog and the manifest.clear() above.)
873 let dirty_index = migrated_pages_to_v6 && ann.is_some();
874
875 // Load the cache directory (v8+ only). Pre-v8 files have
876 // cache_pages == 0 by header layout, so this trivially yields
877 // an empty directory without touching the disk.
878 let cache_dir = load_cache_directory(
879 &mut pager,
880 header.cache_start,
881 header.cache_pages,
882 header.cache_len,
883 )?;
884 let cache_idx = DirectoryIndex::build(&cache_dir);
885
886 Ok(Mnemo {
887 pager,
888 header,
889 catalog,
890 index,
891 path,
892 dimensions,
893 kdf,
894 dirty_catalog: migrated,
895 ann,
896 dirty_index,
897 manifest,
898 // `Mnemo::open` has no [`MnemoConfig`] to read; default the cap
899 // to [`DEFAULT_MAX_SNAPSHOTS`]. Override with
900 // [`Mnemo::set_max_snapshots`] right after open if you need
901 // unlimited retention or a different cap for this handle.
902 max_snapshots: DEFAULT_MAX_SNAPSHOTS,
903 read_only: false,
904 cache_dir,
905 cache_idx,
906 dirty_cache: false,
907 cache_budgets: std::collections::HashMap::new(),
908 cache_flush_policy: CacheFlushPolicy::default(),
909 cache_batch: BatchState::default(),
910 cache_counters: CounterState::default(),
911 })
912 }
913
914 /// Open an existing database in **read-only** mode. Takes an OS shared
915 /// lock; multiple read-only handles can coexist, but any attempt to
916 /// open the same file read-write elsewhere returns
917 /// [`MnemoError::Locked`]. All mutating methods on the returned handle
918 /// return [`MnemoError::ReadOnly`], including `recall` with
919 /// `track_access(true)`.
920 ///
921 /// Two conditions block a read-only open and return
922 /// [`MnemoError::NeedsWriteOpen`]:
923 ///
924 /// 1. The file has a committed-but-uncheckpointed WAL transaction that
925 /// would ordinarily be replayed on open. A read-only handle
926 /// cannot write the recovered pages back.
927 /// 2. The on-disk format version is older than the current
928 /// `format::VERSION`, requiring an on-disk migration.
929 ///
930 /// In either case, open the file read-write once (which recovers /
931 /// migrates in place), close the handle cleanly, and reopen
932 /// read-only. This is documented in the README's Durability section.
933 pub fn open_read_only(path: impl Into<PathBuf>, passphrase: &str) -> Result<Mnemo> {
934 let path: PathBuf = path.into();
935 let mut file = open_with_lock(&path, /*write=*/ false, /*create=*/ false)?;
936
937 let mut hbuf = [0u8; PAGE_SIZE];
938 file.seek(SeekFrom::Start(0))?;
939 file.read_exact(&mut hbuf)?;
940 // A torn header on a read-only open cannot be healed (that requires
941 // writing recovered frames back). Surface as `NeedsWriteOpen` so the
942 // caller knows to reopen read-write once.
943 let header = Header::from_page(&hbuf).map_err(|e| match e {
944 MnemoError::HeaderChecksum => MnemoError::NeedsWriteOpen {
945 reason: "torn header page needs recovery from WAL",
946 },
947 other => other,
948 })?;
949
950 // Refuse to migrate on read-only opens. v6→v7 is header-only
951 // (no page writes required), but even that bumps `version` on
952 // disk and re-seals the header — which needs write access. Keep
953 // the policy uniform across all sub-version bumps.
954 if header.version < VERSION {
955 return Err(MnemoError::NeedsWriteOpen {
956 reason: "pending format-version migration",
957 });
958 }
959
960 // Refuse if there is a committed-but-uncheckpointed transaction
961 // waiting for `wal::recover` to fold it into home pages.
962 if wal::recover(
963 &mut file,
964 header.wal_start,
965 header.wal_pages,
966 header.wal_seq,
967 )?
968 .is_some()
969 {
970 return Err(MnemoError::NeedsWriteOpen {
971 reason: "pending WAL recovery",
972 });
973 }
974
975 let kdf = KdfParams {
976 m_cost: header.m_cost,
977 t_cost: header.t_cost,
978 p_cost: header.p_cost,
979 };
980 let kek = crypto::derive_kek(passphrase.as_bytes(), &header.salt, kdf)?;
981 let dek = crypto::unwrap_dek(&kek, &header.dek_nonce, &header.wrapped_dek)?;
982 header.validate_seal(&dek)?;
983
984 let mut pager = Pager::new(file, dek, header.write_counter, header.version);
985 let dimensions = header.dimensions as usize;
986
987 let catalog = load_catalog(
988 &mut pager,
989 header.catalog_start,
990 header.catalog_pages,
991 header.catalog_len,
992 )?;
993 let index = build_id_index(&catalog);
994 let ann = load_index(
995 &mut pager,
996 header.index_start,
997 header.index_pages,
998 header.index_len,
999 dimensions,
1000 )?;
1001 let manifest = load_manifest(
1002 &mut pager,
1003 header.manifest_start,
1004 header.manifest_pages,
1005 header.manifest_len,
1006 )?;
1007 let cache_dir = load_cache_directory(
1008 &mut pager,
1009 header.cache_start,
1010 header.cache_pages,
1011 header.cache_len,
1012 )?;
1013 let cache_idx = DirectoryIndex::build(&cache_dir);
1014
1015 Ok(Mnemo {
1016 pager,
1017 header,
1018 catalog,
1019 index,
1020 path,
1021 dimensions,
1022 kdf,
1023 // No mutation is possible; leave both dirty flags false so
1024 // `close()` is a genuine no-op even if we later add an
1025 // auto-flush on drop.
1026 dirty_catalog: false,
1027 ann,
1028 dirty_index: false,
1029 manifest,
1030 max_snapshots: DEFAULT_MAX_SNAPSHOTS,
1031 read_only: true,
1032 cache_dir,
1033 cache_idx,
1034 dirty_cache: false,
1035 cache_budgets: std::collections::HashMap::new(),
1036 cache_flush_policy: CacheFlushPolicy::default(),
1037 cache_batch: BatchState::default(),
1038 cache_counters: CounterState::default(),
1039 })
1040 }
1041
1042 /// Override the manifest snapshot cap on this open handle. `0` disables
1043 /// the cap; any positive value keeps the most-recent `max` snapshots and
1044 /// drops the rest on the next flush. See [`MnemoConfig::max_snapshots`].
1045 pub fn set_max_snapshots(&mut self, max: usize) {
1046 self.max_snapshots = max;
1047 }
1048
1049 /// Embedding dimensionality this database expects.
1050 pub fn dimensions(&self) -> usize {
1051 self.dimensions
1052 }
1053
1054 /// Number of live (non-deleted) memories.
1055 pub fn len(&self) -> usize {
1056 self.catalog.iter().filter(|e| !e.deleted).count()
1057 }
1058
1059 /// True if the database holds no live memories.
1060 pub fn is_empty(&self) -> bool {
1061 self.len() == 0
1062 }
1063
1064 // --- internal page/record helpers -------------------------------------
1065
1066 fn write_record(&mut self, bytes: &[u8]) -> Result<(u64, u32)> {
1067 let pc = bytes.len().div_ceil(PAYLOAD).max(1);
1068 let start = self.header.next_page;
1069 self.header.next_page += pc as u64;
1070 for i in 0..pc {
1071 let lo = i * PAYLOAD;
1072 let hi = ((i + 1) * PAYLOAD).min(bytes.len());
1073 self.pager.write_page(start + i as u64, &bytes[lo..hi])?;
1074 }
1075 Ok((start, pc as u32))
1076 }
1077
1078 fn read_record(&mut self, e: &CatalogEntry) -> Result<Vec<u8>> {
1079 let mut buf = Vec::with_capacity(e.len as usize);
1080 for i in 0..e.page_count as u64 {
1081 buf.extend_from_slice(&self.pager.read_page(e.start_page + i)?);
1082 }
1083 buf.truncate(e.len as usize);
1084 Ok(buf)
1085 }
1086
1087 fn read_memory(&mut self, e: &CatalogEntry) -> Result<Memory> {
1088 let bytes = self.read_record(e)?;
1089 let mut m: Memory =
1090 rmp_serde::from_slice(&bytes).map_err(|err| MnemoError::Serialize(err.to_string()))?;
1091 // The catalog is the source of truth for access stats — the values
1092 // in the record body are a stale snapshot from when the record was
1093 // last written. Overwrite them here so consumers always see live
1094 // values regardless of when the record was last serialized.
1095 m.accessed_at = e.accessed_at;
1096 m.access_count = e.access_count;
1097 Ok(m)
1098 }
1099
1100 /// Serialize and store a memory (insert or overwrite by ID).
1101 fn put(&mut self, mut m: Memory) -> Result<Ulid> {
1102 if m.vector.len() != self.dimensions {
1103 return Err(MnemoError::DimensionMismatch {
1104 expected: self.dimensions,
1105 got: m.vector.len(),
1106 });
1107 }
1108 if m.id == Ulid::nil() {
1109 m.id = Ulid::new();
1110 }
1111 let id_u: u128 = m.id.0;
1112 let bytes = rmp_serde::to_vec(&m).map_err(|e| MnemoError::Serialize(e.to_string()))?;
1113 let (start, pc) = self.write_record(&bytes)?;
1114 // For overwrites preserve the existing access stats; for inserts
1115 // seed them from the memory (which carries them in its body).
1116 let (prev_accessed, prev_count) = match self.index.get(&id_u).copied() {
1117 Some(idx) => (
1118 self.catalog[idx].accessed_at,
1119 self.catalog[idx].access_count,
1120 ),
1121 None => (m.accessed_at, m.access_count),
1122 };
1123 let entry = CatalogEntry {
1124 id: id_u,
1125 start_page: start,
1126 page_count: pc,
1127 len: bytes.len() as u32,
1128 deleted: false,
1129 accessed_at: prev_accessed,
1130 access_count: prev_count,
1131 };
1132 match self.index.get(&id_u).copied() {
1133 Some(idx) => self.catalog[idx] = entry,
1134 None => {
1135 self.index.insert(id_u, self.catalog.len());
1136 self.catalog.push(entry);
1137 }
1138 }
1139 self.dirty_catalog = true;
1140
1141 // Keep the ANN index complete: assign the (new or changed) vector to
1142 // its nearest partition. Centroids/codebook stay fixed until rebuild.
1143 if let Some(ann) = &mut self.ann {
1144 ann.add(id_u, &m.vector);
1145 self.dirty_index = true;
1146 }
1147 Ok(m.id)
1148 }
1149
1150 // --- public CRUD ------------------------------------------------------
1151
1152 /// Store a memory. Returns its ULID. If the memory's ID is nil a fresh
1153 /// one is assigned; an existing ID overwrites in place.
1154 pub fn remember(&mut self, memory: Memory) -> Result<Ulid> {
1155 if self.read_only {
1156 return Err(MnemoError::ReadOnly);
1157 }
1158 self.put(memory)
1159 }
1160
1161 /// Fetch a memory by ID.
1162 pub fn get(&mut self, id: &Ulid) -> Result<Memory> {
1163 let idx = *self
1164 .index
1165 .get(&id.0)
1166 .ok_or_else(|| MnemoError::NotFound(id.to_string()))?;
1167 let entry = self.catalog[idx].clone();
1168 if entry.deleted {
1169 return Err(MnemoError::NotFound(id.to_string()));
1170 }
1171 self.read_memory(&entry)
1172 }
1173
1174 /// Soft-delete a memory (tombstoned; space reclaimed by `compact`).
1175 pub fn delete(&mut self, id: &Ulid) -> Result<()> {
1176 if self.read_only {
1177 return Err(MnemoError::ReadOnly);
1178 }
1179 let idx = *self
1180 .index
1181 .get(&id.0)
1182 .ok_or_else(|| MnemoError::NotFound(id.to_string()))?;
1183 if !self.catalog[idx].deleted {
1184 self.catalog[idx].deleted = true;
1185 self.dirty_catalog = true;
1186 }
1187 if let Some(ann) = &mut self.ann {
1188 ann.remove(id.0);
1189 self.dirty_index = true;
1190 }
1191 Ok(())
1192 }
1193
1194 /// Return every live memory (used by tooling and compaction).
1195 pub fn memories(&mut self) -> Result<Vec<Memory>> {
1196 let entries: Vec<CatalogEntry> = self
1197 .catalog
1198 .iter()
1199 .filter(|e| !e.deleted)
1200 .cloned()
1201 .collect();
1202 let mut out = Vec::with_capacity(entries.len());
1203 for e in &entries {
1204 out.push(self.read_memory(e)?);
1205 }
1206 Ok(out)
1207 }
1208
1209 /// Return the database's **self-describing onboarding memories** — the
1210 /// ones tagged `metadata.area = "onboarding"`. This is the engine-level
1211 /// surface for the *single-file philosophy*: an agent who receives a
1212 /// `.mnemo` file (and its passphrase) can call this to learn what the
1213 /// file is, which embedder it expects, the recommended agent id, and any
1214 /// other conventions the file's author chose to record — all without
1215 /// needing any external documentation.
1216 ///
1217 /// Ordering: the canonical manifest (tag `metadata.topic = "manifest"`)
1218 /// always comes first; everything else follows in `importance` descending,
1219 /// then `created_at` ascending for deterministic results.
1220 pub fn about(&mut self) -> Result<Vec<Memory>> {
1221 let mut out: Vec<Memory> = self
1222 .memories()?
1223 .into_iter()
1224 .filter(|m| {
1225 m.metadata
1226 .get("area")
1227 .and_then(|v| v.as_str())
1228 .map(|s| s.eq_ignore_ascii_case("onboarding"))
1229 .unwrap_or(false)
1230 })
1231 .collect();
1232 out.sort_by(|a, b| {
1233 let a_manifest = is_manifest(a);
1234 let b_manifest = is_manifest(b);
1235 // Manifest topic wins first; then importance desc; then created_at asc.
1236 b_manifest
1237 .cmp(&a_manifest)
1238 .then_with(|| b.importance.total_cmp(&a.importance))
1239 .then_with(|| a.created_at.cmp(&b.created_at))
1240 });
1241 Ok(out)
1242 }
1243
1244 // --- retrieval --------------------------------------------------------
1245
1246 /// Phase-1 brute-force search: rank live memories by raw similarity only.
1247 /// Read-only — does not update access statistics.
1248 pub fn search(
1249 &mut self,
1250 query: &[f32],
1251 top_k: usize,
1252 metric: Metric,
1253 ) -> Result<Vec<(Memory, f32)>> {
1254 if query.len() != self.dimensions {
1255 return Err(MnemoError::DimensionMismatch {
1256 expected: self.dimensions,
1257 got: query.len(),
1258 });
1259 }
1260 let now = memory::now_secs();
1261 let entries: Vec<CatalogEntry> = self
1262 .catalog
1263 .iter()
1264 .filter(|e| !e.deleted)
1265 .cloned()
1266 .collect();
1267
1268 let mut scored: Vec<(Memory, f32)> = Vec::new();
1269 for e in &entries {
1270 let m = self.read_memory(e)?;
1271 if m.is_expired(now) {
1272 continue;
1273 }
1274 let sim = memory::similarity(metric, query, &m.vector);
1275 scored.push((m, sim));
1276 }
1277 scored.sort_by(|a, b| b.1.total_cmp(&a.1));
1278 scored.truncate(top_k);
1279 Ok(scored)
1280 }
1281
1282 /// Phase-5 multi-signal recall: rank by
1283 /// `α·similarity + β·recency + γ·importance + δ·ln(freq)`, with type and
1284 /// agent filtering. Updates `accessed_at` / `access_count` on the
1285 /// returned memories (persisted on the next `flush`).
1286 ///
1287 /// When an ANN index has been built ([`Mnemo::build_index`]), recall runs
1288 /// the tiered IVF→PQ→rerank pipeline: it scores only the similarity-nearest
1289 /// `n_rerank` candidates rather than every memory. Without an index it
1290 /// scores every live memory exactly.
1291 pub fn recall(&mut self, req: &RecallRequest) -> Result<Vec<RecallResult>> {
1292 if req.query.len() != self.dimensions {
1293 return Err(MnemoError::DimensionMismatch {
1294 expected: self.dimensions,
1295 got: req.query.len(),
1296 });
1297 }
1298 // Access-tracking bumps `accessed_at` / `access_count` in the
1299 // catalog, which is a mutation. On a read-only handle the caller
1300 // must pass `track_access(false)` (via CLI `--no-track`).
1301 if self.read_only && req.track_access {
1302 return Err(MnemoError::ReadOnly);
1303 }
1304 let now = memory::now_secs();
1305
1306 // Candidate set: ANN-narrowed when an index exists, else everything.
1307 let entries: Vec<CatalogEntry> = if let Some(ann) = &self.ann {
1308 let ids = ann.query(&req.query, req.n_probe, req.n_rerank);
1309 ids.iter()
1310 .filter_map(|id| self.index.get(id).copied())
1311 .map(|i| self.catalog[i].clone())
1312 .filter(|e| !e.deleted)
1313 .collect()
1314 } else {
1315 self.catalog
1316 .iter()
1317 .filter(|e| !e.deleted)
1318 .cloned()
1319 .collect()
1320 };
1321
1322 let mut scored: Vec<RecallResult> = Vec::new();
1323 for e in &entries {
1324 let m = self.read_memory(e)?;
1325 if m.is_expired(now) {
1326 continue;
1327 }
1328 // Type filter.
1329 if let Some(types) = &req.memory_types {
1330 if !types.contains(&m.memory_type) {
1331 continue;
1332 }
1333 }
1334 // Agent-scoping: an agent sees its own memories plus shared ones.
1335 if let Some(agent) = &req.agent_id {
1336 let visible = m.agent_id == *agent || m.scope == Scope::Shared;
1337 if !visible {
1338 continue;
1339 }
1340 }
1341 let sim = memory::similarity(req.metric, &req.query, &m.vector);
1342 let age = (now - m.accessed_at) as f32;
1343 let score = req.weights.score(sim, age, m.importance, m.access_count);
1344 scored.push(RecallResult {
1345 memory: m,
1346 score,
1347 similarity: sim,
1348 });
1349 }
1350 scored.sort_by(|a, b| b.score.total_cmp(&a.score));
1351 scored.truncate(req.top_k);
1352
1353 // Update access statistics for everything we surfaced — but only
1354 // mutate the catalog entry, not the full record on disk. Pre-v5
1355 // this loop called `self.put(r.memory.clone())`, which serialized
1356 // and rewrote every result (vector included) to fresh pages,
1357 // turning a top-K recall into K full record rewrites. Now the
1358 // catalog is the source of truth for these two fields and the
1359 // record body stays untouched until the next real edit.
1360 if req.track_access {
1361 for r in &mut scored {
1362 if let Some(&idx) = self.index.get(&r.memory.id.0) {
1363 let entry = &mut self.catalog[idx];
1364 entry.accessed_at = now;
1365 entry.access_count = entry.access_count.saturating_add(1);
1366 // Propagate the just-updated values onto the returned
1367 // Memory so the caller sees the post-recall state.
1368 r.memory.accessed_at = entry.accessed_at;
1369 r.memory.access_count = entry.access_count;
1370 }
1371 }
1372 // Mark the catalog dirty so the next flush persists these
1373 // bumps. The records themselves are not dirty.
1374 if !scored.is_empty() {
1375 self.dirty_catalog = true;
1376 }
1377 }
1378 Ok(scored)
1379 }
1380
1381 // --- approximate index ------------------------------------------------
1382
1383 /// Build an IVF+PQ approximate-nearest-neighbour index over every live
1384 /// memory, using default tuning. After this, [`Mnemo::recall`] runs the
1385 /// tiered pipeline instead of an exact scan. Persisted on the next
1386 /// `flush`. Returns a snapshot of the index shape.
1387 pub fn build_index(&mut self) -> Result<IndexInfo> {
1388 self.build_index_with(IndexConfig::default())
1389 }
1390
1391 /// Build the ANN index with explicit tuning.
1392 pub fn build_index_with(&mut self, cfg: IndexConfig) -> Result<IndexInfo> {
1393 if self.read_only {
1394 return Err(MnemoError::ReadOnly);
1395 }
1396 let mems = self.memories()?;
1397 if mems.is_empty() {
1398 return Err(MnemoError::Invalid(
1399 "cannot build an index over an empty database".into(),
1400 ));
1401 }
1402 let items: Vec<(u128, &[f32])> =
1403 mems.iter().map(|m| (m.id.0, m.vector.as_slice())).collect();
1404 let idx = IvfPqIndex::build(self.dimensions, &items, cfg)?;
1405 let info = idx.info();
1406 self.ann = Some(idx);
1407 self.dirty_index = true;
1408 Ok(info)
1409 }
1410
1411 /// Rebuild the ANN index from scratch — re-clusters centroids and retrains
1412 /// the PQ codebook, undoing the cluster drift that accumulates as memories
1413 /// are inserted against fixed centroids.
1414 pub fn rebuild_index(&mut self) -> Result<IndexInfo> {
1415 let cfg = match &self.ann {
1416 Some(a) => IndexConfig {
1417 n_probe: a.n_probe(),
1418 n_rerank: a.n_rerank(),
1419 ..Default::default()
1420 },
1421 None => IndexConfig::default(),
1422 };
1423 self.build_index_with(cfg)
1424 }
1425
1426 /// Drop the ANN index; recall reverts to exact scans. Persisted on flush.
1427 /// No-op on a read-only handle (signature is infallible — callers of
1428 /// `build_index` / `flush` on a read-only handle already get a typed
1429 /// `ReadOnly` error, which is the enforceable boundary).
1430 pub fn drop_index(&mut self) {
1431 if self.read_only {
1432 return;
1433 }
1434 if self.ann.is_some() {
1435 self.ann = None;
1436 self.dirty_index = true;
1437 }
1438 }
1439
1440 /// Whether an ANN index is currently loaded.
1441 pub fn has_index(&self) -> bool {
1442 self.ann.is_some()
1443 }
1444
1445 // --- durability & maintenance ----------------------------------------
1446
1447 /// Seal a serialized buffer into a fresh run of encrypted home pages and
1448 /// append a WAL frame for each. Returns the run's `(start_page, pages)`.
1449 /// The pages are *not* written to their home locations here — they go to
1450 /// the WAL and are folded in later by [`Mnemo::checkpoint`].
1451 fn seal_run(&mut self, bytes: &[u8], frames: &mut Vec<wal::Frame>) -> Result<(u64, u32)> {
1452 let pc = bytes.len().div_ceil(PAYLOAD).max(1);
1453 let start = self.header.next_page;
1454 self.header.next_page += pc as u64;
1455 for i in 0..pc {
1456 let lo = i * PAYLOAD;
1457 let hi = ((i + 1) * PAYLOAD).min(bytes.len());
1458 let page_no = start + i as u64;
1459 let sealed = self.pager.seal_page(page_no, &bytes[lo..hi])?;
1460 frames.push((page_no, sealed.to_vec()));
1461 }
1462 Ok((start, pc as u32))
1463 }
1464
1465 /// Grow the WAL region if `frame_pages` worth of frames would not fit.
1466 /// Called at the top of `flush`, where the WAL is always spent (every
1467 /// flush ends with a checkpoint), so relocating it cannot strand a
1468 /// committed transaction.
1469 fn ensure_wal_capacity(&mut self, frame_pages: usize) -> Result<()> {
1470 let need = wal::txn_byte_len(frame_pages, PAGE_SIZE);
1471 if need <= self.header.wal_pages * PAGE_SIZE as u64 {
1472 return Ok(());
1473 }
1474 // Allocate a new, larger region at the file tail (~1.5x headroom).
1475 let req = need.div_ceil(PAGE_SIZE as u64);
1476 let new_pages = (req + req / 2 + 4).max(DEFAULT_WAL_PAGES);
1477 let new_start = self.header.next_page;
1478 self.header.next_page += new_pages;
1479 self.header.wal_start = new_start;
1480 self.header.wal_pages = new_pages;
1481 // Persist the relocation now: an isolated header write over an empty
1482 // WAL. A crash here leaves a consistent state with a bigger WAL.
1483 self.header.write_counter = self.pager.write_counter;
1484 let mut page = self.header.to_page();
1485 self.header.apply_seal(&mut page, self.pager.dek())?;
1486 self.pager.write_raw(0, &page)?;
1487 self.pager.sync()?;
1488 Ok(())
1489 }
1490
1491 /// Fold a committed transaction's WAL frames into their home pages.
1492 fn checkpoint(&mut self, frames: &[wal::Frame]) -> Result<()> {
1493 for (page_no, bytes) in frames {
1494 let mut img = [0u8; PAGE_SIZE];
1495 img.copy_from_slice(bytes);
1496 self.pager.write_sealed(*page_no, &img)?;
1497 }
1498 self.pager.sync()?;
1499 Ok(())
1500 }
1501
1502 /// Persist all pending changes as one **write-ahead-logged transaction**.
1503 ///
1504 /// Sequence:
1505 ///
1506 /// 1. **Prepare.** Serialize the control plane (catalog, ANN index),
1507 /// grow the WAL if needed, and **lease** counter and page slots —
1508 /// bump `header.write_counter` and `header.next_page` by upper
1509 /// bounds on this transaction's consumption and persist that
1510 /// *leased* header to disk before any encrypted page is written.
1511 /// 2. **Data pages.** [`Pager::flush`] writes dirty record pages
1512 /// copy-on-write under fresh nonces and fsyncs them. The orphan-page
1513 /// nonce-reuse window (see [Phase 1.1 of the improvement plan])
1514 /// is closed because the leased header on disk already records a
1515 /// `write_counter` past anything this step can produce.
1516 /// 3. **Control plane.** Seal the new catalog / ANN index / snapshot
1517 /// manifest into fresh home page runs.
1518 /// 4. **Commit.** Log all of step 3's frames plus the final header
1519 /// frame into the WAL and fsync — the durability point.
1520 /// 5. **Checkpoint.** Fold the WAL into the home pages.
1521 ///
1522 /// A crash before step 4 leaves the previous state intact. A crash
1523 /// after step 4 is repaired by [`Mnemo::open`] replaying the WAL.
1524 /// Safe to call repeatedly.
1525 pub fn flush(&mut self) -> Result<()> {
1526 if self.read_only {
1527 return Err(MnemoError::ReadOnly);
1528 }
1529 if !self.dirty_catalog && !self.dirty_index && !self.dirty_cache {
1530 // No control-plane changes pending. Because every `put` that
1531 // dirties a data page also dirties the catalog, and every
1532 // `cache_put` that dirties a data page also dirties
1533 // `dirty_cache`, there can't be dirty data pages either —
1534 // nothing to do.
1535 return Ok(());
1536 }
1537
1538 // 1. Serialize control plane, ensure WAL capacity, lease counter +
1539 // page slots, and persist the leased header *before* any data
1540 // page hits the disk.
1541 let ctx = self.prepare_for_flush()?;
1542
1543 // 2. Record (vector) data pages: copy-on-write to fresh pages, fsynced.
1544 // Safe: the lease guarantees that even if we crash here, reopen
1545 // will see a `write_counter` past every nonce this step uses.
1546 self.pager.flush()?;
1547
1548 // 3. Seal the catalog / index control pages into fresh home runs.
1549 let mut frames: Vec<wal::Frame> = Vec::new();
1550 if let Some(bytes) = &ctx.cat_bytes {
1551 let (start, pages) = self.seal_run(bytes, &mut frames)?;
1552 self.header.catalog_start = start;
1553 self.header.catalog_pages = pages as u64;
1554 self.header.catalog_len = bytes.len() as u64;
1555 }
1556 if self.dirty_index {
1557 match &ctx.idx_bytes {
1558 Some(bytes) => {
1559 let (start, pages) = self.seal_run(bytes, &mut frames)?;
1560 self.header.index_start = start;
1561 self.header.index_pages = pages as u64;
1562 self.header.index_len = bytes.len() as u64;
1563 }
1564 None => {
1565 self.header.index_start = 0;
1566 self.header.index_pages = 0;
1567 self.header.index_len = 0;
1568 }
1569 }
1570 }
1571 // v8+ result-cache directory (Phase 10.1). Seal the serialized
1572 // directory into a fresh page run and update the pointer in the
1573 // header. The seal AAD already covers cache_start/pages/len for
1574 // v8+ (see `header_seal_aad`), so tampering with the pointer
1575 // trips the header seal on the next open.
1576 if self.dirty_cache {
1577 match &ctx.cache_bytes {
1578 Some(bytes) if !self.cache_dir.is_empty() => {
1579 let (start, pages) = self.seal_run(bytes, &mut frames)?;
1580 self.header.cache_start = start;
1581 self.header.cache_pages = pages as u64;
1582 self.header.cache_len = bytes.len() as u64;
1583 }
1584 _ => {
1585 // Either the directory serialized to nothing (all
1586 // entries deleted after tombstone-only migration) or
1587 // it's empty — collapse the pointer to zero.
1588 self.header.cache_start = 0;
1589 self.header.cache_pages = 0;
1590 self.header.cache_len = 0;
1591 }
1592 }
1593 }
1594
1595 // 3b. Record this transaction as a restorable snapshot. The manifest
1596 // update is staged in a local and adopted only once the commit
1597 // below succeeds, so a failed flush leaves no phantom entry.
1598 //
1599 // The cap (`self.max_snapshots`) is applied to the staged local
1600 // before the new entry lands, so after appending the manifest
1601 // holds at most `max_snapshots` entries — the most-recent N.
1602 // `max_snapshots == 0` disables the cap. Pages referenced only
1603 // by pruned snapshots stay on disk until `compact_file`.
1604 let live = self.len() as u64;
1605 let txn_id = self.header.wal_seq + 1;
1606 let mut manifest = self.manifest.clone();
1607 if self.max_snapshots > 0 {
1608 // Drop the oldest entries so there's room for the new one.
1609 while manifest.len() >= self.max_snapshots {
1610 manifest.remove(0);
1611 }
1612 }
1613 manifest.push(Snapshot {
1614 txn_id,
1615 created_at: memory::now_secs(),
1616 catalog_start: self.header.catalog_start,
1617 catalog_pages: self.header.catalog_pages,
1618 catalog_len: self.header.catalog_len,
1619 index_start: self.header.index_start,
1620 index_pages: self.header.index_pages,
1621 index_len: self.header.index_len,
1622 memory_count: live,
1623 });
1624 let man_bytes =
1625 rmp_serde::to_vec(&manifest).map_err(|e| MnemoError::Serialize(e.to_string()))?;
1626 let (m_start, m_pages) = self.seal_run(&man_bytes, &mut frames)?;
1627 self.header.manifest_start = m_start;
1628 self.header.manifest_pages = m_pages as u64;
1629 self.header.manifest_len = man_bytes.len() as u64;
1630
1631 // 3c. The header is the transaction's final frame; stamp the new id
1632 // and the *actual* (post-flush) write counter, which is <= the
1633 // leased value persisted in `prepare_for_flush`. This is the
1634 // value that ultimately replaces the leased header on checkpoint.
1635 self.header.vector_count = live;
1636 self.header.write_counter = self.pager.write_counter;
1637 self.header.wal_seq = txn_id;
1638 let mut hpage = self.header.to_page();
1639 self.header.apply_seal(&mut hpage, self.pager.dek())?;
1640 frames.push((0, hpage.to_vec()));
1641
1642 // 4. COMMIT — log the transaction and fsync. Nothing at a home page
1643 // other than the leased header has changed yet; this single
1644 // fsync is what makes the rest durable.
1645 let (wal_start, wal_pages) = (self.header.wal_start, self.header.wal_pages);
1646 wal::commit(self.pager.file_mut(), wal_start, wal_pages, txn_id, &frames)?;
1647
1648 // 5. Checkpoint — fold the WAL into the home pages.
1649 self.checkpoint(&frames)?;
1650
1651 self.manifest = manifest;
1652 self.dirty_catalog = false;
1653 self.dirty_index = false;
1654 self.dirty_cache = false;
1655 // Batched policy: successful flush consumes the pending batch.
1656 self.cache_batch.reset();
1657 Ok(())
1658 }
1659
1660 /// Pre-flush prelude shared by [`Mnemo::flush`] and the
1661 /// `__crash_partial_flush_for_testing` hook.
1662 ///
1663 /// Serializes the control plane (catalog, ANN index), grows the WAL if
1664 /// needed, then **leases** counter and page slots — bumps a *clone* of
1665 /// the header by upper-bound amounts and persists that leased clone
1666 /// with `pager.write_raw + sync`. The in-memory `self.header` keeps
1667 /// its pre-lease `next_page` so [`Mnemo::seal_run`] continues to
1668 /// allocate from the right slot; the final committed header carries
1669 /// the actual post-flush values and overwrites the leased one on
1670 /// checkpoint.
1671 ///
1672 /// This pre-stamping is what closes the nonce-reuse window flagged in
1673 /// Phase 1.1 of the improvement plan: a crash anywhere between here
1674 /// and the WAL commit leaves a file whose on-disk header records a
1675 /// `write_counter` *past* every nonce the rest of this transaction
1676 /// could produce, so the next reopen starts from a counter that
1677 /// guarantees uniqueness against any orphan data pages still on disk.
1678 fn prepare_for_flush(&mut self) -> Result<FlushPrelude> {
1679 // Serialize.
1680 let cat_bytes = if self.dirty_catalog {
1681 Some(
1682 rmp_serde::to_vec(&self.catalog)
1683 .map_err(|e| MnemoError::Serialize(e.to_string()))?,
1684 )
1685 } else {
1686 None
1687 };
1688 let idx_bytes: Option<Vec<u8>> = if self.dirty_index {
1689 match &self.ann {
1690 Some(ann) => {
1691 Some(rmp_serde::to_vec(ann).map_err(|e| MnemoError::Serialize(e.to_string()))?)
1692 }
1693 None => None,
1694 }
1695 } else {
1696 None
1697 };
1698 // Serialize the result-cache directory when it's dirty. An
1699 // empty directory serializes to a small header + 0-element
1700 // array; still cheap enough that we don't special-case it.
1701 let cache_bytes: Option<Vec<u8>> = if self.dirty_cache {
1702 Some(
1703 rmp_serde::to_vec(&self.cache_dir)
1704 .map_err(|e| MnemoError::Serialize(e.to_string()))?,
1705 )
1706 } else {
1707 None
1708 };
1709
1710 let cat_pc = cat_bytes
1711 .as_ref()
1712 .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
1713 let idx_pc = idx_bytes
1714 .as_ref()
1715 .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
1716 let cache_pc = cache_bytes
1717 .as_ref()
1718 .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
1719 // Upper bound on the manifest run: one extra entry, <=82 bytes each
1720 // (rmp_serde encodes a Snapshot as a 9-element fixarray of i64/u64,
1721 // worst case 81 bytes plus a 1-byte array header).
1722 let man_upper = 9 + (self.manifest.len() + 1) * 82;
1723 let man_pc_est = man_upper.div_ceil(PAYLOAD).max(1);
1724
1725 // Grow WAL if needed (does its own header write+sync). The
1726 // cache directory rides in the same transaction so its page
1727 // count contributes to the reservation.
1728 self.ensure_wal_capacity(cat_pc + idx_pc + cache_pc + man_pc_est + 1)?;
1729
1730 // Lease counter and page slots. counter_lease is an upper bound on
1731 // how many `pager.write_counter` advances this transaction will
1732 // make (one per dirty data page + one per sealed control-plane
1733 // page). page_lease covers only the control-plane allocations,
1734 // because dirty data pages were already allocated past
1735 // `header.next_page` by `write_record` when `remember` ran or
1736 // `cache_put` ran.
1737 let data_dirty = self.pager.dirty_page_count() as u64;
1738 let counter_lease = data_dirty + (cat_pc + idx_pc + cache_pc + man_pc_est) as u64;
1739 let page_lease = (cat_pc + idx_pc + cache_pc + man_pc_est) as u64;
1740
1741 // Persist a leased *clone* of the header — in-memory `self.header`
1742 // keeps its pre-lease `next_page` for `seal_run` to consume from
1743 // the correct slot.
1744 let mut leased = self.header.clone();
1745 leased.write_counter = self.pager.write_counter + counter_lease;
1746 leased.next_page += page_lease;
1747 let mut page = leased.to_page();
1748 leased.apply_seal(&mut page, self.pager.dek())?;
1749 self.pager.write_raw(0, &page)?;
1750 self.pager.sync()?;
1751
1752 Ok(FlushPrelude {
1753 cat_bytes,
1754 idx_bytes,
1755 cache_bytes,
1756 })
1757 }
1758
1759 /// Flush and close. Equivalent to `flush()` on a read-write handle; on
1760 /// a read-only handle it's a no-op (no dirty state to persist). The
1761 /// file is released on drop, which also drops the OS advisory lock.
1762 pub fn close(&mut self) -> Result<()> {
1763 if self.read_only {
1764 return Ok(());
1765 }
1766 self.flush()
1767 }
1768
1769 /// **TEST ONLY — do not use.** Reproduces a crash inside
1770 /// [`Mnemo::flush`] *after* the prelude has leased counter+page slots
1771 /// and persisted the leased header, but *before* the WAL is committed.
1772 /// Runs `prepare_for_flush` + `pager.flush` and returns — leaving
1773 /// data pages and a leased header on disk, with no commit frame.
1774 ///
1775 /// Used by `tests/integration.rs::nonce_unique_after_crashed_data_flush`
1776 /// to verify that this window (Phase 1.1 of the improvement plan) does
1777 /// not enable AES-GCM nonce reuse. Calling this in production code
1778 /// strands data pages; never expose it from a binding.
1779 #[doc(hidden)]
1780 pub fn __crash_partial_flush_for_testing(&mut self) -> Result<()> {
1781 if self.read_only {
1782 return Err(MnemoError::ReadOnly);
1783 }
1784 if !self.dirty_catalog && !self.dirty_index {
1785 return Ok(());
1786 }
1787 let _prelude = self.prepare_for_flush()?;
1788 self.pager.flush()
1789 }
1790
1791 /// Bound the in-memory page cache to `pages` decrypted pages.
1792 ///
1793 /// The cache holds decrypted page payloads to speed repeated reads. By
1794 /// default it is capped at 8192 pages (~64 MiB); lower the cap to trade
1795 /// hit rate for a smaller footprint, or raise it for a hotter cache. The
1796 /// cap governs *clean* pages — pages with un-flushed writes are always
1797 /// retained until [`Mnemo::flush`], regardless of the cap.
1798 pub fn set_cache_capacity(&mut self, pages: usize) {
1799 self.pager.set_cache_capacity(pages);
1800 }
1801
1802 /// Page-cache occupancy: `(pages_cached, capacity)`. Renamed from
1803 /// `cache_stats` in v0.4.0: the result-cache API (Phase 10.1) took
1804 /// that name for its per-namespace [`CacheStats`] summary, and the
1805 /// page-cache pair is the older/less-user-facing of the two, so it
1806 /// gets the more-specific name.
1807 pub fn page_cache_stats(&self) -> (usize, usize) {
1808 self.pager.cache_stats()
1809 }
1810
1811 // === Result cache (Phase 10.1 + 10.3) ============================
1812
1813 /// Insert or overwrite a cache entry at `(namespace, key)`.
1814 ///
1815 /// The engine SHA-256-hashes `key` — callers pass the raw input
1816 /// string. Persists to disk on the next `flush()` under the current
1817 /// flush policy (see [`Mnemo::set_cache_flush_policy`]); in
1818 /// [`CacheFlushPolicy::Batched`] the engine may call `flush()`
1819 /// internally when a threshold trips.
1820 ///
1821 /// On overflow of the namespace's [`CacheBudget`] (default 10,000
1822 /// entries / 64 MiB) the LRU entry is tombstoned before the new
1823 /// insert. Tombstones are reclaimed by
1824 /// [`Mnemo::compact_file`], not the runtime flush.
1825 ///
1826 /// Returns [`MnemoError::ReadOnly`] on a read-only handle.
1827 pub fn cache_put(
1828 &mut self,
1829 namespace: &str,
1830 key: &str,
1831 value: &[u8],
1832 opts: CachePutOpts,
1833 ) -> Result<()> {
1834 if self.read_only {
1835 return Err(MnemoError::ReadOnly);
1836 }
1837 let key_hash = result_cache::hash_key(key);
1838 let key_preview = result_cache::key_preview(key);
1839 let now = memory::now_secs();
1840 let size_bytes = value.len() as u64;
1841
1842 let entry = CacheEntry {
1843 namespace: namespace.to_string(),
1844 key_hash,
1845 key_preview: key_preview.clone(),
1846 value: value.to_vec(),
1847 content_type: opts.content_type.clone(),
1848 created_at: now,
1849 accessed_at: now,
1850 access_count: 0,
1851 ttl_secs: opts.ttl_secs,
1852 cost_hint_ms: opts.cost_hint_ms,
1853 size_bytes,
1854 vector: None,
1855 model: None,
1856 };
1857 let bytes = rmp_serde::to_vec(&entry).map_err(|e| MnemoError::Serialize(e.to_string()))?;
1858 let (start_page, page_count) = self.write_record(&bytes)?;
1859
1860 // Tombstone any prior entry for this key. Its pages remain on
1861 // disk until `compact_file`, exactly like memory-record
1862 // overwrites — the tombstone just hides it from lookups.
1863 if let Some(prev_pos) = self.cache_idx.get(namespace, &key_hash) {
1864 self.cache_dir[prev_pos].deleted = true;
1865 self.cache_idx.remove(namespace, &key_hash);
1866 }
1867
1868 let dir_entry = CacheDirectoryEntry {
1869 namespace: namespace.to_string(),
1870 key_hash,
1871 start_page,
1872 page_count,
1873 len: bytes.len() as u32,
1874 deleted: false,
1875 accessed_at: now,
1876 access_count: 0,
1877 size_bytes,
1878 ttl_secs: opts.ttl_secs,
1879 created_at: now,
1880 vector: None,
1881 model: None,
1882 };
1883 let pos = self.cache_dir.len();
1884 self.cache_dir.push(dir_entry);
1885 self.cache_idx.insert(namespace, key_hash, pos);
1886
1887 // Enforce budget — evict LRU tombstones until the namespace
1888 // fits under both caps.
1889 self.evict_to_budget(namespace)?;
1890
1891 self.dirty_cache = true;
1892 self.cache_batch.record_dirty();
1893
1894 // Batched policy: auto-flush if a threshold tripped. Successful
1895 // `flush()` calls `cache_batch.reset()` for us.
1896 if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
1897 self.flush()?;
1898 }
1899 Ok(())
1900 }
1901
1902 /// Look up `(namespace, key)`. Returns `None` on miss (either the
1903 /// key was never inserted or the entry is TTL-expired). On hit
1904 /// updates catalog-side access stats (mirrors the recall v5-trick);
1905 /// on a read-only handle the stats update is silently skipped.
1906 ///
1907 /// A miss is recorded as a miss in this handle's counters even if
1908 /// the key was TTL-expired — from the caller's perspective the two
1909 /// are indistinguishable.
1910 pub fn cache_get(&mut self, namespace: &str, key: &str) -> Result<Option<CachedValue>> {
1911 let key_hash = result_cache::hash_key(key);
1912 let pos = match self.cache_idx.get(namespace, &key_hash) {
1913 Some(p) => p,
1914 None => {
1915 self.cache_counters.misses += 1;
1916 return Ok(None);
1917 }
1918 };
1919 let now = memory::now_secs();
1920 // Snapshot the directory entry fields we need before any
1921 // mutation — the borrow checker gets grumpy otherwise.
1922 let (start_page, page_count, len, ttl_secs, expired, created_at) = {
1923 let e = &self.cache_dir[pos];
1924 (
1925 e.start_page,
1926 e.page_count,
1927 e.len,
1928 e.ttl_secs,
1929 e.is_expired(now),
1930 e.created_at,
1931 )
1932 };
1933 if expired {
1934 // Treat expired as a miss for the caller. Don't tombstone
1935 // here — cache_purge does that. Just skip.
1936 self.cache_counters.misses += 1;
1937 return Ok(None);
1938 }
1939 // Decrypt and decode the record body.
1940 let mut buf = Vec::with_capacity(len as usize);
1941 for i in 0..page_count as u64 {
1942 buf.extend_from_slice(&self.pager.read_page(start_page + i)?);
1943 }
1944 buf.truncate(len as usize);
1945 let entry: CacheEntry =
1946 rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
1947
1948 // Bump access stats catalog-only (v5 trick — no full record
1949 // rewrite on a hit). Skipped on read-only handles so a shared-lock
1950 // reader can't dirty the catalog.
1951 if !self.read_only {
1952 self.cache_dir[pos].accessed_at = now;
1953 self.cache_dir[pos].access_count = self.cache_dir[pos].access_count.saturating_add(1);
1954 self.dirty_cache = true;
1955 }
1956 self.cache_counters.hits += 1;
1957 // On a mutating handle: access_count reflects the just-bumped
1958 // value. On a read-only handle: the on-disk value is unchanged,
1959 // so returning it verbatim is honest (the caller's reads did
1960 // not, and cannot, alter the persisted counter).
1961 let access_count = self.cache_dir[pos].access_count;
1962 Ok(Some(CachedValue {
1963 value: entry.value,
1964 content_type: entry.content_type,
1965 created_at,
1966 accessed_at: if self.read_only {
1967 entry.accessed_at
1968 } else {
1969 now
1970 },
1971 access_count,
1972 ttl_secs,
1973 }))
1974 }
1975
1976 /// Tombstone the entry at `(namespace, key)`. Silently succeeds if
1977 /// no such entry exists. Space is reclaimed by `compact_file`.
1978 pub fn cache_delete(&mut self, namespace: &str, key: &str) -> Result<()> {
1979 if self.read_only {
1980 return Err(MnemoError::ReadOnly);
1981 }
1982 let key_hash = result_cache::hash_key(key);
1983 if let Some(pos) = self.cache_idx.get(namespace, &key_hash) {
1984 self.cache_dir[pos].deleted = true;
1985 self.cache_idx.remove(namespace, &key_hash);
1986 self.dirty_cache = true;
1987 self.cache_batch.record_dirty();
1988 if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
1989 self.flush()?;
1990 }
1991 }
1992 Ok(())
1993 }
1994
1995 /// Purge cache entries. If `namespace` is `Some`, restrict to that
1996 /// namespace; otherwise purge across the whole cache. If
1997 /// `expired_only` is `true`, only TTL-expired entries are removed;
1998 /// otherwise every live entry in the target scope is tombstoned.
1999 /// Returns the number of entries newly tombstoned.
2000 pub fn cache_purge(&mut self, namespace: Option<&str>, expired_only: bool) -> Result<usize> {
2001 if self.read_only {
2002 return Err(MnemoError::ReadOnly);
2003 }
2004 let now = memory::now_secs();
2005 let mut purged = 0usize;
2006 // Iterate the directory by index; mark matching entries as
2007 // deleted and pull them out of the in-memory index.
2008 for pos in 0..self.cache_dir.len() {
2009 if self.cache_dir[pos].deleted {
2010 continue;
2011 }
2012 if let Some(ns) = namespace {
2013 if self.cache_dir[pos].namespace != ns {
2014 continue;
2015 }
2016 }
2017 if expired_only && !self.cache_dir[pos].is_expired(now) {
2018 continue;
2019 }
2020 let key_hash = self.cache_dir[pos].key_hash;
2021 let ns = self.cache_dir[pos].namespace.clone();
2022 self.cache_dir[pos].deleted = true;
2023 self.cache_idx.remove(&ns, &key_hash);
2024 purged += 1;
2025 }
2026 if purged > 0 {
2027 self.dirty_cache = true;
2028 self.cache_batch.record_dirty();
2029 if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
2030 self.flush()?;
2031 }
2032 }
2033 Ok(purged)
2034 }
2035
2036 /// Statistics for `namespace` (or the whole cache if `None`).
2037 /// `hits`/`misses`/`evictions` are per-handle in-memory counters
2038 /// that reset on reopen; `entries`/`bytes` reflect the durable
2039 /// on-disk state (live entries only, tombstones excluded).
2040 pub fn cache_stats(&self, namespace: Option<&str>) -> CacheStats {
2041 let mut entries = 0usize;
2042 let mut bytes = 0u64;
2043 for e in &self.cache_dir {
2044 if e.deleted {
2045 continue;
2046 }
2047 if let Some(ns) = namespace {
2048 if e.namespace != ns {
2049 continue;
2050 }
2051 }
2052 entries += 1;
2053 bytes += e.size_bytes;
2054 }
2055 let hits = self.cache_counters.hits;
2056 let misses = self.cache_counters.misses;
2057 let total = hits + misses;
2058 let hit_rate = if total > 0 {
2059 hits as f64 / total as f64
2060 } else {
2061 0.0
2062 };
2063 CacheStats {
2064 entries,
2065 bytes,
2066 hits,
2067 misses,
2068 hit_rate,
2069 evictions: self.cache_counters.evictions,
2070 }
2071 }
2072
2073 /// Set (or update) the eviction budget for `namespace`. Applies to
2074 /// subsequent `cache_put` calls; does not retroactively evict
2075 /// entries that already sit under a looser cap.
2076 pub fn set_cache_budget(&mut self, namespace: &str, budget: CacheBudget) {
2077 self.cache_budgets.insert(namespace.to_string(), budget);
2078 }
2079
2080 /// Set the per-handle cache flush policy. Defaults to
2081 /// [`CacheFlushPolicy::Strict`]. Memory writes are never batched;
2082 /// this affects only cache mutations.
2083 pub fn set_cache_flush_policy(&mut self, policy: CacheFlushPolicy) {
2084 self.cache_flush_policy = policy;
2085 // Switching mid-run: if we're moving to Strict and the batch is
2086 // non-empty, the pending count still counts toward dirty_cache
2087 // which the next flush() will consume.
2088 }
2089
2090 /// Evict LRU live entries in `namespace` until it fits under both
2091 /// budget caps. Called from `cache_put` after the insert; the new
2092 /// entry is included in the count so we always leave room.
2093 fn evict_to_budget(&mut self, namespace: &str) -> Result<()> {
2094 let budget = self
2095 .cache_budgets
2096 .get(namespace)
2097 .copied()
2098 .unwrap_or_default();
2099 loop {
2100 // Sum current live count + bytes in the namespace.
2101 let mut live_count = 0usize;
2102 let mut live_bytes = 0u64;
2103 for e in &self.cache_dir {
2104 if !e.deleted && e.namespace == namespace {
2105 live_count += 1;
2106 live_bytes += e.size_bytes;
2107 }
2108 }
2109 if live_count <= budget.max_entries && live_bytes <= budget.max_bytes {
2110 return Ok(());
2111 }
2112 // Find the LRU (oldest accessed_at) live entry.
2113 let victim_pos = self
2114 .cache_dir
2115 .iter()
2116 .enumerate()
2117 .filter(|(_, e)| !e.deleted && e.namespace == namespace)
2118 .min_by_key(|(_, e)| e.accessed_at)
2119 .map(|(i, _)| i);
2120 let Some(pos) = victim_pos else {
2121 // Nothing to evict but still over budget — the single
2122 // just-inserted entry itself is bigger than
2123 // max_bytes. Nothing we can safely do here without
2124 // rejecting the put; leave it and return.
2125 return Ok(());
2126 };
2127 let key_hash = self.cache_dir[pos].key_hash;
2128 let ns = self.cache_dir[pos].namespace.clone();
2129 self.cache_dir[pos].deleted = true;
2130 self.cache_idx.remove(&ns, &key_hash);
2131 self.cache_counters.evictions += 1;
2132 }
2133 }
2134
2135 // === Semantic cache (Phase 10.2) ==================================
2136
2137 /// Insert or overwrite a **semantic** cache entry — same as
2138 /// [`Mnemo::cache_put`] but with a required embedding vector and
2139 /// model identifier. The vector is stored on the directory entry
2140 /// so [`Mnemo::cache_get_semantic`] can score without decrypting
2141 /// record bodies.
2142 ///
2143 /// Vector dimensionality is validated against the database's
2144 /// configured `dimensions` — a mismatch returns
2145 /// [`MnemoError::DimensionMismatch`], the same signal `remember`
2146 /// gives for the memory catalog.
2147 ///
2148 /// The `model` string is matched **exactly** on lookup — a
2149 /// different model's cache is invisible to a given query. Pick
2150 /// canonical identifiers (`"text-embedding-3-small"`,
2151 /// `"bge-large-en-v1.5"`, ...) and stick to them.
2152 pub fn cache_put_semantic(
2153 &mut self,
2154 namespace: &str,
2155 key: &str,
2156 vector: Vec<f32>,
2157 value: &[u8],
2158 opts: SemanticCachePutOpts,
2159 ) -> Result<()> {
2160 if self.read_only {
2161 return Err(MnemoError::ReadOnly);
2162 }
2163 if vector.len() != self.dimensions {
2164 return Err(MnemoError::DimensionMismatch {
2165 expected: self.dimensions,
2166 got: vector.len(),
2167 });
2168 }
2169 let key_hash = result_cache::hash_key(key);
2170 let key_preview = result_cache::key_preview(key);
2171 let now = memory::now_secs();
2172 let size_bytes = value.len() as u64;
2173
2174 let entry = CacheEntry {
2175 namespace: namespace.to_string(),
2176 key_hash,
2177 key_preview: key_preview.clone(),
2178 value: value.to_vec(),
2179 content_type: opts.content_type.clone(),
2180 created_at: now,
2181 accessed_at: now,
2182 access_count: 0,
2183 ttl_secs: opts.ttl_secs,
2184 cost_hint_ms: opts.cost_hint_ms,
2185 size_bytes,
2186 vector: Some(vector.clone()),
2187 model: Some(opts.model.clone()),
2188 };
2189 let bytes = rmp_serde::to_vec(&entry).map_err(|e| MnemoError::Serialize(e.to_string()))?;
2190 let (start_page, page_count) = self.write_record(&bytes)?;
2191
2192 if let Some(prev_pos) = self.cache_idx.get(namespace, &key_hash) {
2193 self.cache_dir[prev_pos].deleted = true;
2194 self.cache_idx.remove(namespace, &key_hash);
2195 }
2196
2197 let dir_entry = CacheDirectoryEntry {
2198 namespace: namespace.to_string(),
2199 key_hash,
2200 start_page,
2201 page_count,
2202 len: bytes.len() as u32,
2203 deleted: false,
2204 accessed_at: now,
2205 access_count: 0,
2206 size_bytes,
2207 ttl_secs: opts.ttl_secs,
2208 created_at: now,
2209 vector: Some(vector),
2210 model: Some(opts.model),
2211 };
2212 let pos = self.cache_dir.len();
2213 self.cache_dir.push(dir_entry);
2214 self.cache_idx.insert(namespace, key_hash, pos);
2215
2216 self.evict_to_budget(namespace)?;
2217
2218 self.dirty_cache = true;
2219 self.cache_batch.record_dirty();
2220 if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
2221 self.flush()?;
2222 }
2223 Ok(())
2224 }
2225
2226 /// Semantic recall over the cache: score every live entry in
2227 /// `namespace` whose `model` matches against `query_vector` using
2228 /// cosine similarity; return the top-1 hit and its similarity iff
2229 /// `sim >= threshold`. `None` otherwise.
2230 ///
2231 /// Uses [`crate::result_cache::DEFAULT_SEMANTIC_THRESHOLD`] if you
2232 /// want a sensible default — pass it explicitly (`0.97` today).
2233 ///
2234 /// Exact-key entries (those without a vector) are transparently
2235 /// skipped, as are entries whose model doesn't match — a
2236 /// different-model hit would be a semantic bug, not a win.
2237 /// TTL-expired entries are also skipped.
2238 ///
2239 /// Access-stat updates on hit follow the same policy as
2240 /// [`Mnemo::cache_get`]: bumped on read-write handles, silently
2241 /// skipped on read-only. Hits and misses are recorded in the
2242 /// per-handle counters just like exact-key lookups.
2243 pub fn cache_get_semantic(
2244 &mut self,
2245 namespace: &str,
2246 query_vector: &[f32],
2247 threshold: f32,
2248 model: &str,
2249 ) -> Result<Option<(CachedValue, f32)>> {
2250 if query_vector.len() != self.dimensions {
2251 return Err(MnemoError::DimensionMismatch {
2252 expected: self.dimensions,
2253 got: query_vector.len(),
2254 });
2255 }
2256 let now = memory::now_secs();
2257
2258 // Scan namespace live entries, filter by model, compute
2259 // cosine, keep the best. Exact-key entries have `vector: None`
2260 // and are skipped by the filter.
2261 let mut best: Option<(usize, f32)> = None;
2262 for (i, e) in self.cache_dir.iter().enumerate() {
2263 if e.deleted || e.namespace != namespace || e.is_expired(now) {
2264 continue;
2265 }
2266 let Some(ref v) = e.vector else { continue };
2267 let Some(ref m) = e.model else { continue };
2268 if m != model {
2269 continue;
2270 }
2271 let sim = memory::cosine(query_vector, v);
2272 if best.map(|(_, s)| sim > s).unwrap_or(true) {
2273 best = Some((i, sim));
2274 }
2275 }
2276
2277 let Some((pos, sim)) = best else {
2278 self.cache_counters.misses += 1;
2279 return Ok(None);
2280 };
2281 if sim < threshold {
2282 self.cache_counters.misses += 1;
2283 return Ok(None);
2284 }
2285
2286 // Extract fields before any mutation to keep the borrow checker
2287 // happy.
2288 let (start_page, page_count, len, ttl_secs, created_at) = {
2289 let e = &self.cache_dir[pos];
2290 (e.start_page, e.page_count, e.len, e.ttl_secs, e.created_at)
2291 };
2292
2293 // Decrypt and decode the record body.
2294 let mut buf = Vec::with_capacity(len as usize);
2295 for i in 0..page_count as u64 {
2296 buf.extend_from_slice(&self.pager.read_page(start_page + i)?);
2297 }
2298 buf.truncate(len as usize);
2299 let entry: CacheEntry =
2300 rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
2301
2302 if !self.read_only {
2303 self.cache_dir[pos].accessed_at = now;
2304 self.cache_dir[pos].access_count = self.cache_dir[pos].access_count.saturating_add(1);
2305 self.dirty_cache = true;
2306 }
2307 self.cache_counters.hits += 1;
2308 let access_count = self.cache_dir[pos].access_count;
2309 Ok(Some((
2310 CachedValue {
2311 value: entry.value,
2312 content_type: entry.content_type,
2313 created_at,
2314 accessed_at: if self.read_only {
2315 entry.accessed_at
2316 } else {
2317 now
2318 },
2319 access_count,
2320 ttl_secs,
2321 },
2322 sim,
2323 )))
2324 }
2325
2326 /// Begin a conversation [`Session`](crate::Session) for `agent_id`.
2327 ///
2328 /// The session borrows the database for its lifetime, records turns as
2329 /// working memory, and consolidates them into episodic memory when closed.
2330 pub fn session(&mut self, agent_id: impl Into<String>) -> crate::session::Session<'_> {
2331 crate::session::Session::new(self, agent_id.into())
2332 }
2333
2334 // --- snapshots & point-in-time recovery ------------------------------
2335
2336 /// Every committed transaction, oldest first — the restore points
2337 /// available to [`Mnemo::restore_to`] and [`Mnemo::restore_to_time`].
2338 ///
2339 /// Each `flush` appends one snapshot. Because the storage engine is
2340 /// append-only, the pages a past flush wrote are still on disk, so any
2341 /// listed snapshot can be reinstated exactly. The history reaches back to
2342 /// the last [`Mnemo::compact_file`], which reclaims space by collapsing
2343 /// it.
2344 pub fn snapshots(&self) -> Vec<SnapshotInfo> {
2345 let mut v: Vec<SnapshotInfo> = self
2346 .manifest
2347 .iter()
2348 .map(|s| SnapshotInfo {
2349 txn_id: s.txn_id,
2350 created_at: s.created_at,
2351 memory_count: s.memory_count,
2352 })
2353 .collect();
2354 v.sort_by_key(|s| s.txn_id);
2355 v
2356 }
2357
2358 /// Load a past snapshot's state and commit it as a new transaction.
2359 fn apply_snapshot(&mut self, snap: &Snapshot) -> Result<()> {
2360 let catalog = load_catalog(
2361 &mut self.pager,
2362 snap.catalog_start,
2363 snap.catalog_pages,
2364 snap.catalog_len,
2365 )?;
2366 let ann = load_index(
2367 &mut self.pager,
2368 snap.index_start,
2369 snap.index_pages,
2370 snap.index_len,
2371 self.dimensions,
2372 )?;
2373 self.index = build_id_index(&catalog);
2374 self.catalog = catalog;
2375 self.ann = ann;
2376 // Re-commit as a fresh transaction: crash-safe, and itself recorded
2377 // as a new snapshot so a restore can always be undone.
2378 self.dirty_catalog = true;
2379 self.dirty_index = true;
2380 self.flush()
2381 }
2382
2383 /// Restore the database to the snapshot produced by transaction `txn_id`.
2384 ///
2385 /// The restore is itself a new committed transaction (and a new
2386 /// snapshot), so it is crash-safe and reversible — restoring forward to a
2387 /// later snapshot afterwards works just as well.
2388 pub fn restore_to(&mut self, txn_id: u64) -> Result<SnapshotInfo> {
2389 if self.read_only {
2390 return Err(MnemoError::ReadOnly);
2391 }
2392 let snap = self
2393 .manifest
2394 .iter()
2395 .find(|s| s.txn_id == txn_id)
2396 .cloned()
2397 .ok_or_else(|| MnemoError::NotFound(format!("snapshot for transaction {txn_id}")))?;
2398 self.apply_snapshot(&snap)?;
2399 Ok(SnapshotInfo {
2400 txn_id: snap.txn_id,
2401 created_at: snap.created_at,
2402 memory_count: snap.memory_count,
2403 })
2404 }
2405
2406 /// Restore the database to the latest snapshot committed at or before
2407 /// `unix_secs`. Returns [`MnemoError::NotFound`] if no snapshot is that
2408 /// old. Like [`Mnemo::restore_to`], the restore is a new transaction.
2409 pub fn restore_to_time(&mut self, unix_secs: i64) -> Result<SnapshotInfo> {
2410 if self.read_only {
2411 return Err(MnemoError::ReadOnly);
2412 }
2413 let snap = self
2414 .manifest
2415 .iter()
2416 .filter(|s| s.created_at <= unix_secs)
2417 .max_by_key(|s| s.txn_id)
2418 .cloned()
2419 .ok_or_else(|| {
2420 MnemoError::NotFound(format!("no snapshot at or before time {unix_secs}"))
2421 })?;
2422 self.apply_snapshot(&snap)?;
2423 Ok(SnapshotInfo {
2424 txn_id: snap.txn_id,
2425 created_at: snap.created_at,
2426 memory_count: snap.memory_count,
2427 })
2428 }
2429
2430 /// Change the passphrase. Cheap: re-derives the KEK and re-wraps the DEK;
2431 /// the encrypted pages are never rewritten.
2432 pub fn rekey(&mut self, new_passphrase: &str, kdf: KdfParams) -> Result<()> {
2433 if self.read_only {
2434 return Err(MnemoError::ReadOnly);
2435 }
2436 self.flush()?;
2437 let new_salt = crypto::random_salt();
2438 let new_kek = crypto::derive_kek(new_passphrase.as_bytes(), &new_salt, kdf)?;
2439 let new_nonce = crypto::random_nonce();
2440 let wrapped = crypto::wrap_dek(&new_kek, &new_nonce, self.pager.dek())?;
2441 if wrapped.len() != WRAPPED_DEK_LEN {
2442 return Err(MnemoError::Crypto("unexpected wrapped DEK length".into()));
2443 }
2444 let mut wrapped_dek = [0u8; WRAPPED_DEK_LEN];
2445 wrapped_dek.copy_from_slice(&wrapped);
2446
2447 self.header.salt = new_salt;
2448 self.header.dek_nonce = new_nonce;
2449 self.header.wrapped_dek = wrapped_dek;
2450 self.header.m_cost = kdf.m_cost;
2451 self.header.t_cost = kdf.t_cost;
2452 self.header.p_cost = kdf.p_cost;
2453 self.header.write_counter = self.pager.write_counter;
2454 self.kdf = kdf;
2455
2456 let mut page = self.header.to_page();
2457 self.header.apply_seal(&mut page, self.pager.dek())?;
2458 self.pager.write_raw(0, &page)?;
2459 self.pager.sync()?;
2460 Ok(())
2461 }
2462
2463 /// Decrypt and validate every live record. Returns the count verified.
2464 pub fn verify(&mut self) -> Result<usize> {
2465 let entries: Vec<CatalogEntry> = self
2466 .catalog
2467 .iter()
2468 .filter(|e| !e.deleted)
2469 .cloned()
2470 .collect();
2471 let n = entries.len();
2472 for e in &entries {
2473 let bytes = self.read_record(e)?;
2474 let _m: Memory = rmp_serde::from_slice(&bytes)
2475 .map_err(|err| MnemoError::Serialize(err.to_string()))?;
2476 }
2477 Ok(n)
2478 }
2479
2480 /// Summary statistics for the open database.
2481 pub fn stats(&mut self) -> Result<Stats> {
2482 let deleted = self.catalog.iter().filter(|e| e.deleted).count();
2483 let mut agents: Vec<String> = self.memories()?.into_iter().map(|m| m.agent_id).collect();
2484 agents.sort();
2485 agents.dedup();
2486 let file_bytes = self.header.next_page * PAGE_SIZE as u64;
2487 Ok(Stats {
2488 memories: self.len(),
2489 deleted,
2490 dimensions: self.dimensions,
2491 file_bytes,
2492 agents,
2493 encrypted: self.header.flags & FLAG_ENCRYPTED != 0,
2494 created_at: self.header.created_at,
2495 index: self.ann.as_ref().map(|a| a.info()),
2496 wal_pages: self.header.wal_pages,
2497 })
2498 }
2499
2500 /// Rewrite the file, dropping tombstoned and expired memories and
2501 /// reclaiming stale pages left by updates. Done out-of-place via a temp
2502 /// file and an atomic rename.
2503 pub fn compact_file(path: &str, passphrase: &str) -> Result<CompactReport> {
2504 let mut old = Mnemo::open(path, passphrase)?;
2505 let dims = old.dimensions;
2506 let kdf = old.kdf;
2507 let tmp = format!("{path}.compact-tmp");
2508
2509 let mut new = Mnemo::create(
2510 &tmp,
2511 passphrase,
2512 MnemoConfig {
2513 dimensions: dims,
2514 kdf,
2515 ..Default::default()
2516 },
2517 )?;
2518 let want_index = old.ann.as_ref().map(|a| (a.n_probe(), a.n_rerank()));
2519 let now = memory::now_secs();
2520 let all = old.memories()?;
2521 let before = all.len();
2522 let mut after = 0;
2523 for m in all {
2524 if m.is_expired(now) {
2525 continue;
2526 }
2527 new.remember(m)?;
2528 after += 1;
2529 }
2530 // Rebuild the index fresh (re-clustered) when the source had one.
2531 if let Some((n_probe, n_rerank)) = want_index {
2532 if after > 0 {
2533 new.build_index_with(IndexConfig {
2534 n_probe,
2535 n_rerank,
2536 ..Default::default()
2537 })?;
2538 }
2539 }
2540 new.flush()?;
2541 drop(new);
2542 drop(old);
2543
2544 std::fs::rename(&tmp, path)?;
2545 Ok(CompactReport { before, after })
2546 }
2547}