Skip to main content

nedb_engine/
db.rs

1// SPDX-FileCopyrightText: 2026 INTERCHAINED LLC
2// SPDX-License-Identifier: BUSL-1.1
3// NEDB · © 2026 INTERCHAINED LLC × Eth-Interchained × Vex (Claude Opus 5)
4
5//! Main DAG database — coordinates ObjectStore, IdIndex, SortedIndexes, GraphStore.
6
7use std::fs;
8use std::path::{Path, PathBuf};
9use std::sync::Arc;
10use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
11use anyhow::Result;
12use dashmap::DashMap;
13use serde_json::Value;
14use parking_lot::RwLock;
15
16use crate::store::{Dek, Node, ObjectStore};
17use crate::index::{IdIndex, OrderedValue, SortedIndexes};
18use crate::graph::GraphStore;
19use crate::migrate;
20
21// ── A note on where diagnostics go ───────────────────────────────────────
22//
23// Every startup and repair message in this crate goes to STDERR, without
24// exception. `nedb-engine` is a LIBRARY, and a library that writes to stdout
25// is corrupting somebody else's output — it just does not find out until a
26// caller needs stdout to mean something.
27//
28// It found out. `nesql --json status` emitted:
29//
30//     [nedbd] cold start — background scan will start after heap allocation
31//     { "ok": true, ... }
32//
33// which is not JSON, so every machine consumer of that command was broken by
34// a progress note. The prints were already SPLIT between the two streams
35// before this — some `println!`, some `eprintln!`, a few lines apart — which
36// is the tell that it was never a decision in the first place.
37//
38// The daemon's own banner stays on stdout. `nedbd` is an application and its
39// stdout belongs to it; `Db::open` is a function anybody may call.
40
41/// MANIFEST: cached {seq, head} written atomically after every write.
42/// On startup, if MANIFEST exists and no sorted indexes need rebuilding,
43/// startup is O(1) — just read this one file instead of scanning all objects.
44#[derive(serde::Serialize, serde::Deserialize)]
45struct Manifest {
46    seq:  u64,
47    head: String,
48    /// Object hash of the highest-seq node at flush time. Lets `tip()` resolve the
49    /// last write O(1) on a warm boot — before any scan repopulates the in-memory
50    /// seq index. `#[serde(default)]` so pre-2.5.43 MANIFESTs (no field) still parse.
51    #[serde(default)]
52    tip_hash: String,
53    /// Per-collection tip: `coll -> object hash of the highest-seq node in that
54    /// collection`. Lets `tip_collection()` resolve O(1) on a warm boot, same
55    /// contract as `tip_hash` for the global head. `#[serde(default)]` so
56    /// pre-this-field MANIFESTs still parse (empty map — self-heals on next write
57    /// or cold scan).
58    #[serde(default)]
59    coll_tips: std::collections::HashMap<String, String>,
60}
61
62/// Default cap for `since()` when the caller passes `limit == 0`. Bounds the
63/// engine primitive itself so a stale/offline consumer can never force an
64/// unbounded materialization — the safety lives in the core, not the HTTP layer.
65pub const DEFAULT_SINCE_LIMIT: usize = 10_000;
66
67/// One page of the changefeed returned by `since()`. The replication contract:
68/// apply `nodes` in ascending seq order, advance your cursor to `to_seq`, and keep
69/// paging while `has_more` is true; then attach to the live `subscribe` edge.
70/// `head_seq` tells the consumer how far the log currently extends (how far behind
71/// it is).
72#[derive(Debug, Clone, serde::Serialize)]
73pub struct SinceBatch {
74    /// Writes in (`from_seq`, `to_seq`], ascending by seq.
75    pub nodes:    Vec<Node>,
76    /// The exclusive cursor this page started from (echoes the request).
77    pub from_seq: u64,
78    /// Seq of the last node in this page — the consumer's next cursor.
79    pub to_seq:   u64,
80    /// Current head seq of the log (latest committed write).
81    pub head_seq: u64,
82    /// True when more writes remain past `to_seq` (the page hit `limit`).
83    pub has_more: bool,
84}
85
86/// Replication readiness snapshot. `scan_complete` is the correctness gate: until
87/// the cold-scan finishes rebuilding the seq index, an old cursor passed to
88/// `since()` can return a PARTIAL page and look (wrongly) like "caught up". A
89/// correctness-critical consumer MUST wait for `scan_complete == true` before
90/// trusting historical catch-up. `indexed_seq_min/max` report the currently
91/// resolvable seq range; `tip_seq` is the log head.
92#[derive(Debug, Clone, serde::Serialize)]
93pub struct ScanStatus {
94    /// Cold-scan finished — historical seqs fully resolvable; catch-up is safe.
95    pub scan_complete:   bool,
96    /// Head seq of the log (latest committed write).
97    pub tip_seq:         u64,
98    /// Lowest seq currently in the seq index (0 if empty).
99    pub indexed_seq_min: u64,
100    /// Highest seq currently in the seq index.
101    pub indexed_seq_max: u64,
102    /// Number of seqs currently resolvable via the index.
103    pub indexed_count:   usize,
104    /// True when the seq index actually covers the log — i.e. `since()` can
105    /// resolve historical seqs. DISTINCT from `scan_complete`: a warm boot is
106    /// "startup complete" in O(1) precisely because it SKIPS the scan, so
107    /// `scan_complete` is true while this is false and `since()` resolves
108    /// nothing. Replication consumers must gate on this field, not on
109    /// `scan_complete`; call `rebuild_id_index()`/`repair()` to populate it.
110    pub seq_index_ready: bool,
111}
112
113pub struct Db {
114    pub objects:        ObjectStore,
115    pub id_index:       IdIndex,
116    /// Deleted id → its tombstone hash. The GRAVEYARD.
117    ///
118    /// `id_index` answers "what is the current version of this key?", so a
119    /// delete has to remove the entry from it or the row would stay visible.
120    /// But that made a deleted document's whole HISTORY unreachable: `AS OF`
121    /// enumerates ids from `id_index`, so the id was never considered at any
122    /// sequence — even one long before the delete. Nothing was lost on disk
123    /// (the tombstone node keeps a `prev` link to the full version chain); it
124    /// was simply unreferenced.
125    ///
126    /// That contradicted the central promise: a `DELETE` is a tombstone, not
127    /// an erasure. So the pointer is not dropped, it is MOVED here — the id
128    /// leaves the land of the living and stays addressable in history.
129    ///
130    /// It is a second `IdIndex` rather than a new namespace inside the first
131    /// because every operation needed — set, get, list, remove, WAL buffering,
132    /// sharded on-disk layout — already exists and is already tested. A
133    /// deliberately boring choice.
134    pub del_index:      IdIndex,
135    pub sorted_indexes: SortedIndexes,
136    pub graph:          GraphStore,
137    pub root:           PathBuf,
138    /// Advisory exclusive lock on the data directory (`LOCK` file), held for
139    /// the Db's lifetime. One process owns a durable store at a time — a
140    /// second opener gets a loud refusal instead of silent split-brain (two
141    /// engines with independent in-memory state on one dir: cross-process
142    /// writes invisible, CAS races — the 2026-07-20 aias multi-worker session
143    /// bug, caught live). Released automatically on drop AND on any process
144    /// death including SIGKILL, because the flock dies with the fd. `None`
145    /// for in-memory databases and under NEDB_SHARED_OPEN=1 (operator
146    /// override for tooling that accepts the risk).
147    _dir_lock:          Option<std::fs::File>,
148    /// Dirty flag — set true when head changes, cleared after manifest flush.
149    /// Decouples flush_manifest from the hot write path so concurrent writes
150    /// don't serialise on 2× file I/O per PUT.
151    manifest_dirty:     Arc<AtomicBool>,
152    pub seq:            AtomicU64,
153    /// Cached Merkle head — updated incrementally on every write (O(1)).
154    head:               RwLock<String>,
155    /// `(seq, object hash)` of the most recent write (highest seq). Mirrors `head`
156    /// but holds the tip's content hash, so `tip()` can resolve the last node O(1)
157    /// on a warm boot when the in-memory `seq_index` is still cold. The seq rides
158    /// along so concurrent writers can settle the tip by HIGHEST SEQ rather than
159    /// arrival order (a slow older put must never clobber a newer tip). Only the
160    /// hash is persisted in MANIFEST — format unchanged.
161    tip_hash:           RwLock<(u64, String)>,
162    /// Per-collection tip: `coll -> (seq, object hash)` of the highest-seq node in
163    /// that collection. Kept current on every write (`update_head`, seq-guarded),
164    /// restored from MANIFEST on warm boot, rebuilt by the cold scan — so
165    /// `tip_collection()` is O(1) and durable across restarts in every startup
166    /// regime, by construction.
167    coll_tip_hash:      Arc<DashMap<String, (u64, String)>>,
168    /// True once startup is fully ready (MANIFEST loaded or cold scan complete).
169    /// Warm starts set this true before returning from open().
170    /// Cold starts set this true in the background thread when scan completes.
171    /// Writes are held with 503 until this is true; reads always proceed.
172    pub startup_ready:  Arc<AtomicBool>,
173    /// Seq → hash lookup for v1 compatibility. Populated by put(), put_batch(),
174    /// and the cold-scan background pass. Only covers nodes from the current
175    /// process session + cold-scan; older seqs not in this map cannot be resolved.
176    seq_index:          Arc<DashMap<u64, String>>,
177    /// Write-time → seq, sorted ascending, for wall-clock `AS OF SYSTEM TIME`
178    /// resolution. Populated alongside `seq_index` (same put/cold-scan paths),
179    /// so it carries the same session-scoped coverage and the same
180    /// `seq_index_ready` gate. A `Vec` of `(ts, seq)` pairs rather than a map:
181    /// the ONLY query it answers is "last seq at or before T", which is one
182    /// binary search over a sorted array — a map would need a full key scan.
183    ts_index:           Arc<std::sync::RwLock<Vec<(f64, u64)>>>,
184    /// Collections already known to be registered, so the common case — every
185    /// write after a collection's first — costs one lock-free map hit instead of
186    /// an index lookup.
187    ///
188    /// A CACHE, never the answer. `collections()` reads the registry in the DAG.
189    /// A stale or empty cache can only cause a redundant registry check, never a
190    /// wrong namespace, which is the asymmetry that makes it safe to keep it
191    /// this simple.
192    known_collections:  Arc<DashMap<String, ()>>,
193}
194
195impl Db {
196    /// Create a pure in-memory database — no disk I/O, no migration, instant startup.
197    /// Perfect for tests, hot-cache layers, and ephemeral sessions.
198    /// All data is lost when the Db is dropped.
199    pub fn in_memory() -> Self {
200        Self {
201            objects:        ObjectStore::in_memory(),
202            id_index:       IdIndex::in_memory(),
203            del_index:      IdIndex::in_memory(),
204            sorted_indexes: SortedIndexes::new(),
205            graph:          GraphStore::in_memory(),
206            root:           std::path::PathBuf::from(":memory:"),
207            _dir_lock:      None,
208            seq:            AtomicU64::new(0),
209            head:           RwLock::new(String::new()),
210            tip_hash:       RwLock::new((0, String::new())),
211            coll_tip_hash:  Arc::new(DashMap::new()),
212            startup_ready:  Arc::new(AtomicBool::new(true)),  // always ready
213            manifest_dirty: Arc::new(AtomicBool::new(false)),
214            seq_index:      Arc::new(DashMap::new()),
215            ts_index:       Arc::new(std::sync::RwLock::new(Vec::new())),
216            known_collections: Arc::new(DashMap::new()),
217        }
218    }
219
220    /// Acquire the exclusive advisory lock on a durable data directory.
221    /// Refuses (with the holder's pid when known) rather than allowing a
222    /// second live engine on the same files. NEDB_SHARED_OPEN=1 skips the
223    /// guard entirely — for tooling that knowingly accepts split-brain risk.
224    fn acquire_dir_lock(db_root: &Path) -> Result<Option<std::fs::File>> {
225        if std::env::var("NEDB_SHARED_OPEN").map(|v| v.trim() == "1").unwrap_or(false) {
226            return Ok(None);
227        }
228        use fs2::FileExt as _;
229        use std::io::Write as _;
230        let lock_path = db_root.join("LOCK");
231        let lock_file = std::fs::OpenOptions::new()
232            .create(true).read(true).write(true).open(&lock_path)?;
233        if lock_file.try_lock_exclusive().is_err() {
234            let holder = std::fs::read_to_string(&lock_path).unwrap_or_default();
235            let holder = holder.trim();
236            anyhow::bail!(
237                "data directory {:?} is locked by another process{} — refusing a \
238                 split-brain open: a second engine on the same files cannot see this \
239                 process's writes (invisible sessions, CAS races). Stop the other \
240                 process, or set NEDB_SHARED_OPEN=1 only if you accept that risk.",
241                db_root,
242                if holder.is_empty() { String::new() } else { format!(" (pid {holder})") }
243            );
244        }
245        // Best-effort: record our pid for the next contender's error message.
246        let _ = lock_file.set_len(0);
247        let _ = writeln!(&lock_file, "{}", std::process::id());
248        let _ = lock_file.sync_all();
249        Ok(Some(lock_file))
250    }
251
252    /// Open (or create) a database. Runs v1→v2 migration automatically if log.aof is present.
253    pub fn open(db_root: &Path, dek: Option<Dek>) -> Result<Self> {
254        std::fs::create_dir_all(db_root)?;
255
256        // Split-brain guard FIRST — refuse before touching any store state.
257        let dir_lock = Self::acquire_dir_lock(db_root)?;
258
259        let objects        = ObjectStore::new(db_root, dek.clone())?;
260        let id_index       = IdIndex::new(db_root)?;
261        // The graveyard lives under its own root so it shares no path with the
262        // live index and cannot be confused with it by any existing reader.
263        let del_index      = IdIndex::new(&db_root.join("graveyard"))?;
264        let sorted_indexes = SortedIndexes::new();
265        let graph          = GraphStore::new(db_root)?;
266
267        let mut db = Self {
268            objects,
269            id_index,
270            del_index,
271            sorted_indexes,
272            graph,
273            root: db_root.to_path_buf(),
274            _dir_lock: dir_lock,
275            seq:  AtomicU64::new(0),
276            head: RwLock::new(String::new()),
277            tip_hash: RwLock::new((0, String::new())),
278            coll_tip_hash: Arc::new(DashMap::new()),
279            startup_ready:  Arc::new(AtomicBool::new(false)),
280            manifest_dirty: Arc::new(AtomicBool::new(false)),
281            seq_index:      Arc::new(DashMap::new()),
282            ts_index:       Arc::new(std::sync::RwLock::new(Vec::new())),
283            known_collections: Arc::new(DashMap::new()),
284        };
285
286        // Auto-migrate v1 → v2 if needed (pass DEK so encrypted AOFs convert correctly)
287        migrate::migrate_if_needed(
288            db_root,
289            &db.objects,
290            &db.id_index,
291            &db.sorted_indexes,
292            &db.graph,
293            dek.as_ref(),
294        )?;
295
296        // Fast startup: load seq+head from MANIFEST if no sorted indexes need rebuilding.
297        // Falls back to full object scan only when necessary (first open, or post-migration).
298        db.startup_rebuild()?;
299
300        Ok(db)
301    }
302
303    /// Smart startup:
304    /// - Warm (MANIFEST exists): O(1) load → startup_ready = true immediately.
305    /// - Cold (no MANIFEST): start server immediately, run scan in background thread.
306    ///   Writes return 503 until scan completes; reads always proceed.
307    fn startup_rebuild(&mut self) -> Result<()> {
308        let manifest_path = self.root.join("MANIFEST");
309        let needs_index_rebuild = !self.sorted_indexes.is_empty();
310
311        // Warm path: MANIFEST + no sorted indexes to rebuild → instant start
312        if manifest_path.exists() && !needs_index_rebuild {
313            if let Some(m) = fs::read_to_string(&manifest_path)
314                .ok()
315                .and_then(|s| serde_json::from_str::<Manifest>(&s).ok())
316            {
317                // Self-heal: MANIFEST with an empty or short head is corrupt/stale.
318                // Fall through to cold scan so the head is rebuilt correctly from objects.
319                if m.head.len() < 8 {
320                    eprintln!("  [nedbd] MANIFEST head invalid (len={}), self-healing via cold scan", m.head.len());
321                } else {
322                    // Pre-2.5.43 MANIFEST (no persisted tip): warm-boot ANYWAY.
323                    //
324                    // The old policy forced a full cold scan "once to upgrade" —
325                    // on multi-million-object embedded stores (itcd -dagv3:
326                    // 1.7M+ objects per database) that scan is hours of random
327                    // reads on seek-bound media, it races the host's own boot
328                    // I/O, and if the process exits before it completes the
329                    // NEXT boot pays it again — a permanent boot tax for
330                    // exactly the deployments that can least afford it. And it
331                    // buys nothing that can't heal lazily: seq + head in the
332                    // old MANIFEST are perfectly valid, and flush_manifest
333                    // writes tip_hash + coll_tips from live state, so the very
334                    // first write + flush after boot upgrades the MANIFEST
335                    // organically. Until then tip()/tip_collection() simply
336                    // return None on this boot — exactly their documented
337                    // behavior for an unresolvable tip — and every other read
338                    // and write path is unaffected.
339                    if m.tip_hash.is_empty() {
340                        eprintln!("  [nedbd] MANIFEST predates durable tip() — warm boot; tip()/tip_collection() heal on first flush (no forced scan)");
341                    }
342                    self.seq.store(m.seq, Ordering::SeqCst); // m.seq is already the next-to-assign counter
343                    *self.head.write() = m.head.clone();
344                    // The tip's seq is the last ASSIGNED seq (m.seq is next-to-assign).
345                    *self.tip_hash.write() = (m.seq.saturating_sub(1), m.tip_hash.clone());
346                    for (coll, hash) in &m.coll_tips {
347                        // Per-coll seqs aren't persisted (MANIFEST format unchanged);
348                        // seed 0 — every future write has seq >= m.seq > 0 and wins,
349                        // and nothing older than the persisted tip can ever arrive
350                        // because the seq counter resumes at m.seq.
351                        self.coll_tip_hash.insert(coll.clone(), (0, hash.clone()));
352                    }
353                    self.startup_ready.store(true, Ordering::SeqCst);
354                    eprintln!("  [nedbd] warm start — seq={} head={}... tip={}...",
355                        m.seq, &m.head[..8],
356                        if m.tip_hash.is_empty() { "(pre-2.5.43, heals on flush)" }
357                        else { &m.tip_hash[..8.min(m.tip_hash.len())] });
358                    return Ok(());
359                }
360            } else {
361                eprintln!("  [nedbd] MANIFEST corrupt or missing, falling back to cold scan");
362            }
363        }
364
365        // Cold path: mark as not ready, return immediately.
366        // The actual background scan is started by Db::start_cold_scan(arc)
367        // which is called from Manager::open_all() AFTER Arc::new(db) — when
368        // the Db is heap-allocated and its field addresses are permanently stable.
369        // Capturing field addresses here would cause UB: Db moves on return.
370        eprintln!("  [nedbd] cold start — background scan will start after heap allocation");
371        Ok(())
372    }
373
374    /// Call this from Manager::open_all() after Arc::new(db).
375    /// Spawns the cold scan background thread with stable heap addresses.
376    /// No-op if startup is already complete (warm start).
377    pub fn start_cold_scan(self_arc: Arc<Self>) {
378        if self_arc.startup_ready.load(Ordering::SeqCst) {
379            return; // warm start — already ready
380        }
381        // Fast path: if the database is empty (new or just created), skip the
382        // background thread entirely. No objects to scan = instant startup.
383        if self_arc.objects.all_hashes().next().is_none() {
384            self_arc.startup_ready.store(true, Ordering::SeqCst);
385            return;
386        }
387        eprintln!("  [nedbd] cold start — background scan starting, server accepting reads now");
388        std::thread::spawn(move || {
389            let db = self_arc;
390            cold_scan_background_arc(db);
391        });
392    }
393
394    /// Rebuild the id index from the object store, synchronously.
395    ///
396    /// Every object carries its own `coll`, `id` and `seq`, so the id index is
397    /// fully derivable: for each (coll, id) the highest seq wins. Use this to
398    /// recover a database whose id-index WAL never reached disk — the objects
399    /// are intact and verify, but `list()`/`get()` return nothing.
400    ///
401    /// Idempotent, and safe on a healthy store (it rewrites the same winners).
402    /// Returns the number of entries written. Flushes before returning.
403    pub fn rebuild_id_index(&self) -> Result<usize> {
404        let hashes: Vec<String> = self.objects.all_hashes().collect();
405        let mut nodes: Vec<Node> = Vec::with_capacity(hashes.len());
406        for h in &hashes {
407            if let Ok(node) = self.objects.read(h) {
408                self.seq_index.insert(node.seq, node.hash.clone());
409                nodes.push(node);
410            }
411        }
412        let written = rebuild_id_index_from_nodes(self, &nodes);
413
414        // The wall-clock index rides the same repair: it is derivable from
415        // the same objects, and a store repaired for `since()` should answer
416        // wall-clock `AS OF` too — not half of its history contract.
417        {
418            let mut pairs: Vec<(f64, u64)> = nodes.iter().map(|n| (n.ts, n.seq)).collect();
419            pairs.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(std::cmp::Ordering::Equal).then(a.1.cmp(&b.1)));
420            if let Ok(mut idx) = self.ts_index.write() {
421                *idx = pairs;
422            }
423        }
424
425        // Per-collection tips, so tip_collection() resolves after a repair.
426        let mut coll_max: std::collections::HashMap<String, (u64, String)> =
427            std::collections::HashMap::new();
428        for node in &nodes {
429            coll_max
430                .entry(node.coll.clone())
431                .and_modify(|cur| {
432                    if node.seq > cur.0 {
433                        *cur = (node.seq, node.hash.clone());
434                    }
435                })
436                .or_insert((node.seq, node.hash.clone()));
437        }
438        for (coll, (seq, hash)) in coll_max {
439            self.coll_tip_hash.insert(coll, (seq, hash));
440        }
441
442        let max_seq = nodes.iter().map(|n| n.seq).max().unwrap_or(0);
443        // Keep the seq counter ahead of everything we just found, so the next
444        // write cannot reuse a seq that already exists in the log.
445        let next = max_seq + 1;
446        if !nodes.is_empty() && self.seq.load(Ordering::SeqCst) < next {
447            self.seq.store(next, Ordering::SeqCst);
448        }
449
450        // Recompute head + tip through the shared implementation, so a repaired
451        // database reopens WARM with a valid MANIFEST instead of coming back up
452        // cold with an empty head (which reads as corruption to the next boot).
453        if !nodes.is_empty() {
454            recompute_head_and_tip(self, hashes, max_seq);
455        }
456
457        self.try_flush_all()?;
458        Ok(written)
459    }
460
461    /// Full repair: rebuild the seq index and the id index from objects, even on
462    /// a WARM store, then flush.
463    ///
464    /// [`start_cold_scan`] deliberately no-ops when startup is already complete,
465    /// which meant the documented repair path ("idempotent — a no-op on a warm
466    /// store, a full self-heal on a stale MANIFEST") could never repair a
467    /// database that had a valid MANIFEST and a damaged id index. This is the
468    /// forcing entry point; `start_cold_scan` keeps its O(1) warm-boot contract.
469    pub fn repair(&self) -> Result<usize> {
470        self.rebuild_id_index()
471    }
472
473    /// Write a document. Returns the new node with its content hash set.
474    ///
475    /// Refuses an unusable or engine-owned collection name, and registers the
476    /// collection if this is its first write — so that "this collection exists"
477    /// becomes a durable fact at the moment it becomes true, rather than an
478    /// inference drawn later from whatever the storage layer happens to have
479    /// lying around.
480    pub fn put(
481        &self,
482        coll: &str,
483        id: &str,
484        data: Value,
485        caused_by: Vec<String>,
486        valid_from: Option<String>,
487        valid_to:   Option<String>,
488    ) -> Result<Node> {
489        crate::namespace::validate_writable(coll)?;
490        self.ensure_collection(coll)?;
491        self.put_unchecked(coll, id, data, caused_by, valid_from, valid_to)
492    }
493
494    /// The write itself, with no namespace policy applied.
495    ///
496    /// Exists so the engine can write its own reserved records through exactly
497    /// the same path user data takes — same object store, same version chain,
498    /// same Merkle head. A registry that was written by a side channel would be
499    /// a registry `verify()` does not cover.
500    pub(crate) fn put_unchecked(
501        &self,
502        coll: &str,
503        id: &str,
504        data: Value,
505        caused_by: Vec<String>,
506        valid_from: Option<String>,
507        valid_to:   Option<String>,
508    ) -> Result<Node> {
509        let seq  = self.seq.fetch_add(1, Ordering::SeqCst);
510        let prev = self.id_index.get(coll, id);
511
512        // Remove old node from sorted indexes (it's being superseded).
513        // Skip the old-object disk read entirely when no sorted index exists —
514        // the read (open + BLAKE2b verify + optional AES-GCM decrypt + JSON
515        // parse) was pure waste in the common unindexed case, ~2x read
516        // amplification on every update (the itcd chainstate shape).
517        if !self.sorted_indexes.is_empty() {
518            if let Some(old_hash) = &prev {
519                if let Ok(old_node) = self.objects.read(old_hash) {
520                    if let Value::Object(ref obj) = old_node.data {
521                        for (field, value) in obj {
522                            self.sorted_indexes.remove(coll, field, value, old_hash);
523                        }
524                    }
525                }
526            }
527        }
528
529        let mut node = Node {
530            id:         id.to_string(),
531            coll:       coll.to_string(),
532            seq,
533            data:       data.clone(),
534            prev,
535            caused_by:  caused_by.clone(),
536            ts:         now(),
537            valid_from,
538            valid_to,
539            hash:       String::new(),
540        };
541
542        // Write to object store (atomic, content-addressed)
543        let hash = self.objects.write(&mut node)?;
544        self.seq_index.insert(seq, hash.clone());
545        self.ts_index_push(node.ts, seq);
546
547        // Update id index (atomic file)
548        self.id_index.set(coll, id, &hash)?;
549
550        // Update sorted indexes
551        if let Value::Object(ref obj) = data {
552            for (field, value) in obj {
553                if self.sorted_indexes.has(coll, field) {
554                    self.sorted_indexes.insert(coll, field, value, &hash);
555                }
556            }
557        }
558
559        // Write causal graph edges
560        for cause in &caused_by {
561            self.graph.add_edge(&hash, "caused_by", cause)?;
562            self.graph.add_edge(cause, "caused_by_rev", &hash)?;
563        }
564
565        // Update running Merkle head: O(1) chain, no full recompute.
566        // new_head = BLAKE2b(prev_head || seq_bytes || new_object_hash)
567        self.update_head(coll, seq, &hash);
568
569        Ok(node)
570    }
571
572    // ── Collection registry ───────────────────────────────────────────────
573    //
574    // See `crate::namespace` for why a collection's existence has to be a
575    // recorded event rather than an inference from storage.
576
577    /// Record that a collection exists, if that is not already recorded.
578    ///
579    /// Idempotent, and cheap after the first write to a given collection: a
580    /// `DashMap` hit. On a miss it consults the registry itself before writing,
581    /// so reopening a database does not re-register everything in it.
582    pub(crate) fn ensure_collection(&self, coll: &str) -> Result<()> {
583        // Fast path: already known, no locking at all. This is every write
584        // after a collection's first.
585        if self.known_collections.contains_key(coll) {
586            return Ok(());
587        }
588
589        // Slow path, taken once per collection per process. The entry lock is
590        // held across the registry write ON PURPOSE: registration has to be
591        // exactly-once, and a check-then-write without it is a race that N
592        // concurrent first-writers all win.
593        //
594        // That race was not hypothetical. Four threads writing into a fresh
595        // collection each saw it as unregistered and each appended a registry
596        // record — harmless to the ANSWER (same id, the version chain just
597        // grows) but four seqs and four nodes spent on one fact, and on a
598        // wide parallel ingest it would be one per writer. A concurrency test
599        // asserting exact sequence counts is what caught it.
600        //
601        // Holding a shard lock across I/O is safe here because nothing in the
602        // write path touches `known_collections`, so there is no path back
603        // into this map to deadlock against.
604        use dashmap::mapref::entry::Entry;
605        match self.known_collections.entry(coll.to_string()) {
606            Entry::Occupied(_) => Ok(()),
607            Entry::Vacant(slot) => {
608                if let Some(rec) = self.get(crate::namespace::COLLECTIONS, coll) {
609                    // Registered in a previous process. Revive it if it was
610                    // dropped and is being written to again — a write is an
611                    // unambiguous assertion that the caller means for this
612                    // collection to exist.
613                    if rec.data.get("dropped").and_then(|v| v.as_bool()).unwrap_or(false) {
614                        self.write_collection_record(coll, false)?;
615                    }
616                } else {
617                    self.write_collection_record(coll, false)?;
618                }
619                slot.insert(());
620                Ok(())
621            }
622        }
623    }
624
625    /// Append a registry record. Creation and drop are the same shape, because
626    /// they are the same kind of event: an assertion, at a sequence, about
627    /// whether a name is currently live. The `prev` chain makes the history of
628    /// that name walkable by exactly the machinery that walks every other
629    /// document's history.
630    fn write_collection_record(&self, coll: &str, dropped: bool) -> Result<()> {
631        let seq = self.seq.load(Ordering::SeqCst);
632        self.put_unchecked(
633            crate::namespace::COLLECTIONS,
634            coll,
635            serde_json::json!({ "name": coll, "dropped": dropped, "at_seq": seq }),
636            vec![], None, None,
637        )?;
638        Ok(())
639    }
640
641    /// Every collection that currently exists.
642    ///
643    /// THE authoritative answer, and the one a state root must commit to.
644    /// Invariant across storage backends and independent of flush timing,
645    /// because it reads recorded events rather than directory entries.
646    ///
647    /// An empty-but-created collection is present here. That is the whole
648    /// point: a database where `orders` was created and then emptied is not the
649    /// same database as one where `orders` never existed, and a root that
650    /// cannot tell them apart is not committing to the namespace.
651    pub fn collections(&self) -> Vec<String> {
652        let mut live: Vec<String> = self.id_index
653            .list_ids(crate::namespace::COLLECTIONS)
654            .into_iter()
655            .filter(|name| {
656                self.get(crate::namespace::COLLECTIONS, name)
657                    .map(|rec| !rec.data.get("dropped")
658                        .and_then(|v| v.as_bool())
659                        .unwrap_or(false))
660                    .unwrap_or(false)
661            })
662            .collect();
663        live.sort();
664        live
665    }
666
667    /// Which collections existed as of a sequence. The namespace is versioned
668    /// for free, because the registry is ordinary documents in the DAG.
669    pub fn collections_as_of(&self, target_seq: u64) -> Vec<String> {
670        let mut live: Vec<String> = self
671            .list_ids_including_deleted(crate::namespace::COLLECTIONS)
672            .into_iter()
673            .filter(|name| {
674                self.get_as_of(crate::namespace::COLLECTIONS, name, target_seq)
675                    .map(|rec| !rec.data.get("dropped")
676                        .and_then(|v| v.as_bool())
677                        .unwrap_or(false))
678                    .unwrap_or(false)
679            })
680            .collect();
681        live.sort();
682        live
683    }
684
685    /// Drop a collection: record that the name is no longer live.
686    ///
687    /// A TOMBSTONE, not an erasure — the same contract `delete` already has for
688    /// documents. The registry keeps the name, marked dropped, so `AS OF`
689    /// before the drop still reports the collection as having existed, and a
690    /// later root can distinguish "dropped" from "never created".
691    ///
692    /// Documents are left where they are. Reclaiming them is `compact`'s job
693    /// and an operator's explicit decision; quietly destroying history behind a
694    /// namespace operation is exactly the behaviour the engine refuses to have.
695    ///
696    /// Returns false when the collection was not live to begin with.
697    pub fn drop_collection(&self, coll: &str) -> Result<bool> {
698        crate::namespace::validate_writable(coll)?;
699        let live = self.get(crate::namespace::COLLECTIONS, coll)
700            .map(|rec| !rec.data.get("dropped")
701                .and_then(|v| v.as_bool())
702                .unwrap_or(false))
703            .unwrap_or(false);
704        if !live {
705            return Ok(false);
706        }
707        self.write_collection_record(coll, true)?;
708        self.known_collections.remove(coll);
709        Ok(true)
710    }
711
712    // ── State roots ───────────────────────────────────────────────────────
713    //
714    // See `crate::root` for the format and for why the leaves are logical
715    // content rather than object hashes.
716
717    /// Every live document, as the material a root is computed from.
718    fn live_records(&self) -> Vec<Node> {
719        let mut out = Vec::new();
720        for coll in self.collections() {
721            for id in self.id_index.list_ids(&coll) {
722                if let Some(n) = self.get(&coll, &id) {
723                    out.push(n);
724                }
725            }
726        }
727        out
728    }
729
730    /// The database's current state root.
731    ///
732    /// A stateless recomputation over live state, not a maintained tree. That
733    /// is a deliberate v1 choice: an incrementally-updated Merkle tree is a
734    /// second source of truth that can silently drift from the first, and the
735    /// cost of being wrong about a root is much higher than the cost of
736    /// recomputing one.
737    pub fn state_root(&self) -> std::result::Result<crate::root::StateRoot, String> {
738        let colls = self.collections();
739        let nodes = self.live_records();
740        let refs: Vec<crate::root::RecordRef<'_>> = nodes.iter()
741            .map(|n| crate::root::RecordRef {
742                coll: &n.coll,
743                id: &n.id,
744                data: &n.data,
745                valid_from: n.valid_from.as_deref(),
746                valid_to: n.valid_to.as_deref(),
747            })
748            .collect();
749        crate::root::compute(&colls, &refs)
750    }
751
752    /// The state root as of a sequence.
753    ///
754    /// Reuses the same enumeration `AS OF` queries already use — live ids plus
755    /// the graveyard — so a historical root sees exactly what a historical
756    /// query would see. Anything else would be a root for a state no query can
757    /// return.
758    ///
759    /// `None` when the material is gone: `compact` prunes superseded versions,
760    /// and a root over history that has been discarded cannot be recomputed.
761    /// Reported as unavailable rather than approximated.
762    pub fn state_root_as_of(&self, target_seq: u64)
763        -> std::result::Result<crate::root::StateRoot, String>
764    {
765        let colls = self.collections_as_of(target_seq);
766        let mut nodes = Vec::new();
767        for coll in &colls {
768            for id in self.list_ids_including_deleted(coll) {
769                if let Some(n) = self.get_as_of(coll, &id, target_seq) {
770                    nodes.push(n);
771                }
772            }
773        }
774        let refs: Vec<crate::root::RecordRef<'_>> = nodes.iter()
775            .map(|n| crate::root::RecordRef {
776                coll: &n.coll,
777                id: &n.id,
778                data: &n.data,
779                valid_from: n.valid_from.as_deref(),
780                valid_to: n.valid_to.as_deref(),
781            })
782            .collect();
783        crate::root::compute(&colls, &refs)
784    }
785
786    // ── Persisted root records ────────────────────────────────────────────
787
788    /// Persist the state root as of a sequence.
789    ///
790    /// Creation and BACKFILL are the same operation with different arguments,
791    /// and they are deliberately not the same COMMAND: `at_seq` at the tip is
792    /// O(live state), while `at_seq` in the past is O(live state) plus a
793    /// version-chain walk per document. Hiding the second behind something
794    /// that looks like the first is how an operator discovers the cost by
795    /// waiting.
796    pub fn create_root_at(&self, at_seq: u64) -> Result<crate::root::RootRecord> {
797        let computed = self.state_root_as_of(at_seq)
798            .map_err(|e| anyhow::anyhow!("compute root at seq {}: {}", at_seq, e))?;
799        let rec = crate::root::RootRecord { at_seq, root: computed };
800        let data = serde_json::to_value(&rec)?;
801        self.put_unchecked(
802            crate::namespace::ROOTS,
803            &crate::namespace::seq_id(at_seq),
804            data, vec![], None, None,
805        )?;
806        Ok(rec)
807    }
808
809    /// Persist the state root at the current tip.
810    pub fn create_root(&self) -> Result<crate::root::RootRecord> {
811        // The tip is the last ASSIGNED seq, so one below the next one out.
812        let tip = self.seq.load(Ordering::SeqCst).saturating_sub(1);
813        self.create_root_at(tip)
814    }
815
816    /// A persisted root record, if one was taken at this sequence.
817    pub fn get_root(&self, at_seq: u64) -> Option<crate::root::RootRecord> {
818        let n = self.get(crate::namespace::ROOTS, &crate::namespace::seq_id(at_seq))?;
819        serde_json::from_value(n.data).ok()
820    }
821
822    /// Every persisted root, oldest first.
823    pub fn list_roots(&self) -> Vec<crate::root::RootRecord> {
824        self.id_index
825            .list_ids(crate::namespace::ROOTS)
826            .into_iter()
827            .filter_map(|id| self.get(crate::namespace::ROOTS, &id))
828            .filter_map(|n| serde_json::from_value::<crate::root::RootRecord>(n.data).ok())
829            .collect()
830    }
831
832    /// Check a persisted root against a fresh recomputation.
833    ///
834    /// Two INDEPENDENT facts, reported independently:
835    ///
836    ///   - the record exists and is well-formed
837    ///   - the history needed to recompute it is still here
838    ///
839    /// A persisted root may outlive the material that produced it — `compact`
840    /// discards superseded versions, and after that a historical root is a
841    /// perfectly valid record of something no longer reconstructable. Folding
842    /// that into PASS would claim a verification that did not happen, and
843    /// folding it into FAIL would report tampering that did not occur. So it
844    /// is neither.
845    pub fn verify_root(&self, at_seq: u64) -> crate::root::RootVerification {
846        let record = match self.get_root(at_seq) {
847            None => return crate::root::RootVerification {
848                at_seq,
849                record: crate::root::RecordStatus::Missing,
850                recomputation: crate::root::Recomputation::NotAttempted,
851                recomputed: None,
852            },
853            Some(r) => r,
854        };
855        if record.root.version != "state_root_v1" {
856            return crate::root::RootVerification {
857                at_seq,
858                record: crate::root::RecordStatus::UnknownVersion(record.root.version.clone()),
859                recomputation: crate::root::Recomputation::NotAttempted,
860                recomputed: None,
861            };
862        }
863        // The floor is the oldest sequence still reconstructable. Below it the
864        // material is gone and a mismatch would say nothing about integrity.
865        if at_seq < self.history_floor() {
866            return crate::root::RootVerification {
867                at_seq,
868                record: crate::root::RecordStatus::Valid,
869                recomputation: crate::root::Recomputation::Unavailable(crate::root::UnavailableReason::HistoryPruned),
870                recomputed: None,
871            };
872        }
873        match self.state_root_as_of(at_seq) {
874            Err(e) => crate::root::RootVerification {
875                at_seq,
876                record: crate::root::RecordStatus::Valid,
877                recomputation: crate::root::Recomputation::Unavailable(crate::root::UnavailableReason::Other(e)),
878                recomputed: None,
879            },
880            Ok(fresh) => {
881                let agrees = fresh.state_root == record.root.state_root;
882                crate::root::RootVerification {
883                    at_seq,
884                    record: crate::root::RecordStatus::Valid,
885                    recomputation: if agrees {
886                        crate::root::Recomputation::Matches
887                    } else {
888                        crate::root::Recomputation::Differs
889                    },
890                    recomputed: Some(fresh),
891                }
892            }
893        }
894    }
895
896    /// The oldest sequence whose state can still be reconstructed.
897    ///
898    /// 0 until something prunes. `compact` records where it cut, because after
899    /// it runs the engine cannot otherwise tell "this sequence had no writes"
900    /// from "this sequence's writes were discarded" — and those two answers
901    /// differ by whether a failed verification means anything.
902    pub fn history_floor(&self) -> u64 {
903        self.get(crate::namespace::META, "history_floor")
904            .and_then(|n| n.data.get("floor").and_then(|v| v.as_u64()))
905            .unwrap_or(0)
906    }
907
908    /// Declare where reconstructable history begins.
909    ///
910    /// Public because pruning is not only something `compact` does: an
911    /// operator who restores from a trimmed backup, or ships a database with
912    /// its early segments removed, has pruned history that the engine has no
913    /// way to notice. Without a way to say so, every historical root in that
914    /// database would fail verification as if it had been tampered with.
915    ///
916    /// MONOTONIC. The floor may rise and may never fall, because lowering it
917    /// asserts that history exists which demonstrably does not — and the first
918    /// thing that assertion does is turn an honest "unavailable" into a
919    /// confident, wrong "mismatch".
920    pub fn set_history_floor(&self, floor: u64) -> Result<()> {
921        let current = self.history_floor();
922        if floor < current {
923            anyhow::bail!(
924                "refusing to lower the history floor from {} to {}: the floor records \
925                 what was DISCARDED, and material does not come back. Lowering it would \
926                 make the engine attempt recomputations it cannot perform and report the \
927                 failures as mismatches.",
928                current, floor
929            );
930        }
931        if floor == current {
932            return Ok(());
933        }
934        self.write_history_floor(floor)
935    }
936
937    fn write_history_floor(&self, floor: u64) -> Result<()> {
938        self.put_unchecked(
939            crate::namespace::META, "history_floor",
940            serde_json::json!({"floor": floor}),
941            vec![], None, None,
942        )?;
943        Ok(())
944    }
945
946    /// Batch put: write N documents in parallel, preserving monotonic seq ordering.
947    /// Pre-allocates N seq numbers atomically, then parallelises object writes and
948    /// id-index updates via Rayon. Each op is independent — safe to parallelise.
949    /// Returns nodes in input order with assigned seq numbers.
950    pub fn put_batch(
951        &self,
952        ops: Vec<(String, String, Value, Vec<String>, Option<String>, Option<String>)>,
953        // (coll, id, data, caused_by, valid_from, valid_to)
954    ) -> Result<Vec<Node>> {
955        use rayon::prelude::*;
956
957        if ops.is_empty() { return Ok(vec![]); }
958
959        // Validate and register EVERY collection before allocating a single
960        // seq. A batch that is going to be refused must be refused before it
961        // has written anything, and registration consumes seqs of its own — so
962        // it cannot happen inside the block that assumes N consecutive ones.
963        for (coll, ..) in ops.iter() {
964            crate::namespace::validate_writable(coll)?;
965        }
966        for coll in ops.iter()
967            .map(|(c, ..)| c.as_str())
968            .collect::<std::collections::BTreeSet<_>>()
969        {
970            self.ensure_collection(coll)?;
971        }
972
973        let n = ops.len() as u64;
974
975        // Pre-allocate N consecutive seq numbers — preserves ordering under concurrency
976        let base_seq = self.seq.fetch_add(n, Ordering::SeqCst);
977        let ts = now();
978
979        // Build nodes with assigned seq numbers
980        let index_live = !self.sorted_indexes.is_empty();
981        let mut nodes: Vec<Node> = ops.into_iter().enumerate().map(|(i, (coll, id, data, caused_by, valid_from, valid_to))| {
982            let prev = self.id_index.get(&coll, &id);
983            // Parity with put(): drop the superseded version's values from any
984            // sorted indexes, so top-k never returns stale hashes after a batch
985            // update. Without this, batch updates left the old version's index
986            // entries in place — ORDER BY surfaced superseded rows alongside
987            // current ones. Only pay the old-object read when an index exists.
988            if index_live {
989                if let Some(old_hash) = &prev {
990                    if let Ok(old_node) = self.objects.read(old_hash) {
991                        if let Value::Object(ref obj) = old_node.data {
992                            for (field, value) in obj {
993                                self.sorted_indexes.remove(&coll, field, value, old_hash);
994                            }
995                        }
996                    }
997                }
998            }
999            Node {
1000                id, coll, seq: base_seq + i as u64,
1001                data, prev, caused_by,
1002                ts, valid_from, valid_to,
1003                hash: String::new(),
1004            }
1005        }).collect();
1006
1007        // Parallel object writes (content-addressed, idempotent, safe to parallelise)
1008        let write_errors: Vec<anyhow::Error> = nodes.par_iter_mut()
1009            .filter_map(|node| self.objects.write(node).err())
1010            .collect();
1011        if let Some(e) = write_errors.into_iter().next() { return Err(e); }
1012
1013        // Parallel id-index updates
1014        let index_errors: Vec<anyhow::Error> = nodes.par_iter()
1015            .filter_map(|node| self.id_index.set(&node.coll, &node.id, &node.hash).err())
1016            .collect();
1017        if let Some(e) = index_errors.into_iter().next() { return Err(e); }
1018
1019        // Sorted indexes + causal graph (sequential — small overhead, usually no indexes)
1020        for node in &nodes {
1021            self.seq_index.insert(node.seq, node.hash.clone());
1022            if let Value::Object(ref obj) = node.data {
1023                for (field, value) in obj {
1024                    if self.sorted_indexes.has(&node.coll, field) {
1025                        self.sorted_indexes.insert(&node.coll, field, value, &node.hash);
1026                    }
1027                }
1028            }
1029            for cause in &node.caused_by {
1030                self.graph.add_edge(&node.hash, "caused_by", cause).ok();
1031                self.graph.add_edge(cause, "caused_by_rev", &node.hash).ok();
1032            }
1033        }
1034
1035        // Single Merkle head update for the whole batch (chain all hashes)
1036        for node in &nodes {
1037            self.update_head(&node.coll, node.seq, &node.hash);
1038        }
1039
1040        Ok(nodes)
1041    }
1042
1043    /// Update the running Merkle head with a new write. O(1); no file I/O — the
1044    /// background ticker flushes MANIFEST.
1045    ///
1046    /// Concurrency contract (this function is reached by parallel `put()`s —
1047    /// the server runs puts on blocking threads):
1048    /// - The head chain is extended under ONE write lock held across the whole
1049    ///   read-modify-write. The old read-then-write shape let two concurrent
1050    ///   writers both read the same prev head; one contribution was silently
1051    ///   dropped from the chain — a corrupted tamper-evidence primitive. The
1052    ///   chain is arrival-ordered under concurrency (a seq-ordered canonical
1053    ///   head is tracked as follow-up work); what this lock guarantees is that
1054    ///   EVERY write is committed into the chain exactly once.
1055    /// - Tip pointers settle by HIGHEST SEQ, not arrival order: concurrent
1056    ///   puts can reach here out of seq order, and "last call wins" could
1057    ///   persist a stale tip into MANIFEST for the next warm boot.
1058    fn update_head(&self, coll: &str, seq: u64, new_hash: &str) {
1059        use blake2::{Blake2b512, Digest};
1060        {
1061            let mut head = self.head.write();
1062            let mut h = Blake2b512::new();
1063            h.update(head.as_bytes());
1064            h.update(seq.to_le_bytes());
1065            h.update(new_hash.as_bytes());
1066            *head = hex::encode(&h.finalize()[..32]);
1067        }
1068        {
1069            let mut tip = self.tip_hash.write();
1070            if seq >= tip.0 {
1071                *tip = (seq, new_hash.to_string());
1072            }
1073        }
1074        self.coll_tip_hash
1075            .entry(coll.to_string())
1076            .and_modify(|t| {
1077                if seq >= t.0 {
1078                    *t = (seq, new_hash.to_string());
1079                }
1080            })
1081            .or_insert_with(|| (seq, new_hash.to_string()));
1082        // Mark dirty — background ticker will flush to MANIFEST (no I/O on write path)
1083        self.manifest_dirty.store(true, Ordering::Release);
1084    }
1085
1086    /// Flush both the id-index WAL and MANIFEST, REPORTING failure.
1087    ///
1088    /// This is the durability boundary: until it returns `Ok(())`, writes that
1089    /// `put()` acknowledged may not be on disk. Callers that must not lose data
1090    /// — anything about to take a destructive or externally-visible action on
1091    /// the strength of a persisted record — should use this, not [`flush_all`].
1092    ///
1093    /// Every stage is attempted even if an earlier one fails (a MANIFEST flush
1094    /// is still worth doing when one index leaf failed), and the first error is
1095    /// returned. Failed id-index entries stay in the WAL for retry.
1096    pub fn try_flush_all(&self) -> Result<()> {
1097        let index_result = self.id_index.try_flush_write_buf()
1098            // The graveyard is as durable as the live index: a tombstone
1099            // pointer lost to a crash would take a document's history back out
1100            // of reach, which is the bug this index exists to prevent.
1101            .and(self.del_index.try_flush_write_buf());
1102        // v3: fsync the active segment (no-op for loose/in-memory stores).
1103        // One durability point per batch instead of one fsync per object.
1104        let sync_result = self.objects.sync();
1105        let manifest_result = self.try_flush_manifest();
1106
1107        index_result.map_err(|e| anyhow::anyhow!("id-index WAL flush failed: {}", e))?;
1108        sync_result.map_err(|e| anyhow::anyhow!("object segment sync failed: {}", e))?;
1109        manifest_result.map_err(|e| anyhow::anyhow!("MANIFEST flush failed: {}", e))?;
1110        Ok(())
1111    }
1112
1113    /// Flush both the id-index WAL and MANIFEST. Used on graceful shutdown.
1114    ///
1115    /// Errors are logged, not returned — kept for back-compat and for the
1116    /// ticker/`Drop` paths that have nowhere to propagate. Prefer
1117    /// [`try_flush_all`] whenever the outcome matters.
1118    pub fn flush_all(&self) {
1119        if let Err(e) = self.try_flush_all() {
1120            eprintln!("nedb: flush_all failed: {}", e);
1121        }
1122    }
1123
1124    /// Compact the v3 packed object store: keep the CURRENT version of every
1125    /// document (from the id-index) and reclaim everything else. No-op unless
1126    /// running with the v3 segment substrate (`--dag-v3` / NEDB_DAG_V3).
1127    ///
1128    /// This is a PRUNING operation: superseded/historical object versions are
1129    /// dropped, so AS OF / TRACE over pruned versions is discarded — that is
1130    /// what reclaims the space. Flushes first so all data is durable on disk
1131    /// before the old segments are deleted.
1132    /// Reclaim space by rewriting the segments with only CURRENT versions.
1133    ///
1134    /// # This discards history. On purpose.
1135    ///
1136    /// The live set is each document's current-version hash and nothing else,
1137    /// so compaction drops every superseded version and every tombstone. After
1138    /// it runs, `AS OF` can no longer reach a prior value and `TRACE` can no
1139    /// longer walk to a pruned ancestor — the rows simply become unavailable
1140    /// rather than wrong, and `verify()` stays clean because what remains is
1141    /// still internally consistent.
1142    ///
1143    /// That is worth stating loudly, because NEDB's headline property is that
1144    /// history is permanent and never garbage-collected — and it is, right up
1145    /// until an operator calls THIS. Nothing calls it automatically: it is not
1146    /// on the HTTP surface, not in the CLI, and not on any timer. It exists for
1147    /// the operator who has decided, explicitly, to trade the audit trail for
1148    /// disk space.
1149    ///
1150    /// A graveyard entry whose tombstone was pruned is left pointing at an
1151    /// object that no longer exists. `get_as_of` degrades to `None` there
1152    /// rather than failing, so a compacted store answers "not available at that
1153    /// sequence" instead of erroring or inventing a value.
1154    ///
1155    /// # Live branches veto it
1156    ///
1157    /// A branch promises a future three-way merge, and a three-way merge needs
1158    /// the BASE side: the parent state as of the branch's fork point. This
1159    /// prunes every superseded version down to the tip, which is exactly the
1160    /// material that base is made of. Running it under a live branch would
1161    /// produce "branch exists, merge ancestry gone" — a branch that can never
1162    /// be reconciled and does not find that out until someone tries.
1163    ///
1164    /// Because compaction here is all-or-nothing to the tip, there is no honest
1165    /// partial answer ("prune down to the pin" is a different algorithm, not a
1166    /// parameter). So the answer is REFUSAL, naming the branches and what they
1167    /// pin. There is deliberately no force flag: a bypass would be reached for
1168    /// exactly when it does the damage, and a silent bypass is the thing this
1169    /// interlock exists to design out. The operator's escape hatch is to merge
1170    /// or abandon the branch — both of which are recorded decisions.
1171    pub fn compact(&self) -> Result<crate::segment::CompactStats> {
1172        // Interlock first: before touching anything, ask what history is spoken
1173        // for. `None` means no live branch, which is the only state in which
1174        // history may be discarded freely.
1175        if let Some(pinned) = crate::branch::minimum_pinned_seq(self) {
1176            anyhow::bail!("{}", crate::branch::compaction_refusal(self, pinned));
1177        }
1178
1179        self.flush_all();
1180        let mut live: std::collections::HashSet<String> = std::collections::HashSet::new();
1181        for coll in self.id_index.collections() {
1182            for id in self.id_index.list_ids(&coll) {
1183                if let Some(h) = self.id_index.get(&coll, &id) {
1184                    live.insert(h);
1185                }
1186            }
1187        }
1188        let stats = self.objects.compact(&live)?;
1189
1190        // Record where history now begins — but ONLY if history was actually
1191        // discarded.
1192        //
1193        // `ObjectStore::compact` is a no-op that returns zeroed stats for the
1194        // loose-object (v2) and in-memory substrates: it prunes nothing at all
1195        // unless the v3 segment store is active. Raising the floor
1196        // unconditionally therefore declared every earlier sequence pruned on
1197        // a database where nothing had been pruned, and every historical root
1198        // became permanently unverifiable with reason HISTORY_PRUNED.
1199        //
1200        // That is a FALSE ALARM, and a false alarm is the one failure this
1201        // three-state verification exists to prevent — an operator who cannot
1202        // trust "unavailable" is back to guessing, which is where PASS/FAIL
1203        // left them. So the floor moves on evidence: objects were dropped.
1204        if stats.dropped_objects > 0 {
1205            let tip = self.seq.load(Ordering::SeqCst).saturating_sub(1);
1206            self.set_history_floor(tip)?;
1207        }
1208        Ok(stats)
1209    }
1210
1211    /// Flush MANIFEST to disk if dirty. No-op for in-memory databases.
1212    pub fn flush_manifest_if_dirty(&self) {
1213        if self.root == std::path::PathBuf::from(":memory:") { return; }
1214        if self.manifest_dirty.compare_exchange(
1215            true, false, Ordering::AcqRel, Ordering::Relaxed
1216        ).is_ok() {
1217            self.flush_manifest();
1218        }
1219    }
1220
1221    /// Atomically persist current seq+head to MANIFEST, reporting failure.
1222    /// No-op (`Ok`) for in-memory databases.
1223    ///
1224    /// A silently failed MANIFEST write is not data loss — the startup
1225    /// self-heal rescans — but it IS a warm-boot regression and, on a full
1226    /// disk, the first symptom that persistence is failing. Callers deserve
1227    /// to know.
1228    pub fn try_flush_manifest(&self) -> std::io::Result<()> {
1229        if self.root == std::path::PathBuf::from(":memory:") { return Ok(()); }
1230        let seq  = self.seq.load(Ordering::SeqCst);
1231        let head = self.head.read().clone();
1232        let tip_hash = self.tip_hash.read().1.clone();
1233        let coll_tips: std::collections::HashMap<String, String> = self.coll_tip_hash
1234            .iter()
1235            .map(|kv| (kv.key().clone(), kv.value().1.clone()))
1236            .collect();
1237        let m = Manifest { seq, head, tip_hash, coll_tips };
1238        let json = serde_json::to_string(&m)
1239            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
1240        let path = self.root.join("MANIFEST");
1241        let tmp  = self.root.join("MANIFEST.tmp");
1242        // fsync the tmp file BEFORE the rename: rename-without-fsync can
1243        // leave a zero-length/partial MANIFEST at the final path after
1244        // power loss (ext4 delayed allocation). The startup self-heal
1245        // (invalid head -> cold scan) catches that, but a full rescan is
1246        // exactly the cost MANIFEST exists to avoid. One fsync per flush,
1247        // and flushes are already off the hot write path (ticker-driven).
1248        let wrote = (|| -> std::io::Result<()> {
1249            use std::io::Write;
1250            let mut f = fs::File::create(&tmp)?;
1251            f.write_all(json.as_bytes())?;
1252            f.sync_all()
1253        })();
1254        if let Err(e) = wrote {
1255            let _ = fs::remove_file(&tmp);
1256            return Err(e);
1257        }
1258        fs::rename(&tmp, &path)?;
1259        // Make the rename itself durable (directory entry). Unix-only;
1260        // on Windows directory handles don't support this and the
1261        // rename is already journaled by NTFS.
1262        #[cfg(unix)]
1263        if let Ok(dir) = fs::File::open(&self.root) {
1264            let _ = dir.sync_all();
1265        }
1266        Ok(())
1267    }
1268
1269    /// Atomically persist current seq+head to MANIFEST. No-op for in-memory databases.
1270    /// Errors are logged; prefer [`try_flush_manifest`] when the outcome matters.
1271    pub fn flush_manifest(&self) {
1272        if let Err(e) = self.try_flush_manifest() {
1273            eprintln!("nedb: MANIFEST flush failed: {}", e);
1274        }
1275    }
1276
1277
1278    /// Start a background thread that flushes both the id-index WAL and MANIFEST
1279    /// every `interval_ms` milliseconds.
1280    /// Call this after Arc::new(db) — the Arc keeps Db alive for the thread's lifetime.
1281    /// Flush cadence for EMBEDDED durable handles (the napi and pyo3 `open()` paths).
1282    ///
1283    /// `nedbd` has always run the manifest ticker at 1 s, so a server flushes the id-index WAL and
1284    /// MANIFEST every second and a hard kill loses at most a second of acknowledged writes. The
1285    /// embedded bindings did not start a ticker at all: their WAL was flushed only by the exit hooks
1286    /// (SIGINT/SIGTERM/atexit) — so an embedded app killed with SIGKILL, OOM-killed, or cut by power
1287    /// lost EVERY write since open, with no bound. Found by CHALK / Sports-Rater on 2026-09-04
1288    /// (acknowledged fan writes gone after `kill -9`). Since 2.8.5 the bindings start the ticker on
1289    /// durable open with this cadence — parity with nedbd.
1290    ///
1291    /// `NEDB_FLUSH_MS` overrides: an integer of milliseconds (min 50), or `0` / `off` to disable
1292    /// (only for hosts that own their own flush cadence). Unset → 1000.
1293    pub fn embedded_flush_interval_ms() -> Option<u64> {
1294        match std::env::var("NEDB_FLUSH_MS") {
1295            Err(_) => Some(1000),
1296            Ok(v) => {
1297                let v = v.trim().to_ascii_lowercase();
1298                if v.is_empty() { return Some(1000); }
1299                if v == "0" || v == "off" || v == "false" || v == "no" { return None; }
1300                match v.parse::<u64>() {
1301                    Ok(ms) => Some(ms.max(50)),
1302                    Err(_) => { eprintln!("nedb: NEDB_FLUSH_MS={:?} is not a number — using 1000", v); Some(1000) }
1303                }
1304            }
1305        }
1306    }
1307
1308    /// Spawn the background flush ticker.
1309    ///
1310    /// The ticker holds a **`Weak<Db>`** and exits the first time the upgrade
1311    /// fails — i.e. as soon as the last real owner drops the database. The
1312    /// caller must therefore keep its own `Arc` alive for as long as it wants
1313    /// ticking; every current caller already does (nedbd stores it in its
1314    /// database map, the napi and pyo3 handles own theirs).
1315    ///
1316    /// It used to hold a strong `Arc` inside an unconditional `loop`, which
1317    /// meant the thread never exited and the `Db` was never dropped. Three
1318    /// consequences, all of them live since 2.8.5:
1319    ///
1320    /// * The exclusive data-dir `LOCK` taken in `Db::open` was never released,
1321    ///   so reopening the same path **in the same process** failed with
1322    ///   "locked by another process (pid N)" where N was the caller's own pid.
1323    /// * Every `open()` leaked a thread and the entire `Db` — indexes, caches,
1324    ///   segment handles — for the lifetime of the process.
1325    /// * `Drop for Db` (flush-on-close) could never fire for embedded users,
1326    ///   exactly as its own doc comment warned: it "only fires once every
1327    ///   owning handle is gone", and an immortal thread always held one.
1328    ///
1329    /// nedbd's `drop_db` was hit by the same thing: removing a database from
1330    /// the map did not free it, and an orphaned ticker went on fsyncing it.
1331    ///
1332    /// The `Arc` is upgraded inside the loop and dropped before the next
1333    /// sleep, so the ticker never extends the database's life across a tick.
1334    /// No final flush is needed here — the owner's `Drop` does it.
1335    pub fn start_manifest_ticker(self_arc: Arc<Self>, interval_ms: u64) {
1336        let weak = Arc::downgrade(&self_arc);
1337        // Do not let this function's own argument keep the database alive.
1338        drop(self_arc);
1339        std::thread::spawn(move || {
1340            loop {
1341                std::thread::sleep(std::time::Duration::from_millis(interval_ms));
1342                // Last owner gone: stop ticking and let the thread die.
1343                let db = match weak.upgrade() {
1344                    Some(db) => db,
1345                    None => break,
1346                };
1347                // Flush id-index WAL to disk (parallel Rayon writes)
1348                db.id_index.flush_write_buf();
1349                db.del_index.flush_write_buf();
1350                // Segment bytes must be durable BEFORE a MANIFEST that
1351                // references them: otherwise power loss can leave MANIFEST
1352                // pointing at a tip whose object bytes were still in the page
1353                // cache — the torn tail is truncated on reopen and the warm
1354                // boot resolves a tip that no longer exists, with the seq
1355                // counter ahead of durable data. Order: sync segments, then
1356                // MANIFEST. Gated on the dirty flag so an idle database pays
1357                // no per-tick fsync. (flush_all already used this order; the
1358                // ticker now matches it.)
1359                if db.manifest_dirty.load(Ordering::Acquire) {
1360                    if let Err(e) = db.objects.sync() {
1361                        eprintln!("nedb: segment sync failed: {}", e);
1362                    }
1363                    db.flush_manifest_if_dirty();
1364                }
1365            }
1366        });
1367    }
1368
1369    /// Return the current Merkle head string. O(1) — read from cache.
1370    pub fn head(&self) -> String {
1371        self.head.read().clone()
1372    }
1373
1374    /// Delete a document — writes a tombstone node and removes the id from the index.
1375    /// The object history is preserved in the DAG; only the live id pointer is cleared.
1376    pub fn delete(&self, coll: &str, id: &str) -> Result<bool> {
1377        crate::namespace::validate_writable(coll)?;
1378        let prev = match self.id_index.get(coll, id) {
1379            None => return Ok(false),   // already gone
1380            Some(h) => h,
1381        };
1382        let seq = self.seq.fetch_add(1, Ordering::SeqCst);
1383        let mut tombstone = Node {
1384            id:         format!("_del_{}", id),
1385            coll:       coll.to_string(),
1386            seq,
1387            data:       serde_json::json!({"_deleted": id, "_prev": prev}),
1388            prev:       Some(prev),
1389            caused_by:  vec![],
1390            ts:         now(),
1391            valid_from: None,
1392            valid_to:   None,
1393            hash:       String::new(),
1394        };
1395        let hash = self.objects.write(&mut tombstone)?;
1396        self.update_head(coll, seq, &hash);
1397        // Remove the live id pointer — doc is now invisible to queries and list()
1398        self.id_index.remove(coll, id)?;
1399        // …and MOVE it to the graveyard, so history stays reachable.
1400        //
1401        // Removing the live pointer without this made the document's whole
1402        // version chain unaddressable: `AS OF` walks ids from `id_index`, so a
1403        // deleted id was skipped at every sequence — including sequences long
1404        // before the delete, where the row demonstrably existed. Nothing was
1405        // lost on disk, only unreferenced, which is the worst kind of data
1406        // loss because `verify()` still counts every object as healthy.
1407        //
1408        // The tombstone hash is the entry point: its `prev` links to the last
1409        // live version, and that chain back to the first write.
1410        self.del_index.set(coll, id, &hash)?;
1411        Ok(true)
1412    }
1413
1414    /// Get the current version of a document by id.
1415    pub fn get(&self, coll: &str, id: &str) -> Option<Node> {
1416        let hash = self.id_index.get(coll, id)?;
1417        self.objects.read(&hash).ok()
1418    }
1419
1420    /// Get a specific version of a document by object hash.
1421    pub fn get_by_hash(&self, hash: &str) -> Option<Node> {
1422        self.objects.read(hash).ok()
1423    }
1424
1425    /// Get a document AS OF a specific sequence number.
1426    /// Walks the version chain (prev links) backward until seq <= target.
1427    ///
1428    /// Reaches DELETED documents too. A delete moves the id's pointer into the
1429    /// graveyard rather than dropping it, so the version chain stays walkable
1430    /// and a row is still readable at a sequence before it was deleted — which
1431    /// is what "a DELETE is a tombstone, not an erasure" has to mean in
1432    /// practice. At or after the tombstone's own sequence the document is
1433    /// correctly absent.
1434    pub fn get_as_of(&self, coll: &str, id: &str, target_seq: u64) -> Option<Node> {
1435        // The live chain first: the common case, and the only one for an id
1436        // that was never deleted.
1437        if let Some(hash) = self.id_index.get(coll, id) {
1438            if let Some(node) = self.walk_back_to(&hash, target_seq) {
1439                return Some(node);
1440            }
1441            // Falling through matters for a RE-CREATED id. A `put` after a
1442            // delete starts a fresh chain with no `prev`, so the live chain
1443            // cannot reach a sequence from before the delete — but the
1444            // graveyard still can.
1445        }
1446        let tomb_hash = self.del_index.get(coll, id)?;
1447        let tomb = self.objects.read(&tomb_hash).ok()?;
1448        // As of the tombstone's own sequence the document is deleted. Returning
1449        // the tombstone node itself would surface `{_deleted, _prev}` as if it
1450        // were the document.
1451        if tomb.seq <= target_seq {
1452            return None;
1453        }
1454        self.walk_back_to(tomb.prev.as_deref()?, target_seq)
1455    }
1456
1457    /// Walk `prev` links back from `hash` to the newest version at or before
1458    /// `target_seq`. `None` when the chain starts after it.
1459    fn walk_back_to(&self, hash: &str, target_seq: u64) -> Option<Node> {
1460        let mut current = self.objects.read(hash).ok()?;
1461        loop {
1462            if current.seq <= target_seq {
1463                return Some(current);
1464            }
1465            let prev_hash = current.prev.as_deref()?;
1466            current = self.objects.read(prev_hash).ok()?;
1467        }
1468    }
1469
1470    /// Every id in a collection that AS OF must consider: the live ones, plus
1471    /// the deleted ones whose history is still addressable.
1472    ///
1473    /// Order is stable (sorted, deduplicated) so a historical query answers the
1474    /// same way run to run.
1475    pub fn list_ids_including_deleted(&self, coll: &str) -> Vec<String> {
1476        let mut ids = self.id_index.list_ids(coll);
1477        ids.extend(self.del_index.list_ids(coll));
1478        ids.sort_unstable();
1479        ids.dedup();
1480        ids
1481    }
1482
1483    /// List all documents in a collection, returning current versions.
1484    pub fn list(&self, coll: &str) -> Vec<Node> {
1485        self.id_index
1486            .list_ids(coll)
1487            .into_iter()
1488            .filter_map(|id| self.get(coll, &id))
1489            .collect()
1490    }
1491
1492    /// Candidate nodes whose `field` falls in the given range, via the sorted
1493    /// index. `None` when no index covers (coll, field) — the caller must then
1494    /// fall back to a scan.
1495    ///
1496    /// Returns CURRENT versions only (the index drops a superseded hash on
1497    /// overwrite), so this must not be used to serve an `AS OF` query.
1498    pub fn range_scan(
1499        &self,
1500        coll: &str,
1501        field: &str,
1502        low: Option<&Value>,
1503        high: Option<&Value>,
1504        low_incl: bool,
1505        high_incl: bool,
1506    ) -> Option<Vec<Node>> {
1507        if !self.sorted_indexes.has(coll, field) {
1508            return None;
1509        }
1510        Some(
1511            self.sorted_indexes
1512                .range(coll, field, low, high, low_incl, high_incl)
1513                .into_iter()
1514                .filter_map(|h| self.objects.read(&h).ok())
1515                .collect(),
1516        )
1517    }
1518
1519    /// Candidate nodes whose `field` equals any of `values` — the indexed path
1520    /// for `=` and for `IN (...)`. `None` when no index covers the field.
1521    pub fn index_lookup(&self, coll: &str, field: &str, values: &[Value]) -> Option<Vec<Node>> {
1522        if !self.sorted_indexes.has(coll, field) {
1523            return None;
1524        }
1525        // A value may legitimately appear in several arms of an IN list, and a
1526        // hash must not be returned twice.
1527        let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
1528        let mut out = vec![];
1529        for v in values {
1530            for h in self.sorted_indexes.exact(coll, field, v) {
1531                if seen.insert(h.clone()) {
1532                    if let Ok(node) = self.objects.read(&h) {
1533                        out.push(node);
1534                    }
1535                }
1536            }
1537        }
1538        Some(out)
1539    }
1540
1541    /// How many rows an indexed range covers, without reading any of them.
1542    /// `None` when no index covers the field.
1543    pub fn range_cardinality(
1544        &self,
1545        coll: &str,
1546        field: &str,
1547        low: Option<&Value>,
1548        high: Option<&Value>,
1549        low_incl: bool,
1550        high_incl: bool,
1551    ) -> Option<usize> {
1552        if !self.sorted_indexes.has(coll, field) {
1553            return None;
1554        }
1555        Some(self.sorted_indexes.range_len(coll, field, low, high, low_incl, high_incl))
1556    }
1557
1558    /// True when a sorted index covers (coll, field).
1559    pub fn has_sorted_index(&self, coll: &str, field: &str) -> bool {
1560        self.sorted_indexes.has(coll, field)
1561    }
1562
1563    /// ORDER BY field ASC LIMIT n — uses sorted index if available, else falls back to full scan.
1564    pub fn order_by_asc(&self, coll: &str, field: &str, limit: usize) -> Vec<Node> {
1565        if self.sorted_indexes.has(coll, field) {
1566            self.sorted_indexes
1567                .top_k_asc(coll, field, limit)
1568                .into_iter()
1569                .filter_map(|h| self.objects.read(&h).ok())
1570                .collect()
1571        } else {
1572            let mut docs = self.list(coll);
1573            docs.sort_by(|a, b| {
1574                let av = a.data.get(field).map(OrderedValue::from).unwrap_or(OrderedValue::Null);
1575                let bv = b.data.get(field).map(OrderedValue::from).unwrap_or(OrderedValue::Null);
1576                av.cmp(&bv)
1577            });
1578            docs.truncate(limit);
1579            docs
1580        }
1581    }
1582
1583    /// ORDER BY field DESC LIMIT n
1584    pub fn order_by_desc(&self, coll: &str, field: &str, limit: usize) -> Vec<Node> {
1585        if self.sorted_indexes.has(coll, field) {
1586            self.sorted_indexes
1587                .top_k_desc(coll, field, limit)
1588                .into_iter()
1589                .filter_map(|h| self.objects.read(&h).ok())
1590                .collect()
1591        } else {
1592            let mut docs = self.list(coll);
1593            docs.sort_by(|a, b| {
1594                let av = a.data.get(field).map(OrderedValue::from).unwrap_or(OrderedValue::Null);
1595                let bv = b.data.get(field).map(OrderedValue::from).unwrap_or(OrderedValue::Null);
1596                bv.cmp(&av)
1597            });
1598            docs.truncate(limit);
1599            docs
1600        }
1601    }
1602
1603    /// TRACE caused_by — walk causal graph from a node.
1604    pub fn trace(&self, hash: &str, reverse: bool, limit: usize) -> Vec<Node> {
1605        self.graph
1606            .trace(hash, "caused_by", reverse, limit)
1607            .into_iter()
1608            .filter_map(|h| self.objects.read(&h).ok())
1609            .collect()
1610    }
1611
1612    /// Verify tamper-evidence of all objects.
1613    pub fn verify(&self) -> (usize, Vec<String>) {
1614        self.objects.verify_all()
1615    }
1616
1617    /// Create a sorted index for a (coll, field) pair.
1618    pub fn create_sorted_index(&self, coll: &str, field: &str) {
1619        self.sorted_indexes.ensure(coll, field);
1620        // Backfill from existing objects
1621        for id in self.id_index.list_ids(coll) {
1622            if let Some(node) = self.get(coll, &id) {
1623                if let Value::Object(ref obj) = node.data {
1624                    if let Some(value) = obj.get(field) {
1625                        self.sorted_indexes.insert(coll, field, value, &node.hash);
1626                    }
1627                }
1628            }
1629        }
1630    }
1631
1632    /// Resolve a sequence number to its content hash (v1 compatibility).
1633    /// Only covers nodes written in the current process session + cold-scan nodes.
1634    pub fn get_hash_by_seq(&self, seq: u64) -> Option<String> {
1635        self.seq_index.get(&seq).map(|r| r.clone())
1636    }
1637
1638    // ── wall-clock AS OF resolution ────────────────────────────────────────
1639    //
1640    // `AS OF SYSTEM TIME <ts>` (a datetime, not a bare integer) asks "state as
1641    // known at wall-clock moment T". The log is seq-ordered and ts is monotonic
1642    // (single-writer sequencer — every put stamps its own `now()`), so the
1643    // answer is one binary search: the last seq whose write-time is <= T.
1644    //
1645    // The index carries the SAME coverage as `seq_index` — this session's
1646    // writes + whatever the cold scan has indexed so far — and is gated by the
1647    // SAME `seq_index_ready` flag. On a warm boot (scan skipped) a wall-clock
1648    // moment that resolves into un-indexed history reports "could not
1649    // determine" rather than guessing; that is the same honesty `since()` was
1650    // taught, and the fix is the same: rebuild/repair to index history.
1651
1652    /// Append one (ts, seq) pair to the wall-clock index. Called from the put
1653    /// paths; appends only, because the sequencer's timestamps are monotonic.
1654    /// A caller-supplied out-of-order ts (never produced by the engine) would
1655    /// corrupt the sort — so the push asserts monotonicity and falls back to a
1656    /// full re-sort, the cheap path being the common one.
1657    fn ts_index_push(&self, ts: f64, seq: u64) {
1658        if let Ok(mut idx) = self.ts_index.write() {
1659            if idx.last().map(|(last_ts, _)| ts >= *last_ts).unwrap_or(true) {
1660                idx.push((ts, seq));
1661            } else {
1662                idx.push((ts, seq));
1663                idx.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(std::cmp::Ordering::Equal).then(a.1.cmp(&b.1)));
1664            }
1665        }
1666    }
1667
1668    /// The newest seq whose write-time is at or before `ts` — the seq a
1669    /// wall-clock `AS OF SYSTEM TIME '<datetime>'` resolves to.
1670    ///
1671    /// `None` = "could not determine": either nothing was written at or before
1672    /// `ts` within indexed history (ts before the store's first write), or the
1673    /// relevant history is not indexed on this boot. Callers report these
1674    /// distinctly: `ts_index_ready()` false means repair/rebuild can help;
1675    /// true with `None` means the moment genuinely precedes indexed history.
1676    /// (After compaction, a pruned moment is "history no longer available" —
1677    /// the floor check the caller does against `history_floor()`.)
1678    pub fn seq_at(&self, ts: f64) -> Option<u64> {
1679        let idx = self.ts_index.read().ok()?;
1680        if idx.is_empty() {
1681            return None;
1682        }
1683        // Binary search: rightmost entry with entry.ts <= ts.
1684        let mut lo = 0usize;
1685        let mut hi = idx.len();
1686        while lo < hi {
1687            let mid = (lo + hi) / 2;
1688            if idx[mid].0 <= ts {
1689                lo = mid + 1;
1690            } else {
1691                hi = mid;
1692            }
1693        }
1694        if lo == 0 {
1695            return None; // every indexed write happened after ts
1696        }
1697        Some(idx[lo - 1].1)
1698    }
1699
1700    /// Whether the wall-clock index covers the store's history (same gate as
1701    /// the seq index: true once the cold scan has indexed, or nothing needs
1702    /// indexing). `false` + a failed `seq_at` = "not indexed on this boot".
1703    pub fn ts_index_ready(&self) -> bool {
1704        self.scan_status().seq_index_ready
1705    }
1706
1707    /// The tip — the most recently written node (highest seq), or `None` if the
1708    /// database is empty. O(1): `self.seq` is the next-to-assign counter, so the
1709    /// latest write sits at `seq - 1`; we resolve it through the same
1710    /// seq_index → object-store path a normal read uses, so the returned Node is
1711    /// byte-identical to one fetched by id or hash (it carries its own seq, hash,
1712    /// causal links, and valid-time). This is the cheap "give me the latest write"
1713    /// primitive — the head of the log, not an aggregate.
1714    pub fn tip(&self) -> Option<Node> {
1715        let next = self.seq.load(Ordering::SeqCst);
1716        if next == 0 {
1717            return None; // nothing written yet
1718        }
1719        // Fast path: resolve the head seq through the in-memory seq index
1720        // (populated by this session's writes or by the cold scan).
1721        if let Some(hash) = self.get_hash_by_seq(next - 1) {
1722            return self.get_by_hash(&hash);
1723        }
1724        // Warm-boot fallback: the seq index is still cold (warm start skips the
1725        // scan), but the tip's object hash was persisted in MANIFEST and restored
1726        // on open. O(1), no scan — this is what makes tip() survive a restart.
1727        let th = self.tip_hash.read().1.clone();
1728        if !th.is_empty() {
1729            return self.get_by_hash(&th);
1730        }
1731        None
1732    }
1733
1734    /// The collection-local tip — the most recent write into `coll` (highest seq in
1735    /// that collection), or `None` if the collection has no writes. O(1): resolves
1736    /// through `coll_tip_hash`, a dedicated per-collection map kept current on every
1737    /// write (`update_head`), restored from MANIFEST on warm boot, and rebuilt by the
1738    /// cold scan — durable across restarts by construction, same contract as `tip()`
1739    /// for the global head. Conceptually a different index than the global `tip()`
1740    /// (global head vs collection head), kept as a separate method so each is
1741    /// explicit — parity with the Python reference's `tip(coll)`. Lets a consumer
1742    /// resume one chain (e.g. blocks / tx / utxo) without pulling global tip and
1743    /// filtering.
1744    pub fn tip_collection(&self, coll: &str) -> Option<Node> {
1745        let hash = self.coll_tip_hash.get(coll)?.1.clone();
1746        self.get_by_hash(&hash)
1747    }
1748
1749    /// Changefeed page: up to `limit` nodes written AFTER `after_seq` (EXCLUSIVE),
1750    /// ascending by seq, wrapped in a `SinceBatch` cursor envelope. `after_seq` is
1751    /// the cursor you last applied (a prior `tip()` seq or `to_seq`). `limit` bounds
1752    /// the page — `0` means DEFAULT_SINCE_LIMIT, so the engine primitive can never
1753    /// materialize an unbounded batch even when embedders call it directly (the
1754    /// safety is here, not only in the HTTP layer). Drain by paging while
1755    /// `has_more`, advancing your cursor to `to_seq`, then hand off to the live
1756    /// `subscribe` edge. The append-only log IS the changefeed, so this is an
1757    /// O(page) walk; unresolved seqs (outside seq_index coverage — see
1758    /// `scan_status()`) are skipped rather than faked.
1759    pub fn since(&self, after_seq: u64, limit: usize) -> SinceBatch {
1760        let next = self.seq.load(Ordering::SeqCst);          // head + 1
1761        let head_seq = next.saturating_sub(1);
1762        let cap = if limit == 0 { DEFAULT_SINCE_LIMIT } else { limit };
1763        let mut nodes: Vec<Node> = Vec::new();
1764        let mut to_seq = after_seq;
1765        let mut hit_limit = false;
1766        let mut s = after_seq.saturating_add(1);
1767        while s < next {
1768            if nodes.len() >= cap { hit_limit = true; break; }
1769            if let Some(hash) = self.get_hash_by_seq(s) {
1770                if let Some(node) = self.get_by_hash(&hash) {
1771                    to_seq = node.seq;
1772                    nodes.push(node);
1773                }
1774            }
1775            s += 1;
1776        }
1777        // `has_more` must never say "caught up" while the cursor is behind the
1778        // log head. Before 2.8.6 this was `hit_limit` alone, so any page whose
1779        // seqs could not be resolved (the whole range, on a warm boot: the warm
1780        // path skips the scan, leaving seq_index empty) returned zero nodes with
1781        // has_more=false — indistinguishable from genuinely up to date. A
1782        // consumer following the documented drain loop stopped forever, one call
1783        // in, on a database with every record unread.
1784        let has_more = hit_limit || to_seq < head_seq;
1785        SinceBatch { nodes, from_seq: after_seq, to_seq, head_seq, has_more }
1786    }
1787
1788    /// Replication readiness — see `ScanStatus`. `scan_complete` gates safe
1789    /// historical catch-up: a consumer pulling an old cursor right after a cold
1790    /// start must wait for it, or `since()` may hand back a partial page that looks
1791    /// like "caught up". Computes the indexed range by scanning the in-memory seq
1792    /// index (O(index)) — intended for periodic status polls, not the per-write
1793    /// hot path.
1794    pub fn scan_status(&self) -> ScanStatus {
1795        let next = self.seq.load(Ordering::SeqCst);
1796        let mut min = u64::MAX;
1797        let mut max = 0u64;
1798        let mut count = 0usize;
1799        for kv in self.seq_index.iter() {
1800            let s = *kv.key();
1801            if s < min { min = s; }
1802            if s > max { max = s; }
1803            count += 1;
1804        }
1805        if count == 0 { min = 0; }
1806        ScanStatus {
1807            scan_complete:   self.startup_ready.load(Ordering::SeqCst),
1808            tip_seq:         next.saturating_sub(1),
1809            indexed_seq_min: min,
1810            indexed_seq_max: max,
1811            indexed_count:   count,
1812            // The seq index covers the log when it resolves as many seqs as the
1813            // log has entries. On a warm boot it is empty while the log is not.
1814            seq_index_ready: count > 0 && (count as u64) >= next.saturating_sub(1),
1815        }
1816    }
1817
1818    /// Add an explicit named relation edge between two documents.
1819    /// Add an explicit named relation between two "coll:id" nodes.
1820    /// Relations stored as __links__ documents — NQL-queryable, time-travelable,
1821    /// consistent with the PyO3 binding which uses the same __links__ convention.
1822    pub fn link(&self, frm: &str, rel: &str, to: &str) -> Result<()> {
1823        let (frm_coll, frm_id) = frm.split_once(':')
1824            .ok_or_else(|| anyhow::anyhow!("link frm must be 'coll:id', got: {}", frm))?;
1825        let (to_coll, to_id) = to.split_once(':')
1826            .ok_or_else(|| anyhow::anyhow!("link to must be 'coll:id', got: {}", to))?;
1827        if self.id_index.get(frm_coll, frm_id).is_none() {
1828            anyhow::bail!("link: frm not found: {}", frm);
1829        }
1830        if self.id_index.get(to_coll, to_id).is_none() {
1831            anyhow::bail!("link: to not found: {}", to);
1832        }
1833        let link_id = format!("{}|{}|{}", frm, rel, to);
1834        let doc = serde_json::json!({"_from": frm, "_rel": rel, "_to": to});
1835        self.put("__links__", &link_id, doc, vec![], None, None)?;
1836        Ok(())
1837    }
1838
1839    /// Remove a named relation (deletes the __links__ document).
1840    pub fn unlink(&self, frm: &str, rel: &str, to: &str) -> Result<bool> {
1841        let link_id = format!("{}|{}|{}", frm, rel, to);
1842        self.delete("__links__", &link_id)
1843    }
1844
1845    /// Get neighbor nodes via a named relation.
1846    /// Queries __links__ — consistent with the PyO3 binding.
1847    pub fn neighbors(&self, frm: &str, rel: &str) -> Vec<Node> {
1848        self.id_index
1849            .list_ids("__links__")
1850            .into_iter()
1851            .filter_map(|id| self.get("__links__", &id))
1852            .filter(|node| {
1853                node.data.get("_from").and_then(|v| v.as_str()) == Some(frm)
1854                    && node.data.get("_rel").and_then(|v| v.as_str()) == Some(rel)
1855            })
1856            .filter_map(|node| {
1857                let to = node.data.get("_to")?.as_str()?;
1858                let (to_coll, to_id) = to.split_once(':')?;
1859                self.get(to_coll, to_id)
1860            })
1861            .collect()
1862    }
1863}
1864
1865impl Drop for Db {
1866    /// Flush buffered state when the database is closed so a write-then-drop
1867    /// sequence is durable without an explicit `flush_all()`.
1868    ///
1869    /// `IdIndex::set` only stages updates in the in-memory WAL `write_buf`;
1870    /// disk persistence happens in `flush_write_buf()`, normally driven by the
1871    /// manifest ticker. A short-lived `Db` (a library user's `{ let db =
1872    /// Db::open(p)?; db.put(..)?; }` block, or a test) has no ticker, so without
1873    /// this its writes would be silently lost on reopen. Flushing on drop
1874    /// mirrors the flush-on-close contract of other embedded stores (sled,
1875    /// RocksDB).
1876    ///
1877    /// In production this is a harmless safety net, not the primary durability
1878    /// path: the manifest ticker thread holds an `Arc<Db>` for the process
1879    /// lifetime, so `Drop` only fires once every owning handle is gone. No-op
1880    /// for in-memory databases (`flush_all` short-circuits on `:memory:`).
1881    fn drop(&mut self) {
1882        self.flush_all();
1883    }
1884}
1885
1886/// Background cold-scan worker. Takes Arc<Db> — safe, Db is on the heap.
1887fn cold_scan_background_arc(db: Arc<Db>) {
1888    use rayon::prelude::*;
1889
1890    let objects        = &db.objects;
1891    let seq_atomic     = &db.seq;
1892    let sorted_indexes = &db.sorted_indexes;
1893    let seq_index      = &db.seq_index;
1894    let ready_flag     = Arc::clone(&db.startup_ready);
1895    // (ts, seq) pairs gathered by the parallel readers, merged into `ts_index`
1896    // once — sorted + deduped — after the collect. Rayon workers push to their
1897    // own vectors; the index itself is built in one pass below.
1898    let ts_pairs: std::sync::Mutex<Vec<(f64, u64)>> = std::sync::Mutex::new(Vec::new());
1899
1900    let hashes: Vec<String> = objects.all_hashes().collect();
1901    let total = hashes.len();
1902
1903    if total == 0 {
1904        ready_flag.store(true, Ordering::SeqCst);
1905        return;
1906    }
1907
1908    eprintln!("  [nedbd] background scan — {} objects...", total);
1909    let t0 = std::time::Instant::now();
1910    let step = (total / 10).max(1000);
1911
1912    // Populate the seq index AS objects are read here, not in a second pass
1913    // afterward: this loop is the slow, disk-I/O-bound phase (verifying and
1914    // parsing every object), and it can run for minutes on a multi-million
1915    // object store. `scan_status().indexed_count` reads `seq_index`'s size, so
1916    // inserting here — not after `.collect()` — is what makes that a real, live
1917    // progress signal through the phase that actually takes the time, instead
1918    // of reporting a flat 0 until this whole pass finishes. Safe: DashMap
1919    // supports concurrent inserts, and every parallel worker here inserts a
1920    // disjoint key (each object has its own seq).
1921    let nodes: Vec<Node> = hashes.par_iter()
1922        .enumerate()
1923        .filter_map(|(i, h)| {
1924            if i > 0 && i % step == 0 {
1925                let pct     = i * 100 / total;
1926                let elapsed = t0.elapsed().as_secs_f32();
1927                let rate    = i as f32 / elapsed;
1928                let eta     = (total - i) as f32 / rate;
1929                eprint!("\r  [nedbd]   {:>3}%  {:>8} / {:>8}  ({:>8.0}/s  eta {:.0}s)   ",
1930                    pct, i, total, rate, eta);
1931            }
1932            let node = objects.read(h).ok()?;
1933            seq_index.insert(node.seq, node.hash.clone());
1934            if let Ok(mut tp) = ts_pairs.lock() { tp.push((node.ts, node.seq)); }
1935            Some(node)
1936        })
1937        .collect();
1938
1939    eprintln!("\r  [nedbd]   100%  {:>8} / {:>8}  ({:.1}s)                        ",
1940        total, total, t0.elapsed().as_secs_f32());
1941
1942    let max_seq = nodes.iter().map(|n| n.seq).max().unwrap_or(0);
1943    seq_atomic.store(max_seq + 1, Ordering::SeqCst);
1944
1945    // Merge the gathered (ts, seq) pairs into the sorted wall-clock index.
1946    if let Ok(mut tp) = ts_pairs.lock() {
1947        let mut pairs = std::mem::take(&mut *tp);
1948        pairs.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(std::cmp::Ordering::Equal).then(a.1.cmp(&b.1)));
1949        pairs.dedup_by(|a, b| a.1 == b.1 && a.0 == b.0);
1950        if let Ok(mut idx) = db.ts_index.write() { *idx = pairs; }
1951    }
1952
1953    // Per-collection tip: highest-seq node's hash, per coll. `nodes` is NOT
1954    // seq-ordered here (it comes from an unordered object-hash scan), so this
1955    // must track the max explicitly — unlike the live write path's "last call
1956    // wins" (which relies on ascending call order that a scan doesn't have).
1957    let mut coll_max: std::collections::HashMap<String, (u64, String)> = std::collections::HashMap::new();
1958
1959    for node in &nodes {
1960        // seq_index was already populated above, during the read pass.
1961        coll_max.entry(node.coll.clone())
1962            .and_modify(|(s, h)| if node.seq > *s { *s = node.seq; *h = node.hash.clone(); })
1963            .or_insert_with(|| (node.seq, node.hash.clone()));
1964        if let Value::Object(ref obj) = node.data {
1965            for (field, value) in obj {
1966                if sorted_indexes.has(&node.coll, field) {
1967                    sorted_indexes.insert(&node.coll, field, value, &node.hash);
1968                }
1969            }
1970        }
1971    }
1972
1973    for (coll, (seq, hash)) in coll_max {
1974        db.coll_tip_hash.insert(coll, (seq, hash));
1975    }
1976
1977    // Rebuild the id index when it has no collections at all — the lost-WAL
1978    // case. Until 2.8.6 the cold scan restored seq_index, coll_tips, head and
1979    // MANIFEST but NEVER the id index, so a database whose id-index WAL never
1980    // reached disk came back with every object present and verifying while
1981    // `list()` and `get()` returned nothing — and `nedb-cli repair`, whose whole
1982    // job is this, reported success without fixing it.
1983    //
1984    // Gated on "no collections" so a normal cold boot of a healthy store (itcd:
1985    // millions of objects) does not pay N extra index writes. A partially lost
1986    // index is repaired by the explicit `rebuild_id_index()` path.
1987    if db.id_index.collections().is_empty() && !nodes.is_empty() {
1988        let restored = rebuild_id_index_from_nodes(&db, &nodes);
1989        eprintln!("  [nedbd] id index was empty — rebuilt {} entries from objects", restored);
1990    }
1991
1992    // Merkle head + tip, through the one shared implementation so the cold scan
1993    // and the explicit repair path can never drift apart.
1994    recompute_head_and_tip(&db, hashes, max_seq);
1995
1996    // Write MANIFEST through the one canonical writer. The hand-rolled write
1997    // this replaces stored `seq: max_seq` (the last USED seq) — but the warm
1998    // boot loads `m.seq` as the NEXT-TO-ASSIGN counter, so a restart right
1999    // after a quiet cold scan handed the next write the tip's seq: a duplicate
2000    // seq in the log (seq_index overwrite, wrong since() page). flush_manifest
2001    // reads the live counter (already max_seq + 1) — correct by construction.
2002    db.flush_manifest();
2003
2004    // Signal server: writes can now proceed
2005    ready_flag.store(true, Ordering::SeqCst);
2006    eprintln!("  [nedbd] background scan complete — seq={} objects={} MANIFEST written", max_seq, total);
2007}
2008
2009/// Recompute the Merkle head and the tip hash from the full object-hash set.
2010///
2011/// Shared by the cold scan and by `repair()` so the two can never disagree
2012/// about what the head of a rebuilt database is. `hashes` must be every object
2013/// hash in the store; `max_seq` the highest seq observed.
2014fn recompute_head_and_tip(db: &Db, hashes: Vec<String>, max_seq: u64) {
2015    use blake2::{Blake2b512, Digest};
2016    let mut sorted_hashes = hashes;
2017    sorted_hashes.sort();
2018    let mut h = Blake2b512::new();
2019    h.update(max_seq.to_le_bytes());
2020    for hash_str in &sorted_hashes {
2021        h.update(hash_str.as_bytes());
2022    }
2023    *db.head.write() = hex::encode(&h.finalize()[..32]);
2024
2025    // Tip = the highest-seq object indexed. Persisting its hash lets tip()
2026    // resolve O(1) on the next warm boot, before any scan repopulates seq_index.
2027    let tip_hash = db.seq_index.iter()
2028        .max_by_key(|kv| *kv.key())
2029        .map(|kv| kv.value().clone())
2030        .unwrap_or_default();
2031    *db.tip_hash.write() = (max_seq, tip_hash);
2032}
2033
2034/// Reconstruct id-index entries from already-read nodes: for every (coll, id),
2035/// the winner is the HIGHEST seq, which is exactly what `put()` would have left
2036/// behind. Returns the number of entries written.
2037///
2038/// The id index is fully derivable from the object store because every object
2039/// carries its own `coll`, `id` and `seq` — so a lost WAL is recoverable, and
2040/// nothing here invents data.
2041fn rebuild_id_index_from_nodes(db: &Db, nodes: &[Node]) -> usize {
2042    let mut winner: std::collections::HashMap<(String, String), (u64, String)> =
2043        std::collections::HashMap::new();
2044    for node in nodes {
2045        let key = (node.coll.clone(), node.id.clone());
2046        winner
2047            .entry(key)
2048            .and_modify(|cur| {
2049                if node.seq > cur.0 {
2050                    *cur = (node.seq, node.hash.clone());
2051                }
2052            })
2053            .or_insert((node.seq, node.hash.clone()));
2054    }
2055    let mut written = 0usize;
2056    for ((coll, id), (_seq, hash)) in &winner {
2057        if db.id_index.set(coll, id, hash).is_ok() {
2058            written += 1;
2059        }
2060    }
2061    // Persist immediately: a rebuild that only lands in the WAL would be lost
2062    // again by the very crash class this recovers from.
2063    if let Err(e) = db.id_index.try_flush_write_buf() {
2064        eprintln!("nedb: id-index rebuild flush failed: {}", e);
2065    }
2066    written
2067}
2068
2069fn now() -> f64 {
2070    std::time::SystemTime::now()
2071        .duration_since(std::time::UNIX_EPOCH)
2072        .map(|d| d.as_secs_f64())
2073        .unwrap_or(0.0)
2074}
2075
2076#[cfg(test)]
2077mod tests {
2078    use super::*;
2079    use tempfile::tempdir;
2080
2081    #[test]
2082    fn put_and_get() {
2083        let dir = tempdir().unwrap();
2084        let db = Db::open(dir.path(), None).unwrap();
2085        db.put(
2086            "blocks", "618000",
2087            serde_json::json!({"height": 618000, "hash": "0000abc"}),
2088            vec![], None, None,
2089        ).unwrap();
2090        let node = db.get("blocks", "618000").unwrap();
2091        assert_eq!(node.id, "618000");
2092        assert_eq!(node.data["height"], 618000);
2093    }
2094
2095    #[test]
2096    fn order_by_with_sorted_index() {
2097        let dir = tempdir().unwrap();
2098        let db = Db::open(dir.path(), None).unwrap();
2099        db.create_sorted_index("blocks", "height");
2100        for h in [3u64, 1, 5, 2, 4] {
2101            db.put("blocks", &h.to_string(),
2102                serde_json::json!({"height": h}),
2103                vec![], None, None).unwrap();
2104        }
2105        let asc = db.order_by_asc("blocks", "height", 3);
2106        let heights: Vec<u64> = asc.iter()
2107            .filter_map(|n| n.data["height"].as_u64())
2108            .collect();
2109        assert_eq!(heights, vec![1, 2, 3]);
2110    }
2111
2112    #[test]
2113    fn causal_trace() {
2114        let dir = tempdir().unwrap();
2115        let db = Db::open(dir.path(), None).unwrap();
2116        let a = db.put("ops", "a", serde_json::json!({"op": "create"}), vec![], None, None).unwrap();
2117        let b = db.put("ops", "b", serde_json::json!({"op": "transfer"}), vec![a.hash.clone()], None, None).unwrap();
2118        let c = db.put("ops", "c", serde_json::json!({"op": "burn"}), vec![b.hash.clone()], None, None).unwrap();
2119
2120        let trace = db.trace(&c.hash, false, 10);
2121        assert_eq!(trace.len(), 3);  // c → b → a
2122    }
2123
2124    #[test]
2125    fn as_of() {
2126        let dir = tempdir().unwrap();
2127        let db = Db::open(dir.path(), None).unwrap();
2128        let v1 = db.put("docs", "x", serde_json::json!({"v": 1}), vec![], None, None).unwrap();
2129        let _v2 = db.put("docs", "x", serde_json::json!({"v": 2}), vec![], None, None).unwrap();
2130
2131        let at_v1 = db.get_as_of("docs", "x", v1.seq).unwrap();
2132        assert_eq!(at_v1.data["v"], 1);
2133        let current = db.get("docs", "x").unwrap();
2134        assert_eq!(current.data["v"], 2);
2135    }
2136
2137    #[test]
2138    fn wall_clock_as_of_resolves_through_seq_at() {
2139        // The full chain a datetime AS OF rides: put stamps its ts, the ts
2140        // index holds it, seq_at binary-searches back to the right write.
2141        let dir = tempdir().unwrap();
2142        let db = Db::open(dir.path(), None).unwrap();
2143
2144        let v1 = db.put("docs", "x", serde_json::json!({"v": 1}), vec![], None, None).unwrap();
2145        let v2 = db.put("docs", "x", serde_json::json!({"v": 2}), vec![], None, None).unwrap();
2146        let v3 = db.put("docs", "x", serde_json::json!({"v": 3}), vec![], None, None).unwrap();
2147
2148        // Ground truth: each node's own ts resolves to ITSELF (the boundary
2149        // case — "at or before" includes the write that happened exactly then).
2150        assert_eq!(db.seq_at(v1.ts), Some(v1.seq));
2151        assert_eq!(db.seq_at(v2.ts), Some(v2.seq));
2152        assert_eq!(db.seq_at(v3.ts), Some(v3.seq));
2153
2154        // One microsecond BEFORE v2's ts resolves to v1 — "state as known at
2155        // that moment", not "state as of the next write".
2156        assert_eq!(db.seq_at(v2.ts - 0.000001), Some(v1.seq));
2157        assert_eq!(db.seq_at(v3.ts - 0.000001), Some(v2.seq));
2158
2159        // Between writes: still the last write at or before.
2160        assert_eq!(db.seq_at((v1.ts + v2.ts) / 2.0), Some(v1.seq));
2161
2162        // Before the store existed: could-not-determine, not a guess.
2163        assert_eq!(db.seq_at(0.0), None);
2164        // Long after: clamps to the newest write (the tip's seq).
2165        assert_eq!(db.seq_at(9_999_999_999.0), Some(v3.seq));
2166    }
2167
2168    #[test]
2169    fn ts_index_survives_reopen_via_cold_scan() {
2170        // Warm starts skip the scan and the index comes back empty — the
2171        // SAME session-scoped coverage seq_index has, gated by the same
2172        // flag. A cold start (fresh open of the same dir in a new Db) fills
2173        // it back. This is the reopen half of the contract.
2174        let dir = tempdir().unwrap();
2175        let ts_to_seq;
2176        {
2177            let db = Db::open(dir.path(), None).unwrap();
2178            let v1 = db.put("docs", "x", serde_json::json!({"v": 1}), vec![], None, None).unwrap();
2179            ts_to_seq = (v1.ts, v1.seq);
2180            db.flush_all();
2181        }
2182        {
2183            let db = Db::open(dir.path(), None).unwrap();
2184            // A reopen of a healthy store is a WARM boot — the scan is skipped
2185            // by design and the wall-clock index is empty with it. The
2186            // contract: not-ready gate + could-not-determine, never a guess.
2187            // (A deployment that needs wall-clock AS OF after warm boots runs
2188            // `nedb-cli repair`, exactly as it would for `since()`.)
2189            if !db.ts_index_ready() {
2190                assert_eq!(db.seq_at(ts_to_seq.0), None,
2191                    "an unindexed moment must answer could-not-determine, never guess");
2192            }
2193            // The forced cold path fills it: explicit repair semantics.
2194            let restored = db.rebuild_id_index().expect("rebuild runs");
2195            assert!(restored >= 1);
2196            let got = db.seq_at(ts_to_seq.0);
2197            assert_eq!(got, Some(ts_to_seq.1), "after the rebuild, the moment resolves");
2198            // And it resolves to the WRITE, not to one-after: the boundary.
2199            let node = db.get_as_of("docs", "x", got.unwrap()).unwrap();
2200            assert_eq!(node.data["v"], 1);
2201        }
2202    }
2203
2204    #[test]
2205    fn monotonic_put_keeps_the_index_sorted_without_resorting() {
2206        // The common path: every put appends a strictly-later ts. The index
2207        // must stay sorted by construction — a binary search over an
2208        // unsorted array answers randomly, which is worse than answering
2209        // nothing.
2210        let dir = tempdir().unwrap();
2211        let db = Db::open(dir.path(), None).unwrap();
2212        let mut last_ts = 0.0f64;
2213        for i in 0..50 {
2214            let n = db.put("docs", &format!("id-{}", i), serde_json::json!({"i": i}), vec![], None, None).unwrap();
2215            assert!(n.ts >= last_ts, "the sequencer stamps monotonically");
2216            last_ts = n.ts;
2217        }
2218        // Every write resolves to ITSELF — only true if the index is sorted
2219        // with seq tie-breaks (rapid puts share one clock tick; a binary
2220        // search over unsorted equal-ts runs answers arbitrarily).
2221        for seq in 1..=50u64 {
2222            let n = db
2223                .get_hash_by_seq(seq)
2224                .and_then(|h| db.objects.read(&h).ok())
2225                .unwrap_or_else(|| panic!("seq {} must resolve", seq));
2226            if seq <= 12 {
2227                if let Ok(idx) = db.ts_index.read() {
2228                    eprintln!("DBG seq {} ts {} idx[seq]={:?} idx_len {}",
2229                        seq, n.ts, idx.get(seq as usize), idx.len());
2230                }
2231            }
2232            assert_eq!(
2233                db.seq_at(n.ts),
2234                Some(n.seq),
2235                "write at seq {} (ts {}) must resolve to itself",
2236                n.seq,
2237                n.ts
2238            );
2239        }
2240    }
2241}
2242
2243#[cfg(test)]
2244mod tests_v2 {
2245    use super::*;
2246    use tempfile::tempdir;
2247
2248    // ── a DELETE is a tombstone, not an erasure ─────────────────────────────
2249    //
2250    // `delete()` used to just remove the live id pointer, which made the
2251    // document's whole history unreachable: `AS OF` enumerates ids from
2252    // `id_index`, so a deleted id was skipped at EVERY sequence — including
2253    // sequences long before the delete, where the row demonstrably existed.
2254    //
2255    // Nothing was lost on disk. The tombstone node keeps a `prev` link to the
2256    // full version chain and `verify()` counted every object as healthy — which
2257    // makes it the worst kind of data loss, the kind that passes its own audit.
2258    // The pointer is now MOVED to the graveyard instead of dropped.
2259
2260    #[test]
2261    fn a_deleted_documents_history_is_still_readable_before_the_delete() {
2262        let db = Db::in_memory();
2263        let v1 = db.put("o", "a", serde_json::json!({"t": 55}), vec![], None, None).unwrap();
2264        let v2 = db.put("o", "a", serde_json::json!({"t": 66}), vec![], None, None).unwrap();
2265        assert!(db.delete("o", "a").unwrap());
2266
2267        // Gone from the present — a delete must still delete.
2268        assert!(db.get("o", "a").is_none(), "a deleted doc must not be visible now");
2269
2270        // …and readable at each sequence it existed at.
2271        let at_v1 = db.get_as_of("o", "a", v1.seq).expect("the ORIGINAL value survives");
2272        assert_eq!(at_v1.data["t"], serde_json::json!(55));
2273        let at_v2 = db.get_as_of("o", "a", v2.seq).expect("the UPDATED value survives");
2274        assert_eq!(at_v2.data["t"], serde_json::json!(66));
2275    }
2276
2277    #[test]
2278    fn as_of_the_tombstone_or_later_reports_the_document_absent() {
2279        let db = Db::in_memory();
2280        db.put("o", "a", serde_json::json!({"t": 55}), vec![], None, None).unwrap();
2281        db.delete("o", "a").unwrap();
2282        let tomb_seq = db.tip().expect("the tombstone is the tip").seq;
2283
2284        assert!(db.get_as_of("o", "a", tomb_seq).is_none(),
2285                "at the delete's own sequence the document is gone");
2286        assert!(db.get_as_of("o", "a", tomb_seq + 10).is_none(), "and after it");
2287        // Never the tombstone node itself: `{_deleted, _prev}` is bookkeeping,
2288        // and surfacing it would look like a document with strange fields.
2289        for s in 0..=tomb_seq + 1 {
2290            if let Some(n) = db.get_as_of("o", "a", s) {
2291                assert!(n.data.get("_deleted").is_none(),
2292                        "seq {} surfaced the tombstone as a document: {:?}", s, n.data);
2293            }
2294        }
2295    }
2296
2297    #[test]
2298    fn an_as_of_query_lists_deleted_ids_alongside_live_ones() {
2299        let db = Db::in_memory();
2300        db.put("o", "keep", serde_json::json!({"n": 1}), vec![], None, None).unwrap();
2301        let gone = db.put("o", "gone", serde_json::json!({"n": 2}), vec![], None, None).unwrap();
2302        db.delete("o", "gone").unwrap();
2303
2304        assert_eq!(db.id_index.list_ids("o"), vec!["keep".to_string()],
2305                   "the live index holds only the living");
2306        assert_eq!(db.list_ids_including_deleted("o"),
2307                   vec!["gone".to_string(), "keep".to_string()],
2308                   "AS OF must consider both, in a stable order");
2309
2310        // The query path, end to end — this is what actually regressed.
2311        let (rows, _) = crate::nql::query(&db, &format!("FROM o AS OF {}", gone.seq)).unwrap();
2312        let ids: Vec<&str> = rows.iter().filter_map(|r| r["_id"].as_str()).collect();
2313        assert!(ids.contains(&"gone"), "AS OF must see the deleted row: {:?}", ids);
2314        assert!(ids.contains(&"keep"), "{:?}", ids);
2315
2316        // And the present must not.
2317        let (now, _) = crate::nql::query(&db, "FROM o").unwrap();
2318        let ids: Vec<&str> = now.iter().filter_map(|r| r["_id"].as_str()).collect();
2319        assert_eq!(ids, vec!["keep"], "a delete still deletes");
2320    }
2321
2322    #[test]
2323    fn a_recreated_id_keeps_the_history_from_before_its_delete() {
2324        // The edge case the graveyard fallback exists for: a `put` after a
2325        // delete starts a FRESH chain with no `prev`, so the live chain cannot
2326        // reach a sequence from before the delete. Only the graveyard can.
2327        let db = Db::in_memory();
2328        let old = db.put("o", "a", serde_json::json!({"era": "first"}), vec![], None, None).unwrap();
2329        db.delete("o", "a").unwrap();
2330        let new = db.put("o", "a", serde_json::json!({"era": "second"}), vec![], None, None).unwrap();
2331
2332        assert_eq!(db.get("o", "a").unwrap().data["era"], serde_json::json!("second"));
2333        assert_eq!(db.get_as_of("o", "a", new.seq).unwrap().data["era"],
2334                   serde_json::json!("second"));
2335        assert_eq!(db.get_as_of("o", "a", old.seq).expect("the FIRST era survives").data["era"],
2336                   serde_json::json!("first"),
2337                   "re-creating an id must not orphan what came before it");
2338    }
2339
2340    #[test]
2341    fn the_graveyard_survives_a_reopen() {
2342        // A tombstone pointer lost to a restart would put the history back out
2343        // of reach — the exact bug, just deferred. So it is flushed with the
2344        // live index and read back from disk.
2345        let dir = tempdir().unwrap();
2346        let seq = {
2347            let db = Db::open(dir.path(), None).unwrap();
2348            let v1 = db.put("o", "a", serde_json::json!({"t": 7}), vec![], None, None).unwrap();
2349            db.delete("o", "a").unwrap();
2350            db.try_flush_all().expect("flush must succeed");
2351            v1.seq
2352        };
2353        let db = Db::open(dir.path(), None).unwrap();
2354        assert!(db.get("o", "a").is_none(), "still deleted after a reopen");
2355        assert_eq!(db.get_as_of("o", "a", seq).expect("history survives a reopen").data["t"],
2356                   serde_json::json!(7));
2357        assert_eq!(db.list_ids_including_deleted("o"), vec!["a".to_string()]);
2358    }
2359
2360    #[test]
2361    fn the_graveyard_is_invisible_to_everything_that_enumerates_the_store() {
2362        // It adds a directory to the data dir, so the risk is that it shows up
2363        // as a phantom COLLECTION or a phantom OBJECT. Both enumerations are
2364        // rooted at their own subdirectory rather than at the data dir, which
2365        // is why it cannot — but that is exactly the kind of reasoning worth
2366        // pinning, because a stray "graveyard" collection would be nasty and
2367        // would only surface in someone's UI.
2368        let dir = tempdir().unwrap();
2369        let db = Db::open(dir.path(), None).unwrap();
2370        db.put("orders", "a", serde_json::json!({"t": 1}), vec![], None, None).unwrap();
2371        // A surviving sibling. This used to be load-bearing: with `a` alone,
2372        // `orders` had no index entries left and so no directory to enumerate,
2373        // and the test would have asserted the wrong thing for a reason that
2374        // had nothing to do with the graveyard. The collection registry fixed
2375        // that — an emptied collection stays in the namespace — so the sibling
2376        // is now just a second row.
2377        db.put("orders", "b", serde_json::json!({"t": 2}), vec![], None, None).unwrap();
2378        db.delete("orders", "a").unwrap();
2379        db.try_flush_all().unwrap();
2380
2381        let colls = db.collections();
2382        assert!(!colls.iter().any(|c| c == "graveyard"),
2383                "the graveyard must not look like a collection: {:?}", colls);
2384        assert_eq!(colls, vec!["orders".to_string()]);
2385
2386        let (_checked, tampered) = db.verify();
2387        assert!(tampered.is_empty(), "{:?}", tampered);
2388    }
2389
2390    #[test]
2391    fn a_delete_leaves_the_hash_chain_verifiable() {
2392        // The graveyard is an index, not a second source of truth: it must not
2393        // be able to make `verify()` disagree with the objects on disk.
2394        let db = Db::in_memory();
2395        db.put("o", "a", serde_json::json!({"t": 1}), vec![], None, None).unwrap();
2396        db.put("o", "b", serde_json::json!({"t": 2}), vec![], None, None).unwrap();
2397        db.delete("o", "a").unwrap();
2398        let (checked, tampered) = db.verify();
2399        assert!(tampered.is_empty(), "a delete must not break verify(): {:?}", tampered);
2400        assert!(checked >= 3, "the tombstone is an object too, got {}", checked);
2401    }
2402
2403    #[test]
2404    fn deleting_a_missing_id_stays_a_no_op() {
2405        let db = Db::in_memory();
2406        assert!(!db.delete("o", "nope").unwrap(), "nothing to delete");
2407        assert!(db.list_ids_including_deleted("o").is_empty(),
2408                "a failed delete must not put anything in the graveyard");
2409    }
2410
2411    #[test]
2412    fn seq_index_populated_on_put() {
2413        let db = Db::in_memory();
2414        let a = db.put("item", "a", serde_json::json!({"x": 1}), vec![], None, None).unwrap();
2415        let b = db.put("item", "b", serde_json::json!({"x": 2}), vec![], None, None).unwrap();
2416        assert_eq!(db.get_hash_by_seq(a.seq), Some(a.hash.clone()));
2417        assert_eq!(db.get_hash_by_seq(b.seq), Some(b.hash.clone()));
2418        assert_eq!(db.get_hash_by_seq(9999), None);
2419    }
2420
2421    #[test]
2422    fn tip_and_since() {
2423        let db = Db::in_memory();
2424        // Empty db: no tip, empty changefeed.
2425        assert!(db.tip().is_none());
2426        assert!(db.since(0, 0).nodes.is_empty());
2427
2428        let a = db.put("item", "a", serde_json::json!({"x": 1}), vec![], None, None).unwrap();
2429        let b = db.put("item", "b", serde_json::json!({"x": 2}), vec![], None, None).unwrap();
2430
2431        // tip() = the most recent write (highest seq), returned as a full node.
2432        let t = db.tip().expect("tip after writes");
2433        assert_eq!(t.seq, b.seq);
2434        assert_eq!(t.id, "b");
2435        assert_eq!(t.hash, b.hash);
2436
2437        // since(after_seq, limit) — EXCLUSIVE cursor, bounded page + envelope.
2438        let after_a = db.since(a.seq, 0);
2439        assert_eq!(after_a.nodes.len(), 1);
2440        assert_eq!(after_a.nodes[0].id, "b");
2441        assert_eq!(after_a.from_seq, a.seq);
2442        assert_eq!(after_a.to_seq, b.seq);
2443        assert_eq!(after_a.head_seq, b.seq);
2444        assert!(!after_a.has_more);
2445
2446        // Nothing written after the tip.
2447        assert!(db.since(b.seq, 0).nodes.is_empty());
2448
2449        // `limit` bounds the page and sets has_more; resume from to_seq.
2450        let c = db.put("item", "c", serde_json::json!({"x": 3}), vec![], None, None).unwrap();
2451        let page = db.since(a.seq, 1);             // (a..] capped at 1 -> [b], more pending
2452        assert_eq!(page.nodes.len(), 1);
2453        assert_eq!(page.nodes[0].id, "b");
2454        assert_eq!(page.to_seq, b.seq);
2455        assert!(page.has_more);
2456        let page2 = db.since(page.to_seq, 1);      // resume from b -> [c], done
2457        assert_eq!(page2.nodes.len(), 1);
2458        assert_eq!(page2.nodes[0].id, "c");
2459        assert_eq!(page2.to_seq, c.seq);
2460        assert!(!page2.has_more);
2461    }
2462
2463    #[test]
2464    fn tip_collection_per_chain() {
2465        // The ITC sync-client case: separate chains in separate collections; a
2466        // consumer resumes ONE without pulling global tip and filtering.
2467        let db = Db::in_memory();
2468        assert!(db.tip_collection("blocks").is_none());
2469
2470        db.put("blocks", "b0", serde_json::json!({"h": 0}), vec![], None, None).unwrap();
2471        db.put("tx",     "t0", serde_json::json!({"v": 1}), vec![], None, None).unwrap();
2472        let b1 = db.put("blocks", "b1", serde_json::json!({"h": 1}), vec![], None, None).unwrap();
2473        let t1 = db.put("tx",     "t1", serde_json::json!({"v": 2}), vec![], None, None).unwrap();
2474
2475        // global tip = latest write overall (t1)
2476        assert_eq!(db.tip().unwrap().id, "t1");
2477        // collection-local tips = latest write in each collection
2478        let bt = db.tip_collection("blocks").expect("blocks tip");
2479        assert_eq!(bt.id, "b1");
2480        assert_eq!(bt.seq, b1.seq);
2481        assert_eq!(db.tip_collection("tx").unwrap().seq, t1.seq);
2482        assert!(db.tip_collection("absent").is_none());
2483    }
2484
2485    #[test]
2486    fn seq_index_survives_batch() {
2487        let db = Db::in_memory();
2488        let nodes = db.put_batch(vec![
2489            ("item".into(), "x".into(), serde_json::json!({"v": 1}), vec![], None, None),
2490            ("item".into(), "y".into(), serde_json::json!({"v": 2}), vec![], None, None),
2491        ]).unwrap();
2492        for node in &nodes {
2493            assert_eq!(db.get_hash_by_seq(node.seq), Some(node.hash.clone()));
2494        }
2495    }
2496
2497    /// Regression: put_batch must remove the superseded version's sorted-index
2498    /// entries, exactly like put() does. Old behavior left the old hashes in
2499    /// the BTree — ORDER BY returned superseded rows alongside current ones
2500    /// (they resolve fine through the content-addressed store, which made the
2501    /// stale rows look legitimate).
2502    #[test]
2503    fn put_batch_removes_superseded_sorted_index_entries() {
2504        let db = Db::in_memory();
2505        db.create_sorted_index("blocks", "height");
2506        db.put("blocks", "x", serde_json::json!({"height": 1}), vec![], None, None).unwrap();
2507        db.put_batch(vec![
2508            ("blocks".into(), "x".into(), serde_json::json!({"height": 99}), vec![], None, None),
2509        ]).unwrap();
2510
2511        let asc = db.order_by_asc("blocks", "height", 10);
2512        assert_eq!(asc.len(), 1, "stale index entry for the superseded version must be gone");
2513        assert_eq!(asc[0].data["height"], 99);
2514        assert_eq!(asc[0].id, "x");
2515    }
2516
2517    /// Updates without any sorted index must keep full version-chain semantics
2518    /// (guards the new skip-old-object-read fast path in put()).
2519    #[test]
2520    fn update_without_indexes_preserves_chain() {
2521        let db = Db::in_memory();
2522        let v1 = db.put("docs", "x", serde_json::json!({"v": 1}), vec![], None, None).unwrap();
2523        let v2 = db.put("docs", "x", serde_json::json!({"v": 2}), vec![], None, None).unwrap();
2524        assert_eq!(v2.prev.as_deref(), Some(v1.hash.as_str()), "prev chain must survive the fast path");
2525        assert_eq!(db.get("docs", "x").unwrap().data["v"], 2);
2526        assert_eq!(db.get_as_of("docs", "x", v1.seq).unwrap().data["v"], 1);
2527    }
2528
2529    #[test]
2530    fn link_and_neighbors() {
2531        let db = Db::in_memory();
2532        db.put("driver", "d1", serde_json::json!({"name": "Bob"}),   vec![], None, None).unwrap();
2533        db.put("driver", "d2", serde_json::json!({"name": "Carol"}), vec![], None, None).unwrap();
2534        db.put("trip",   "t1", serde_json::json!({"status": "req"}), vec![], None, None).unwrap();
2535        db.put("trip",   "t2", serde_json::json!({"status": "req"}), vec![], None, None).unwrap();
2536
2537        db.link("driver:d1", "handles", "trip:t1").unwrap();
2538        db.link("driver:d1", "handles", "trip:t2").unwrap();
2539        db.link("driver:d2", "handles", "trip:t1").unwrap();
2540
2541        let d1_trips = db.neighbors("driver:d1", "handles");
2542        assert_eq!(d1_trips.len(), 2);
2543        let ids: std::collections::HashSet<&str> = d1_trips.iter().map(|n| n.id.as_str()).collect();
2544        assert!(ids.contains("t1") && ids.contains("t2"));
2545
2546        let d2_trips = db.neighbors("driver:d2", "handles");
2547        assert_eq!(d2_trips.len(), 1);
2548        assert_eq!(d2_trips[0].id, "t1");
2549    }
2550
2551    #[test]
2552    fn link_stored_in_links_collection() {
2553        // Links are stored as __links__ documents, not as graph edges.
2554        // The __links__ collection is NQL-queryable and consistent with the PyO3 binding.
2555        let db = Db::in_memory();
2556        db.put("driver", "d1", serde_json::json!({"name": "Bob"}),   vec![], None, None).unwrap();
2557        db.put("trip",   "t1", serde_json::json!({"status": "req"}), vec![], None, None).unwrap();
2558        db.link("driver:d1", "handles", "trip:t1").unwrap();
2559        // Verify the __links__ document was created
2560        let link_doc = db.get("__links__", "driver:d1|handles|trip:t1");
2561        assert!(link_doc.is_some(), "__links__ doc should exist");
2562        let doc = link_doc.unwrap();
2563        assert_eq!(doc.data["_from"], "driver:d1");
2564        assert_eq!(doc.data["_rel"],  "handles");
2565        assert_eq!(doc.data["_to"],   "trip:t1");
2566        // neighbors() resolves to the target node
2567        let nb = db.neighbors("driver:d1", "handles");
2568        assert_eq!(nb.len(), 1);
2569        assert_eq!(nb[0].id, "t1");
2570    }
2571
2572    /// A lost id-index WAL must be recoverable: the objects carry coll/id/seq,
2573    /// so `repair()` can reconstruct every row, and the repaired database must
2574    /// reopen WARM with a valid head.
2575    ///
2576    /// Regression for 2.8.5, where the cold scan rebuilt seq_index, coll_tips,
2577    /// head and MANIFEST but never the id index — so a database in this state
2578    /// returned 0 rows from `list()` while `verify()` reported every object
2579    /// healthy, and `nedb-cli repair` printed success without fixing anything.
2580    #[test]
2581    fn repair_rebuilds_id_index_after_lost_wal() {
2582        let dir = tempdir().unwrap();
2583        {
2584            let db = Db::open(dir.path(), None).unwrap();
2585            for i in 0..25 {
2586                db.put("rows", &format!("r{}", i), serde_json::json!({"i": i}), vec![], None, None)
2587                    .unwrap();
2588            }
2589            db.put("rows", "r0", serde_json::json!({"i": 0, "v": 2}), vec![], None, None).unwrap();
2590            db.try_flush_all().unwrap();
2591        }
2592
2593        // Simulate the lost WAL: objects survive, the id index does not.
2594        std::fs::remove_dir_all(dir.path().join("indexes")).unwrap();
2595
2596        {
2597            let db = Db::open(dir.path(), None).unwrap();
2598            assert_eq!(db.list("rows").len(), 0, "precondition: rows unreachable");
2599            let (ok, bad) = db.verify();
2600            assert!(ok > 0 && bad.is_empty(), "objects must still be intact and verifying");
2601
2602            let written = db.repair().unwrap();
2603            // 25 rows plus the one `_nedb.collections` record that registered
2604            // the collection. The registry is written through the ordinary
2605            // object path precisely so that repair, verify and replication
2606            // cover it without knowing it is special.
2607            assert_eq!(written, 26, "one entry per distinct (coll, id)");
2608            assert_eq!(db.list("rows").len(), 25, "every row must come back");
2609
2610            // The winner for a re-put id is the HIGHEST seq, matching put().
2611            let r0 = db.get("rows", "r0").expect("r0 present");
2612            assert_eq!(r0.data.get("v").and_then(|v| v.as_i64()), Some(2),
2613                "repair must restore the latest version, not an older one");
2614        }
2615
2616        // A repaired database must reopen warm with a real head.
2617        let db3 = Db::open(dir.path(), None).unwrap();
2618        assert_eq!(db3.list("rows").len(), 25);
2619        assert!(!db3.head().is_empty(), "repair must leave a valid MANIFEST head");
2620        assert!(db3.tip_collection("rows").is_some(), "tip_collection must resolve after repair");
2621    }
2622
2623    /// `since()` must never report "caught up" while the cursor is behind head.
2624    ///
2625    /// Regression for 2.8.5: on a warm boot the seq index is empty by design
2626    /// (the warm path skips the scan), so every seq lookup missed and `since()`
2627    /// returned zero nodes with `has_more = false` — identical to genuinely up
2628    /// to date. A consumer following the documented drain loop stopped one call
2629    /// in, on a database with every record unread.
2630    #[test]
2631    fn since_never_reports_caught_up_while_behind_head() {
2632        let dir = tempdir().unwrap();
2633        {
2634            let db = Db::open(dir.path(), None).unwrap();
2635            for i in 0..10 {
2636                db.put("rows", &format!("r{}", i), serde_json::json!({"i": i}), vec![], None, None)
2637                    .unwrap();
2638            }
2639            db.try_flush_all().unwrap();
2640        }
2641
2642        // Warm reopen: startup is "complete" in O(1) because the scan is skipped.
2643        let db2 = Db::open(dir.path(), None).unwrap();
2644        let st = db2.scan_status();
2645        assert!(st.tip_seq > 0, "log has entries");
2646        assert!(
2647            !st.seq_index_ready,
2648            "warm boot leaves the seq index cold — that is the honest signal"
2649        );
2650
2651        let batch = db2.since(0, 100);
2652        assert!(
2653            batch.to_seq < batch.head_seq,
2654            "cursor is behind the log head in this state"
2655        );
2656        assert!(
2657            batch.has_more,
2658            "has_more must be true while the cursor is behind head — otherwise the \
2659             consumer reads 'caught up' and stops with every record unread"
2660        );
2661
2662        // After a repair the index resolves and the drain actually completes.
2663        db2.repair().unwrap();
2664        assert!(db2.scan_status().seq_index_ready);
2665        let drained = db2.since(0, 100);
2666        assert!(!drained.has_more, "genuinely caught up reports has_more=false");
2667
2668        // KNOWN SHARP EDGE, pinned here deliberately: the cursor is EXCLUSIVE
2669        // and seqs start at 0, so `since(0, _)` returns (0, head] and whatever
2670        // holds seq 0 is not reachable through any cursor value. Changing the
2671        // cursor convention would break existing replication consumers, so this
2672        // is documented rather than silently altered.
2673        //
2674        // The collection registry softened it by accident and in the right
2675        // direction: seq 0 is now the `_nedb.collections` record rather than a
2676        // user's first row, so all 10 writes drain. A replica seeded from
2677        // since() alone is still one record short — but the record it misses is
2678        // one it can re-derive, instead of somebody's data.
2679        assert_eq!(
2680            drained.nodes.len(),
2681            10,
2682            "since(0) is exclusive of seq 0 — see the sharp edge noted above"
2683        );
2684        assert!(
2685            drained.nodes.iter().all(|n| n.seq >= 1),
2686            "seq 0 is unreachable via since()"
2687        );
2688    }
2689
2690    #[test]
2691    fn link_missing_node_errors() {
2692        let db = Db::in_memory();
2693        db.put("driver", "d1", serde_json::json!({}), vec![], None, None).unwrap();
2694        assert!(db.link("driver:d1", "handles", "trip:ghost").is_err());
2695    }
2696
2697    #[test]
2698    fn link_durable_survives_reopen() {
2699        let dir = tempdir().unwrap();
2700        {
2701            let db = Db::open(dir.path(), None).unwrap();
2702            db.put("driver", "d1", serde_json::json!({"name": "Bob"}),   vec![], None, None).unwrap();
2703            db.put("trip",   "t1", serde_json::json!({"status": "req"}), vec![], None, None).unwrap();
2704            db.link("driver:d1", "handles", "trip:t1").unwrap();
2705        }
2706        let db2 = Db::open(dir.path(), None).unwrap();
2707        db2.startup_ready.store(true, std::sync::atomic::Ordering::SeqCst);
2708        let trips = db2.neighbors("driver:d1", "handles");
2709        assert_eq!(trips.len(), 1);
2710        assert_eq!(trips[0].id, "t1");
2711    }
2712
2713    #[test]
2714    fn tip_survives_warm_restart() {
2715        // v2.5.43: tip() returns the last written object AND survives a warm restart.
2716        // On reopen the seq_index is cold (warm start skips the scan), so tip() must
2717        // resolve the last write via the MANIFEST tip_hash fallback — no scan.
2718        let dir = tempdir().unwrap();
2719        {
2720            let db = Db::open(dir.path(), None).unwrap();
2721            db.put("blocks", "b1", serde_json::json!({"h": 1}), vec![], None, None).unwrap();
2722            db.put("blocks", "b2", serde_json::json!({"h": 2}), vec![], None, None).unwrap();
2723            db.flush_all(); // persists MANIFEST incl. tip_hash
2724            assert_eq!(db.tip().expect("tip in-session").id, "b2");
2725        }
2726        // Warm reopen: MANIFEST present -> no cold scan -> seq_index cold.
2727        let db2 = Db::open(dir.path(), None).unwrap();
2728        assert!(db2.get_hash_by_seq(1).is_none(), "seq_index is cold on a warm boot");
2729        let tip = db2.tip().expect("tip() must survive a warm restart");
2730        assert_eq!(tip.id, "b2");
2731        assert_eq!(tip.data.get("h").and_then(|v| v.as_i64()), Some(2));
2732    }
2733
2734    #[test]
2735    fn tip_collection_survives_warm_restart() {
2736        // Same contract as tip(), per collection: itc-node-rs resumes headers /
2737        // blocks / l2_receipts independently, so each must be its own durable
2738        // resume point — not just the global tip.
2739        let dir = tempdir().unwrap();
2740        {
2741            let db = Db::open(dir.path(), None).unwrap();
2742            db.put("blocks", "b1", serde_json::json!({"h": 1}), vec![], None, None).unwrap();
2743            db.put("tx",     "t1", serde_json::json!({"v": 1}), vec![], None, None).unwrap();
2744            let b2 = db.put("blocks", "b2", serde_json::json!({"h": 2}), vec![], None, None).unwrap();
2745            db.flush_all(); // persists MANIFEST incl. coll_tips
2746            assert_eq!(db.tip_collection("blocks").unwrap().id, "b2");
2747            assert_eq!(db.tip_collection("blocks").unwrap().seq, b2.seq);
2748        }
2749        // Warm reopen: MANIFEST present -> no cold scan -> seq_index cold.
2750        let db2 = Db::open(dir.path(), None).unwrap();
2751        assert!(db2.get_hash_by_seq(0).is_none(), "seq_index is cold on a warm boot");
2752        let blocks_tip = db2.tip_collection("blocks").expect("tip_collection must survive a warm restart");
2753        assert_eq!(blocks_tip.id, "b2");
2754        assert_eq!(blocks_tip.data.get("h").and_then(|v| v.as_i64()), Some(2));
2755        let tx_tip = db2.tip_collection("tx").expect("tx tip must also survive");
2756        assert_eq!(tx_tip.id, "t1");
2757        assert!(db2.tip_collection("absent").is_none());
2758    }
2759
2760    #[test]
2761    fn cold_scan_indexes_every_object_and_reports_completion() {
2762        // Regression guard for the cold-scan refactor: seq_index is now populated
2763        // DURING the parallel read pass (for live scan_status().indexed_count
2764        // progress — see cold_scan_background_arc), not in a second pass
2765        // afterward. This asserts the end state is unchanged: every written
2766        // object is indexed, tip()/tip_collection() are correct, and
2767        // scan_complete eventually reports true.
2768        let dir = tempdir().unwrap();
2769        let n = 25u64;
2770        {
2771            let db = Db::open(dir.path(), None).unwrap();
2772            for i in 0..n {
2773                db.put("things", &i.to_string(), serde_json::json!({"i": i}), vec![], None, None).unwrap();
2774            }
2775            db.flush_all();
2776        }
2777        // Force a COLD start regardless of the MANIFEST nedb-v2 itself would
2778        // have written: delete it so startup_rebuild() takes the cold path and
2779        // start_cold_scan() actually spawns the background scan this test needs
2780        // to exercise.
2781        std::fs::remove_file(dir.path().join("MANIFEST")).unwrap();
2782
2783        let db = Db::open(dir.path(), None).unwrap();
2784        assert!(!db.scan_status().scan_complete, "should be cold immediately after open");
2785        let db = std::sync::Arc::new(db);
2786        Db::start_cold_scan(std::sync::Arc::clone(&db));
2787
2788        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
2789        while !db.scan_status().scan_complete {
2790            assert!(std::time::Instant::now() < deadline, "cold scan did not complete in time");
2791            std::thread::sleep(std::time::Duration::from_millis(5));
2792        }
2793
2794        let status = db.scan_status();
2795        // n rows + the collection registry record for "things".
2796        assert_eq!(status.indexed_count, n as usize + 1, "every written object must be indexed");
2797        assert!(status.scan_complete);
2798
2799        let tip = db.tip().expect("tip resolves after cold scan");
2800        assert_eq!(tip.data.get("i").and_then(|v| v.as_u64()), Some(n - 1));
2801        let coll_tip = db.tip_collection("things").expect("tip_collection resolves after cold scan");
2802        assert_eq!(coll_tip.id, tip.id);
2803    }
2804
2805    /// Concurrent writers must settle the tip at the HIGHEST SEQ, and that tip
2806    /// must survive a warm restart. Before the seq-guarded tip fix, update_head
2807    /// was "last call wins": a slower thread carrying an OLDER seq could
2808    /// overwrite tip_hash after a newer write, and MANIFEST then persisted the
2809    /// stale tip for the next warm boot (flaky by nature — this pins the
2810    /// contract deterministically for the fixed code).
2811    #[test]
2812    fn concurrent_puts_tip_resolves_to_highest_seq_after_warm_restart() {
2813        let dir = tempdir().unwrap();
2814        let total: u64 = 100;
2815        {
2816            let db = std::sync::Arc::new(Db::open(dir.path(), None).unwrap());
2817            let mut handles = vec![];
2818            for t in 0..4u64 {
2819                let db2 = std::sync::Arc::clone(&db);
2820                handles.push(std::thread::spawn(move || {
2821                    for i in 0..25u64 {
2822                        db2.put("c", &format!("{}-{}", t, i),
2823                                serde_json::json!({"t": t, "i": i}),
2824                                vec![], None, None).unwrap();
2825                    }
2826                }));
2827            }
2828            for h in handles { h.join().unwrap(); }
2829            // In-session: tip must be the highest assigned seq.
2830            let expected = db.seq.load(std::sync::atomic::Ordering::SeqCst) - 1;
2831            // `total` user writes plus one registry record for collection "c",
2832            // so the highest assigned seq is `total`, not `total - 1`.
2833            assert_eq!(expected, total, "exactly {} writes expected", total);
2834            assert_eq!(db.tip().expect("in-session tip").seq, expected);
2835            db.flush_all(); // persist MANIFEST incl. tip_hash
2836        }
2837        // Warm reopen: seq_index cold; tip() resolves via MANIFEST tip_hash.
2838        let db2 = Db::open(dir.path(), None).unwrap();
2839        let tip = db2.tip().expect("tip must survive warm restart after concurrent writes");
2840        assert_eq!(tip.seq, total, "warm-boot tip must be the highest-seq write");
2841        // Per-collection tip: same contract.
2842        let ct = db2.tip_collection("c").expect("coll tip survives");
2843        assert_eq!(ct.seq, total);
2844    }
2845
2846    /// Pre-2.5.43 MANIFESTs (no tip_hash) must warm-boot, NOT force a cold
2847    /// scan. The old "cold scan once to upgrade" policy was hours of random
2848    /// reads on multi-million-object seek-bound stores (itcd -dagv3), re-paid
2849    /// on every boot if the process exited before the scan finished. seq+head
2850    /// in the old MANIFEST are valid; tip()/tip_collection() return None until
2851    /// the first write+flush organically rewrites MANIFEST with a tip.
2852    #[test]
2853    fn pre_durable_tip_manifest_warm_boots_and_heals_lazily() {
2854        let dir = tempdir().unwrap();
2855        {
2856            let db = Db::open(dir.path(), None).unwrap();
2857            for i in 0..5u64 {
2858                db.put("things", &i.to_string(), serde_json::json!({"i": i}), vec![], None, None).unwrap();
2859            }
2860            db.flush_all();
2861        }
2862        // Rewrite MANIFEST in the pre-2.5.43 shape: seq + head only.
2863        let manifest_path = dir.path().join("MANIFEST");
2864        let m: serde_json::Value =
2865            serde_json::from_str(&std::fs::read_to_string(&manifest_path).unwrap()).unwrap();
2866        let old_format = serde_json::json!({ "seq": m["seq"], "head": m["head"] });
2867        std::fs::write(&manifest_path, serde_json::to_string(&old_format).unwrap()).unwrap();
2868
2869        // Reopen: must be WARM (startup_ready immediately — no cold scan gate).
2870        let db2 = Db::open(dir.path(), None).unwrap();
2871        assert!(db2.startup_ready.load(std::sync::atomic::Ordering::SeqCst),
2872                "pre-2.5.43 MANIFEST must warm-boot, not fall to a cold scan");
2873        // tip() unresolvable this boot — documented None, not a panic or scan.
2874        assert!(db2.tip().is_none(), "tip() is None until the manifest heals");
2875        // seq continuity: a new write gets a FRESH seq (no reuse).
2876        let n = db2.put("things", "next", serde_json::json!({"fresh": true}), vec![], None, None).unwrap();
2877        assert_eq!(n.seq, m["seq"].as_u64().unwrap(), "next write takes the persisted next-to-assign seq");
2878        db2.flush_all(); // organic upgrade: MANIFEST now carries tip_hash
2879        drop(db2);
2880
2881        // Healed: next boot is warm AND tip() resolves.
2882        let db3 = Db::open(dir.path(), None).unwrap();
2883        assert!(db3.startup_ready.load(std::sync::atomic::Ordering::SeqCst));
2884        let tip = db3.tip().expect("tip() must resolve after the organic upgrade");
2885        assert_eq!(tip.id, "next");
2886    }
2887
2888    /// Regression for the cold-scan MANIFEST seq off-by-one. The scan's old
2889    /// hand-rolled MANIFEST stored `seq: max_seq` (the last USED seq), but the
2890    /// warm boot loads `m.seq` as the NEXT-TO-ASSIGN counter — so a restart
2891    /// right after a quiet cold scan handed the next write the tip's seq:
2892    /// a DUPLICATE seq in the log (seq_index overwrite, wrong since() page).
2893    /// The scan now writes MANIFEST via flush_manifest(), which reads the live
2894    /// counter (max_seq + 1).
2895    #[test]
2896    fn manifest_after_cold_scan_does_not_reuse_tip_seq() {
2897        let dir = tempdir().unwrap();
2898        let old_tip_seq;
2899        {
2900            let db = Db::open(dir.path(), None).unwrap();
2901            for i in 0..5u64 {
2902                db.put("things", &i.to_string(), serde_json::json!({"i": i}), vec![], None, None).unwrap();
2903            }
2904            db.flush_all();
2905            old_tip_seq = db.tip().unwrap().seq;
2906        }
2907        // Force a cold start: remove MANIFEST so the background scan runs and
2908        // writes a fresh MANIFEST itself.
2909        std::fs::remove_file(dir.path().join("MANIFEST")).unwrap();
2910        {
2911            let db = std::sync::Arc::new(Db::open(dir.path(), None).unwrap());
2912            Db::start_cold_scan(std::sync::Arc::clone(&db));
2913            let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
2914            while !db.scan_status().scan_complete {
2915                assert!(std::time::Instant::now() < deadline, "cold scan did not complete");
2916                std::thread::sleep(std::time::Duration::from_millis(5));
2917            }
2918            // No further writes — the scan's own MANIFEST is what the next boot sees.
2919        }
2920        // Warm reopen from the scan-written MANIFEST: the next write must get a
2921        // FRESH seq, never the tip's.
2922        let db3 = Db::open(dir.path(), None).unwrap();
2923        let tip_before = db3.tip().expect("tip survives scan-written MANIFEST");
2924        assert_eq!(tip_before.seq, old_tip_seq, "tip identity preserved across the scan");
2925        let new_node = db3.put("things", "next", serde_json::json!({"fresh": true}),
2926                               vec![], None, None).unwrap();
2927        assert!(new_node.seq > old_tip_seq,
2928                "new write reused seq {} (tip was {}) — duplicate seq in the log",
2929                new_node.seq, old_tip_seq);
2930    }
2931
2932    /// Regression: the flush ticker must NOT pin the database.
2933    ///
2934    /// Before this was fixed, `start_manifest_ticker` held a strong `Arc<Db>`
2935    /// in an unconditional `loop`, so the thread never exited, the `Db` was
2936    /// never dropped, and the exclusive data-dir `LOCK` from `Db::open` was
2937    /// never released. Reopening the same path in the SAME PROCESS then failed
2938    /// with "locked by another process (pid N)" — where N was the caller's own
2939    /// pid. Live in every release from 2.8.5 through 3.1.0, and invisible
2940    /// because no CI ran the suite (tests/test_native.py) that hit it.
2941    ///
2942    /// Put the strong `Arc` back in the ticker and this test fails.
2943    #[test]
2944    fn ticker_does_not_pin_the_db_across_a_reopen() {
2945        let dir = tempdir().unwrap();
2946        {
2947            let db = std::sync::Arc::new(Db::open(dir.path(), None).unwrap());
2948            Db::start_manifest_ticker(std::sync::Arc::clone(&db), 25);
2949            db.put("t", "a", serde_json::json!({"v": 1}), vec![], None, None).unwrap();
2950            // Let the ticker run at least a couple of times while the db lives.
2951            std::thread::sleep(std::time::Duration::from_millis(90));
2952        } // last owner dropped here -> Drop flushes -> LOCK released
2953
2954        // The ticker upgrades its Weak for the duration of a tick, so at any
2955        // given instant it may legitimately hold a transient strong reference.
2956        // Release is therefore "eventual, within about one interval", not
2957        // instantaneous -- poll for it.
2958        //
2959        // The first version of this test sampled Arc::strong_count once and
2960        // asserted it was 1. That passed on an idle machine and failed the
2961        // first time it met a loaded CI runner, because the sample landed
2962        // mid-tick. A leak still fails this test deterministically: if the
2963        // ticker holds a strong Arc forever the LOCK is never released and
2964        // the deadline expires.
2965        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
2966        let db2 = loop {
2967            match Db::open(dir.path(), None) {
2968                Ok(db) => break db,
2969                Err(e) => {
2970                    assert!(std::time::Instant::now() < deadline,
2971                            "reopen never succeeded -- the ticker is pinning the Db: {e}");
2972                    std::thread::sleep(std::time::Duration::from_millis(25));
2973                }
2974            }
2975        };
2976        assert!(db2.get("t", "a").is_some(), "the write survived close/reopen");
2977    }
2978
2979    /// The ticker thread must actually terminate, not merely stop pinning.
2980    #[test]
2981    fn ticker_thread_exits_when_the_last_owner_drops() {
2982        let dir = tempdir().unwrap();
2983        let weak = {
2984            let db = std::sync::Arc::new(Db::open(dir.path(), None).unwrap());
2985            Db::start_manifest_ticker(std::sync::Arc::clone(&db), 25);
2986            db.put("t", "a", serde_json::json!({"v": 1}), vec![], None, None).unwrap();
2987            std::thread::sleep(std::time::Duration::from_millis(60));
2988            std::sync::Arc::downgrade(&db)
2989        };
2990        // Same reasoning as above: a tick in flight holds a real strong
2991        // reference for a few microseconds, so this is an eventual property.
2992        // A genuine leak never releases and blows the deadline.
2993        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
2994        while weak.upgrade().is_some() {
2995            assert!(std::time::Instant::now() < deadline,
2996                    "the Db outlived its last owner — the ticker is leaking it");
2997            std::thread::sleep(std::time::Duration::from_millis(25));
2998        }
2999    }
3000}
3001
3002/// Collection identity: does the database know which collections exist,
3003/// independently of how and when it happened to store them?
3004///
3005/// The three tests that used to fail are the first three here. They failed
3006/// like this, on the running engine:
3007///
3008/// ```text
3009/// disk, flush between   : ["orders"]
3010/// disk, one tick        : []
3011/// memory                : []
3012/// ```
3013#[cfg(test)]
3014mod collection_identity {
3015    use super::*;
3016    use tempfile::tempdir;
3017
3018    fn j(v: u64) -> serde_json::Value { serde_json::json!({"v": v}) }
3019
3020    /// Create a collection, then empty it — flushing BETWEEN the two.
3021    fn disk_emptied_with_flush_between() -> Vec<String> {
3022        let dir = tempdir().unwrap();
3023        let db = Db::open(dir.path(), None).unwrap();
3024        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3025        db.flush_all();
3026        db.delete("orders", "1").unwrap();
3027        db.flush_all();
3028        db.collections()
3029    }
3030
3031    /// The same logical history, with no flush in between. Before the registry
3032    /// this returned `[]`, because the WAL buffer is keyed by `(coll, id)` and
3033    /// the tombstone overwrote the PUT before any directory was created.
3034    fn disk_emptied_within_one_tick() -> Vec<String> {
3035        let dir = tempdir().unwrap();
3036        let db = Db::open(dir.path(), None).unwrap();
3037        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3038        db.delete("orders", "1").unwrap();
3039        db.flush_all();
3040        db.collections()
3041    }
3042
3043    fn memory_emptied() -> Vec<String> {
3044        let db = Db::in_memory();
3045        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3046        db.delete("orders", "1").unwrap();
3047        db.collections()
3048    }
3049
3050    /// A background timer is not a fact about the data.
3051    #[test]
3052    fn the_same_history_yields_the_same_namespace_regardless_of_flush_timing() {
3053        assert_eq!(
3054            disk_emptied_with_flush_between(),
3055            disk_emptied_within_one_tick(),
3056            "a 1-second flush ticker decided the namespace"
3057        );
3058    }
3059
3060    /// A root computed on a disk replica and on a memory replica of the same
3061    /// database has to be the same root.
3062    #[test]
3063    fn the_namespace_does_not_depend_on_the_storage_backend() {
3064        assert_eq!(
3065            disk_emptied_with_flush_between(),
3066            memory_emptied(),
3067            "disk and memory disagree about which collections exist"
3068        );
3069    }
3070
3071    /// The property the Oracle named: an empty-but-durable collection must not
3072    /// be indistinguishable from one that never existed.
3073    #[test]
3074    fn an_emptied_collection_is_not_the_same_as_one_that_never_existed() {
3075        assert_eq!(memory_emptied(), vec!["orders".to_string()]);
3076
3077        let never = Db::in_memory();
3078        assert!(never.collections().is_empty());
3079    }
3080
3081    #[test]
3082    fn a_dropped_collection_is_gone_but_a_merely_empty_one_is_not() {
3083        let db = Db::in_memory();
3084        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3085        db.delete("orders", "1").unwrap();
3086        assert_eq!(db.collections(), vec!["orders".to_string()], "emptying is not dropping");
3087
3088        assert!(db.drop_collection("orders").unwrap());
3089        assert!(db.collections().is_empty());
3090
3091        // Dropping twice is not an error, it is just not a second event.
3092        assert!(!db.drop_collection("orders").unwrap());
3093    }
3094
3095    #[test]
3096    fn writing_to_a_dropped_collection_revives_it() {
3097        let db = Db::in_memory();
3098        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3099        db.drop_collection("orders").unwrap();
3100        assert!(db.collections().is_empty());
3101
3102        db.put("orders", "2", j(2), vec![], None, None).unwrap();
3103        assert_eq!(db.collections(), vec!["orders".to_string()]);
3104    }
3105
3106    /// The namespace is versioned, because the registry is ordinary documents.
3107    #[test]
3108    fn the_namespace_can_be_read_as_of_a_sequence() {
3109        let db = Db::in_memory();
3110        let a = db.put("alpha", "1", j(1), vec![], None, None).unwrap();
3111        let b = db.put("beta", "1", j(1), vec![], None, None).unwrap();
3112
3113        assert_eq!(db.collections_as_of(a.seq), vec!["alpha".to_string()]);
3114        assert_eq!(
3115            db.collections_as_of(b.seq),
3116            vec!["alpha".to_string(), "beta".to_string()]
3117        );
3118    }
3119
3120    #[test]
3121    fn a_drop_is_visible_as_a_drop_in_history_not_as_an_absence() {
3122        let db = Db::in_memory();
3123        let a = db.put("orders", "1", j(1), vec![], None, None).unwrap();
3124        db.drop_collection("orders").unwrap();
3125
3126        assert!(db.collections().is_empty(), "not live now");
3127        assert_eq!(
3128            db.collections_as_of(a.seq), vec!["orders".to_string()],
3129            "but it existed then, and history says so"
3130        );
3131    }
3132
3133    #[test]
3134    fn the_registry_does_not_list_itself() {
3135        let db = Db::in_memory();
3136        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3137        assert_eq!(db.collections(), vec!["orders".to_string()]);
3138        assert!(
3139            !db.collections().iter().any(|c| crate::namespace::is_reserved(c)),
3140            "an engine-owned collection is not part of the user's namespace"
3141        );
3142    }
3143
3144    #[test]
3145    fn a_client_cannot_write_to_the_registry() {
3146        let db = Db::in_memory();
3147        assert!(db.put(crate::namespace::COLLECTIONS, "forged", j(1), vec![], None, None).is_err());
3148        assert!(db.put("_nedb.anything", "x", j(1), vec![], None, None).is_err());
3149        assert!(db.delete(crate::namespace::COLLECTIONS, "orders").is_err());
3150        assert!(db.drop_collection(crate::namespace::COLLECTIONS).is_err());
3151    }
3152
3153    #[test]
3154    fn a_collection_name_cannot_escape_the_data_directory() {
3155        let db = Db::in_memory();
3156        for escape in ["../etc", "a/b", "..", ""] {
3157            assert!(
3158                db.put(escape, "x", j(1), vec![], None, None).is_err(),
3159                "{:?} must not be usable as a collection name", escape
3160            );
3161        }
3162    }
3163
3164    #[test]
3165    fn registration_survives_a_reopen_without_re_registering() {
3166        let dir = tempdir().unwrap();
3167        let seq_after_first_open;
3168        {
3169            let db = Db::open(dir.path(), None).unwrap();
3170            db.put("orders", "1", j(1), vec![], None, None).unwrap();
3171            db.put("orders", "2", j(2), vec![], None, None).unwrap();
3172            db.flush_all();
3173            seq_after_first_open = db.seq.load(Ordering::SeqCst);
3174        }
3175        let db = Db::open(dir.path(), None).unwrap();
3176        assert_eq!(db.collections(), vec!["orders".to_string()]);
3177        db.put("orders", "3", j(3), vec![], None, None).unwrap();
3178        assert_eq!(
3179            db.seq.load(Ordering::SeqCst), seq_after_first_open + 1,
3180            "reopening and writing again must not append a second registry record"
3181        );
3182    }
3183
3184    #[test]
3185    fn a_batch_registers_every_collection_it_touches_exactly_once() {
3186        let db = Db::in_memory();
3187        db.put_batch(vec![
3188            ("a".into(), "1".into(), j(1), vec![], None, None),
3189            ("b".into(), "1".into(), j(1), vec![], None, None),
3190            ("a".into(), "2".into(), j(2), vec![], None, None),
3191        ]).unwrap();
3192        assert_eq!(db.collections(), vec!["a".to_string(), "b".to_string()]);
3193        assert_eq!(
3194            db.id_index.list_ids(crate::namespace::COLLECTIONS).len(), 2,
3195            "three writes across two collections is two registry records"
3196        );
3197    }
3198
3199    #[test]
3200    fn a_batch_naming_a_reserved_collection_writes_nothing_at_all() {
3201        let db = Db::in_memory();
3202        let before = db.seq.load(Ordering::SeqCst);
3203        let r = db.put_batch(vec![
3204            ("ok".into(), "1".into(), j(1), vec![], None, None),
3205            (crate::namespace::COLLECTIONS.into(), "forged".into(), j(1), vec![], None, None),
3206        ]);
3207        assert!(r.is_err(), "a batch with a refused collection must be refused");
3208        assert_eq!(
3209            db.seq.load(Ordering::SeqCst), before,
3210            "and must not have written the acceptable half of itself first"
3211        );
3212        assert!(db.collections().is_empty());
3213    }
3214}
3215
3216/// State roots against a live engine: does the root actually track state, and
3217/// does verification tell the truth about what it could and could not check?
3218#[cfg(test)]
3219mod state_roots {
3220    use super::*;
3221    use crate::root::{RecordStatus, Recomputation, UnavailableReason};
3222    use tempfile::tempdir;
3223
3224    fn j(v: u64) -> serde_json::Value { serde_json::json!({"v": v}) }
3225
3226    #[test]
3227    fn an_empty_database_has_a_stable_nonzero_root() {
3228        let a = Db::in_memory().state_root().unwrap();
3229        let b = Db::in_memory().state_root().unwrap();
3230        assert_eq!(a, b);
3231        assert_ne!(a.state_root, "0".repeat(64));
3232        assert_eq!(a.collection_count, 0);
3233        assert_eq!(a.record_count, 0);
3234    }
3235
3236    /// The invariance the whole format exists for.
3237    #[test]
3238    fn disk_and_memory_agree_on_the_root_of_the_same_history() {
3239        let dir = tempdir().unwrap();
3240        let disk = Db::open(dir.path(), None).unwrap();
3241        let mem = Db::in_memory();
3242        for db in [&disk, &mem] {
3243            db.put("orders", "1", j(1), vec![], None, None).unwrap();
3244            db.put("orders", "2", j(2), vec![], None, None).unwrap();
3245            db.put("users", "u", j(9), vec![], None, None).unwrap();
3246        }
3247        assert_eq!(disk.state_root().unwrap(), mem.state_root().unwrap());
3248    }
3249
3250    /// Encryption changes object hashes; it must not change the root.
3251    #[test]
3252    fn an_encrypted_replica_has_the_same_root_as_a_plaintext_one() {
3253        let plain_dir = tempdir().unwrap();
3254        let enc_dir = tempdir().unwrap();
3255        let plain = Db::open(plain_dir.path(), None).unwrap();
3256        let enc = Db::open(enc_dir.path(), Some(crate::store::Dek([7u8; 32]))).unwrap();
3257        for db in [&plain, &enc] {
3258            db.put("orders", "1", serde_json::json!({"total": 100}), vec![], None, None).unwrap();
3259        }
3260        assert_ne!(
3261            plain.get("orders", "1").unwrap().hash,
3262            enc.get("orders", "1").unwrap().hash,
3263            "precondition: encryption really does change the object hash"
3264        );
3265        assert_eq!(
3266            plain.state_root().unwrap(), enc.state_root().unwrap(),
3267            "but the root commits to logical content, so it must not move"
3268        );
3269    }
3270
3271    #[test]
3272    fn the_root_moves_when_the_state_moves_and_not_otherwise() {
3273        let db = Db::in_memory();
3274        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3275        let a = db.state_root().unwrap().state_root;
3276
3277        // A no-op rewrite of the same value: new node, new seq, same state.
3278        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3279        assert_eq!(db.state_root().unwrap().state_root, a,
3280            "the root commits to state, not to how many times you wrote it");
3281
3282        db.put("orders", "1", j(2), vec![], None, None).unwrap();
3283        assert_ne!(db.state_root().unwrap().state_root, a);
3284    }
3285
3286    #[test]
3287    fn a_delete_removes_a_record_but_keeps_the_collection() {
3288        let db = Db::in_memory();
3289        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3290        db.delete("orders", "1").unwrap();
3291        let r = db.state_root().unwrap();
3292        assert_eq!(r.record_count, 0, "a tombstoned document is not live state");
3293        assert_eq!(r.collection_count, 1, "but its collection still exists");
3294
3295        let never = Db::in_memory();
3296        assert_ne!(r.state_root, never.state_root().unwrap().state_root);
3297    }
3298
3299    #[test]
3300    fn a_historical_root_matches_what_the_tip_root_was_at_that_time() {
3301        let db = Db::in_memory();
3302        let a = db.put("orders", "1", j(1), vec![], None, None).unwrap();
3303        let then = db.state_root().unwrap();
3304        db.put("orders", "2", j(2), vec![], None, None).unwrap();
3305        assert_ne!(db.state_root().unwrap().state_root, then.state_root);
3306        assert_eq!(
3307            db.state_root_as_of(a.seq).unwrap().state_root, then.state_root,
3308            "AS OF the first write is the state after the first write"
3309        );
3310    }
3311
3312    #[test]
3313    fn a_persisted_root_verifies_against_a_fresh_recomputation() {
3314        let db = Db::in_memory();
3315        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3316        let rec = db.create_root().unwrap();
3317
3318        let v = db.verify_root(rec.at_seq);
3319        assert_eq!(v.record, RecordStatus::Valid);
3320        assert_eq!(v.recomputation, Recomputation::Matches);
3321        assert!(v.is_verified());
3322        assert!(!v.is_mismatch());
3323        assert_eq!(v.exit_code(), 0);
3324    }
3325
3326    /// Taking a root must not change the state it describes.
3327    #[test]
3328    fn taking_a_root_does_not_change_the_root() {
3329        let db = Db::in_memory();
3330        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3331        let before = db.state_root().unwrap().state_root.clone();
3332        db.create_root().unwrap();
3333        db.create_root().unwrap();
3334        assert_eq!(db.state_root().unwrap().state_root, before,
3335            "root records are reserved, so they are not part of the state");
3336    }
3337
3338    #[test]
3339    fn later_writes_do_not_retroactively_change_an_old_root() {
3340        let db = Db::in_memory();
3341        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3342        let rec = db.create_root().unwrap();
3343        db.put("orders", "2", j(2), vec![], None, None).unwrap();
3344        db.put("orders", "1", j(99), vec![], None, None).unwrap();
3345
3346        let v = db.verify_root(rec.at_seq);
3347        assert!(v.is_verified(), "a root is a statement about a sequence, not about now");
3348    }
3349
3350    #[test]
3351    fn a_missing_root_is_reported_as_missing_not_as_a_failure() {
3352        let db = Db::in_memory();
3353        let v = db.verify_root(42);
3354        assert_eq!(v.record, RecordStatus::Missing);
3355        assert_eq!(v.recomputation, Recomputation::NotAttempted);
3356        assert!(!v.is_verified());
3357        assert!(!v.is_mismatch(), "absent is not wrong");
3358        assert_eq!(v.exit_code(), 4);
3359    }
3360
3361    /// Compaction must not raise the floor when it pruned nothing.
3362    ///
3363    /// `ObjectStore::compact` is a NO-OP returning zeroed stats on the
3364    /// loose-object and in-memory substrates. An unconditional floor bump
3365    /// after it declared every earlier sequence pruned on a database where
3366    /// nothing had been — turning every historical root permanently
3367    /// unverifiable for a reason that was not true. A false alarm defeats the
3368    /// whole point of having an "unavailable" state.
3369    #[test]
3370    fn a_compaction_that_prunes_nothing_does_not_raise_the_floor() {
3371        let dir = tempdir().unwrap();
3372        let db = Db::open(dir.path(), None).unwrap();
3373        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3374        let rec = db.create_root().unwrap();
3375        db.put("orders", "1", j(2), vec![], None, None).unwrap();
3376        db.flush_all();
3377
3378        let stats = db.compact().expect("compact");
3379        assert_eq!(stats.dropped_objects, 0, "precondition: v2 compaction prunes nothing");
3380        assert_eq!(db.history_floor(), 0, "so no history was lost, and the floor must not move");
3381        assert!(
3382            db.verify_root(rec.at_seq).is_verified(),
3383            "the root must still verify — nothing was discarded"
3384        );
3385    }
3386
3387    /// The distinction the Oracle asked for.
3388    ///
3389    /// Driven through `set_history_floor` rather than a real prune because the
3390    /// only substrate that prunes is selected by the process-global
3391    /// `NEDB_DAG_V3` environment variable, and tests run threaded in one
3392    /// process — setting it here changed the substrate under every other test
3393    /// that opened a database at the same moment. The end-to-end prune is
3394    /// covered in `tests/v3_integration.rs`, which is its own process.
3395    #[test]
3396    fn a_pruned_history_reports_unavailable_rather_than_pass_or_fail() {
3397        let db = Db::in_memory();
3398        db.put("orders", "1", j(1), vec![], None, None).unwrap();
3399        let rec = db.create_root().unwrap();
3400        db.put("orders", "1", j(2), vec![], None, None).unwrap();
3401
3402        assert!(db.verify_root(rec.at_seq).is_verified(), "verifiable before the prune");
3403
3404        let tip = db.seq.load(Ordering::SeqCst).saturating_sub(1);
3405        db.set_history_floor(tip).unwrap();
3406
3407        let v = db.verify_root(rec.at_seq);
3408        assert_eq!(v.record, RecordStatus::Valid, "the record itself is still fine");
3409        assert_eq!(
3410            v.recomputation,
3411            Recomputation::Unavailable(UnavailableReason::HistoryPruned),
3412            "and the engine says plainly that it could not check it"
3413        );
3414        assert!(!v.is_verified(), "unavailable is not verified");
3415        assert!(!v.is_mismatch(), "and it is not a mismatch either");
3416        assert_eq!(v.exit_code(), 3, "its own exit code, distinct from pass and fail");
3417    }
3418
3419    #[test]
3420    fn roots_are_listed_in_sequence_order() {
3421        let db = Db::in_memory();
3422        for i in 0..12u64 {
3423            db.put("c", &i.to_string(), j(i), vec![], None, None).unwrap();
3424            db.create_root().unwrap();
3425        }
3426        let seqs: Vec<u64> = db.list_roots().iter().map(|r| r.at_seq).collect();
3427        let mut sorted = seqs.clone();
3428        sorted.sort();
3429        assert_eq!(seqs, sorted, "zero-padded ids must order numerically");
3430        assert_eq!(seqs.len(), 12);
3431    }
3432
3433    #[test]
3434    fn a_root_survives_a_reopen(){
3435        let dir = tempdir().unwrap();
3436        let at;
3437        let expected;
3438        {
3439            let db = Db::open(dir.path(), None).unwrap();
3440            db.put("orders", "1", j(1), vec![], None, None).unwrap();
3441            let r = db.create_root().unwrap();
3442            at = r.at_seq;
3443            expected = r.root.state_root.clone();
3444            db.flush_all();
3445        }
3446        let db = Db::open(dir.path(), None).unwrap();
3447        let got = db.get_root(at).expect("root record survives a reopen");
3448        assert_eq!(got.root.state_root, expected);
3449        assert!(db.verify_root(at).is_verified());
3450    }
3451}