Skip to main content

mongreldb_core/
run_lookup.rs

1//! RowId → run-set lookup directory (TODO §1.2).
2//!
3//! Derived, rebuildable, **never** authoritative. Missing, stale, corrupt, or
4//! incomplete directory state falls back to the existing range-scan path in
5//! `Table::get`. The directory is rebuilt from sorted-run system columns and
6//! published atomically alongside the manifest, so a torn publication never
7//! leaves an inconsistent directory visible.
8//!
9//! On-disk shape:
10//! - delta-encoded `RowId` keys (varint);
11//! - for each key, a newest-first list of `RunLocator`s (run_id u128, epoch
12//!   min/max, optional HLC min/max, unstamped flag);
13//! - a fingerprint of the exact active run-set + schema/index generation that
14//!   produced the directory; rejected on reopen if the active manifest
15//!   diverges;
16//! - a CRC32C footer for corruption detection.
17//!
18//! Large tables use a sharded checkpoint layout (one file per `RowId` range).
19//! The single-file API is retained for tests and small tables; the
20//! production open path uses [`RunLookupDirectory::read_checkpoint_sharded`].
21
22use std::collections::BTreeMap;
23use std::fs::{File, OpenOptions};
24use std::io::{self, Read, Write};
25use std::path::Path;
26
27use crate::epoch::{Epoch, Snapshot};
28use crate::manifest::RunRef;
29use crate::rowid::RowId;
30use crate::{Result, Table};
31use crc::{Crc, CRC_32_ISCSI};
32use mongreldb_types::hlc::HlcTimestamp;
33
34/// On-disk filename of the single-file directory checkpoint. Production tables
35/// use the sharded layout (see `directory_shard_path`).
36pub const DIRECTORY_FILENAME: &str = "directory.bin";
37/// Prefix for sharded directory files (`directory.shard-<start>.bin`).
38pub const DIRECTORY_SHARD_PREFIX: &str = "directory.shard-";
39pub const DIRECTORY_SHARD_SUFFIX: &str = ".bin";
40/// Default max bytes per shard when publishing a large directory. Keeps the
41/// in-memory buffer under 16 MiB so rebuilds/checkpoint publishes never need
42/// the entire directory resident at once on a 256-run workload.
43pub const DEFAULT_SHARD_BYTES: u64 = 16 * 1024 * 1024;
44
45/// Magic bytes that prefix the on-disk checkpoint (`MLKP` = Mongrel Lookup).
46const LOOKUP_MAGIC: [u8; 4] = *b"MLKP";
47/// CRC32C (Castagnoli) over the body — same algorithm as the WAL.
48const CRC32C: Crc<u32> = Crc::<u32>::new(&CRC_32_ISCSI);
49const FOOTER_LEN: usize = 4;
50const HLC_BYTES: usize = 16;
51
52/// What one run-level posting says about a `RowId`: enough metadata to
53/// conservatively decide whether the run can contain a version visible to a
54/// supplied snapshot, without opening the run.
55#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56pub struct RunLocator {
57    /// Stable identifier of the run, matched against the manifest's `RunRef`.
58    pub run_id: u128,
59    pub min_epoch: Epoch,
60    pub max_epoch: Epoch,
61    /// `Some` when the run carries HLC-stamped rows.
62    pub min_hlc: Option<HlcTimestamp>,
63    pub max_hlc: Option<HlcTimestamp>,
64    /// `true` when the run also contains legacy unstamped (epoch-only) rows.
65    pub contains_unstamped_versions: bool,
66}
67
68/// A complete, directly-consultable view of the directory for a single
69/// `RowId`. Returned by [`RunLookupDirectory::locate`].
70#[derive(Debug, Clone, Default)]
71pub struct RunLocatorList {
72    pub locators: Vec<RunLocator>,
73}
74
75impl RunLocatorList {
76    pub fn is_empty(&self) -> bool {
77        self.locators.is_empty()
78    }
79
80    pub fn len(&self) -> usize {
81        self.locators.len()
82    }
83}
84
85/// What a `Table::get` directory consultation concluded with.
86///
87/// `CompleteMiss` is the authoritative answer "this row has no postings in
88/// the directory" — the memtable and mutable-run tiers have already been
89/// searched, so an empty complete lookup means no immutable run can host a
90/// visible version. The caller opens zero readers.
91///
92/// `Candidates` means the directory returned at least one locator; the
93/// caller walks them with the conservative snapshot filter and the
94/// [`RunLocator::can_contain_version_newer_than`] early-stop proof.
95///
96/// `UnavailableOrStale` means the directory was missing, corrupt, or had a
97/// stale fingerprint; the caller falls back to the existing range-scan
98/// path. Distinct from `CompleteMiss` because the directory gave no
99/// definitive answer and the row *could* still exist in an immutable run.
100#[derive(Debug, Clone)]
101pub(crate) enum DirectoryLookupDecision {
102    CompleteMiss,
103    Candidates(Vec<RunLocator>),
104    UnavailableOrStale,
105}
106
107/// A compact snapshot of the (epoch, hlc) coordinates of a row version. Used
108/// by the safe early-stop proof so the caller can ask "can this locator
109/// still beat the current winner?" without exposing the full [`crate::memtable::Row`]
110/// in the public surface.
111#[derive(Debug, Clone, Copy, PartialEq, Eq)]
112pub(crate) struct VersionStamp {
113    pub epoch: Epoch,
114    pub hlc: Option<HlcTimestamp>,
115}
116
117/// Per-`RowId` posting list. Backed by a `BTreeMap` for the in-memory
118/// representation; the on-disk checkpoint uses delta-encoded `RowId` keys.
119#[derive(Debug, Default, Clone)]
120pub struct RunLookupDirectory {
121    /// `(RowId -> newest-first RunLocator list)`.
122    postings: BTreeMap<RowId, RunLocatorList>,
123    /// Fingerprint of the active run-set + schema/index generation that
124    /// produced this directory. `None` for an in-memory rebuild not yet
125    /// checkpointed.
126    fingerprint: Option<u64>,
127}
128
129impl RunLookupDirectory {
130    /// Construct an empty directory with no fingerprint.
131    pub fn empty() -> Self {
132        Self::default()
133    }
134
135    /// Look up the candidate run locators for `row_id`. Returns an empty list
136    /// when the row has no postings (caller may treat this as a miss).
137    pub fn locate(&self, row_id: RowId) -> &RunLocatorList {
138        static EMPTY: RunLocatorList = RunLocatorList {
139            locators: Vec::new(),
140        };
141        self.postings.get(&row_id).unwrap_or(&EMPTY)
142    }
143
144    /// Decide how the read path should consult this directory for `row_id`.
145    /// Distinguishes a complete directory miss (no postings → open zero
146    /// immutable readers) from the absence of any usable directory
147    /// (`UnavailableOrStale` → range-scan fallback).
148    pub(crate) fn decide(&self, row_id: RowId) -> DirectoryLookupDecision {
149        match self.postings.get(&row_id) {
150            None => DirectoryLookupDecision::CompleteMiss,
151            Some(list) if list.is_empty() => DirectoryLookupDecision::CompleteMiss,
152            Some(list) => DirectoryLookupDecision::Candidates(list.locators.clone()),
153        }
154    }
155
156    /// Insert or replace a `RunLocator` for `row_id`. Locators are kept
157    /// newest-first; the most recent write wins for ties.
158    pub fn insert(&mut self, row_id: RowId, locator: RunLocator) {
159        debug_assert!(
160            locator.validate_metadata().is_ok(),
161            "RunLookupDirectory::insert with invalid locator: {locator:?}"
162        );
163        let entry = self.postings.entry(row_id).or_default();
164        // Newest-first: locate the first locator whose `max_epoch`/`max_hlc`
165        // is older than the new one and insert before it. Falls back to
166        // append when the new locator is the oldest.
167        let pos = entry
168            .locators
169            .iter()
170            .position(|existing| locator_is_newer(locator, *existing));
171        match pos {
172            Some(idx) => entry.locators.insert(idx, locator),
173            None => entry.locators.push(locator),
174        }
175    }
176
177    /// Drop every locator whose `run_id` no longer exists in the active
178    /// manifest. Returns the number of removed postings.
179    pub fn retain_active_runs(&mut self, active_runs: &[u128]) -> usize {
180        let mut removed = 0;
181        for list in self.postings.values_mut() {
182            let before = list.locators.len();
183            list.locators.retain(|l| active_runs.contains(&l.run_id));
184            removed += before - list.locators.len();
185        }
186        removed
187    }
188
189    /// Set the fingerprint of the active run-set + schema/index generation.
190    pub fn set_fingerprint(&mut self, fp: u64) {
191        self.fingerprint = Some(fp);
192    }
193
194    pub fn fingerprint(&self) -> Option<u64> {
195        self.fingerprint
196    }
197
198    /// Number of `(RowId, posting)` entries. Used for memory/footprint metrics.
199    pub fn total_locators(&self) -> usize {
200        self.postings.values().map(|l| l.locators.len()).sum()
201    }
202
203    /// Number of distinct `RowId` keys.
204    pub fn key_count(&self) -> usize {
205        self.postings.len()
206    }
207
208    /// Maximum number of locators for any single `RowId`.
209    pub fn max_locators_per_row(&self) -> usize {
210        self.postings
211            .values()
212            .map(|l| l.locators.len())
213            .max()
214            .unwrap_or(0)
215    }
216
217    /// Average number of locators per `RowId` (0 when the directory is empty).
218    pub fn avg_locators_per_row(&self) -> f64 {
219        if self.postings.is_empty() {
220            return 0.0;
221        }
222        self.total_locators() as f64 / self.postings.len() as f64
223    }
224
225    /// Persist the directory to `path` as a single atomic file (write to temp
226    /// sibling, fsync, rename). A `fingerprint` must have been set; without
227    /// one the on-disk file is meaningless on reopen. Layout: magic, fingerprint,
228    /// key count, delta-encoded `RowId` keys with newest-first per-key locator
229    /// lists, then a CRC32C footer.
230    pub fn write_checkpoint(&self, path: &Path) -> io::Result<()> {
231        let fp = self.fingerprint.ok_or_else(|| {
232            io::Error::new(
233                io::ErrorKind::InvalidData,
234                "run-lookup directory has no fingerprint; refusing to checkpoint",
235            )
236        })?;
237        let mut body: Vec<u8> = Vec::new();
238        body.extend_from_slice(&LOOKUP_MAGIC);
239        body.extend_from_slice(&fp.to_le_bytes());
240        let key_count = self.postings.len() as u32;
241        body.extend_from_slice(&key_count.to_le_bytes());
242        let mut prev: u64 = 0;
243        for (row_id, list) in &self.postings {
244            let cur = row_id.0;
245            let delta = cur.checked_sub(prev).ok_or_else(|| {
246                io::Error::new(
247                    io::ErrorKind::InvalidData,
248                    "run-lookup row_id sequence is not non-decreasing",
249                )
250            })?;
251            encode_varint(delta, &mut body);
252            let locator_count = list.locators.len() as u32;
253            body.extend_from_slice(&locator_count.to_le_bytes());
254            for loc in &list.locators {
255                body.extend_from_slice(&loc.run_id.to_le_bytes());
256                body.extend_from_slice(&loc.min_epoch.0.to_le_bytes());
257                body.extend_from_slice(&loc.max_epoch.0.to_le_bytes());
258                write_optional_hlc(loc.min_hlc, &mut body);
259                write_optional_hlc(loc.max_hlc, &mut body);
260                body.push(if loc.contains_unstamped_versions {
261                    1
262                } else {
263                    0
264                });
265            }
266            prev = cur;
267        }
268        let crc = CRC32C.checksum(&body);
269        body.extend_from_slice(&crc.to_le_bytes());
270
271        let parent = path.parent().unwrap_or_else(|| Path::new("."));
272        let staging = parent.join(format!(
273            ".{}.{}.{}.staging",
274            path.file_name()
275                .and_then(|name| name.to_str())
276                .unwrap_or("checkpoint"),
277            std::process::id(),
278            std::time::SystemTime::now()
279                .duration_since(std::time::UNIX_EPOCH)
280                .map(|d| d.as_nanos())
281                .unwrap_or(0),
282        ));
283        {
284            // Best-effort cleanup of any stale staging file left behind by a
285            // crashed previous attempt at the same nanos tick. The window is
286            // vanishingly small but the truncate avoids an `AlreadyExists`
287            // failure that would mask a real I/O error.
288            let _ = std::fs::remove_file(&staging);
289            let mut file = OpenOptions::new()
290                .create(true)
291                .write(true)
292                .truncate(true)
293                .open(&staging)?;
294            file.write_all(&body)?;
295            file.sync_all()?;
296        }
297        match std::fs::rename(&staging, path) {
298            Ok(()) => {}
299            Err(error) => {
300                let _ = std::fs::remove_file(&staging);
301                return Err(error);
302            }
303        }
304        if let Ok(dir) = OpenOptions::new().read(true).open(parent) {
305            let _ = dir.sync_all();
306        }
307        Ok(())
308    }
309
310    /// Read a checkpoint file, validating magic, CRC32C, and that the stored
311    /// fingerprint equals the supplied `fingerprint`. Any mismatch is returned
312    /// as an `io::Error`; the caller may then fall back to `rebuild_from_runs`.
313    pub fn read_checkpoint(path: &Path, fingerprint: u64) -> io::Result<Self> {
314        let mut file = File::open(path)?;
315        let len = file.metadata()?.len();
316        const MAX_CHECKPOINT_BYTES: u64 = 64 * 1024 * 1024;
317        if len < 4 + 8 + 4 + FOOTER_LEN as u64 || len > MAX_CHECKPOINT_BYTES {
318            return Err(io::Error::new(
319                io::ErrorKind::InvalidData,
320                "run-lookup checkpoint length is out of range",
321            ));
322        }
323        let mut bytes = Vec::with_capacity(len as usize);
324        file.read_to_end(&mut bytes)?;
325        let split = bytes.len() - FOOTER_LEN;
326        let (body, footer) = bytes.split_at(split);
327        let expected = u32::from_le_bytes(footer.try_into().unwrap());
328        let actual = CRC32C.checksum(body);
329        if expected != actual {
330            return Err(io::Error::new(
331                io::ErrorKind::InvalidData,
332                "run-lookup checkpoint CRC mismatch",
333            ));
334        }
335        if body.len() < 4 + 8 + 4 {
336            return Err(io::Error::new(
337                io::ErrorKind::InvalidData,
338                "run-lookup checkpoint body too short",
339            ));
340        }
341        if body[..4] != LOOKUP_MAGIC {
342            return Err(io::Error::new(
343                io::ErrorKind::InvalidData,
344                "run-lookup checkpoint magic mismatch",
345            ));
346        }
347        let stored_fp = u64::from_le_bytes(body[4..12].try_into().unwrap());
348        if stored_fp != fingerprint {
349            return Err(io::Error::new(
350                io::ErrorKind::InvalidData,
351                "run-lookup checkpoint fingerprint does not match active run set",
352            ));
353        }
354        let key_count = u32::from_le_bytes(body[12..16].try_into().unwrap());
355        let mut cursor = &body[16..];
356        let mut postings: BTreeMap<RowId, RunLocatorList> = BTreeMap::new();
357        let mut prev: u64 = 0;
358        for _ in 0..key_count {
359            let (delta, consumed) = decode_varint(cursor)?;
360            cursor = &cursor[consumed..];
361            let cur = prev.checked_add(delta).ok_or_else(|| {
362                io::Error::new(
363                    io::ErrorKind::InvalidData,
364                    "run-lookup checkpoint varint overflows u64",
365                )
366            })?;
367            if cur < prev {
368                return Err(io::Error::new(
369                    io::ErrorKind::InvalidData,
370                    "run-lookup checkpoint row_id sequence is not monotonic",
371                ));
372            }
373            if cursor.len() < 4 {
374                return Err(io::Error::new(
375                    io::ErrorKind::UnexpectedEof,
376                    "run-lookup checkpoint truncates locator count",
377                ));
378            }
379            let locator_count = u32::from_le_bytes(cursor[..4].try_into().unwrap()) as usize;
380            cursor = &cursor[4..];
381            let mut list = RunLocatorList {
382                locators: Vec::with_capacity(locator_count),
383            };
384            for _ in 0..locator_count {
385                // Per-locator fixed prefix: run_id(16) + min_epoch(8) +
386                // max_epoch(8) + min_hlc_presence(1) + max_hlc_presence(1)
387                // + contains_unstamped(1) = 35. The two 16-byte HLC bodies
388                // are checked separately via read_optional_hlc.
389                let need = 16 + 8 + 8 + 1 + 1 + 1;
390                if cursor.len() < need {
391                    return Err(io::Error::new(
392                        io::ErrorKind::UnexpectedEof,
393                        "run-lookup checkpoint truncates locator fields",
394                    ));
395                }
396                let run_id_bytes: [u8; 16] = cursor[..16].try_into().unwrap();
397                let run_id = u128::from_le_bytes(run_id_bytes);
398                cursor = &cursor[16..];
399                let min_epoch = u64::from_le_bytes(cursor[..8].try_into().unwrap());
400                cursor = &cursor[8..];
401                let max_epoch = u64::from_le_bytes(cursor[..8].try_into().unwrap());
402                cursor = &cursor[8..];
403                let (min_hlc, used) = read_optional_hlc(cursor)?;
404                cursor = &cursor[used..];
405                let (max_hlc, used) = read_optional_hlc(cursor)?;
406                cursor = &cursor[used..];
407                if cursor.is_empty() {
408                    return Err(io::Error::new(
409                        io::ErrorKind::UnexpectedEof,
410                        "run-lookup checkpoint truncates unstamped flag",
411                    ));
412                }
413                let contains_unstamped = cursor[0] != 0;
414                cursor = &cursor[1..];
415                let locator = RunLocator {
416                    run_id,
417                    min_epoch: Epoch(min_epoch),
418                    max_epoch: Epoch(max_epoch),
419                    min_hlc,
420                    max_hlc,
421                    contains_unstamped_versions: contains_unstamped,
422                };
423                if let Err(err) = locator.validate_metadata() {
424                    return Err(io::Error::new(
425                        io::ErrorKind::InvalidData,
426                        format!("run-lookup locator metadata invalid: {err:?}"),
427                    ));
428                }
429                list.locators.push(locator);
430            }
431            postings.insert(RowId(cur), list);
432            prev = cur;
433        }
434        if !cursor.is_empty() {
435            return Err(io::Error::new(
436                io::ErrorKind::InvalidData,
437                "run-lookup checkpoint has trailing bytes after declared key count",
438            ));
439        }
440        Ok(Self {
441            postings,
442            fingerprint: Some(fingerprint),
443        })
444    }
445
446    /// Rebuild the directory by scanning each run's system columns. One
447    /// `RunLocator` is emitted per `(RowId, run_id)` pair, aggregating
448    /// min/max epoch + HLC and the unstamped-row presence flag across every
449    /// version that run carries for that row.
450    ///
451    /// The implementation is the system-column-only iterator
452    /// ([`crate::sorted_run::RunReader::for_each_system`]); the full-row
453    /// fallback is intentionally absent so directory rebuilds never pay to
454    /// materialize user columns. See spec §8.4 step 4.
455    pub fn rebuild_from_runs(runs: &[RunRef], table: &Table) -> Result<Self> {
456        rebuild_from_runs_system_only(runs, table)
457    }
458
459    /// Persist the directory as a sequence of shard files in `base_dir`, one
460    /// per `RowId` range, bounded by `max_shard_bytes` each (default
461    /// [`DEFAULT_SHARD_BYTES`]). Any previous `directory.shard-*.bin` files
462    /// in `base_dir` are unlinked first so a torn-tail publication cannot
463    /// leave a stale shard from the previous run.
464    pub fn write_checkpoint_sharded(
465        &self,
466        base_dir: &Path,
467        max_shard_bytes: u64,
468    ) -> io::Result<()> {
469        let _ = self.fingerprint.ok_or_else(|| {
470            io::Error::new(
471                io::ErrorKind::InvalidData,
472                "run-lookup directory has no fingerprint; refusing to checkpoint",
473            )
474        })?;
475        // Best-effort: clear any prior shards before publishing new ones.
476        clear_shards(base_dir)?;
477        let max = max_shard_bytes.max(1024 * 1024); // sanity floor
478        let mut iter = self.postings.iter().peekable();
479        let mut shard_no: usize = 0;
480        while let Some((first_row_id, _)) = iter.peek() {
481            shard_no += 1;
482            let shard_start = first_row_id.0;
483            // The shard spans at least one entry. Determine a clean upper
484            // boundary in row_id space by accumulating entries until the
485            // projected body length would exceed `max`.
486            let mut shard_entries: Vec<(RowId, RunLocatorList)> = Vec::new();
487            let mut body_len: usize = 4 + 8 + 4; // magic + fp + key_count
488            while let Some((_row_id, list)) = iter.peek() {
489                let entry_cost = 1 + 8 // varint delta
490                    + 4 // locator count
491                    + list.locators.len()
492                        * (16 + 8 + 8 + 1 + 1 + 16 + 16 + 1); // worst case per locator
493                if !shard_entries.is_empty() && body_len + entry_cost > max as usize {
494                    break;
495                }
496                body_len += entry_cost;
497                let (rid, list) = iter.next().unwrap();
498                shard_entries.push((*rid, list.clone()));
499            }
500            let last = shard_entries
501                .last()
502                .map(|(rid, _)| rid.0)
503                .unwrap_or(shard_start);
504            let path = if iter.peek().is_some() {
505                directory_shard_path(base_dir, shard_start, Some(last))
506            } else {
507                directory_shard_path(base_dir, shard_start, None)
508            };
509            write_single_shard(&path, &shard_entries, self.fingerprint.unwrap())?;
510        }
511        if shard_no == 0 {
512            // Empty directory: still emit one zero-row shard so the open path
513            // can confirm the checkpoint was published atomically.
514            let path = directory_shard_path(base_dir, 0, None);
515            write_single_shard(&path, &[], self.fingerprint.unwrap())?;
516        }
517        // Final best-effort directory sync.
518        if let Ok(dir) = OpenOptions::new().read(true).open(base_dir) {
519            let _ = dir.sync_all();
520        }
521        Ok(())
522    }
523
524    /// Read the directory from the sharded layout in `base_dir`. Validates
525    /// every shard's magic, CRC, and fingerprint independently. Returns the
526    /// merged directory; a missing or unreadable shard is propagated as
527    /// `io::Error` so the caller can fall back to rebuild-from-runs.
528    pub fn read_checkpoint_sharded(base_dir: &Path, fingerprint: u64) -> io::Result<Self> {
529        let mut paths: Vec<PathBuf> = Vec::new();
530        for entry in std::fs::read_dir(base_dir)? {
531            let entry = entry?;
532            let name = entry.file_name();
533            let Some(name) = name.to_str() else {
534                continue;
535            };
536            // Shards use the `directory.shard-` prefix; the last shard is
537            // published as `directory.shard-<start>-open` (no `.bin` suffix)
538            // so the open boundary is unambiguous. Both layouts are matched
539            // here.
540            if name.starts_with(DIRECTORY_SHARD_PREFIX)
541                && (name.ends_with(DIRECTORY_SHARD_SUFFIX) || name.ends_with("-open"))
542            {
543                paths.push(entry.path());
544            }
545        }
546        paths.sort();
547        if paths.is_empty() {
548            return Err(io::Error::new(
549                io::ErrorKind::NotFound,
550                "no run-lookup directory shards present",
551            ));
552        }
553        let mut merged: BTreeMap<RowId, RunLocatorList> = BTreeMap::new();
554        for path in paths {
555            let shard = read_single_shard(&path, fingerprint)?;
556            for (rid, list) in shard {
557                merged.insert(rid, list);
558            }
559        }
560        Ok(Self {
561            postings: merged,
562            fingerprint: Some(fingerprint),
563        })
564    }
565}
566
567fn clear_shards(base_dir: &Path) -> io::Result<()> {
568    if !base_dir.exists() {
569        return Ok(());
570    }
571    for entry in std::fs::read_dir(base_dir)? {
572        let entry = entry?;
573        let name = entry.file_name();
574        let Some(name) = name.to_str() else {
575            continue;
576        };
577        if name.starts_with(DIRECTORY_SHARD_PREFIX)
578            && (name.ends_with(DIRECTORY_SHARD_SUFFIX) || name.ends_with("-open"))
579        {
580            let _ = std::fs::remove_file(entry.path());
581        }
582    }
583    Ok(())
584}
585
586fn write_single_shard(
587    path: &Path,
588    entries: &[(RowId, RunLocatorList)],
589    fingerprint: u64,
590) -> io::Result<()> {
591    let mut body: Vec<u8> = Vec::new();
592    body.extend_from_slice(&LOOKUP_MAGIC);
593    body.extend_from_slice(&fingerprint.to_le_bytes());
594    let key_count = entries.len() as u32;
595    body.extend_from_slice(&key_count.to_le_bytes());
596    let mut prev: u64 = 0;
597    for (row_id, list) in entries {
598        let cur = row_id.0;
599        let delta = cur.checked_sub(prev).ok_or_else(|| {
600            io::Error::new(
601                io::ErrorKind::InvalidData,
602                "run-lookup row_id sequence is not non-decreasing",
603            )
604        })?;
605        encode_varint(delta, &mut body);
606        let locator_count = list.locators.len() as u32;
607        body.extend_from_slice(&locator_count.to_le_bytes());
608        for loc in &list.locators {
609            body.extend_from_slice(&loc.run_id.to_le_bytes());
610            body.extend_from_slice(&loc.min_epoch.0.to_le_bytes());
611            body.extend_from_slice(&loc.max_epoch.0.to_le_bytes());
612            write_optional_hlc(loc.min_hlc, &mut body);
613            write_optional_hlc(loc.max_hlc, &mut body);
614            body.push(if loc.contains_unstamped_versions {
615                1
616            } else {
617                0
618            });
619        }
620        prev = cur;
621    }
622    let crc = CRC32C.checksum(&body);
623    body.extend_from_slice(&crc.to_le_bytes());
624
625    let parent = path.parent().unwrap_or_else(|| Path::new("."));
626    let staging = parent.join(format!(
627        ".{}.{}.{}.staging",
628        path.file_name()
629            .and_then(|name| name.to_str())
630            .unwrap_or("shard"),
631        std::process::id(),
632        std::time::SystemTime::now()
633            .duration_since(std::time::UNIX_EPOCH)
634            .map(|d| d.as_nanos())
635            .unwrap_or(0),
636    ));
637    {
638        let _ = std::fs::remove_file(&staging);
639        let mut file = OpenOptions::new()
640            .create(true)
641            .write(true)
642            .truncate(true)
643            .open(&staging)?;
644        file.write_all(&body)?;
645        file.sync_all()?;
646    }
647    match std::fs::rename(&staging, path) {
648        Ok(()) => {}
649        Err(error) => {
650            let _ = std::fs::remove_file(&staging);
651            return Err(error);
652        }
653    }
654    Ok(())
655}
656
657fn read_single_shard(path: &Path, fingerprint: u64) -> io::Result<BTreeMap<RowId, RunLocatorList>> {
658    let mut file = File::open(path)?;
659    let len = file.metadata()?.len();
660    // Shards are bounded by `max_shard_bytes`. A pathologically large shard
661    // is treated as corruption so the caller can fall back to rebuild.
662    if len < 4 + 8 + 4 + FOOTER_LEN as u64 || len > MAX_SHARD_BYTES * 4 {
663        return Err(io::Error::new(
664            io::ErrorKind::InvalidData,
665            "run-lookup shard length is out of range",
666        ));
667    }
668    let mut bytes = Vec::with_capacity(len as usize);
669    file.read_to_end(&mut bytes)?;
670    let split = bytes.len() - FOOTER_LEN;
671    let (body, footer) = bytes.split_at(split);
672    let expected = u32::from_le_bytes(footer.try_into().unwrap());
673    let actual = CRC32C.checksum(body);
674    if expected != actual {
675        return Err(io::Error::new(
676            io::ErrorKind::InvalidData,
677            "run-lookup shard CRC mismatch",
678        ));
679    }
680    if body.len() < 4 + 8 + 4 {
681        return Err(io::Error::new(
682            io::ErrorKind::InvalidData,
683            "run-lookup shard body too short",
684        ));
685    }
686    if body[..4] != LOOKUP_MAGIC {
687        return Err(io::Error::new(
688            io::ErrorKind::InvalidData,
689            "run-lookup shard magic mismatch",
690        ));
691    }
692    let stored_fp = u64::from_le_bytes(body[4..12].try_into().unwrap());
693    if stored_fp != fingerprint {
694        return Err(io::Error::new(
695            io::ErrorKind::InvalidData,
696            "run-lookup shard fingerprint does not match active run set",
697        ));
698    }
699    let key_count = u32::from_le_bytes(body[12..16].try_into().unwrap());
700    let mut cursor = &body[16..];
701    let mut postings: BTreeMap<RowId, RunLocatorList> = BTreeMap::new();
702    let mut prev: u64 = 0;
703    for _ in 0..key_count {
704        let (delta, consumed) = decode_varint(cursor)?;
705        cursor = &cursor[consumed..];
706        let cur = prev.checked_add(delta).ok_or_else(|| {
707            io::Error::new(
708                io::ErrorKind::InvalidData,
709                "run-lookup shard varint overflows u64",
710            )
711        })?;
712        if cur < prev {
713            return Err(io::Error::new(
714                io::ErrorKind::InvalidData,
715                "run-lookup shard row_id sequence is not monotonic",
716            ));
717        }
718        if cursor.len() < 4 {
719            return Err(io::Error::new(
720                io::ErrorKind::UnexpectedEof,
721                "run-lookup shard truncates locator count",
722            ));
723        }
724        let locator_count = u32::from_le_bytes(cursor[..4].try_into().unwrap()) as usize;
725        cursor = &cursor[4..];
726        let mut list = RunLocatorList {
727            locators: Vec::with_capacity(locator_count),
728        };
729        for _ in 0..locator_count {
730            let need = 16 + 8 + 8 + 1 + 1 + 1;
731            if cursor.len() < need {
732                return Err(io::Error::new(
733                    io::ErrorKind::UnexpectedEof,
734                    "run-lookup shard truncates locator fields",
735                ));
736            }
737            let run_id_bytes: [u8; 16] = cursor[..16].try_into().unwrap();
738            let run_id = u128::from_le_bytes(run_id_bytes);
739            cursor = &cursor[16..];
740            let min_epoch = u64::from_le_bytes(cursor[..8].try_into().unwrap());
741            cursor = &cursor[8..];
742            let max_epoch = u64::from_le_bytes(cursor[..8].try_into().unwrap());
743            cursor = &cursor[8..];
744            let (min_hlc, used) = read_optional_hlc(cursor)?;
745            cursor = &cursor[used..];
746            let (max_hlc, used) = read_optional_hlc(cursor)?;
747            cursor = &cursor[used..];
748            if cursor.is_empty() {
749                return Err(io::Error::new(
750                    io::ErrorKind::UnexpectedEof,
751                    "run-lookup shard truncates unstamped flag",
752                ));
753            }
754            let contains_unstamped = cursor[0] != 0;
755            cursor = &cursor[1..];
756            let locator = RunLocator {
757                run_id,
758                min_epoch: Epoch(min_epoch),
759                max_epoch: Epoch(max_epoch),
760                min_hlc,
761                max_hlc,
762                contains_unstamped_versions: contains_unstamped,
763            };
764            if let Err(err) = locator.validate_metadata() {
765                return Err(io::Error::new(
766                    io::ErrorKind::InvalidData,
767                    format!("run-lookup locator metadata invalid: {err:?}"),
768                ));
769            }
770            list.locators.push(locator);
771        }
772        postings.insert(RowId(cur), list);
773        prev = cur;
774    }
775    if !cursor.is_empty() {
776        return Err(io::Error::new(
777            io::ErrorKind::InvalidData,
778            "run-lookup shard has trailing bytes after declared key count",
779        ));
780    }
781    Ok(postings)
782}
783
784// PathBuf is referenced by `read_checkpoint_sharded`; import for clarity.
785use std::path::PathBuf;
786
787fn encode_varint(mut value: u64, out: &mut Vec<u8>) {
788    loop {
789        let mut byte = (value & 0x7F) as u8;
790        value >>= 7;
791        if value != 0 {
792            byte |= 0x80;
793        }
794        out.push(byte);
795        if value == 0 {
796            break;
797        }
798    }
799}
800
801fn decode_varint(input: &[u8]) -> io::Result<(u64, usize)> {
802    let mut value: u64 = 0;
803    let mut shift: u32 = 0;
804    let mut consumed = 0;
805    for byte in input.iter().take(10) {
806        consumed += 1;
807        let low = (byte & 0x7F) as u64;
808        value |= low.checked_shl(shift).ok_or_else(|| {
809            io::Error::new(
810                io::ErrorKind::InvalidData,
811                "run-lookup varint overflows u64",
812            )
813        })?;
814        shift += 7;
815        if byte & 0x80 == 0 {
816            return Ok((value, consumed));
817        }
818    }
819    Err(io::Error::new(
820        io::ErrorKind::InvalidData,
821        "run-lookup varint is truncated or too long",
822    ))
823}
824
825fn write_optional_hlc(ts: Option<HlcTimestamp>, out: &mut Vec<u8>) {
826    match ts {
827        None => out.push(0),
828        Some(ts) => {
829            out.push(1);
830            out.extend_from_slice(&ts.physical_micros.to_le_bytes());
831            out.extend_from_slice(&ts.logical.to_le_bytes());
832            out.extend_from_slice(&ts.node_tiebreaker.to_le_bytes());
833        }
834    }
835}
836
837fn read_optional_hlc(input: &[u8]) -> io::Result<(Option<HlcTimestamp>, usize)> {
838    if input.is_empty() {
839        return Err(io::Error::new(
840            io::ErrorKind::UnexpectedEof,
841            "run-lookup checkpoint truncates hlc presence flag",
842        ));
843    }
844    let present = input[0] != 0;
845    if !present {
846        return Ok((None, 1));
847    }
848    if input.len() < 1 + HLC_BYTES {
849        return Err(io::Error::new(
850            io::ErrorKind::UnexpectedEof,
851            "run-lookup checkpoint truncates hlc body",
852        ));
853    }
854    let physical_micros = u64::from_le_bytes(input[1..9].try_into().unwrap());
855    let logical = u32::from_le_bytes(input[9..13].try_into().unwrap());
856    let node_tiebreaker = u32::from_le_bytes(input[13..17].try_into().unwrap());
857    Ok((
858        Some(HlcTimestamp {
859            physical_micros,
860            logical,
861            node_tiebreaker,
862        }),
863        1 + HLC_BYTES,
864    ))
865}
866
867/// `true` when `newer` is strictly newer than `older` by either epoch or
868/// HLC. Ties compare as `false` (preserve existing order).
869fn locator_is_newer(newer: RunLocator, older: RunLocator) -> bool {
870    if newer.max_epoch > older.max_epoch {
871        return true;
872    }
873    if newer.max_epoch < older.max_epoch {
874        return false;
875    }
876    match (newer.max_hlc, older.max_hlc) {
877        (Some(a), Some(b)) => a > b,
878        (Some(_), None) => true,
879        (None, Some(_)) => false,
880        (None, None) => false,
881    }
882}
883
884/// Why a [`RunLocator`]'s metadata is contradictory or incomplete (B468-08).
885#[derive(Debug, Clone, Copy, PartialEq, Eq)]
886pub enum RunLocatorMetadataError {
887    MinEpochAfterMaxEpoch,
888    PartialHlcEnvelope,
889    MinHlcAfterMaxHlc,
890    EmptyVersionEnvelope,
891}
892
893impl RunLocator {
894    /// Validate locator metadata invariants (B468-08).
895    ///
896    /// - `min_epoch <= max_epoch`
897    /// - HLC bounds are both present or both absent
898    /// - when both present: `min_hlc <= max_hlc`
899    /// - when both absent: `contains_unstamped_versions` must be true
900    pub fn validate_metadata(&self) -> std::result::Result<(), RunLocatorMetadataError> {
901        if self.min_epoch > self.max_epoch {
902            return Err(RunLocatorMetadataError::MinEpochAfterMaxEpoch);
903        }
904        match (self.min_hlc, self.max_hlc) {
905            (Some(min), Some(max)) => {
906                if min > max {
907                    return Err(RunLocatorMetadataError::MinHlcAfterMaxHlc);
908                }
909            }
910            (None, None) => {
911                if !self.contains_unstamped_versions {
912                    return Err(RunLocatorMetadataError::EmptyVersionEnvelope);
913                }
914            }
915            _ => return Err(RunLocatorMetadataError::PartialHlcEnvelope),
916        }
917        Ok(())
918    }
919
920    /// `true` when this locator *cannot* contain a version visible to
921    /// `snapshot`. Conservative: when in doubt, return `false` so the caller
922    /// opens the run. The proof is split into the stamped and unstamped
923    /// possibilities and never uses positional tuple fields (REM-A):
924    ///
925    /// - a stamped version may be visible when the locator's **minimum**
926    ///   HLC is at or below the snapshot (`min_hlc`, not `max_hlc` — a run
927    ///   can span the snapshot and still host an older visible version);
928    /// - an unstamped version may be visible when the locator's minimum
929    ///   epoch is at or below the snapshot epoch.
930    ///
931    /// Contradictory/malformed metadata never proves impossibility (B468-08).
932    pub fn is_impossible_for(&self, snapshot: Snapshot) -> bool {
933        if self.validate_metadata().is_err() {
934            return false;
935        }
936        let stamped_possible = self.may_have_visible_stamped_version(snapshot);
937        let unstamped_possible = self.may_have_visible_unstamped_version(snapshot);
938
939        !stamped_possible && !unstamped_possible
940    }
941
942    /// `true` when a stamped version inside this locator may be visible to
943    /// `snapshot`. HLC-authoritative snapshots compare HLC bounds; epoch-only
944    /// snapshots compare epoch bounds (stamped rows remain epoch-visible
945    /// under the dual-model rule). The locator's `min_epoch` is aggregated
946    /// across stamped and unstamped rows, so a false positive (opening a run
947    /// that could have been skipped) is possible — that is safe; a false
948    /// negative (skipping a run that contains a visible row) is not.
949    fn may_have_visible_stamped_version(&self, snapshot: Snapshot) -> bool {
950        let Some(min_hlc) = self.min_hlc else {
951            return false;
952        };
953
954        if snapshot.uses_hlc_authority() {
955            min_hlc <= snapshot.commit_ts
956        } else {
957            self.min_epoch <= snapshot.epoch
958        }
959    }
960
961    /// `true` when an unstamped version inside this locator may be visible to
962    /// `snapshot`. Unstamped visibility is epoch-based under both snapshot
963    /// modes.
964    fn may_have_visible_unstamped_version(&self, snapshot: Snapshot) -> bool {
965        self.contains_unstamped_versions && self.min_epoch <= snapshot.epoch
966    }
967
968    /// Safe early-stop proof for `Table::get`. Returns `true` when this
969    /// locator **might** still contain a version strictly newer than the
970    /// current winner `best` and visible to `snapshot`. Returns `false`
971    /// only when the locator provably cannot — the caller may then skip
972    /// opening this run and increment the early-stop counter.
973    ///
974    /// The proof asks the stamped and unstamped questions independently
975    /// (REM-B): a locator is skippable only when neither a stamped nor an
976    /// unstamped version inside it can beat `best`. The recency rule of
977    /// [`Snapshot::version_is_newer`] is honored exactly — HLC recency
978    /// applies whenever both candidates are stamped, regardless of whether
979    /// the snapshot itself is HLC-authoritative.
980    ///
981    /// Conservative: when the metadata is too sparse to prove impossibility,
982    /// returns `true` so the caller opens the run (correctness wins over the
983    /// optimization).
984    pub(crate) fn can_contain_version_newer_than(
985        &self,
986        best: VersionStamp,
987        snapshot: Snapshot,
988    ) -> bool {
989        // Unknown metadata: the locator records no usable HLC envelope and
990        // claims no unstamped versions, so neither proof below can establish
991        // anything. Contradictory-by-construction metadata must stay
992        // conservative — open the run.
993        if (self.min_hlc.is_none() || self.max_hlc.is_none()) && !self.contains_unstamped_versions {
994            return true;
995        }
996        self.stamped_version_may_beat(best, snapshot)
997            || self.unstamped_version_may_beat(best, snapshot)
998    }
999
1000    /// `true` when a **stamped** version inside this locator may be visible
1001    /// to `snapshot` and strictly newer than `best`.
1002    ///
1003    /// - Stamped candidate versus stamped best compares HLC — under an
1004    ///   HLC-authoritative snapshot visibility caps the locator's max HLC at
1005    ///   `snapshot.commit_ts`; under an epoch-only snapshot visibility is
1006    ///   epoch-based but recency is still HLC-based, so the proof only
1007    ///   requires some stamped row to be epoch-visible while the locator's
1008    ///   max HLC exceeds the best (the locator does not preserve the
1009    ///   per-version epoch↔HLC correlation, so this is conservative).
1010    /// - Stamped candidate versus unstamped best compares epoch. Under an
1011    ///   HLC-authoritative snapshot the candidate is HLC-visible regardless
1012    ///   of its local epoch, so `max_epoch` is **not** capped by
1013    ///   `snapshot.epoch`; under an epoch-only snapshot both visibility and
1014    ///   recency are epoch-based and the cap applies.
1015    fn stamped_version_may_beat(&self, best: VersionStamp, snapshot: Snapshot) -> bool {
1016        let (Some(min_hlc), Some(max_hlc)) = (self.min_hlc, self.max_hlc) else {
1017            return false;
1018        };
1019
1020        match best.hlc {
1021            Some(best_hlc) => {
1022                if snapshot.uses_hlc_authority() {
1023                    max_hlc.min(snapshot.commit_ts) > best_hlc
1024                } else {
1025                    self.min_epoch <= snapshot.epoch && max_hlc > best_hlc
1026                }
1027            }
1028            None => {
1029                if snapshot.uses_hlc_authority() {
1030                    min_hlc <= snapshot.commit_ts && self.max_epoch > best.epoch
1031                } else {
1032                    self.max_epoch.min(snapshot.epoch) > best.epoch
1033                }
1034            }
1035        }
1036    }
1037
1038    /// `true` when an **unstamped** version inside this locator may be
1039    /// visible to `snapshot` and strictly newer than `best`. One side of the
1040    /// comparison is unstamped, so recency uses epoch; unstamped visibility
1041    /// is also epoch-based under both snapshot modes. `max_epoch` includes
1042    /// stamped versions too, so this may be a false positive — it cannot
1043    /// produce a false negative.
1044    fn unstamped_version_may_beat(&self, best: VersionStamp, snapshot: Snapshot) -> bool {
1045        self.contains_unstamped_versions && self.max_epoch.min(snapshot.epoch) > best.epoch
1046    }
1047}
1048
1049/// Deterministic fingerprint of the run-set + schema/index generations that
1050/// produced this directory. `format_version` is the directory's own wire
1051/// version so a future format bump invalidates older checkpoints even if the
1052/// run set is unchanged. Uses xxh3 (already a workspace dependency) with a
1053/// fixed seed, so the value is reproducible across processes/restarts.
1054pub fn compute_fingerprint(
1055    run_ids_ordered: &[u128],
1056    schema_id: u64,
1057    index_generation: u64,
1058    run_generation: u64,
1059    format_version: u32,
1060) -> u64 {
1061    let mut buf: Vec<u8> = Vec::with_capacity(8 + 8 + 8 + 4 + run_ids_ordered.len() * (16 + 2));
1062    buf.extend_from_slice(&schema_id.to_le_bytes());
1063    buf.extend_from_slice(&index_generation.to_le_bytes());
1064    buf.extend_from_slice(&run_generation.to_le_bytes());
1065    buf.extend_from_slice(&format_version.to_le_bytes());
1066    // Include the run level alongside the run id so a compacted-down topology
1067    // (level 0 → level 1 promotion) doesn't accidentally share a fingerprint
1068    // with the pre-compaction set.
1069    for run in run_ids_ordered {
1070        let run_id_bytes = run.to_le_bytes();
1071        buf.extend_from_slice(&run_id_bytes);
1072    }
1073    // Stable, fixed seed (NOT process-local / time-based).
1074    xxhash_rust::xxh3::xxh3_64_with_seed(&buf, 0x9E37_79B1_854A_0001)
1075}
1076
1077/// Build the file path of a single sharded directory checkpoint for the
1078/// `RowId` range `[start, end_inclusive]`. `end_inclusive == None` means
1079/// the trailing shard (last open-ended range).
1080pub fn directory_shard_path(
1081    base_dir: &Path,
1082    start: u64,
1083    end_inclusive: Option<u64>,
1084) -> std::path::PathBuf {
1085    match end_inclusive {
1086        Some(end) => base_dir.join(format!(
1087            "{DIRECTORY_SHARD_PREFIX}{start:020}-{end:020}{DIRECTORY_SHARD_SUFFIX}"
1088        )),
1089        None => base_dir.join(format!("{DIRECTORY_SHARD_PREFIX}{start:020}-open")),
1090    }
1091}
1092
1093/// Build a `RunLookupDirectory` for a single sorted run without materializing
1094/// any user columns. Uses [`crate::sorted_run::RunReader::for_each_system`]
1095/// (a system-column-only iterator) and folds the result into one
1096/// [`RunLocator`] per `RowId` that the run carries. The caller is expected
1097/// to call this from inside an `Arc<Table>` or equivalent — `table` is
1098/// borrowed for `open_reader`.
1099pub fn rebuild_from_runs_system_only(runs: &[RunRef], table: &Table) -> Result<RunLookupDirectory> {
1100    let mut dir = RunLookupDirectory::empty();
1101    for run_ref in runs {
1102        let mut reader = table.open_reader(run_ref.run_id)?;
1103        // Per-row agg: min epoch, max epoch, min hlc, max hlc,
1104        // contains_unstamped.
1105        type RunRowAgg = (
1106            Epoch,
1107            Epoch,
1108            Option<HlcTimestamp>,
1109            Option<HlcTimestamp>,
1110            bool,
1111        );
1112        let mut per_row: BTreeMap<RowId, RunRowAgg> = BTreeMap::new();
1113        reader.for_each_system(|row_id, committed_epoch, commit_ts, deleted| {
1114            let entry = per_row.entry(row_id).or_insert((
1115                committed_epoch,
1116                committed_epoch,
1117                commit_ts,
1118                commit_ts,
1119                commit_ts.is_none(),
1120            ));
1121            if committed_epoch < entry.0 {
1122                entry.0 = committed_epoch;
1123            }
1124            if committed_epoch > entry.1 {
1125                entry.1 = committed_epoch;
1126            }
1127            match (commit_ts, entry.2) {
1128                (Some(ts), Some(prev)) if ts < prev => entry.2 = Some(ts),
1129                (Some(_), None) => entry.2 = commit_ts,
1130                _ => {}
1131            }
1132            match (commit_ts, entry.3) {
1133                (Some(ts), Some(prev)) if ts > prev => entry.3 = Some(ts),
1134                (Some(_), None) => entry.3 = commit_ts,
1135                _ => {}
1136            }
1137            if commit_ts.is_none() {
1138                entry.4 = true;
1139            }
1140            // Tombstones do not change the locator's min/max envelope (they
1141            // share the row's commit epoch/HLC with the live version that
1142            // preceded them), so the deleted flag is intentionally ignored
1143            // for envelope construction. The HOT/manifest already removes
1144            // tombstones that are superseded by a newer live version.
1145            let _ = deleted;
1146            Ok(())
1147        })?;
1148        for (row_id, (min_epoch, max_epoch, min_hlc, max_hlc, contains_unstamped)) in per_row {
1149            let locator = RunLocator {
1150                run_id: run_ref.run_id,
1151                min_epoch,
1152                max_epoch,
1153                min_hlc,
1154                max_hlc,
1155                contains_unstamped_versions: contains_unstamped,
1156            };
1157            // B468-08 §12.5: builder must not emit contradictory metadata.
1158            debug_assert!(
1159                locator.validate_metadata().is_ok(),
1160                "rebuild_from_runs_system_only produced invalid locator: {locator:?}"
1161            );
1162            if let Err(err) = locator.validate_metadata() {
1163                return Err(crate::error::MongrelError::InvalidArgument(format!(
1164                    "run-lookup builder emitted invalid locator metadata: {err:?}"
1165                )));
1166            }
1167            dir.insert(row_id, locator);
1168        }
1169    }
1170    Ok(dir)
1171}
1172
1173/// Format version of the on-disk directory. Bumped when the wire format
1174/// changes in a way that older readers cannot safely interpret.
1175pub const DIRECTORY_FORMAT_VERSION: u32 = 1;
1176/// Hard upper bound for a single shard, raised from the 64 MiB legacy limit so
1177/// very large tables (e.g. 256-run benchmarks) can checkpoint without
1178/// truncation. The reader enforces only a soft sanity check.
1179pub const MAX_SHARD_BYTES: u64 = 16 * 1024 * 1024;
1180
1181#[cfg(test)]
1182mod tests {
1183    use super::*;
1184    use tempfile::tempdir;
1185
1186    fn hlc_from_phys(physical_micros: u64) -> HlcTimestamp {
1187        HlcTimestamp {
1188            physical_micros,
1189            logical: 0,
1190            node_tiebreaker: 0,
1191        }
1192    }
1193
1194    fn loc(run_id: u128, epoch: u64, hlc: Option<u64>) -> RunLocator {
1195        RunLocator {
1196            run_id,
1197            min_epoch: Epoch(epoch),
1198            max_epoch: Epoch(epoch),
1199            min_hlc: hlc.map(hlc_from_phys),
1200            max_hlc: hlc.map(hlc_from_phys),
1201            contains_unstamped_versions: hlc.is_none(),
1202        }
1203    }
1204
1205    #[test]
1206    fn empty_directory_returns_empty_list() {
1207        let dir = RunLookupDirectory::empty();
1208        assert!(dir.locate(RowId(42)).is_empty());
1209        assert_eq!(dir.key_count(), 0);
1210    }
1211
1212    #[test]
1213    fn insert_orders_locators_newest_first() {
1214        let mut dir = RunLookupDirectory::empty();
1215        dir.insert(RowId(1), loc(0xA, 1, None));
1216        dir.insert(RowId(1), loc(0xB, 3, None));
1217        dir.insert(RowId(1), loc(0xC, 2, None));
1218        let list = dir.locate(RowId(1));
1219        let ids: Vec<u128> = list.locators.iter().map(|l| l.run_id).collect();
1220        assert_eq!(ids, vec![0xB, 0xC, 0xA]);
1221    }
1222
1223    #[test]
1224    fn retain_active_runs_drops_stale_locators() {
1225        let mut dir = RunLookupDirectory::empty();
1226        dir.insert(RowId(1), loc(10, 1, None));
1227        dir.insert(RowId(1), loc(20, 2, None));
1228        dir.insert(RowId(2), loc(20, 1, None));
1229        let removed = dir.retain_active_runs(&[10]);
1230        assert_eq!(removed, 2);
1231        assert!(dir.locate(RowId(1)).locators.iter().all(|l| l.run_id == 10));
1232        assert!(dir.locate(RowId(2)).is_empty());
1233    }
1234
1235    #[test]
1236    fn hlc_tie_break_resolves_above_epoch_only() {
1237        let mut dir = RunLookupDirectory::empty();
1238        dir.insert(RowId(1), loc(0xA, 5, None));
1239        dir.insert(RowId(1), loc(0xB, 5, Some(100)));
1240        let list = dir.locate(RowId(1));
1241        assert_eq!(list.locators[0].run_id, 0xB, "HLC newer wins on tie");
1242    }
1243
1244    #[test]
1245    fn memory_metrics_report_postings() {
1246        let mut dir = RunLookupDirectory::empty();
1247        dir.insert(RowId(1), loc(1, 1, None));
1248        dir.insert(RowId(1), loc(2, 2, None));
1249        dir.insert(RowId(2), loc(1, 1, None));
1250        assert_eq!(dir.total_locators(), 3);
1251        assert_eq!(dir.key_count(), 2);
1252        assert_eq!(dir.max_locators_per_row(), 2);
1253        assert!((dir.avg_locators_per_row() - 1.5).abs() < 1e-9);
1254    }
1255
1256    #[test]
1257    fn checkpoint_roundtrip_preserves_all_locators() {
1258        let dir = tempdir().unwrap();
1259        let mut src = RunLookupDirectory::empty();
1260        src.set_fingerprint(0x0102_0304_0506_0708);
1261        src.insert(RowId(7), loc(0x11, 1, None));
1262        src.insert(RowId(7), loc(0x22, 5, Some(100)));
1263        src.insert(RowId(8), loc(0x33, 3, Some(50)));
1264        src.insert(RowId(1_000_000), loc(0x44, 9, None));
1265        let path = dir.path().join("lookup.ckpt");
1266        src.write_checkpoint(&path).unwrap();
1267
1268        let restored = RunLookupDirectory::read_checkpoint(&path, 0x0102_0304_0506_0708).unwrap();
1269        assert_eq!(restored.fingerprint(), Some(0x0102_0304_0506_0708));
1270        assert_eq!(restored.key_count(), 3);
1271        assert_eq!(restored.total_locators(), 4);
1272        for (row_id, expected) in [
1273            (RowId(7), vec![0x22, 0x11]),
1274            (RowId(8), vec![0x33]),
1275            (RowId(1_000_000), vec![0x44]),
1276        ] {
1277            let actual: Vec<u128> = restored
1278                .locate(row_id)
1279                .locators
1280                .iter()
1281                .map(|l| l.run_id)
1282                .collect();
1283            assert_eq!(actual, expected, "row_id {row_id} locator order");
1284        }
1285        let list7 = restored.locate(RowId(7));
1286        assert_eq!(
1287            list7.locators[0].min_hlc.map(|t| t.physical_micros),
1288            Some(100)
1289        );
1290        assert!(list7.locators[1].contains_unstamped_versions);
1291    }
1292
1293    #[test]
1294    fn read_checkpoint_rejects_mismatched_fingerprint() {
1295        let dir = tempdir().unwrap();
1296        let mut src = RunLookupDirectory::empty();
1297        src.set_fingerprint(42);
1298        src.insert(RowId(1), loc(7, 1, None));
1299        let path = dir.path().join("lookup.ckpt");
1300        src.write_checkpoint(&path).unwrap();
1301        let err = RunLookupDirectory::read_checkpoint(&path, 43).unwrap_err();
1302        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
1303        assert!(err.to_string().contains("fingerprint"));
1304    }
1305
1306    fn stamped_loc(run_id: u128, epoch: u64, hlc: u64) -> RunLocator {
1307        RunLocator {
1308            run_id,
1309            min_epoch: Epoch(epoch),
1310            max_epoch: Epoch(epoch),
1311            min_hlc: Some(hlc_from_phys(hlc)),
1312            max_hlc: Some(hlc_from_phys(hlc)),
1313            contains_unstamped_versions: false,
1314        }
1315    }
1316
1317    fn mixed_loc(run_id: u128, epoch: u64, hlc: u64) -> RunLocator {
1318        RunLocator {
1319            run_id,
1320            min_epoch: Epoch(epoch),
1321            max_epoch: Epoch(epoch),
1322            min_hlc: Some(hlc_from_phys(hlc)),
1323            max_hlc: Some(hlc_from_phys(hlc)),
1324            contains_unstamped_versions: true,
1325        }
1326    }
1327
1328    #[test]
1329    fn decide_distinguishes_complete_miss_from_candidates() {
1330        let mut dir = RunLookupDirectory::empty();
1331        // No insert for RowId(99): a complete miss.
1332        match dir.decide(RowId(99)) {
1333            DirectoryLookupDecision::CompleteMiss => {}
1334            other => panic!("expected CompleteMiss, got {other:?}"),
1335        }
1336        // Empty posting list is also a complete miss (defensive — should
1337        // not happen in production, but the contract still says miss).
1338        dir.insert(RowId(7), stamped_loc(1, 1, 100));
1339        dir.insert(RowId(7), stamped_loc(2, 2, 200));
1340        match dir.decide(RowId(7)) {
1341            DirectoryLookupDecision::Candidates(list) => assert_eq!(list.len(), 2),
1342            other => panic!("expected Candidates, got {other:?}"),
1343        }
1344    }
1345
1346    #[test]
1347    fn early_stop_epoch_only_proves_no_beat() {
1348        // Locator max_epoch = 5; best = epoch 10; cannot beat.
1349        let l = loc(1, 5, None);
1350        let snap = Snapshot::at(Epoch(20));
1351        let best = VersionStamp {
1352            epoch: Epoch(10),
1353            hlc: None,
1354        };
1355        assert!(
1356            !l.can_contain_version_newer_than(best, snap),
1357            "max_epoch <= best.epoch must early-stop"
1358        );
1359    }
1360
1361    #[test]
1362    fn early_stop_epoch_only_proves_beat() {
1363        // Locator max_epoch = 15; best = epoch 10; visible cap 20.
1364        let l = loc(1, 15, None);
1365        let snap = Snapshot::at(Epoch(20));
1366        let best = VersionStamp {
1367            epoch: Epoch(10),
1368            hlc: None,
1369        };
1370        assert!(
1371            l.can_contain_version_newer_than(best, snap),
1372            "max_epoch > best.epoch must not early-stop"
1373        );
1374    }
1375
1376    #[test]
1377    fn early_stop_pure_hlc_proves_no_beat() {
1378        // Locator max HLC = 100; best HLC = 200; snapshot HLC = 1000.
1379        // Visible cap = min(100, 1000) = 100 <= best.hlc=200 → no beat.
1380        let l = stamped_loc(1, 5, 100);
1381        let snap = Snapshot::at_hlc(Epoch(20), hlc_from_phys(1000));
1382        let best = VersionStamp {
1383            epoch: Epoch(5),
1384            hlc: Some(hlc_from_phys(200)),
1385        };
1386        assert!(
1387            !l.can_contain_version_newer_than(best, snap),
1388            "pure HLC: max visible HLC <= best HLC must early-stop"
1389        );
1390    }
1391
1392    #[test]
1393    fn early_stop_pure_hlc_proves_beat() {
1394        // Locator max HLC = 500; best HLC = 200; snapshot HLC = 1000.
1395        // Visible cap = min(500, 1000) = 500 > best.hlc=200 → beat possible.
1396        let l = stamped_loc(1, 5, 500);
1397        let snap = Snapshot::at_hlc(Epoch(20), hlc_from_phys(1000));
1398        let best = VersionStamp {
1399            epoch: Epoch(5),
1400            hlc: Some(hlc_from_phys(200)),
1401        };
1402        assert!(
1403            l.can_contain_version_newer_than(best, snap),
1404            "pure HLC: max visible HLC > best HLC must not early-stop"
1405        );
1406    }
1407
1408    #[test]
1409    fn early_stop_pure_hlc_respects_snapshot_cap() {
1410        // Locator max HLC = 5000; best HLC = 200; snapshot HLC = 100.
1411        // Visibility caps at snapshot, so locator cannot contain anything
1412        // strictly newer than best AND visible. Note: this is the case
1413        // where `is_impossible_for` would skip the run entirely; the early-
1414        // stop proof must agree when best is already pinned.
1415        let l = stamped_loc(1, 5, 5000);
1416        let snap = Snapshot::at_hlc(Epoch(20), hlc_from_phys(100));
1417        let best = VersionStamp {
1418            epoch: Epoch(5),
1419            hlc: Some(hlc_from_phys(200)),
1420        };
1421        assert!(
1422            !l.can_contain_version_newer_than(best, snap),
1423            "snapshot HLC cap below locator max must early-stop when best > cap"
1424        );
1425    }
1426
1427    #[test]
1428    fn early_stop_mixed_with_unstamped_falls_back_to_epoch() {
1429        // Locator has stamped HLC=100 AND unstamped; best has high epoch
1430        // 200 but no HLC. Mixed comparison falls back to epoch. Even
1431        // though the locator's max HLC is below any sensible best, the
1432        // unstamped rows can beat on epoch alone.
1433        let l = mixed_loc(1, 250, 100);
1434        let snap = Snapshot::at_hlc(Epoch(300), hlc_from_phys(1000));
1435        let best = VersionStamp {
1436            epoch: Epoch(200),
1437            hlc: None,
1438        };
1439        assert!(
1440            l.can_contain_version_newer_than(best, snap),
1441            "mixed: unstamped epoch beat path must not early-stop"
1442        );
1443        // Inverse: best's epoch exceeds the locator's max.
1444        let l = mixed_loc(1, 100, 5000);
1445        let best = VersionStamp {
1446            epoch: Epoch(200),
1447            hlc: None,
1448        };
1449        assert!(
1450            !l.can_contain_version_newer_than(best, snap),
1451            "mixed: max epoch <= best.epoch must early-stop"
1452        );
1453    }
1454
1455    #[test]
1456    fn early_stop_hlc_snap_with_unstamped_best_uses_epoch_path() {
1457        // best has no HLC; even under HLC-authoritative snap, comparison
1458        // falls back to epoch (mixed comparison rule).
1459        let l = stamped_loc(1, 250, 999);
1460        let snap = Snapshot::at_hlc(Epoch(300), hlc_from_phys(1000));
1461        let best = VersionStamp {
1462            epoch: Epoch(200),
1463            hlc: None,
1464        };
1465        assert!(
1466            l.can_contain_version_newer_than(best, snap),
1467            "unstamped best: epoch path wins even under HLC-authority snap"
1468        );
1469    }
1470
1471    #[test]
1472    fn early_stop_unknown_metadata_stays_conservative() {
1473        // Spec §36.4: when locator metadata cannot prove a remaining run
1474        // is unable to beat the winner, open the run. Construct a locator
1475        // whose epoch bound is *equal* to the best — strict inequality is
1476        // required to prove the locator cannot beat, so a tie must stay
1477        // conservative.
1478        let l = RunLocator {
1479            run_id: 1,
1480            min_epoch: Epoch(10),
1481            max_epoch: Epoch(10),
1482            min_hlc: Some(hlc_from_phys(50)),
1483            max_hlc: Some(hlc_from_phys(50)),
1484            contains_unstamped_versions: false,
1485        };
1486        let snap = Snapshot::at_hlc(Epoch(20), hlc_from_phys(100));
1487        let best = VersionStamp {
1488            epoch: Epoch(10),
1489            hlc: Some(hlc_from_phys(50)),
1490        };
1491        // Strict equality is not a beat — but the proof is conservative
1492        // and never asserts a tie is a win.
1493        assert!(
1494            !l.can_contain_version_newer_than(best, snap),
1495            "strict tie (equal epoch+HLC) is not a beat; safe to early-stop"
1496        );
1497
1498        // Truly unknown: locator with no HLC bounds AND no
1499        // contains_unstamped flag — metadata is contradictory and we
1500        // cannot prove anything in either direction. The proof must stay
1501        // conservative and keep the run open.
1502        let l_unknown = RunLocator {
1503            run_id: 2,
1504            min_epoch: Epoch(10),
1505            max_epoch: Epoch(10),
1506            min_hlc: None,
1507            max_hlc: None,
1508            contains_unstamped_versions: false,
1509        };
1510        let best_lower = VersionStamp {
1511            epoch: Epoch(5),
1512            hlc: Some(hlc_from_phys(50)),
1513        };
1514        assert!(
1515            l_unknown.can_contain_version_newer_than(best_lower, snap),
1516            "unknown metadata with higher possible epoch must stay conservative"
1517        );
1518    }
1519
1520    // ------------------------------------------------------------------
1521    // REM-A (spec §5.6): the HLC impossibility proof must use `min_hlc`.
1522    // ------------------------------------------------------------------
1523
1524    #[test]
1525    fn locator_spanning_hlc_snapshot_is_not_impossible() {
1526        // One run carries an old visible version (HLC 100) and a newer
1527        // invisible version (HLC 1,000); the snapshot sits between them.
1528        let locator = RunLocator {
1529            run_id: 1,
1530            min_epoch: Epoch(10),
1531            max_epoch: Epoch(20),
1532            min_hlc: Some(hlc_from_phys(100)),
1533            max_hlc: Some(hlc_from_phys(1_000)),
1534            contains_unstamped_versions: false,
1535        };
1536        let snapshot = Snapshot::at_hlc(Epoch(15), hlc_from_phys(500));
1537        assert!(
1538            !locator.is_impossible_for(snapshot),
1539            "min_hlc 100 <= snapshot 500: the run spans the snapshot and must be opened"
1540        );
1541    }
1542
1543    #[test]
1544    fn locator_entirely_after_hlc_snapshot_is_impossible() {
1545        let locator = RunLocator {
1546            run_id: 1,
1547            min_epoch: Epoch(10),
1548            max_epoch: Epoch(20),
1549            min_hlc: Some(hlc_from_phys(600)),
1550            max_hlc: Some(hlc_from_phys(1_000)),
1551            contains_unstamped_versions: false,
1552        };
1553        let snapshot = Snapshot::at_hlc(Epoch(15), hlc_from_phys(500));
1554        assert!(
1555            locator.is_impossible_for(snapshot),
1556            "min_hlc 600 > snapshot 500: every stamped version is too new"
1557        );
1558    }
1559
1560    #[test]
1561    fn locator_entirely_before_hlc_snapshot_is_possible() {
1562        let locator = RunLocator {
1563            run_id: 1,
1564            min_epoch: Epoch(10),
1565            max_epoch: Epoch(20),
1566            min_hlc: Some(hlc_from_phys(100)),
1567            max_hlc: Some(hlc_from_phys(200)),
1568            contains_unstamped_versions: false,
1569        };
1570        let snapshot = Snapshot::at_hlc(Epoch(15), hlc_from_phys(500));
1571        assert!(
1572            !locator.is_impossible_for(snapshot),
1573            "locator entirely below the snapshot is obviously possible"
1574        );
1575    }
1576
1577    #[test]
1578    fn mixed_locator_is_impossible_only_when_both_paths_are_excluded() {
1579        // Stamped path open (min_hlc visible) even though the epoch bound
1580        // exceeds the snapshot epoch.
1581        let stamped_side_open = RunLocator {
1582            run_id: 1,
1583            min_epoch: Epoch(60),
1584            max_epoch: Epoch(70),
1585            min_hlc: Some(hlc_from_phys(100)),
1586            max_hlc: Some(hlc_from_phys(900)),
1587            contains_unstamped_versions: true,
1588        };
1589        let snap = Snapshot::at_hlc(Epoch(50), hlc_from_phys(500));
1590        assert!(
1591            !stamped_side_open.is_impossible_for(snap),
1592            "stamped min_hlc 100 <= 500 keeps the mixed locator possible"
1593        );
1594        // Unstamped path open (min_epoch visible) even though every HLC is
1595        // above the snapshot.
1596        let unstamped_side_open = RunLocator {
1597            run_id: 2,
1598            min_epoch: Epoch(40),
1599            max_epoch: Epoch(70),
1600            min_hlc: Some(hlc_from_phys(600)),
1601            max_hlc: Some(hlc_from_phys(900)),
1602            contains_unstamped_versions: true,
1603        };
1604        assert!(
1605            !unstamped_side_open.is_impossible_for(snap),
1606            "unstamped min_epoch 40 <= 50 keeps the mixed locator possible"
1607        );
1608        // Both paths excluded: min_hlc above the snapshot AND min_epoch
1609        // above the snapshot epoch.
1610        let fully_after = RunLocator {
1611            run_id: 3,
1612            min_epoch: Epoch(60),
1613            max_epoch: Epoch(70),
1614            min_hlc: Some(hlc_from_phys(600)),
1615            max_hlc: Some(hlc_from_phys(900)),
1616            contains_unstamped_versions: true,
1617        };
1618        assert!(
1619            fully_after.is_impossible_for(snap),
1620            "every stamped version is HLC-invisible and every unstamped version is epoch-invisible"
1621        );
1622    }
1623
1624    // ------------------------------------------------------------------
1625    // REM-B (spec §6.12): authority-matrix tests for the split
1626    // stamped/unstamped early-stop proof.
1627    // ------------------------------------------------------------------
1628
1629    /// Spec §6.5: a mixed locator (unstamped epoch 50 + stamped epoch 40 /
1630    /// HLC 300) must not be epoch-pruned against a stamped best (epoch 100 /
1631    /// HLC 200) — the stamped candidate wins by HLC.
1632    #[test]
1633    fn mixed_locator_stamped_candidate_can_beat_stamped_best_by_hlc() {
1634        let locator = RunLocator {
1635            run_id: 1,
1636            min_epoch: Epoch(40),
1637            max_epoch: Epoch(50),
1638            min_hlc: Some(hlc_from_phys(300)),
1639            max_hlc: Some(hlc_from_phys(300)),
1640            contains_unstamped_versions: true,
1641        };
1642        let snapshot = Snapshot::at_hlc(Epoch(200), hlc_from_phys(400));
1643        let best = VersionStamp {
1644            epoch: Epoch(100),
1645            hlc: Some(hlc_from_phys(200)),
1646        };
1647        assert!(
1648            locator.can_contain_version_newer_than(best, snapshot),
1649            "stamped candidate HLC 300 > best HLC 200; the locator must be opened"
1650        );
1651    }
1652
1653    /// Spec §6.6: under an epoch-only snapshot, two stamped candidates are
1654    /// still ordered by HLC — the epoch-only snapshot must not disable HLC
1655    /// recency.
1656    #[test]
1657    fn epoch_snapshot_stamped_candidate_can_beat_stamped_best_by_hlc() {
1658        let locator = stamped_loc(1, 50, 300);
1659        let snapshot = Snapshot::at(Epoch(200));
1660        let best = VersionStamp {
1661            epoch: Epoch(100),
1662            hlc: Some(hlc_from_phys(200)),
1663        };
1664        assert!(
1665            locator.can_contain_version_newer_than(best, snapshot),
1666            "stamped vs stamped recency is HLC-based even under an epoch-only snapshot"
1667        );
1668        // Negative cell: the locator's max HLC does not exceed the best.
1669        let older = stamped_loc(2, 50, 150);
1670        assert!(
1671            !older.can_contain_version_newer_than(best, snapshot),
1672            "max_hlc 150 <= best HLC 200 and no unstamped rows: provably cannot beat"
1673        );
1674        // Negative cell: no stamped row can be epoch-visible.
1675        let future = RunLocator {
1676            run_id: 3,
1677            min_epoch: Epoch(300),
1678            max_epoch: Epoch(400),
1679            min_hlc: Some(hlc_from_phys(300)),
1680            max_hlc: Some(hlc_from_phys(300)),
1681            contains_unstamped_versions: false,
1682        };
1683        assert!(
1684            !future.can_contain_version_newer_than(best, snapshot),
1685            "min_epoch 300 > snapshot epoch 200: no stamped row is epoch-visible"
1686        );
1687    }
1688
1689    /// Spec §6.7: a stamped candidate under HLC visibility is visible
1690    /// regardless of its local epoch, so its epoch must NOT be capped by
1691    /// `snapshot.epoch` when the best is unstamped.
1692    #[test]
1693    fn hlc_snapshot_stamped_candidate_can_beat_unstamped_best_by_epoch() {
1694        let locator = stamped_loc(1, 500, 200);
1695        let snapshot = Snapshot::at_hlc(Epoch(300), hlc_from_phys(300));
1696        let best = VersionStamp {
1697            epoch: Epoch(400),
1698            hlc: None,
1699        };
1700        assert!(
1701            locator.can_contain_version_newer_than(best, snapshot),
1702            "candidate HLC 200 <= snapshot 300 and epoch 500 > best epoch 400: must open"
1703        );
1704        // Negative cell: the stamped candidate is HLC-invisible.
1705        let invisible = stamped_loc(2, 500, 400);
1706        assert!(
1707            !invisible.can_contain_version_newer_than(best, snapshot),
1708            "min_hlc 400 > snapshot 300: no stamped version is visible"
1709        );
1710        // Negative cell: visible, but its epoch cannot beat the best.
1711        let older = stamped_loc(3, 350, 200);
1712        assert!(
1713            !older.can_contain_version_newer_than(best, snapshot),
1714            "max_epoch 350 <= best epoch 400: provably cannot beat"
1715        );
1716    }
1717
1718    /// Unstamped candidates use epoch visibility and epoch recency under
1719    /// both snapshot modes.
1720    #[test]
1721    fn unstamped_candidate_uses_epoch_visibility_under_hlc_snapshot() {
1722        let locator = loc(1, 250, None); // unstamped-only, epoch 250
1723        let snapshot = Snapshot::at_hlc(Epoch(300), hlc_from_phys(1_000));
1724        let best = VersionStamp {
1725            epoch: Epoch(200),
1726            hlc: Some(hlc_from_phys(50)),
1727        };
1728        assert!(
1729            locator.can_contain_version_newer_than(best, snapshot),
1730            "unstamped epoch 250 > best epoch 200 and visible at snapshot epoch 300"
1731        );
1732        // Visibility cap: the snapshot epoch hides the whole locator.
1733        let tight_snap = Snapshot::at_hlc(Epoch(240), hlc_from_phys(1_000));
1734        let tight_best = VersionStamp {
1735            epoch: Epoch(240),
1736            hlc: Some(hlc_from_phys(50)),
1737        };
1738        assert!(
1739            !locator.can_contain_version_newer_than(tight_best, tight_snap),
1740            "min(250, 240) = 240 <= best epoch 240: no visible unstamped version can beat"
1741        );
1742        // Epoch-only snapshot behaves identically for unstamped candidates.
1743        let epoch_snap = Snapshot::at(Epoch(300));
1744        assert!(
1745            locator.can_contain_version_newer_than(best, epoch_snap),
1746            "unstamped epoch comparison does not depend on the snapshot mode"
1747        );
1748    }
1749
1750    /// Metadata that cannot prove anything in either direction must keep the
1751    /// run open.
1752    #[test]
1753    fn unknown_metadata_is_conservative() {
1754        let best = VersionStamp {
1755            epoch: Epoch(5),
1756            hlc: Some(hlc_from_phys(50)),
1757        };
1758        let hlc_snap = Snapshot::at_hlc(Epoch(20), hlc_from_phys(100));
1759        let epoch_snap = Snapshot::at(Epoch(20));
1760        // Contradictory: claims fully stamped but records no HLC envelope.
1761        let no_envelope = RunLocator {
1762            run_id: 1,
1763            min_epoch: Epoch(10),
1764            max_epoch: Epoch(10),
1765            min_hlc: None,
1766            max_hlc: None,
1767            contains_unstamped_versions: false,
1768        };
1769        for snap in [hlc_snap, epoch_snap] {
1770            assert!(
1771                no_envelope.can_contain_version_newer_than(best, snap),
1772                "missing HLC envelope without the unstamped flag is unknown: stay conservative"
1773            );
1774        }
1775        // Partial envelope (min without max) is equally unprovable.
1776        let partial = RunLocator {
1777            min_hlc: Some(hlc_from_phys(10)),
1778            ..no_envelope
1779        };
1780        assert!(
1781            partial.can_contain_version_newer_than(best, hlc_snap),
1782            "a partial HLC envelope cannot prove impossibility: stay conservative"
1783        );
1784    }
1785
1786    /// Strict inequality is required everywhere: equal HLC or equal epoch
1787    /// never counts as newer.
1788    #[test]
1789    fn equal_authority_does_not_count_as_strictly_newer() {
1790        let hlc_snap = Snapshot::at_hlc(Epoch(20), hlc_from_phys(100));
1791        // Stamped vs stamped, equal HLC.
1792        let stamped = stamped_loc(1, 10, 50);
1793        let stamped_best = VersionStamp {
1794            epoch: Epoch(10),
1795            hlc: Some(hlc_from_phys(50)),
1796        };
1797        assert!(
1798            !stamped.can_contain_version_newer_than(stamped_best, hlc_snap),
1799            "equal HLC is not strictly newer"
1800        );
1801        // Unstamped vs unstamped, equal epoch.
1802        let unstamped = loc(2, 10, None);
1803        let unstamped_best = VersionStamp {
1804            epoch: Epoch(10),
1805            hlc: None,
1806        };
1807        assert!(
1808            !unstamped.can_contain_version_newer_than(unstamped_best, hlc_snap),
1809            "equal epoch is not strictly newer"
1810        );
1811        // Stamped vs unstamped best under an epoch-only snapshot, equal
1812        // epoch.
1813        let epoch_snap = Snapshot::at(Epoch(20));
1814        assert!(
1815            !stamped.can_contain_version_newer_than(unstamped_best, epoch_snap),
1816            "equal epoch is not strictly newer on the stamped-vs-unstamped path"
1817        );
1818        // Unstamped candidate vs stamped best, equal epoch.
1819        assert!(
1820            !unstamped.can_contain_version_newer_than(stamped_best, epoch_snap),
1821            "equal epoch is not strictly newer on the unstamped path"
1822        );
1823    }
1824
1825    #[test]
1826    fn missing_hlc_and_not_unstamped_is_conservative_for_is_impossible() {
1827        let snap = Snapshot::at(Epoch(20));
1828        let bad = RunLocator {
1829            run_id: 1,
1830            min_epoch: Epoch(1),
1831            max_epoch: Epoch(10),
1832            min_hlc: None,
1833            max_hlc: None,
1834            contains_unstamped_versions: false,
1835        };
1836        assert!(bad.validate_metadata().is_err());
1837        assert!(
1838            !bad.is_impossible_for(snap),
1839            "contradictory empty envelope must open the run"
1840        );
1841    }
1842
1843    #[test]
1844    fn only_min_hlc_present_is_rejected() {
1845        let loc = RunLocator {
1846            run_id: 1,
1847            min_epoch: Epoch(1),
1848            max_epoch: Epoch(2),
1849            min_hlc: Some(hlc_from_phys(10)),
1850            max_hlc: None,
1851            contains_unstamped_versions: false,
1852        };
1853        assert_eq!(
1854            loc.validate_metadata(),
1855            Err(RunLocatorMetadataError::PartialHlcEnvelope)
1856        );
1857        assert!(!loc.is_impossible_for(Snapshot::at(Epoch(5))));
1858    }
1859
1860    #[test]
1861    fn only_max_hlc_present_is_rejected() {
1862        let loc = RunLocator {
1863            run_id: 1,
1864            min_epoch: Epoch(1),
1865            max_epoch: Epoch(2),
1866            min_hlc: None,
1867            max_hlc: Some(hlc_from_phys(10)),
1868            contains_unstamped_versions: true,
1869        };
1870        assert_eq!(
1871            loc.validate_metadata(),
1872            Err(RunLocatorMetadataError::PartialHlcEnvelope)
1873        );
1874    }
1875
1876    #[test]
1877    fn min_hlc_greater_than_max_hlc_is_rejected() {
1878        let loc = RunLocator {
1879            run_id: 1,
1880            min_epoch: Epoch(1),
1881            max_epoch: Epoch(2),
1882            min_hlc: Some(hlc_from_phys(50)),
1883            max_hlc: Some(hlc_from_phys(10)),
1884            contains_unstamped_versions: false,
1885        };
1886        assert_eq!(
1887            loc.validate_metadata(),
1888            Err(RunLocatorMetadataError::MinHlcAfterMaxHlc)
1889        );
1890        assert!(!loc.is_impossible_for(Snapshot::at_hlc(Epoch(5), hlc_from_phys(20))));
1891    }
1892
1893    #[test]
1894    fn min_epoch_greater_than_max_epoch_is_rejected() {
1895        let loc = RunLocator {
1896            run_id: 1,
1897            min_epoch: Epoch(9),
1898            max_epoch: Epoch(2),
1899            min_hlc: None,
1900            max_hlc: None,
1901            contains_unstamped_versions: true,
1902        };
1903        assert_eq!(
1904            loc.validate_metadata(),
1905            Err(RunLocatorMetadataError::MinEpochAfterMaxEpoch)
1906        );
1907    }
1908
1909    #[test]
1910    fn valid_unstamped_only_locator_is_accepted() {
1911        let loc = RunLocator {
1912            run_id: 1,
1913            min_epoch: Epoch(1),
1914            max_epoch: Epoch(2),
1915            min_hlc: None,
1916            max_hlc: None,
1917            contains_unstamped_versions: true,
1918        };
1919        assert!(loc.validate_metadata().is_ok());
1920    }
1921
1922    #[test]
1923    fn checkpoint_with_invalid_locator_is_rejected() {
1924        // B468-08 §12.3/12.6: loader must reject contradictory locator metadata.
1925        let dir = tempfile::tempdir().unwrap();
1926        let mut src = RunLookupDirectory::empty();
1927        src.set_fingerprint(42);
1928        // Insert a valid unstamped-only locator, then mutate in-memory to the
1929        // contradictory empty envelope before writing the checkpoint.
1930        src.insert(
1931            RowId(1),
1932            RunLocator {
1933                run_id: 7,
1934                min_epoch: Epoch(1),
1935                max_epoch: Epoch(2),
1936                min_hlc: None,
1937                max_hlc: None,
1938                contains_unstamped_versions: true,
1939            },
1940        );
1941        src.postings.get_mut(&RowId(1)).unwrap().locators[0].contains_unstamped_versions = false;
1942        let path = dir.path().join("lookup-bad.ckpt");
1943        src.write_checkpoint(&path).unwrap();
1944        let err = RunLookupDirectory::read_checkpoint(&path, 42).unwrap_err();
1945        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
1946        assert!(
1947            err.to_string().contains("locator metadata invalid")
1948                || err.to_string().contains("EmptyVersionEnvelope"),
1949            "unexpected error: {err}"
1950        );
1951    }
1952}