1use 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
34pub const DIRECTORY_FILENAME: &str = "directory.bin";
37pub const DIRECTORY_SHARD_PREFIX: &str = "directory.shard-";
39pub const DIRECTORY_SHARD_SUFFIX: &str = ".bin";
40pub const DEFAULT_SHARD_BYTES: u64 = 16 * 1024 * 1024;
44
45const LOOKUP_MAGIC: [u8; 4] = *b"MLKP";
47const CRC32C: Crc<u32> = Crc::<u32>::new(&CRC_32_ISCSI);
49const FOOTER_LEN: usize = 4;
50const HLC_BYTES: usize = 16;
51
52#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56pub struct RunLocator {
57 pub run_id: u128,
59 pub min_epoch: Epoch,
60 pub max_epoch: Epoch,
61 pub min_hlc: Option<HlcTimestamp>,
63 pub max_hlc: Option<HlcTimestamp>,
64 pub contains_unstamped_versions: bool,
66}
67
68#[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#[derive(Debug, Clone)]
101pub(crate) enum DirectoryLookupDecision {
102 CompleteMiss,
103 Candidates(Vec<RunLocator>),
104 UnavailableOrStale,
105}
106
107#[derive(Debug, Clone, Copy, PartialEq, Eq)]
112pub(crate) struct VersionStamp {
113 pub epoch: Epoch,
114 pub hlc: Option<HlcTimestamp>,
115}
116
117#[derive(Debug, Default, Clone)]
120pub struct RunLookupDirectory {
121 postings: BTreeMap<RowId, RunLocatorList>,
123 fingerprint: Option<u64>,
127}
128
129impl RunLookupDirectory {
130 pub fn empty() -> Self {
132 Self::default()
133 }
134
135 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 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 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 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 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 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 pub fn total_locators(&self) -> usize {
200 self.postings.values().map(|l| l.locators.len()).sum()
201 }
202
203 pub fn key_count(&self) -> usize {
205 self.postings.len()
206 }
207
208 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 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 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 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 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 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 pub fn rebuild_from_runs(runs: &[RunRef], table: &Table) -> Result<Self> {
456 rebuild_from_runs_system_only(runs, table)
457 }
458
459 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 clear_shards(base_dir)?;
477 let max = max_shard_bytes.max(1024 * 1024); 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 let mut shard_entries: Vec<(RowId, RunLocatorList)> = Vec::new();
487 let mut body_len: usize = 4 + 8 + 4; while let Some((_row_id, list)) = iter.peek() {
489 let entry_cost = 1 + 8 + 4 + list.locators.len()
492 * (16 + 8 + 8 + 1 + 1 + 16 + 16 + 1); 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 let path = directory_shard_path(base_dir, 0, None);
515 write_single_shard(&path, &[], self.fingerprint.unwrap())?;
516 }
517 if let Ok(dir) = OpenOptions::new().read(true).open(base_dir) {
519 let _ = dir.sync_all();
520 }
521 Ok(())
522 }
523
524 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 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 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
784use 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
867fn 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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
886pub enum RunLocatorMetadataError {
887 MinEpochAfterMaxEpoch,
888 PartialHlcEnvelope,
889 MinHlcAfterMaxHlc,
890 EmptyVersionEnvelope,
891}
892
893impl RunLocator {
894 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 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 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 fn may_have_visible_unstamped_version(&self, snapshot: Snapshot) -> bool {
965 self.contains_unstamped_versions && self.min_epoch <= snapshot.epoch
966 }
967
968 pub(crate) fn can_contain_version_newer_than(
985 &self,
986 best: VersionStamp,
987 snapshot: Snapshot,
988 ) -> bool {
989 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 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 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
1049pub 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 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 xxhash_rust::xxh3::xxh3_64_with_seed(&buf, 0x9E37_79B1_854A_0001)
1075}
1076
1077pub 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
1093pub 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 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 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 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
1173pub const DIRECTORY_FORMAT_VERSION: u32 = 1;
1176pub 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 match dir.decide(RowId(99)) {
1333 DirectoryLookupDecision::CompleteMiss => {}
1334 other => panic!("expected CompleteMiss, got {other:?}"),
1335 }
1336 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 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 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 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 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 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 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 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 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 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 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 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 #[test]
1525 fn locator_spanning_hlc_snapshot_is_not_impossible() {
1526 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 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 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 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 #[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 #[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 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 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 #[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 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 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 #[test]
1721 fn unstamped_candidate_uses_epoch_visibility_under_hlc_snapshot() {
1722 let locator = loc(1, 250, None); 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 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 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 #[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 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 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 #[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 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 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 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 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 let dir = tempfile::tempdir().unwrap();
1926 let mut src = RunLookupDirectory::empty();
1927 src.set_fingerprint(42);
1928 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}