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}