Skip to main content

nedb_engine/
index.rs

1//! Index store for NEDB v2.
2//!
3//! Two index types:
4//!
5//! 1. **ID index** (`indexes/{coll}/id/{doc_id}` → object hash)
6//!    Atomic file-per-document. Reading is a single `fs::read_to_string`.
7//!    Writing is atomic (write .tmp → rename). Parallel reads are lock-free.
8//!
9//! 2. **Sorted index** (`indexes/{coll}/{field}.sorted` → in-memory BTreeMap)
10//!    Rebuilt from object store on startup. Persisted as a compact binary
11//!    file for fast cold start. Used for ORDER BY field ASC/DESC LIMIT n.
12
13use std::collections::BTreeMap;
14use std::fs;
15use std::path::{Path, PathBuf};
16use std::sync::Arc;
17use anyhow::Result;
18use dashmap::DashMap;
19use serde_json::Value;
20
21/// Ordered JSON value for BTree indexes (null < bool < number < string < array < object).
22#[derive(Debug, Clone, PartialEq)]
23pub enum OrderedValue {
24    Null,
25    Bool(bool),
26    Number(f64),   // NaN-safe comparison via total_cmp
27    Str(String),
28    Array(Vec<OrderedValue>),
29    Object,        // objects are all equal in ordering (sort by insertion order falls back to hash)
30}
31
32impl Eq for OrderedValue {}
33
34impl PartialOrd for OrderedValue {
35    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
36        Some(self.cmp(other))
37    }
38}
39
40impl Ord for OrderedValue {
41    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
42        use OrderedValue::*;
43        use std::cmp::Ordering::*;
44        match (self, other) {
45            (Null, Null)       => Equal,
46            (Null, _)          => Less,
47            (_, Null)          => Greater,
48            (Bool(a), Bool(b)) => a.cmp(b),
49            (Bool(_), _)       => Less,
50            (_, Bool(_))       => Greater,
51            (Number(a), Number(b)) => a.total_cmp(b),
52            (Number(_), _)     => Less,
53            (_, Number(_))     => Greater,
54            (Str(a), Str(b))   => a.cmp(b),
55            (Str(_), _)        => Less,
56            (_, Str(_))        => Greater,
57            (Array(a), Array(b)) => a.cmp(b),
58            (Array(_), _)      => Less,
59            (_, Array(_))      => Greater,
60            (Object, Object)   => Equal,
61        }
62    }
63}
64
65impl From<&Value> for OrderedValue {
66    fn from(v: &Value) -> Self {
67        match v {
68            Value::Null        => OrderedValue::Null,
69            Value::Bool(b)     => OrderedValue::Bool(*b),
70            Value::Number(n)   => OrderedValue::Number(n.as_f64().unwrap_or(f64::NAN)),
71            Value::String(s)   => OrderedValue::Str(s.clone()),
72            Value::Array(a)    => OrderedValue::Array(a.iter().map(|x| x.into()).collect()),
73            Value::Object(_)   => OrderedValue::Object,
74        }
75    }
76}
77
78/// Compute a 2-char hex shard prefix from a document id.
79/// Distributes files across 256 subdirectories to avoid flat-directory
80/// slowdown on ext4/xfs when a collection has >50k documents.
81fn id_shard(id: &str) -> String {
82    // FNV-1a 32-bit — fast, no crypto needed, deterministic
83    let mut hash: u32 = 2166136261;
84    for b in id.bytes() {
85        hash ^= b as u32;
86        hash = hash.wrapping_mul(16777619);
87    }
88    format!("{:02x}", hash & 0xff)
89}
90
91/// Encode a document id into a filesystem-safe leaf filename.
92///
93/// The id-index stores one file per document, and the id is the filename. Raw
94/// ids work on case-sensitive POSIX filesystems, but ids containing bytes that
95/// are illegal in Windows filenames (`: | / \ < > " ? *`, control chars) — most
96/// notably link ids like `driver:d1|handles|trip:t1` — cannot be written there,
97/// so the write silently fails and the entry is lost on reopen.
98///
99/// We percent-escape every byte that isn't unreserved (`A-Z a-z 0-9 - _ .`).
100/// `%` itself is escaped so decoding is unambiguous. Safe ids (block heights,
101/// hex hashes, utxo keys) are all-unreserved and return UNCHANGED, so existing
102/// chainstate paths are byte-for-byte identical and the hot path is unaffected.
103fn encode_id(id: &str) -> String {
104    fn is_unreserved(b: u8) -> bool {
105        b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.')
106    }
107    if id.bytes().all(is_unreserved) {
108        return id.to_string();
109    }
110    let mut out = String::with_capacity(id.len() + 8);
111    for &b in id.as_bytes() {
112        if is_unreserved(b) {
113            out.push(b as char);
114        } else {
115            out.push_str(&format!("%{:02X}", b));
116        }
117    }
118    out
119}
120
121/// Inverse of `encode_id`. A name with no `%` (a safe id, or a legacy raw id
122/// written by an older version on a POSIX filesystem) is returned unchanged, so
123/// `list_ids` recovers the right id for both new and pre-upgrade files.
124fn decode_id(name: &str) -> String {
125    if !name.contains('%') {
126        return name.to_string();
127    }
128    fn hexval(b: u8) -> Option<u8> {
129        match b {
130            b'0'..=b'9' => Some(b - b'0'),
131            b'A'..=b'F' => Some(b - b'A' + 10),
132            b'a'..=b'f' => Some(b - b'a' + 10),
133            _ => None,
134        }
135    }
136    let bytes = name.as_bytes();
137    let mut out: Vec<u8> = Vec::with_capacity(bytes.len());
138    let mut i = 0;
139    while i < bytes.len() {
140        if bytes[i] == b'%' && i + 2 < bytes.len() {
141            if let (Some(hi), Some(lo)) = (hexval(bytes[i + 1]), hexval(bytes[i + 2])) {
142                out.push(hi * 16 + lo);
143                i += 3;
144                continue;
145            }
146        }
147        out.push(bytes[i]);
148        i += 1;
149    }
150    String::from_utf8_lossy(&out).into_owned()
151}
152
153/// Per-document ID index — atomic file-per-doc, sharded across 256 subdirs.
154///
155/// Write path: updates go to `write_buf` (DashMap, zero I/O, lock-free).
156/// Background ticker calls `flush_write_buf()` every 1s — Rayon-parallel disk writes.
157/// Read path: `write_buf` checked first (latest value), then disk.
158/// This eliminates per-PUT `fs::rename` from the hot path, fixing concurrent write contention.
159pub struct IdIndex {
160    root:      PathBuf,
161    /// In-memory store: (coll, id) → hash. None = disk-backed (normal mode).
162    mem:       Option<Arc<dashmap::DashMap<(String, String), String>>>,
163    /// WAL write buffer — disk-backed mode buffers here, flushed to disk periodically.
164    write_buf: Arc<dashmap::DashMap<(String, String), Option<String>>>,  // None = tombstone
165}
166
167impl IdIndex {
168    pub fn new(db_root: &Path) -> Result<Self> {
169        let root = db_root.join("indexes");
170        fs::create_dir_all(&root)?;
171        Ok(Self { root, mem: None, write_buf: Arc::new(dashmap::DashMap::new()) })
172    }
173
174    /// Create a pure in-memory id index — no disk I/O.
175    pub fn in_memory() -> Self {
176        Self {
177            root:      PathBuf::from(":memory:"),
178            mem:       Some(Arc::new(dashmap::DashMap::new())),
179            write_buf: Arc::new(dashmap::DashMap::new()),
180        }
181    }
182
183    /// Flush the WAL write buffer to disk in parallel. Called by the background ticker.
184    /// No-op for in-memory databases. Safe to call concurrently with writes.
185    pub fn flush_write_buf(&self) {
186        if self.mem.is_some() || self.write_buf.is_empty() { return; }
187        use rayon::prelude::*;
188        // Drain all pending entries and write them in parallel
189        let entries: Vec<((String, String), Option<String>)> = self.write_buf
190            .iter()
191            .map(|e| (e.key().clone(), e.value().clone()))
192            .collect();
193        entries.par_iter().for_each(|((coll, id), hash_opt)| {
194            match hash_opt {
195                Some(hash) => {
196                    // Write/update: tmp → rename
197                    let path = self.path(coll, id);
198                    if let Some(parent) = path.parent() {
199                        let _ = fs::create_dir_all(parent);
200                    }
201                    let tmp = path.with_extension("tmp");
202                    if fs::write(&tmp, hash).is_ok() {
203                        let _ = fs::rename(&tmp, &path);
204                    }
205                }
206                None => {
207                    // Tombstone: remove the file (encoded leaf + legacy raw if distinct)
208                    let path = self.path(coll, id);
209                    let _ = fs::remove_file(&path);
210                    let raw = self.raw_path(coll, id);
211                    if raw != path { let _ = fs::remove_file(&raw); }
212                }
213            }
214        });
215        // Clear flushed entries — but ONLY when the buffered value is still the
216        // exact value we flushed. An unconditional remove() here would delete a
217        // NEWER value written between the snapshot above and this point: that
218        // write would never reach disk (the file holds the stale hash we just
219        // wrote) and get() would serve the old version once the buffer check
220        // misses — a silent lost update. remove_if closes the race; a newer
221        // value simply stays buffered and flushes on the next tick.
222        for (key, flushed_val) in &entries {
223            self.write_buf.remove_if(key, |_, current| current == flushed_val);
224        }
225    }
226
227    fn path(&self, coll: &str, id: &str) -> PathBuf {
228        // Shard across 256 subdirectories using first 2 hex chars of a simple
229        // hash of the id. Prevents flat-directory slowdown (ext4 htree degrades
230        // past ~50k files per directory) for large collections like kv.
231        // Format: indexes/{coll}/id/{shard}/{encode_id(id)}
232        // Shard on the RAW id (stable across versions); only the leaf filename
233        // is encoded so it is legal on every filesystem (incl. Windows).
234        let shard = id_shard(id);
235        self.root.join(coll).join("id").join(&shard).join(encode_id(id))
236    }
237
238    /// Legacy path: the raw id as the leaf filename (pre-`encode_id`). Used only
239    /// as a read/cleanup fallback so id-index entries written by older versions
240    /// on POSIX filesystems stay readable after upgrade. On Windows a raw path
241    /// with illegal chars simply fails to open (→ treated as absent).
242    fn raw_path(&self, coll: &str, id: &str) -> PathBuf {
243        let shard = id_shard(id);
244        self.root.join(coll).join("id").join(&shard).join(id)
245    }
246
247    /// Get the current object hash for a document.
248    /// Checks WAL write buffer first (most recent), then disk.
249    pub fn get(&self, coll: &str, id: &str) -> Option<String> {
250        if let Some(ref mem) = self.mem {
251            return mem.get(&(coll.to_string(), id.to_string())).map(|v| v.clone());
252        }
253        // Check WAL buffer first — may have an unflushed write or tombstone
254        let key = (coll.to_string(), id.to_string());
255        if let Some(entry) = self.write_buf.get(&key) {
256            return entry.value().clone();  // None = tombstoned
257        }
258        // Fall through to disk: encoded filename first, then the legacy raw
259        // filename (pre-upgrade data). For safe ids the two paths are identical,
260        // so this is a single read on the hot path.
261        let p = self.path(coll, id);
262        let content = match fs::read_to_string(&p) {
263            Ok(c) => c,
264            Err(_) => {
265                let raw = self.raw_path(coll, id);
266                if raw == p { return None; }
267                fs::read_to_string(&raw).ok()?
268            }
269        };
270        let h = content.trim().to_string();
271        if h.is_empty() { None } else { Some(h) }
272    }
273
274    /// Set the current object hash for a document.
275    /// Disk mode: writes to WAL buffer only (zero I/O on hot path).
276    /// Background ticker flushes WAL to disk every 1s via Rayon.
277    pub fn set(&self, coll: &str, id: &str, hash: &str) -> Result<()> {
278        if let Some(ref mem) = self.mem {
279            mem.insert((coll.to_string(), id.to_string()), hash.to_string());
280            return Ok(());
281        }
282        // WAL: buffer the update, no disk I/O here
283        self.write_buf.insert(
284            (coll.to_string(), id.to_string()),
285            Some(hash.to_string()),
286        );
287        Ok(())
288    }
289
290    /// List all doc IDs in a collection (memory map or disk + WAL merge).
291    pub fn list_ids(&self, coll: &str) -> Vec<String> {
292        if let Some(ref mem) = self.mem {
293            // DashMap iteration order is also unspecified — sort here too, so
294            // memory mode and disk mode agree.
295            let mut ids: Vec<String> = mem.iter()
296                .filter(|e| e.key().0 == coll)
297                .map(|e| e.key().1.clone())
298                .collect();
299            ids.sort_unstable();
300            return ids;
301        }
302        // Read from disk then overlay WAL (adds buffered writes, removes tombstones)
303        let id_root = self.root.join(coll).join("id");
304        // Each entry in id_root is a 2-char hex shard dir
305        let mut ids: Vec<String> = fs::read_dir(&id_root)
306            .into_iter()
307            .flatten()
308            .filter_map(|e| e.ok())
309            .filter(|e| e.file_type().map(|t| t.is_dir()).unwrap_or(false))
310            .flat_map(|shard_dir| {
311                fs::read_dir(shard_dir.path())
312                    .into_iter()
313                    .flatten()
314                    .filter_map(|e| e.ok())
315                    .filter_map(|e| {
316                        let name = e.file_name().to_string_lossy().to_string();
317                        if name.ends_with(".tmp") { return None; }
318                        // Decode the on-disk filename back to the document id
319                        // (encoded for new files; identity for legacy/safe ids).
320                        Some(decode_id(&name))
321                    })
322                    .collect::<Vec<_>>()
323            })
324            .collect::<std::collections::HashSet<_>>()
325            .into_iter()
326            // Overlay WAL: add buffered writes, remove tombstones
327            .chain(
328                self.write_buf.iter()
329                    .filter(|e| e.key().0 == coll && e.value().is_some())
330                    .map(|e| e.key().1.clone())
331            )
332            .collect::<std::collections::HashSet<_>>()
333            .into_iter()
334            .filter(|id| {
335                // Exclude WAL tombstones
336                self.write_buf.get(&(coll.to_string(), id.clone()))
337                    .map(|v| v.is_some())
338                    .unwrap_or(true)
339            })
340            .collect::<Vec<_>>();
341
342        // Deterministic order. The dedup above runs through HashSet, and Rust
343        // seeds its hasher randomly PER PROCESS — so without this the same query
344        // over unchanged data returns rows in a different order on every restart:
345        //
346        //   run 1:  o5 o8 o7 o4 o6 ...
347        //   run 2:  o7 o5 o2 o6 o8 ...
348        //
349        // Cosmetic for a full scan, but not for `LIMIT 5` with no ORDER BY,
350        // which then returns an arbitrary 5 of 8 and calls it an answer. It also
351        // makes any snapshot/diff test flaky for reasons that look like data
352        // corruption.
353        //
354        // Sorted by id, which for the common case of sequential ids is also
355        // insertion order. Callers that want a different order say ORDER BY.
356        ids.sort_unstable();
357        ids
358    }
359
360    /// Remove the id index entry for a document (tombstone / delete).
361    /// Disk mode: writes a tombstone to the WAL buffer; flushed to disk on next ticker.
362    pub fn remove(&self, coll: &str, id: &str) -> Result<()> {
363        if let Some(ref mem) = self.mem {
364            mem.remove(&(coll.to_string(), id.to_string()));
365            return Ok(());
366        }
367        // WAL tombstone: None value means "delete this file on flush"
368        self.write_buf.insert((coll.to_string(), id.to_string()), None);
369        Ok(())
370    }
371
372    /// List all known collections.
373    ///
374    /// Overlays the WAL, exactly as `ids()` does. A collection whose first write
375    /// is still sitting in `write_buf` has no directory on disk yet, so a
376    /// read_dir-only implementation reports it as absent for up to a full flush
377    /// tick.
378    ///
379    /// That was a real bug, and a nasty one because it was invisible to a human
380    /// at a terminal: type a PUT, type a query, and the 1s ticker has already
381    /// fired in between. Only an automated caller — one that writes and reads in
382    /// the same millisecond — ever sees the empty list. It surfaced through
383    /// `/cast`, which checks the generated collection against this list and
384    /// returned "collection does not exist" for a collection that had just been
385    /// written successfully.
386    ///
387    /// Tombstoned entries are excluded, but only when the collection has no
388    /// surviving documents anywhere — a delete of one document must not hide the
389    /// whole collection.
390    pub fn collections(&self) -> Vec<String> {
391        if let Some(ref mem) = self.mem {
392            let mut colls: Vec<String> = mem.iter()
393                .map(|e| e.key().0.clone())
394                .collect::<std::collections::HashSet<_>>()
395                .into_iter().collect();
396            colls.sort();
397            return colls;
398        }
399
400        let mut set: std::collections::HashSet<String> = fs::read_dir(&self.root)
401            .into_iter()
402            .flatten()
403            .filter_map(|e| e.ok())
404            .filter(|e| e.file_type().map(|t| t.is_dir()).unwrap_or(false))
405            .map(|e| e.file_name().to_string_lossy().to_string())
406            .collect();
407
408        // Overlay WAL: a buffered live write makes its collection visible now.
409        for e in self.write_buf.iter() {
410            if e.value().is_some() {
411                set.insert(e.key().0.clone());
412            }
413        }
414
415        let mut colls: Vec<String> = set.into_iter().collect();
416        colls.sort();
417        colls
418    }
419}
420
421/// In-memory sorted index per (collection, field).
422/// Rebuilt from object store on startup. O(log n) ORDER BY queries.
423pub struct SortedIndexes {
424    /// (coll, field) → BTreeMap<value, Vec<hash>>
425    inner: DashMap<(String, String), BTreeMap<OrderedValue, Vec<String>>>,
426}
427
428impl SortedIndexes {
429    pub fn new() -> Self {
430        Self { inner: DashMap::new() }
431    }
432
433    /// Register a field as sorted-indexed for a collection.
434    /// Must be called before any puts for that field to be indexed.
435    pub fn ensure(&self, coll: &str, field: &str) {
436        self.inner
437            .entry((coll.to_string(), field.to_string()))
438            .or_default();
439    }
440
441    /// Insert (or update) a value → hash mapping.
442    pub fn insert(&self, coll: &str, field: &str, value: &Value, hash: &str) {
443        let key = (coll.to_string(), field.to_string());
444        if let Some(mut idx) = self.inner.get_mut(&key) {
445            let ov = OrderedValue::from(value);
446            idx.entry(ov)
447               .or_default()
448               .push(hash.to_string());
449        }
450    }
451
452    /// Remove a hash from the index (on overwrite/delete of a doc version).
453    pub fn remove(&self, coll: &str, field: &str, value: &Value, hash: &str) {
454        let key = (coll.to_string(), field.to_string());
455        if let Some(mut idx) = self.inner.get_mut(&key) {
456            let ov = OrderedValue::from(value);
457            if let Some(hashes) = idx.get_mut(&ov) {
458                hashes.retain(|h| h != hash);
459                if hashes.is_empty() { idx.remove(&ov); }
460            }
461        }
462    }
463
464    /// Return the top-k hashes ordered by field ASC.
465    pub fn top_k_asc(&self, coll: &str, field: &str, k: usize) -> Vec<String> {
466        let key = (coll.to_string(), field.to_string());
467        self.inner.get(&key).map(|idx| {
468            idx.values().flat_map(|v| v.iter().cloned()).take(k).collect()
469        }).unwrap_or_default()
470    }
471
472    /// Return the top-k hashes ordered by field DESC.
473    pub fn top_k_desc(&self, coll: &str, field: &str, k: usize) -> Vec<String> {
474        let key = (coll.to_string(), field.to_string());
475        self.inner.get(&key).map(|idx| {
476            idx.values().rev().flat_map(|v| v.iter().cloned()).take(k).collect()
477        }).unwrap_or_default()
478    }
479
480    /// Check if a sorted index exists for a (coll, field) pair.
481    pub fn has(&self, coll: &str, field: &str) -> bool {
482        self.inner.contains_key(&(coll.to_string(), field.to_string()))
483    }
484
485    /// True if no sorted indexes have been registered yet.
486    pub fn is_empty(&self) -> bool {
487        self.inner.is_empty()
488    }
489}
490
491#[cfg(test)]
492mod tests {
493    use super::*;
494    use tempfile::tempdir;
495
496    #[test]
497    fn id_index_roundtrip() {
498        let dir = tempdir().unwrap();
499        let idx = IdIndex::new(dir.path()).unwrap();
500        idx.set("blocks", "618000", "abcdef1234").unwrap();
501        assert_eq!(idx.get("blocks", "618000"), Some("abcdef1234".to_string()));
502    }
503
504    #[test]
505    fn encode_decode_id_bijective() {
506        // Safe ids pass through unchanged (chainstate paths stay identical).
507        for safe in ["618000", "utxo-000000042", "abc_DEF.123", "deadBEEF"] {
508            assert_eq!(encode_id(safe), safe, "safe id must be identity");
509            assert_eq!(decode_id(&encode_id(safe)), safe);
510        }
511        // FS-unsafe ids (link ids, paths) round-trip and contain no illegal
512        // Windows filename chars once encoded.
513        for weird in ["driver:d1|handles|trip:t1", "a/b\\c", "x<y>z?\"*", "100%done"] {
514            let enc = encode_id(weird);
515            assert!(
516                !enc.chars().any(|c| matches!(c,
517                    ':' | '|' | '/' | '\\' | '<' | '>' | '?' | '"' | '*')),
518                "encoded leaf must be filesystem-safe: {}", enc);
519            assert_eq!(decode_id(&enc), weird, "encode/decode must round-trip");
520        }
521    }
522
523    #[test]
524    fn id_index_fs_unsafe_id_survives_disk_roundtrip() {
525        // Regression: link ids contain ':' and '|', illegal in Windows filenames.
526        // They must persist to the on-disk id-index and read back after reopen.
527        let dir = tempdir().unwrap();
528        let weird = "driver:d1|handles|trip:t1";
529        {
530            let idx = IdIndex::new(dir.path()).unwrap();
531            idx.set("__links__", weird, "deadbeefcafe").unwrap();
532            idx.flush_write_buf(); // persist WAL → disk (encoded leaf filename)
533        }
534        // Cold reopen: nothing in the WAL, must come from disk.
535        let idx2 = IdIndex::new(dir.path()).unwrap();
536        assert_eq!(idx2.get("__links__", weird), Some("deadbeefcafe".to_string()),
537                   "FS-unsafe id must be readable from disk after reopen");
538        assert_eq!(idx2.list_ids("__links__"), vec![weird.to_string()],
539                   "list_ids must decode the on-disk filename back to the id");
540    }
541
542    #[test]
543    fn ordered_value_ordering() {
544        use OrderedValue::*;
545        assert!(Null < Bool(false));
546        assert!(Bool(false) < Bool(true));
547        assert!(Bool(true) < Number(0.0));
548        assert!(Number(1.0) < Number(2.0));
549        assert!(Number(2.0) < Str("a".to_string()));
550        assert!(Str("a".to_string()) < Str("b".to_string()));
551    }
552
553    #[test]
554    fn sorted_index_top_k() {
555        let idx = SortedIndexes::new();
556        idx.ensure("blocks", "height");
557        idx.insert("blocks", "height", &serde_json::json!(3), "hash3");
558        idx.insert("blocks", "height", &serde_json::json!(1), "hash1");
559        idx.insert("blocks", "height", &serde_json::json!(2), "hash2");
560        let asc = idx.top_k_asc("blocks", "height", 2);
561        assert_eq!(asc, vec!["hash1", "hash2"]);
562        let desc = idx.top_k_desc("blocks", "height", 2);
563        assert_eq!(desc, vec!["hash3", "hash2"]);
564    }
565
566    /// Regression stress test for the flush_write_buf lost-update race.
567    ///
568    /// Old behavior: flush snapshotted the buffer, wrote files in parallel, then
569    /// UNCONDITIONALLY removed each snapshotted key. A set() landing between the
570    /// snapshot and the remove was deleted from the buffer without ever being
571    /// flushed — disk kept the stale hash and (with no later write to re-insert
572    /// the key) the newer value was lost forever.
573    ///
574    /// Shape: every key is written exactly twice (v1 then v2) while a flusher
575    /// thread spins. Under the old code, keys whose v1 was snapshotted and whose
576    /// v2 arrived during the parallel disk-write phase get their v2 dropped by
577    /// the unconditional remove — the final assert catches them on disk at v1.
578    /// With remove_if, a superseded snapshot entry leaves the newer value
579    /// buffered for the next flush, so every key must read v2 at the end.
580    /// (Probabilistic by nature, but the race window — thousands of parallel
581    /// file writes — is wide; with 2000 keys the old code fails reliably.)
582    #[test]
583    fn flush_never_drops_a_concurrent_newer_write() {
584        use std::sync::Arc;
585        use std::sync::atomic::{AtomicBool, Ordering};
586
587        let dir = tempdir().unwrap();
588        let idx = Arc::new(IdIndex::new(dir.path()).unwrap());
589        let stop = Arc::new(AtomicBool::new(false));
590        const N: usize = 2000;
591
592        let flusher = {
593            let idx = Arc::clone(&idx);
594            let stop = Arc::clone(&stop);
595            std::thread::spawn(move || {
596                while !stop.load(Ordering::Relaxed) {
597                    idx.flush_write_buf();
598                }
599            })
600        };
601
602        // v1 for every key, then v2 for every key — the flusher races both passes.
603        for i in 0..N {
604            idx.set("c", &format!("k{}", i), "v1").unwrap();
605        }
606        for i in 0..N {
607            idx.set("c", &format!("k{}", i), "v2").unwrap();
608        }
609
610        stop.store(true, Ordering::Relaxed);
611        flusher.join().unwrap();
612        // Drain anything still buffered (remove_if leaves superseded entries in).
613        idx.flush_write_buf();
614        idx.flush_write_buf();
615
616        // Every key must be v2 — from this handle AND from a cold reopen (disk).
617        for i in 0..N {
618            let k = format!("k{}", i);
619            assert_eq!(idx.get("c", &k), Some("v2".to_string()),
620                       "key {} lost its newer write (buffer path)", k);
621        }
622        let cold = IdIndex::new(dir.path()).unwrap();
623        for i in 0..N {
624            let k = format!("k{}", i);
625            assert_eq!(cold.get("c", &k), Some("v2".to_string()),
626                       "key {} lost its newer write (disk path)", k);
627        }
628    }
629
630    /// Regression: a collection must be visible the instant it is written, not
631    /// one flush tick later.
632    ///
633    /// Old behavior: `collections()` did a bare `read_dir` of the object root. A
634    /// brand-new collection lives only in `write_buf` until the 1s ticker fires,
635    /// so it was reported as ABSENT for up to a full second after a successful
636    /// write. Every other read path (`get`, `list_ids`) already overlaid the WAL;
637    /// this one silently did not.
638    ///
639    /// Why it hid for so long: a human at a terminal cannot reproduce it. Typing
640    /// a PUT and then a query leaves hundreds of milliseconds in between, and the
641    /// ticker has already run. Only a caller that writes and reads within the
642    /// same millisecond sees the empty list — which is exactly what an automated
643    /// test does. It surfaced through `/cast`, which validates the model's chosen
644    /// collection against this list and rejected a collection that had just been
645    /// written.
646    ///
647    /// NOTE the deliberate absence of any flush below. Calling flush_write_buf()
648    /// here would make this test pass against the OLD code and assert nothing.
649    #[test]
650    fn collections_are_visible_before_flush() {
651        let dir = tempdir().unwrap();
652        let idx = IdIndex::new(dir.path()).unwrap();
653
654        idx.set("orders", "o1", "hash1").unwrap();
655        let colls = idx.collections();
656        assert!(
657            colls.contains(&"orders".to_string()),
658            "collection invisible before flush: {colls:?}"
659        );
660
661        // Still correct once it does reach disk — no duplicates from the overlay.
662        idx.flush_write_buf();
663        let after = idx.collections();
664        assert_eq!(after, vec!["orders".to_string()], "after flush: {after:?}");
665
666        // Second collection, same story, and the first must not vanish.
667        idx.set("stylists", "s1", "hash2").unwrap();
668        let both = idx.collections();
669        assert_eq!(both, vec!["orders".to_string(), "stylists".to_string()],
670                   "expected both collections, got {both:?}");
671    }
672
673    /// Regression: `list_ids` must return a STABLE order.
674    ///
675    /// The dedup path runs through `HashSet`, and Rust seeds its hasher randomly
676    /// per process. Observed on a real daemon — same query, same data, three
677    /// consecutive runs:
678    ///
679    /// ```text
680    ///   o5 o8 o7 o4 o6 ...
681    ///   o7 o5 o2 o6 o8 ...
682    ///   o6 o1 o5 o4 o7 ...
683    /// ```
684    ///
685    /// Cosmetic on a full scan. NOT cosmetic for `LIMIT 5` with no `ORDER BY`,
686    /// which then hands back an arbitrary 5 of 8 as though it were an answer.
687    ///
688    /// NOTE the id set below. Sequential ids (`o1`..`o8`) can land in a
689    /// consistent order by chance, which would let this pass against the old
690    /// code. These are deliberately hash-scattered strings, and 24 of them, so a
691    /// single unsorted run being accidentally sorted is vanishingly unlikely.
692    #[test]
693    fn list_ids_order_is_stable() {
694        let dir = tempdir().unwrap();
695        let idx = IdIndex::new(dir.path()).unwrap();
696
697        let ids: Vec<String> = (0..24).map(|i| format!("zq{:x}-{}", i * 7919, i)).collect();
698        for id in &ids {
699            idx.set("orders", id, "h").unwrap();
700        }
701
702        // Buffered (pre-flush) and on-disk (post-flush) must BOTH be sorted, and
703        // must agree with each other — a flush is not a reordering event.
704        let mut want = ids.clone();
705        want.sort_unstable();
706
707        let before = idx.list_ids("orders");
708        assert_eq!(before, want, "unsorted before flush");
709
710        idx.flush_write_buf();
711        let after = idx.list_ids("orders");
712        assert_eq!(after, want, "unsorted after flush");
713        assert_eq!(before, after, "flush changed the order");
714
715        // Repeat reads within a process must not drift either.
716        for _ in 0..5 {
717            assert_eq!(idx.list_ids("orders"), want, "order varied between reads");
718        }
719    }
720
721    /// A tombstoned document must not resurrect its collection.
722    #[test]
723    fn collections_excludes_tombstone_only_writes() {
724        let dir = tempdir().unwrap();
725        let idx = IdIndex::new(dir.path()).unwrap();
726        idx.remove("ghosts", "g1").unwrap();
727        let colls = idx.collections();
728        assert!(!colls.contains(&"ghosts".to_string()),
729                "a tombstone conjured a collection: {colls:?}");
730    }
731
732}