1use std::fs::File;
4#[cfg(unix)]
5use std::os::fd::AsRawFd;
6use std::path::Path;
7#[cfg(any(test, feature = "test-seams"))]
8use std::path::PathBuf;
9#[cfg(unix)]
10use std::ptr::NonNull;
11use std::sync::atomic::{AtomicU64, Ordering};
12use std::sync::{Arc, Mutex, OnceLock};
13
14use xxhash_rust::xxh3::xxh3_64;
15
16use crate::format::frame::{
17 FILE_HEADER_LEN, FILE_TRAILER_LEN, FormatCheck, FormatError, decode_header, read_u16, read_u32,
18 read_u64,
19};
20use crate::format::{FormatFamily, FormatRegistry};
21use crate::graph::block::{GraphNodeBlocks, ValidatedGraphNodeBlocks, decode_node_blocks};
22use crate::ingest::{DocId, DocumentVersion, Revision};
23use crate::meta::{AliveSet, ColumnStore};
24use crate::quant::Bit4Factors;
25use crate::vfs::Vfs;
26
27use super::layout::{
28 CHECKSUM_CHUNK_BYTES, Cursor, Int8Factors, REGION_ALIGNMENT, REGION_ENTRY_LEN, RegionEntry,
29 RegionKind, SEGMENT_PREFIX_LEN, VECTOR_HEADER_LEN, align_up, decode_alive, decode_columns,
30 decode_vector_header,
31};
32use super::{SegmentError, SegmentId, SegmentMeta};
33
34const MAX_SEGMENT_HEADER_BYTES: usize = 1024 * 1024;
35
36thread_local! {
37 static DATA_READ_AUDIT: std::cell::RefCell<Option<Arc<AtomicU64>>> = const {
38 std::cell::RefCell::new(None)
39 };
40}
41
42#[cfg(any(test, feature = "test-seams"))]
43thread_local! {
44 static SEGMENT_COST_AUDIT: std::cell::RefCell<Option<Arc<SegmentCostCounters>>> = const {
45 std::cell::RefCell::new(None)
46 };
47}
48
49#[cfg(any(test, feature = "test-seams"))]
50#[derive(Default)]
51struct SegmentCostCounters {
52 identity_hash_bytes: AtomicU64,
53 rescore_hash_bytes: AtomicU64,
54 int8_factor_decode_bytes: AtomicU64,
55 postings_hash_bytes: AtomicU64,
56 postings_decode_bytes: AtomicU64,
57 columns_hash_bytes: AtomicU64,
58 columns_decode_bytes: AtomicU64,
59 alive_hash_bytes: AtomicU64,
60 alive_decode_bytes: AtomicU64,
61}
62
63#[cfg(any(test, feature = "test-seams"))]
65#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
66pub struct SegmentCostSnapshot {
67 pub identity_hash_bytes: u64,
69 pub rescore_hash_bytes: u64,
71 pub int8_factor_decode_bytes: u64,
73 pub postings_hash_bytes: u64,
75 pub postings_decode_bytes: u64,
77 pub columns_hash_bytes: u64,
79 pub columns_decode_bytes: u64,
81 pub alive_hash_bytes: u64,
83 pub alive_decode_bytes: u64,
85}
86
87#[cfg(any(test, feature = "test-seams"))]
89#[derive(Clone, Default)]
90pub struct SegmentCostAudit {
91 counters: Arc<SegmentCostCounters>,
92}
93
94#[cfg(any(test, feature = "test-seams"))]
95struct SegmentCostAuditGuard {
96 previous: Option<Arc<SegmentCostCounters>>,
97}
98
99#[cfg(any(test, feature = "test-seams"))]
100impl Drop for SegmentCostAuditGuard {
101 fn drop(&mut self) {
102 let previous = self.previous.take();
103 SEGMENT_COST_AUDIT.with(|slot| {
104 slot.replace(previous);
105 });
106 }
107}
108
109#[cfg(any(test, feature = "test-seams"))]
110impl SegmentCostAudit {
111 #[must_use]
113 pub fn new() -> Self {
114 Self::default()
115 }
116
117 pub fn measure<T>(&self, operation: impl FnOnce() -> T) -> T {
119 let previous =
120 SEGMENT_COST_AUDIT.with(|slot| slot.replace(Some(Arc::clone(&self.counters))));
121 let _guard = SegmentCostAuditGuard { previous };
122 operation()
123 }
124
125 #[must_use]
127 pub fn snapshot(&self) -> SegmentCostSnapshot {
128 SegmentCostSnapshot {
129 identity_hash_bytes: self.counters.identity_hash_bytes.load(Ordering::Relaxed),
130 rescore_hash_bytes: self.counters.rescore_hash_bytes.load(Ordering::Relaxed),
131 int8_factor_decode_bytes: self
132 .counters
133 .int8_factor_decode_bytes
134 .load(Ordering::Relaxed),
135 postings_hash_bytes: self.counters.postings_hash_bytes.load(Ordering::Relaxed),
136 postings_decode_bytes: self.counters.postings_decode_bytes.load(Ordering::Relaxed),
137 columns_hash_bytes: self.counters.columns_hash_bytes.load(Ordering::Relaxed),
138 columns_decode_bytes: self.counters.columns_decode_bytes.load(Ordering::Relaxed),
139 alive_hash_bytes: self.counters.alive_hash_bytes.load(Ordering::Relaxed),
140 alive_decode_bytes: self.counters.alive_decode_bytes.load(Ordering::Relaxed),
141 }
142 }
143}
144
145#[cfg(any(test, feature = "test-seams"))]
146fn account_region_hash(kind: RegionKind, bytes: usize) {
147 let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
148 SEGMENT_COST_AUDIT.with(|slot| {
149 let Some(counters) = slot.borrow().as_ref().cloned() else {
150 return;
151 };
152 let counter = match kind {
153 RegionKind::DocumentVersions => &counters.identity_hash_bytes,
154 RegionKind::VectorRescore => &counters.rescore_hash_bytes,
155 RegionKind::Postings => &counters.postings_hash_bytes,
156 RegionKind::Columns => &counters.columns_hash_bytes,
157 RegionKind::Alive => &counters.alive_hash_bytes,
158 _ => return,
159 };
160 let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
161 Some(current.saturating_add(bytes))
162 });
163 });
164}
165
166#[cfg(any(test, feature = "test-seams"))]
167fn account_region_decode(kind: RegionKind, bytes: usize) {
168 let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
169 SEGMENT_COST_AUDIT.with(|slot| {
170 let Some(counters) = slot.borrow().as_ref().cloned() else {
171 return;
172 };
173 let counter = match kind {
174 RegionKind::VectorFactors => &counters.int8_factor_decode_bytes,
175 RegionKind::Postings => &counters.postings_decode_bytes,
176 RegionKind::Columns => &counters.columns_decode_bytes,
177 RegionKind::Alive => &counters.alive_decode_bytes,
178 _ => return,
179 };
180 let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
181 Some(current.saturating_add(bytes))
182 });
183 });
184}
185
186#[cfg(any(test, feature = "test-seams"))]
187pub(crate) fn account_active_postings_decode(bytes: usize) {
188 let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
189 SEGMENT_COST_AUDIT.with(|slot| {
190 if let Some(counters) = slot.borrow().as_ref() {
191 let _ = counters.postings_decode_bytes.fetch_update(
192 Ordering::Relaxed,
193 Ordering::Relaxed,
194 |current| Some(current.saturating_add(bytes)),
195 );
196 }
197 });
198}
199
200fn region_hash(_kind: RegionKind, bytes: &[u8]) -> u64 {
201 #[cfg(any(test, feature = "test-seams"))]
202 account_region_hash(_kind, bytes.len());
203 xxh3_64(bytes)
204}
205
206pub(crate) struct DataReadAuditGuard {
207 previous: Option<Arc<AtomicU64>>,
208}
209
210impl Drop for DataReadAuditGuard {
211 fn drop(&mut self) {
212 let previous = self.previous.take();
213 DATA_READ_AUDIT.with(|slot| {
214 slot.replace(previous);
215 });
216 }
217}
218
219pub(crate) fn install_data_read_audit(counter: Option<Arc<AtomicU64>>) -> DataReadAuditGuard {
220 let previous = DATA_READ_AUDIT.with(|slot| slot.replace(counter));
221 DataReadAuditGuard { previous }
222}
223
224fn account_data_read(bytes: usize) {
225 let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
226 DATA_READ_AUDIT.with(|slot| {
227 if let Some(counter) = slot.borrow().as_ref() {
228 let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
229 Some(current.saturating_add(bytes))
230 });
231 }
232 });
233}
234
235#[cfg(unix)]
236struct MappedFile {
237 _file: File,
238 pointer: NonNull<u8>,
239 length: usize,
240}
241
242#[cfg(unix)]
243impl MappedFile {
244 fn open(file: File, path: &Path) -> Result<Self, SegmentError> {
245 let length = usize::try_from(
246 file.metadata()
247 .map_err(|error| SegmentError::io(path, error))?
248 .len(),
249 )
250 .map_err(|_| SegmentError::Geometry("mapped file length exceeds usize".to_owned()))?;
251 if length == 0 {
252 return Err(FormatError::new(
253 path.display().to_string(),
254 FormatCheck::Length,
255 "empty segment file",
256 )
257 .into());
258 }
259 let mapped = unsafe {
263 libc::mmap(
264 std::ptr::null_mut(),
265 length,
266 libc::PROT_READ,
267 libc::MAP_PRIVATE,
268 file.as_raw_fd(),
269 0,
270 )
271 };
272 if mapped == libc::MAP_FAILED {
273 return Err(SegmentError::io(path, std::io::Error::last_os_error()));
274 }
275 let pointer = NonNull::new(mapped.cast::<u8>()).ok_or_else(|| {
276 SegmentError::Geometry("mmap returned a null non-failure pointer".to_owned())
277 })?;
278 Ok(Self {
279 _file: file,
280 pointer,
281 length,
282 })
283 }
284
285 fn as_bytes(&self) -> &[u8] {
286 unsafe { std::slice::from_raw_parts(self.pointer.as_ptr(), self.length) }
289 }
290
291 const fn length(&self) -> usize {
293 self.length
294 }
295
296 fn resident_bytes(&self) -> std::io::Result<u64> {
298 crate::sys::memory::mincore_resident_bytes(self.as_bytes())
299 }
300
301 #[cfg(any(test, feature = "test-seams"))]
307 fn corrupt_byte(&mut self, offset: usize, mask: u8) -> std::io::Result<()> {
308 if offset >= self.length {
309 return Err(std::io::Error::other(
310 "corruption offset is outside the mapping",
311 ));
312 }
313 let page_size_raw = unsafe {
314 libc::sysconf(libc::_SC_PAGESIZE)
316 };
317 let page_size = usize::try_from(page_size_raw)
318 .ok()
319 .filter(|size| *size != 0)
320 .ok_or_else(|| std::io::Error::other("sysconf returned an invalid page size"))?;
321 let page_offset = offset - offset % page_size;
322 let page = self
323 .pointer
324 .as_ptr()
325 .wrapping_add(page_offset)
326 .cast::<libc::c_void>();
327 if unsafe { libc::mprotect(page, page_size, libc::PROT_READ | libc::PROT_WRITE) } != 0 {
330 return Err(std::io::Error::last_os_error());
331 }
332 unsafe {
335 let byte = self.pointer.as_ptr().add(offset);
336 byte.write(byte.read() ^ mask);
337 }
338 if unsafe { libc::mprotect(page, page_size, libc::PROT_READ) } != 0 {
340 return Err(std::io::Error::last_os_error());
341 }
342 Ok(())
343 }
344}
345
346#[cfg(unix)]
347impl Drop for MappedFile {
348 fn drop(&mut self) {
349 let _ = unsafe { libc::munmap(self.pointer.as_ptr().cast(), self.length) };
352 }
353}
354
355#[cfg(unix)]
357unsafe impl Send for MappedFile {}
358#[cfg(unix)]
360unsafe impl Sync for MappedFile {}
361
362#[cfg(windows)]
370struct MappedFile {
371 inner: crate::sys::windows::FileMapping,
372}
373
374#[cfg(windows)]
375impl MappedFile {
376 fn open(file: File, path: &Path) -> Result<Self, SegmentError> {
377 let length = file
378 .metadata()
379 .map_err(|error| SegmentError::io(path, error))?
380 .len();
381 if length == 0 {
382 return Err(FormatError::new(
383 path.display().to_string(),
384 FormatCheck::Length,
385 "empty segment file",
386 )
387 .into());
388 }
389 let inner = crate::sys::windows::FileMapping::from_file(file, length)
390 .map_err(|error| SegmentError::io(path, error))?;
391 Ok(Self { inner })
392 }
393
394 fn as_bytes(&self) -> &[u8] {
395 self.inner.as_bytes()
396 }
397
398 const fn length(&self) -> usize {
400 self.inner.length()
401 }
402
403 fn resident_bytes(&self) -> std::io::Result<u64> {
406 self.inner.resident_bytes()
407 }
408
409 #[cfg(any(test, feature = "test-seams"))]
412 fn corrupt_byte(&mut self, offset: usize, mask: u8) -> std::io::Result<()> {
413 self.inner.corrupt_byte(offset, mask)
414 }
415}
416
417impl crate::graph::search::RescoreValidator for SegmentReader {
418 fn validate_rows(&self, rows: &[u32]) -> Result<(), String> {
419 self.validate_rescore_rows(rows).map_err(|source| {
420 format!(
421 "exact scores unavailable for segment {}: {source}",
422 self.meta.id
423 )
424 })
425 }
426}
427
428struct ParsedHeader {
429 meta: SegmentMeta,
430 header_length: usize,
431 entries: Vec<RegionEntry>,
432}
433
434pub struct SegmentReader {
436 mapping: MappedFile,
437 meta: SegmentMeta,
438 #[cfg(any(test, feature = "test-seams"))]
439 storage_store_directory: PathBuf,
440 header_length: usize,
441 entries: Vec<RegionEntry>,
442 document_versions_validation: OnceLock<Result<(), DocumentVersionValidationError>>,
443 stored_text_validated: OnceLock<()>,
447 stored_metadata_validated: OnceLock<()>,
450 vector_codes_validated: OnceLock<()>,
451 vector_factors_validated: OnceLock<()>,
452 rescore_valid_chunks: Mutex<Box<[u64]>>,
453 query_accounting: Option<Arc<crate::lifecycle::stats::Accounting>>,
454 int8_factors_cache: OnceLock<CachedQueryValue<Arc<Vec<crate::scan::Int8Factors>>>>,
455 postings_cache: OnceLock<CachedQueryValue<Arc<crate::fts::sealed::SealedSegment>>>,
456 columns_cache: OnceLock<CachedQueryValue<Arc<ColumnStore>>>,
457 collection_schema: Option<Arc<crate::meta::Schema>>,
461 alive_cache: OnceLock<CachedQueryValue<Arc<AliveSet>>>,
462 document_version_index_cache: OnceLock<CachedQueryValue<Arc<Box<[u32]>>>>,
465 pub(crate) graph_search_cache: crate::lifecycle::graph_cache::SegmentGraphSearchCache,
468 vector_ceiling_norm_range: OnceLock<crate::graph::search::GraphSegmentNormRange>,
473}
474
475struct CachedQueryValue<T> {
476 value: T,
477 _accounting: Option<crate::lifecycle::stats::AccountedCounter>,
478}
479
480#[derive(Clone, Debug)]
481enum DocumentVersionValidationError {
482 Geometry(String),
483 Checksum,
484}
485
486impl DocumentVersionValidationError {
487 fn into_segment_error(self, segment_id: SegmentId) -> SegmentError {
488 match self {
489 Self::Geometry(detail) => SegmentError::Geometry(detail),
490 Self::Checksum => FormatError::new(
491 format!("segment:{segment_id}:document-versions"),
492 FormatCheck::BlockChecksum,
493 "document-version region checksum mismatch",
494 )
495 .into(),
496 }
497 }
498}
499
500#[derive(Clone, Copy, Debug)]
502pub struct StoredMetadataRows<'a> {
503 offsets: &'a [u8],
504 bytes: &'a [u8],
505 row_count: usize,
506}
507
508impl<'a> StoredMetadataRows<'a> {
509 #[must_use]
511 pub const fn row_count(self) -> usize {
512 self.row_count
513 }
514
515 #[must_use]
517 pub fn row(self, row: usize) -> Option<&'a [u8]> {
518 if row >= self.row_count {
519 return None;
520 }
521 let start_index = row.checked_mul(8)?;
522 let end_index = row.checked_add(1)?.checked_mul(8)?;
523 let start = self
524 .offsets
525 .get(start_index..start_index.checked_add(8)?)?
526 .try_into()
527 .ok()
528 .map(u64::from_le_bytes)
529 .and_then(|value| usize::try_from(value).ok())?;
530 let end = self
531 .offsets
532 .get(end_index..end_index.checked_add(8)?)?
533 .try_into()
534 .ok()
535 .map(u64::from_le_bytes)
536 .and_then(|value| usize::try_from(value).ok())?;
537 self.bytes.get(start..end)
538 }
539}
540
541#[derive(Clone, Copy, Debug)]
543pub struct StoredTextRows<'a> {
544 present: &'a [u8],
545 offsets: &'a [u8],
546 bytes: &'a [u8],
547 row_count: usize,
548}
549
550impl<'a> StoredTextRows<'a> {
551 #[must_use]
553 pub const fn row_count(self) -> usize {
554 self.row_count
555 }
556
557 #[must_use]
560 pub fn row(self, row: usize) -> Option<Option<&'a str>> {
561 if row >= self.row_count {
562 return None;
563 }
564 let end_index = row.checked_mul(8)?;
565 let end = self
566 .offsets
567 .get(end_index..end_index.checked_add(8)?)?
568 .try_into()
569 .ok()
570 .map(u64::from_le_bytes)
571 .and_then(|value| usize::try_from(value).ok())?;
572 let start = if row == 0 {
573 0
574 } else {
575 let start_index = row.checked_sub(1)?.checked_mul(8)?;
576 self.offsets
577 .get(start_index..start_index.checked_add(8)?)?
578 .try_into()
579 .ok()
580 .map(u64::from_le_bytes)
581 .and_then(|value| usize::try_from(value).ok())?
582 };
583 let byte = *self.present.get(row / 8)?;
584 if byte & (1 << (row % 8)) == 0 {
585 return (start == end).then_some(None);
586 }
587 let text = std::str::from_utf8(self.bytes.get(start..end)?).ok()?;
588 Some(Some(text))
589 }
590}
591
592static QUERY_CHECKSUMS: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(0);
595
596#[cfg(any(test, feature = "test-seams"))]
601pub fn set_query_checksum_verification(enabled: bool) {
602 QUERY_CHECKSUMS.store(
603 if enabled { 2 } else { 1 },
604 std::sync::atomic::Ordering::Relaxed,
605 );
606}
607
608impl SegmentReader {
609 pub fn open(vfs: &dyn Vfs, path: &Path, expected_id: SegmentId) -> Result<Self, SegmentError> {
611 let file = vfs
612 .open_for_map(path)
613 .map_err(|error| SegmentError::io(path, error))?;
614 let mapping = MappedFile::open(file, path)?;
615 let artifact = path.display().to_string();
616 let parsed = parse_segment_header(
617 &artifact,
618 mapping.as_bytes(),
619 mapping.length() as u64,
620 expected_id,
621 )?;
622 let rescore_valid_chunks = rescore_validation_words(&parsed.entries)?;
623 Ok(Self {
624 mapping,
625 meta: parsed.meta,
626 #[cfg(any(test, feature = "test-seams"))]
627 storage_store_directory: path.parent().unwrap_or(Path::new("")).to_path_buf(),
628 header_length: parsed.header_length,
629 entries: parsed.entries,
630 document_versions_validation: OnceLock::new(),
631 stored_text_validated: OnceLock::new(),
632 stored_metadata_validated: OnceLock::new(),
633 vector_codes_validated: OnceLock::new(),
634 vector_factors_validated: OnceLock::new(),
635 rescore_valid_chunks: Mutex::new(vec![0_u64; rescore_valid_chunks].into_boxed_slice()),
636 query_accounting: None,
637 int8_factors_cache: OnceLock::new(),
638 postings_cache: OnceLock::new(),
639 columns_cache: OnceLock::new(),
640 collection_schema: None,
641 alive_cache: OnceLock::new(),
642 document_version_index_cache: OnceLock::new(),
643 graph_search_cache: crate::lifecycle::graph_cache::SegmentGraphSearchCache::new(),
644 vector_ceiling_norm_range: OnceLock::new(),
645 })
646 }
647
648 pub(crate) fn open_accounted(
649 vfs: &dyn Vfs,
650 path: &Path,
651 expected: &SegmentMeta,
652 accounting: &Arc<crate::lifecycle::stats::Accounting>,
653 collection_schema: &Arc<crate::meta::Schema>,
654 mut before_reader_allocation: impl FnMut(usize) -> Result<(), crate::lifecycle::StoreError>,
655 ) -> Result<Self, crate::lifecycle::StoreError> {
656 let file = vfs
657 .open_for_map(path)
658 .map_err(|error| SegmentError::io(path, error))
659 .map_err(crate::lifecycle::StoreError::Segment)?;
660 let mapping =
661 MappedFile::open(file, path).map_err(crate::lifecycle::StoreError::Segment)?;
662 let artifact = path.display().to_string();
663 let region_count =
664 preflight_region_count(&artifact, mapping.as_bytes(), mapping.length() as u64)
665 .map_err(crate::lifecycle::StoreError::Segment)?;
666 let directory_bytes = region_count
667 .checked_mul(std::mem::size_of::<RegionEntry>())
668 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
669 needed: u64::MAX,
670 budget: u64::MAX,
671 component: "snapshot",
672 })?;
673 before_reader_allocation(directory_bytes)?;
674 let parsed = parse_segment_header(
675 &artifact,
676 mapping.as_bytes(),
677 mapping.length() as u64,
678 expected.id,
679 )
680 .map_err(crate::lifecycle::StoreError::Segment)?;
681 if !parsed.meta.same_segment_file(expected) {
682 return Err(crate::lifecycle::StoreError::Segment(
683 SegmentError::Geometry(format!(
684 "manifest metadata {expected:?}, mapped header metadata {:?}",
685 parsed.meta
686 )),
687 ));
688 }
689 let rescore_valid_chunks = rescore_validation_words(&parsed.entries)
690 .map_err(crate::lifecycle::StoreError::Segment)?;
691 let validation_bytes = rescore_valid_chunks
692 .checked_mul(std::mem::size_of::<u64>())
693 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
694 needed: u64::MAX,
695 budget: u64::MAX,
696 component: "snapshot",
697 })?;
698 before_reader_allocation(validation_bytes)?;
699 Ok(Self {
700 mapping,
701 meta: expected.clone(),
702 #[cfg(any(test, feature = "test-seams"))]
703 storage_store_directory: path.parent().unwrap_or(Path::new("")).to_path_buf(),
704 header_length: parsed.header_length,
705 entries: parsed.entries,
706 document_versions_validation: OnceLock::new(),
707 stored_text_validated: OnceLock::new(),
708 stored_metadata_validated: OnceLock::new(),
709 vector_codes_validated: OnceLock::new(),
710 vector_factors_validated: OnceLock::new(),
711 rescore_valid_chunks: Mutex::new(vec![0_u64; rescore_valid_chunks].into_boxed_slice()),
712 query_accounting: Some(Arc::clone(accounting)),
713 int8_factors_cache: OnceLock::new(),
714 postings_cache: OnceLock::new(),
715 columns_cache: OnceLock::new(),
716 collection_schema: Some(Arc::clone(collection_schema)),
717 alive_cache: OnceLock::new(),
718 document_version_index_cache: OnceLock::new(),
719 graph_search_cache: crate::lifecycle::graph_cache::SegmentGraphSearchCache::new(),
720 vector_ceiling_norm_range: OnceLock::new(),
721 })
722 }
723
724 #[must_use]
727 pub(crate) fn with_collection_schema(mut self, schema: Arc<crate::meta::Schema>) -> Self {
728 self.collection_schema = Some(schema);
729 self
730 }
731
732 #[must_use]
734 pub const fn meta(&self) -> &SegmentMeta {
735 &self.meta
736 }
737
738 #[must_use]
740 pub const fn header_length(&self) -> usize {
741 self.header_length
742 }
743
744 #[must_use]
746 pub(crate) const fn mapped_bytes(&self) -> usize {
747 self.mapping.length()
748 }
749
750 pub(crate) fn mapped_resident_bytes(&self) -> std::io::Result<u64> {
752 self.mapping.resident_bytes()
753 }
754
755 pub(crate) fn file_bytes(&self) -> &[u8] {
760 self.mapping.as_bytes()
761 }
762
763 #[cfg(test)]
764 pub(crate) fn retained_validation_bytes(&self) -> usize {
765 self.rescore_valid_chunks
766 .lock()
767 .map_or(0, |words| std::mem::size_of_val(words.as_ref()))
768 }
769
770 #[cfg(any(test, feature = "test-seams"))]
774 #[doc(hidden)]
775 pub fn corrupt_mapped_byte_for_test(&mut self, offset: usize, mask: u8) -> std::io::Result<()> {
776 self.mapping.corrupt_byte(offset, mask)
777 }
778
779 #[cfg(any(test, feature = "test-seams"))]
780 #[doc(hidden)]
781 pub fn retained_query_view_bytes(&self) -> u64 {
782 let int8_factors = self.int8_factors_cache.get().and_then(|cached| {
783 cached
784 .value
785 .capacity()
786 .checked_mul(std::mem::size_of::<crate::scan::Int8Factors>())
787 .and_then(arc_resident_bytes::<Vec<crate::scan::Int8Factors>>)
788 });
789 let postings = self.postings_cache.get().and_then(|cached| {
790 cached
791 .value
792 .resident_bytes()
793 .ok()
794 .and_then(arc_resident_bytes::<crate::fts::sealed::SealedSegment>)
795 });
796 let columns = self.columns_cache.get().and_then(|cached| {
797 cached
798 .value
799 .resident_bytes()
800 .and_then(arc_resident_bytes::<ColumnStore>)
801 });
802 let alive = self.alive_cache.get().and_then(|cached| {
803 cached
804 .value
805 .resident_bytes()
806 .and_then(arc_resident_bytes::<AliveSet>)
807 });
808 let document_version_index = self.document_version_index_cache.get().and_then(|cached| {
809 cached
810 .value
811 .len()
812 .checked_mul(std::mem::size_of::<u32>())
813 .and_then(arc_resident_bytes::<Box<[u32]>>)
814 });
815 int8_factors
816 .into_iter()
817 .chain(postings)
818 .chain(columns)
819 .chain(alive)
820 .chain(document_version_index)
821 .fold(0_u64, |total, bytes| {
822 total.saturating_add(u64::try_from(bytes).unwrap_or(u64::MAX))
823 })
824 }
825
826 #[must_use]
828 pub fn directory(&self) -> &[RegionEntry] {
829 &self.entries
830 }
831
832 #[must_use]
834 pub fn unknown_region_ids(&self) -> Vec<u16> {
835 self.entries
836 .iter()
837 .filter_map(|entry| {
838 RegionKind::from_id(entry.kind)
839 .is_none()
840 .then_some(entry.kind)
841 })
842 .collect()
843 }
844
845 pub fn region(&self, kind: RegionKind) -> Result<&[u8], SegmentError> {
847 let entry = self.entry(kind)?;
848 let bytes = self.region_slice(entry)?;
849 let actual = region_hash(kind, bytes);
850 if actual != entry.checksum {
851 #[cfg(any(test, feature = "test-seams"))]
852 crate::lifecycle::record_storage_segment_checksum_fault(
853 &self.storage_store_directory,
854 self.meta.id,
855 kind,
856 0,
857 entry.checksum,
858 actual,
859 );
860 return Err(FormatError::checksum_mismatch(
861 format!("segment:{}:{kind:?}", self.meta.id),
862 FormatCheck::BlockChecksum,
863 entry.checksum,
864 actual,
865 )
866 .into());
867 }
868 Ok(bytes)
869 }
870
871 pub(crate) fn region_by_id(&self, kind: u16) -> Result<&[u8], SegmentError> {
873 let entry = self
874 .entries
875 .iter()
876 .find(|entry| entry.kind == kind)
877 .ok_or_else(|| {
878 SegmentError::Geometry(format!("segment is missing region kind {kind}"))
879 })?;
880 let bytes = self.region_slice(entry)?;
881 let actual = xxh3_64(bytes);
882 if actual != entry.checksum {
883 return Err(FormatError::checksum_mismatch(
884 format!("segment:{}:region-{kind}", self.meta.id),
885 FormatCheck::BlockChecksum,
886 entry.checksum,
887 actual,
888 )
889 .into());
890 }
891 Ok(bytes)
892 }
893
894 pub fn region_chunk(&self, kind: RegionKind, chunk_index: u32) -> Result<&[u8], SegmentError> {
896 let entry = self.entry(kind)?;
897 let region = self.region_slice(entry)?;
898 let start = (chunk_index as usize)
899 .checked_mul(CHECKSUM_CHUNK_BYTES)
900 .ok_or_else(|| SegmentError::Geometry("chunk offset overflow".to_owned()))?;
901 let end = start.saturating_add(CHECKSUM_CHUNK_BYTES).min(region.len());
902 let chunk = region.get(start..end).ok_or_else(|| {
903 SegmentError::Geometry(format!("chunk {chunk_index} is outside {kind:?}"))
904 })?;
905 if chunk.is_empty() {
906 return Err(SegmentError::Geometry(format!(
907 "chunk {chunk_index} is outside {kind:?}"
908 )));
909 }
910 let expected = self.chunk_checksum(kind, chunk_index)?;
911 let actual = region_hash(kind, chunk);
912 if actual != expected {
913 #[cfg(any(test, feature = "test-seams"))]
914 crate::lifecycle::record_storage_segment_checksum_fault(
915 &self.storage_store_directory,
916 self.meta.id,
917 kind,
918 chunk_index,
919 expected,
920 actual,
921 );
922 return Err(FormatError::checksum_mismatch(
923 format!("segment:{}:{kind:?}:chunk-{chunk_index}", self.meta.id),
924 FormatCheck::BlockChecksum,
925 expected,
926 actual,
927 )
928 .into());
929 }
930 Ok(chunk)
931 }
932
933 pub fn bit4_codes(&self) -> Result<&[u8], SegmentError> {
935 let (header, payload) = self.vector_payload(RegionKind::VectorCodes)?;
936 self.bit4_codes_from(header, payload)
937 }
938
939 fn bit4_codes_unchecked(&self) -> Result<&[u8], SegmentError> {
940 let (header, payload) = self.vector_payload_unchecked(RegionKind::VectorCodes)?;
941 self.bit4_codes_from(header, payload)
942 }
943
944 fn bit4_codes_from<'a>(
945 &'a self,
946 header: super::layout::VectorHeader,
947 payload: &'a [u8],
948 ) -> Result<&'a [u8], SegmentError> {
949 if self.meta.scheme != 4 {
950 return Err(SegmentError::Geometry(format!(
951 "Bit4 codes requested for scheme {}",
952 self.meta.scheme
953 )));
954 }
955 let expected = (self.meta.dims as usize)
956 .div_ceil(2)
957 .checked_mul(self.meta.row_count as usize)
958 .ok_or_else(|| SegmentError::Geometry("Bit4 code length overflow".to_owned()))?;
959 if header.row_stride_bytes as usize != (self.meta.dims as usize).div_ceil(2)
960 || payload.len() != expected
961 {
962 return Err(SegmentError::Geometry(format!(
963 "Bit4 code stride/length {}/{}, expected {}/{}",
964 header.row_stride_bytes,
965 payload.len(),
966 (self.meta.dims as usize).div_ceil(2),
967 expected
968 )));
969 }
970 Ok(payload)
971 }
972
973 pub fn f32_codes(&self) -> Result<&[f32], SegmentError> {
975 let (header, payload) = self.vector_payload(RegionKind::VectorCodes)?;
976 self.f32_codes_from(header, payload)
977 }
978
979 pub(crate) fn query_f32_codes(&self) -> Result<&[f32], SegmentError> {
981 self.validate_vector_region_once(RegionKind::VectorCodes, &self.vector_codes_validated)?;
982 let (header, payload) = self.vector_payload_unchecked(RegionKind::VectorCodes)?;
983 self.f32_codes_from(header, payload)
984 }
985
986 fn f32_codes_from<'a>(
987 &'a self,
988 header: super::layout::VectorHeader,
989 payload: &'a [u8],
990 ) -> Result<&'a [f32], SegmentError> {
991 if self.meta.scheme != 0 {
992 return Err(SegmentError::Geometry(format!(
993 "F32 codes requested for scheme {}",
994 self.meta.scheme
995 )));
996 }
997 let expected = (self.meta.dims as usize)
998 .checked_mul(self.meta.row_count as usize)
999 .ok_or_else(|| SegmentError::Geometry("F32 code length overflow".to_owned()))?;
1000 let expected_bytes = expected
1001 .checked_mul(std::mem::size_of::<f32>())
1002 .ok_or_else(|| SegmentError::Geometry("F32 code byte length overflow".to_owned()))?;
1003 let expected_stride = (self.meta.dims as usize)
1004 .checked_mul(std::mem::size_of::<f32>())
1005 .ok_or_else(|| SegmentError::Geometry("F32 code stride overflow".to_owned()))?;
1006 if header.row_stride_bytes as usize != expected_stride || payload.len() != expected_bytes {
1007 return Err(SegmentError::Geometry(format!(
1008 "F32 code stride/length {}/{}, expected {expected_stride}/{expected_bytes}",
1009 header.row_stride_bytes,
1010 payload.len()
1011 )));
1012 }
1013 cast_slice::<f32>(payload, expected, "F32 codes")
1014 }
1015
1016 pub fn int8_codes(&self) -> Result<&[i8], SegmentError> {
1018 let (header, payload) = self.vector_payload(RegionKind::VectorCodes)?;
1019 self.int8_codes_from(header, payload)
1020 }
1021
1022 fn int8_codes_unchecked(&self) -> Result<&[i8], SegmentError> {
1023 let (header, payload) = self.vector_payload_unchecked(RegionKind::VectorCodes)?;
1024 self.int8_codes_from(header, payload)
1025 }
1026
1027 fn int8_codes_from<'a>(
1028 &'a self,
1029 header: super::layout::VectorHeader,
1030 payload: &'a [u8],
1031 ) -> Result<&'a [i8], SegmentError> {
1032 if self.meta.scheme != 2 {
1033 return Err(SegmentError::Geometry(format!(
1034 "Int8 codes requested for scheme {}",
1035 self.meta.scheme
1036 )));
1037 }
1038 let expected = (self.meta.dims as usize)
1039 .checked_mul(self.meta.row_count as usize)
1040 .ok_or_else(|| SegmentError::Geometry("Int8 code length overflow".to_owned()))?;
1041 if header.row_stride_bytes as usize != self.meta.dims as usize || payload.len() != expected
1042 {
1043 return Err(SegmentError::Geometry(format!(
1044 "Int8 code stride/length {}/{}, expected {}/{}",
1045 header.row_stride_bytes,
1046 payload.len(),
1047 self.meta.dims,
1048 expected
1049 )));
1050 }
1051 cast_slice::<i8>(payload, expected, "Int8 codes")
1052 }
1053
1054 pub fn bit4_factors(&self) -> Result<&[Bit4Factors], SegmentError> {
1056 let (header, payload) = self.vector_payload(RegionKind::VectorFactors)?;
1057 self.bit4_factors_from(header, payload)
1058 }
1059
1060 fn bit4_factors_unchecked(&self) -> Result<&[Bit4Factors], SegmentError> {
1061 let (header, payload) = self.vector_payload_unchecked(RegionKind::VectorFactors)?;
1062 self.bit4_factors_from(header, payload)
1063 }
1064
1065 fn bit4_factors_from<'a>(
1066 &'a self,
1067 header: super::layout::VectorHeader,
1068 payload: &'a [u8],
1069 ) -> Result<&'a [Bit4Factors], SegmentError> {
1070 if self.meta.scheme != 4 {
1071 return Err(SegmentError::Geometry(format!(
1072 "Bit4 factors requested for scheme {}",
1073 self.meta.scheme
1074 )));
1075 }
1076 if header.factor_stride_bytes != 12 {
1077 return Err(SegmentError::Geometry(format!(
1078 "Bit4 factor stride {}, expected 12",
1079 header.factor_stride_bytes
1080 )));
1081 }
1082 cast_slice::<Bit4Factors>(payload, self.meta.row_count as usize, "Bit4 factors")
1083 }
1084
1085 pub fn int8_factors(&self) -> Result<&[Int8Factors], SegmentError> {
1087 let (header, payload) = self.vector_payload(RegionKind::VectorFactors)?;
1088 self.int8_factors_from(header, payload)
1089 }
1090
1091 fn int8_factors_unchecked(&self) -> Result<&[Int8Factors], SegmentError> {
1092 let (header, payload) = self.vector_payload_unchecked(RegionKind::VectorFactors)?;
1093 self.int8_factors_from(header, payload)
1094 }
1095
1096 fn int8_factors_from<'a>(
1097 &'a self,
1098 header: super::layout::VectorHeader,
1099 payload: &'a [u8],
1100 ) -> Result<&'a [Int8Factors], SegmentError> {
1101 if self.meta.scheme != 2 {
1102 return Err(SegmentError::Geometry(format!(
1103 "Int8 factors requested for scheme {}",
1104 self.meta.scheme
1105 )));
1106 }
1107 if header.factor_stride_bytes != 8 {
1108 return Err(SegmentError::Geometry(format!(
1109 "Int8 factor stride {}, expected 8",
1110 header.factor_stride_bytes
1111 )));
1112 }
1113 #[cfg(any(test, feature = "test-seams"))]
1114 account_region_decode(RegionKind::VectorFactors, payload.len());
1115 cast_slice::<Int8Factors>(payload, self.meta.row_count as usize, "Int8 factors")
1116 }
1117
1118 fn query_checksums_enabled() -> bool {
1128 match QUERY_CHECKSUMS.load(std::sync::atomic::Ordering::Relaxed) {
1129 0 => {
1130 let enabled =
1131 std::env::var_os("ZE_VERIFY_QUERY_CHECKSUMS").is_some_and(|value| value != "0");
1132 QUERY_CHECKSUMS.store(
1133 if enabled { 2 } else { 1 },
1134 std::sync::atomic::Ordering::Relaxed,
1135 );
1136 enabled
1137 }
1138 1 => false,
1139 _ => true,
1140 }
1141 }
1142
1143 fn validate_vector_region_once(
1153 &self,
1154 kind: RegionKind,
1155 gate: &OnceLock<()>,
1156 ) -> Result<(), SegmentError> {
1157 if gate.get().is_some() || !Self::query_checksums_enabled() {
1158 return Ok(());
1159 }
1160 let _ = self.region(kind)?;
1161 let _ = gate.set(());
1162 Ok(())
1163 }
1164
1165 pub(crate) fn query_bit4_codes(&self) -> Result<&[u8], SegmentError> {
1167 self.validate_vector_region_once(RegionKind::VectorCodes, &self.vector_codes_validated)?;
1168 self.bit4_codes_unchecked()
1169 }
1170
1171 pub(crate) fn query_int8_codes(&self) -> Result<&[i8], SegmentError> {
1173 self.validate_vector_region_once(RegionKind::VectorCodes, &self.vector_codes_validated)?;
1174 self.int8_codes_unchecked()
1175 }
1176
1177 pub(crate) fn query_int8_factors_slice(&self) -> Result<&[Int8Factors], SegmentError> {
1179 self.validate_vector_region_once(
1180 RegionKind::VectorFactors,
1181 &self.vector_factors_validated,
1182 )?;
1183 self.int8_factors_unchecked()
1184 }
1185
1186 pub(crate) fn cached_vector_ceiling_norm_range(
1192 &self,
1193 ) -> Option<crate::graph::search::GraphSegmentNormRange> {
1194 self.vector_ceiling_norm_range.get().copied()
1195 }
1196
1197 pub(crate) fn remember_vector_ceiling_norm_range(
1199 &self,
1200 range: crate::graph::search::GraphSegmentNormRange,
1201 ) {
1202 if self.vector_ceiling_norm_range.set(range).is_err() {
1203 }
1206 }
1207
1208 pub(crate) fn query_bit4_factors(&self) -> Result<&[Bit4Factors], SegmentError> {
1210 self.validate_vector_region_once(
1211 RegionKind::VectorFactors,
1212 &self.vector_factors_validated,
1213 )?;
1214 self.bit4_factors_unchecked()
1215 }
1216
1217 pub(crate) fn query_int8_factors(
1218 &self,
1219 ) -> Result<Arc<Vec<crate::scan::Int8Factors>>, crate::lifecycle::StoreError> {
1220 if let Some(cached) = self.int8_factors_cache.get() {
1221 return Ok(Arc::clone(&cached.value));
1222 }
1223 let factors = self
1224 .int8_factors()
1225 .map_err(crate::lifecycle::StoreError::Segment)?
1226 .iter()
1227 .map(|factor| crate::scan::Int8Factors::new(factor.scale, factor.offset))
1228 .collect::<Result<Vec<_>, _>>()
1229 .map_err(|_| {
1230 crate::lifecycle::StoreError::Segment(SegmentError::Geometry(
1231 "Int8 factor is not finite and non-negative".to_owned(),
1232 ))
1233 })?;
1234 let resident_bytes = factors
1235 .capacity()
1236 .checked_mul(std::mem::size_of::<crate::scan::Int8Factors>())
1237 .and_then(arc_resident_bytes::<Vec<crate::scan::Int8Factors>>)
1238 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
1239 needed: u64::MAX,
1240 budget: u64::MAX,
1241 component: "snapshot",
1242 })?;
1243 let cached = CachedQueryValue {
1244 value: Arc::new(factors),
1245 _accounting: self.account_query_cache(resident_bytes)?,
1246 };
1247 if self.int8_factors_cache.set(cached).is_err() {
1248 }
1250 self.int8_factors_cache
1251 .get()
1252 .map(|cached| Arc::clone(&cached.value))
1253 .ok_or(crate::lifecycle::StoreError::Synchronization {
1254 component: "Int8 factors query cache",
1255 })
1256 }
1257
1258 pub fn rescore_f32(&self) -> Result<&[f32], SegmentError> {
1260 let (header, payload) = self.vector_payload(RegionKind::VectorRescore)?;
1261 if header.scheme != 0 {
1262 return Err(SegmentError::Geometry(format!(
1263 "v1 rescore scheme must be F32/0, got {}",
1264 header.scheme
1265 )));
1266 }
1267 let count = (self.meta.row_count as usize)
1268 .checked_mul(self.meta.dims as usize)
1269 .ok_or_else(|| SegmentError::Geometry("rescore count overflow".to_owned()))?;
1270 cast_slice::<f32>(payload, count, "f32 rescore")
1271 }
1272
1273 pub(crate) fn query_rescore_f32(&self) -> Result<&[f32], SegmentError> {
1274 self.validate_rescore_byte_range(0, VECTOR_HEADER_LEN)?;
1275 let (header, payload) = self.vector_payload_unchecked(RegionKind::VectorRescore)?;
1276 if header.scheme != 0 {
1277 return Err(SegmentError::Geometry(format!(
1278 "v1 rescore scheme must be F32/0, got {}",
1279 header.scheme
1280 )));
1281 }
1282 let count = (self.meta.row_count as usize)
1283 .checked_mul(self.meta.dims as usize)
1284 .ok_or_else(|| SegmentError::Geometry("rescore count overflow".to_owned()))?;
1285 cast_slice::<f32>(payload, count, "f32 rescore")
1286 }
1287
1288 pub(crate) fn validate_rescore_rows(&self, rows: &[u32]) -> Result<(), SegmentError> {
1289 let row_bytes = (self.meta.dims as usize)
1290 .checked_mul(std::mem::size_of::<f32>())
1291 .ok_or_else(|| SegmentError::Geometry("rescore row byte length overflow".to_owned()))?;
1292 for row in rows {
1293 if *row >= self.meta.row_count {
1294 return Err(SegmentError::Geometry(format!(
1295 "rescore row {row} is outside {} rows",
1296 self.meta.row_count
1297 )));
1298 }
1299 let start = (*row as usize)
1300 .checked_mul(row_bytes)
1301 .and_then(|offset| offset.checked_add(VECTOR_HEADER_LEN))
1302 .ok_or_else(|| SegmentError::Geometry("rescore row offset overflow".to_owned()))?;
1303 self.validate_rescore_byte_range(start, row_bytes)?;
1304 }
1305 Ok(())
1306 }
1307
1308 pub fn columns(&self) -> Result<ColumnStore, SegmentError> {
1313 let region = self.region(RegionKind::Columns)?;
1314 #[cfg(any(test, feature = "test-seams"))]
1315 account_region_decode(RegionKind::Columns, region.len());
1316 let columns = decode_columns(region, self.collection_schema.as_deref())?;
1317 if columns.row_count() != self.meta.row_count {
1318 return Err(SegmentError::Geometry(format!(
1319 "column rows {}, header rows {}",
1320 columns.row_count(),
1321 self.meta.row_count
1322 )));
1323 }
1324 Ok(columns)
1325 }
1326
1327 pub(crate) fn query_columns(&self) -> Result<Arc<ColumnStore>, crate::lifecycle::StoreError> {
1328 if let Some(cached) = self.columns_cache.get() {
1329 return Ok(Arc::clone(&cached.value));
1330 }
1331 let region = self
1332 .region(RegionKind::Columns)
1333 .map_err(crate::lifecycle::StoreError::Segment)?;
1334 #[cfg(any(test, feature = "test-seams"))]
1335 account_region_decode(RegionKind::Columns, region.len());
1336 let mut columns = decode_columns(region, self.collection_schema.as_deref())
1337 .map_err(crate::lifecycle::StoreError::Segment)?;
1338 if columns.row_count() != self.meta.row_count {
1339 return Err(crate::lifecycle::StoreError::Segment(
1340 SegmentError::Geometry(format!(
1341 "column rows {}, header rows {}",
1342 columns.row_count(),
1343 self.meta.row_count
1344 )),
1345 ));
1346 }
1347 columns.compact_for_cache();
1348 let resident_bytes = columns
1349 .resident_bytes()
1350 .and_then(arc_resident_bytes::<ColumnStore>)
1351 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
1352 needed: u64::MAX,
1353 budget: u64::MAX,
1354 component: "snapshot",
1355 })?;
1356 let cached = CachedQueryValue {
1357 value: Arc::new(columns),
1358 _accounting: self.account_query_cache(resident_bytes)?,
1359 };
1360 if self.columns_cache.set(cached).is_err() {
1361 }
1363 self.columns_cache
1364 .get()
1365 .map(|cached| Arc::clone(&cached.value))
1366 .ok_or(crate::lifecycle::StoreError::Synchronization {
1367 component: "columns query cache",
1368 })
1369 }
1370
1371 pub fn postings(&self) -> Result<Option<crate::fts::sealed::SealedSegment>, SegmentError> {
1373 if !self
1374 .entries
1375 .iter()
1376 .any(|entry| entry.kind == RegionKind::Postings.id())
1377 {
1378 return Ok(None);
1379 }
1380 let region = self.region(RegionKind::Postings)?;
1381 #[cfg(any(test, feature = "test-seams"))]
1382 account_region_decode(RegionKind::Postings, region.len());
1383 let postings = crate::fts::sealed::SealedSegment::decode_region(region)?;
1384 if postings.row_count() != self.meta.row_count {
1385 return Err(SegmentError::Geometry(format!(
1386 "postings rows {}, header rows {}",
1387 postings.row_count(),
1388 self.meta.row_count
1389 )));
1390 }
1391 Ok(Some(postings))
1392 }
1393
1394 pub(crate) fn lexical_postings_identity(&self) -> Result<(u64, u64), SegmentError> {
1397 let entry = self.entry(RegionKind::Postings)?;
1398 Ok((entry.length, entry.checksum))
1399 }
1400
1401 pub(crate) fn query_postings(
1402 &self,
1403 ) -> Result<Option<Arc<crate::fts::sealed::SealedSegment>>, crate::lifecycle::StoreError> {
1404 if !self
1405 .entries
1406 .iter()
1407 .any(|entry| entry.kind == RegionKind::Postings.id())
1408 {
1409 return Ok(None);
1410 }
1411 if let Some(cached) = self.postings_cache.get() {
1412 return Ok(Some(Arc::clone(&cached.value)));
1413 }
1414 let region = self
1415 .region(RegionKind::Postings)
1416 .map_err(crate::lifecycle::StoreError::Segment)?;
1417 #[cfg(any(test, feature = "test-seams"))]
1418 account_region_decode(RegionKind::Postings, region.len());
1419 let postings = crate::fts::sealed::SealedSegment::decode_region(region)
1420 .map_err(SegmentError::from)
1421 .map_err(crate::lifecycle::StoreError::Segment)?;
1422 if postings.row_count() != self.meta.row_count {
1423 return Err(crate::lifecycle::StoreError::Segment(
1424 SegmentError::Geometry(format!(
1425 "postings rows {}, header rows {}",
1426 postings.row_count(),
1427 self.meta.row_count
1428 )),
1429 ));
1430 }
1431 let resident_bytes = postings
1432 .resident_bytes()
1433 .map_err(SegmentError::from)
1434 .map_err(crate::lifecycle::StoreError::Segment)?
1435 .checked_add(std::mem::size_of::<crate::fts::sealed::SealedSegment>())
1436 .and_then(|bytes| bytes.checked_add(2 * std::mem::size_of::<usize>()))
1437 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
1438 needed: u64::MAX,
1439 budget: u64::MAX,
1440 component: "snapshot",
1441 })?;
1442 let mut cache_accounting = self
1443 .query_accounting
1444 .as_ref()
1445 .map(|accounting| {
1446 crate::lifecycle::stats::AccountedCounter::new(
1447 accounting,
1448 crate::lifecycle::stats::AllocationComponent::Snapshot,
1449 )
1450 })
1451 .transpose()?;
1452 if let Some(counter) = cache_accounting.as_mut() {
1453 counter.set(resident_bytes)?;
1454 }
1455 let cached = CachedQueryValue {
1456 value: Arc::new(postings),
1457 _accounting: cache_accounting,
1458 };
1459 if self.postings_cache.set(cached).is_err() {
1460 }
1462 self.postings_cache
1463 .get()
1464 .map(|cached| Some(Arc::clone(&cached.value)))
1465 .ok_or(crate::lifecycle::StoreError::Synchronization {
1466 component: "postings query cache",
1467 })
1468 }
1469
1470 pub fn alive(&self) -> Result<AliveSet, SegmentError> {
1472 let region = self.region(RegionKind::Alive)?;
1473 #[cfg(any(test, feature = "test-seams"))]
1474 account_region_decode(RegionKind::Alive, region.len());
1475 let alive = decode_alive(region)?;
1476 if alive.row_count() != self.meta.row_count {
1477 return Err(SegmentError::Geometry(format!(
1478 "alive rows {}, header rows {}",
1479 alive.row_count(),
1480 self.meta.row_count
1481 )));
1482 }
1483 Ok(alive)
1484 }
1485
1486 pub(crate) fn query_alive(&self) -> Result<Arc<AliveSet>, crate::lifecycle::StoreError> {
1487 if let Some(cached) = self.alive_cache.get() {
1488 return Ok(Arc::clone(&cached.value));
1489 }
1490 let region = self
1491 .region(RegionKind::Alive)
1492 .map_err(crate::lifecycle::StoreError::Segment)?;
1493 #[cfg(any(test, feature = "test-seams"))]
1494 account_region_decode(RegionKind::Alive, region.len());
1495 let mut alive = decode_alive(region).map_err(crate::lifecycle::StoreError::Segment)?;
1496 if alive.row_count() != self.meta.row_count {
1497 return Err(crate::lifecycle::StoreError::Segment(
1498 SegmentError::Geometry(format!(
1499 "alive rows {}, header rows {}",
1500 alive.row_count(),
1501 self.meta.row_count
1502 )),
1503 ));
1504 }
1505 alive.compact_for_cache();
1506 let resident_bytes = alive
1507 .resident_bytes()
1508 .and_then(arc_resident_bytes::<AliveSet>)
1509 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
1510 needed: u64::MAX,
1511 budget: u64::MAX,
1512 component: "snapshot",
1513 })?;
1514 let cached = CachedQueryValue {
1515 value: Arc::new(alive),
1516 _accounting: self.account_query_cache(resident_bytes)?,
1517 };
1518 if self.alive_cache.set(cached).is_err() {
1519 }
1521 self.alive_cache
1522 .get()
1523 .map(|cached| Arc::clone(&cached.value))
1524 .ok_or(crate::lifecycle::StoreError::Synchronization {
1525 component: "alive query cache",
1526 })
1527 }
1528
1529 fn document_versions_region(&self) -> Result<Option<&[u8]>, SegmentError> {
1532 let Some(entry) = self
1533 .entries
1534 .iter()
1535 .find(|entry| entry.kind == RegionKind::DocumentVersions.id())
1536 else {
1537 return Ok(None);
1538 };
1539 let validation = self.document_versions_validation.get_or_init(|| {
1540 let bytes = self
1541 .region_slice(entry)
1542 .map_err(|error| DocumentVersionValidationError::Geometry(error.to_string()))?;
1543 if region_hash(RegionKind::DocumentVersions, bytes) != entry.checksum {
1544 return Err(DocumentVersionValidationError::Checksum);
1545 }
1546 let expected = (self.meta.row_count as usize)
1547 .checked_mul(24)
1548 .ok_or_else(|| {
1549 DocumentVersionValidationError::Geometry(
1550 "document-version length overflow".to_owned(),
1551 )
1552 })?;
1553 if bytes.len() != expected {
1554 return Err(DocumentVersionValidationError::Geometry(format!(
1555 "document-version bytes {}, expected {expected}",
1556 bytes.len()
1557 )));
1558 }
1559 Ok(())
1560 });
1561 if let Err(error) = validation {
1562 return Err(error.clone().into_segment_error(self.meta.id));
1563 }
1564 self.region_slice_unaccounted(entry).map(Some)
1565 }
1566
1567 pub fn document_version(&self, row: usize) -> Result<Option<DocumentVersion>, SegmentError> {
1569 let Some(bytes) = self.document_versions_region()? else {
1570 return Ok(None);
1571 };
1572 let start = row
1573 .checked_mul(24)
1574 .ok_or_else(|| SegmentError::Geometry("document-version row overflow".to_owned()))?;
1575 let doc_end = start
1576 .checked_add(16)
1577 .ok_or_else(|| SegmentError::Geometry("document id end overflow".to_owned()))?;
1578 let revision_end = doc_end
1579 .checked_add(8)
1580 .ok_or_else(|| SegmentError::Geometry("revision end overflow".to_owned()))?;
1581 let doc_id = bytes
1582 .get(start..doc_end)
1583 .and_then(|value| value.try_into().ok())
1584 .map(u128::from_le_bytes)
1585 .ok_or_else(|| SegmentError::Geometry(format!("document id row {row} is missing")))?;
1586 let revision = bytes
1587 .get(doc_end..revision_end)
1588 .and_then(|value| value.try_into().ok())
1589 .map(u64::from_le_bytes)
1590 .ok_or_else(|| SegmentError::Geometry(format!("revision row {row} is missing")))?;
1591 Ok(Some(DocumentVersion::new(
1592 DocId::new(doc_id),
1593 Revision::new(revision),
1594 )))
1595 }
1596
1597 pub fn row_for_document_version(
1602 &self,
1603 version: DocumentVersion,
1604 ) -> Result<Option<usize>, SegmentError> {
1605 let Some(bytes) = self.document_versions_region()? else {
1606 return Ok(None);
1607 };
1608 let needle = encode_document_version(version);
1609 Ok(bytes
1610 .chunks_exact(DOCUMENT_VERSION_BYTES)
1611 .position(|entry| entry == needle))
1612 }
1613
1614 pub(crate) fn query_row_for_document_version(
1619 &self,
1620 version: DocumentVersion,
1621 ) -> Result<Option<usize>, crate::lifecycle::StoreError> {
1622 let Some(bytes) = self
1623 .document_versions_region()
1624 .map_err(crate::lifecycle::StoreError::Segment)?
1625 else {
1626 return Ok(None);
1627 };
1628 let Some(index) = self.query_document_version_index(bytes)? else {
1629 return Ok(None);
1630 };
1631 let needle = encode_document_version(version);
1632 let position =
1633 index.partition_point(|&row| document_version_entry(bytes, row) < needle.as_slice());
1634 Ok(index
1635 .get(position)
1636 .copied()
1637 .filter(|&row| document_version_entry(bytes, row) == needle)
1638 .map(|row| row as usize))
1639 }
1640
1641 pub(crate) fn query_rows_for_doc_id(
1644 &self,
1645 doc_id: DocId,
1646 ) -> Result<Vec<usize>, crate::lifecycle::StoreError> {
1647 let matching = self.query_rows_with_doc_id(doc_id)?;
1648 let alive = self.query_alive()?;
1649 let mut rows = Vec::new();
1650 rows.try_reserve_exact(matching.len()).map_err(|_| {
1651 crate::lifecycle::StoreError::AllocationFailed {
1652 needed: matching
1653 .len()
1654 .checked_mul(std::mem::size_of::<usize>())
1655 .and_then(|bytes| u64::try_from(bytes).ok())
1656 .unwrap_or(u64::MAX),
1657 component: "document-id rows",
1658 }
1659 })?;
1660 for row in &matching {
1661 if alive.is_alive(*row) {
1662 rows.push(*row as usize);
1663 }
1664 }
1665 Ok(rows)
1666 }
1667
1668 pub(crate) fn query_rows_with_doc_id(
1672 &self,
1673 doc_id: DocId,
1674 ) -> Result<Vec<u32>, crate::lifecycle::StoreError> {
1675 let Some(bytes) = self
1676 .document_versions_region()
1677 .map_err(crate::lifecycle::StoreError::Segment)?
1678 else {
1679 return Ok(Vec::new());
1680 };
1681 let Some(index) = self.query_document_version_index(bytes)? else {
1682 return Ok(Vec::new());
1683 };
1684 let needle = doc_id.get().to_le_bytes();
1685 let start = index.partition_point(|&row| document_id_entry(bytes, row) < needle.as_slice());
1686 let end = index.partition_point(|&row| document_id_entry(bytes, row) <= needle.as_slice());
1687 let matching = index.get(start..end).ok_or_else(|| {
1688 crate::lifecycle::StoreError::Segment(SegmentError::Geometry(
1689 "document-id index range is invalid".to_owned(),
1690 ))
1691 })?;
1692 let mut rows = Vec::new();
1693 rows.try_reserve_exact(matching.len()).map_err(|_| {
1694 crate::lifecycle::StoreError::AllocationFailed {
1695 needed: matching
1696 .len()
1697 .checked_mul(std::mem::size_of::<u32>())
1698 .and_then(|bytes| u64::try_from(bytes).ok())
1699 .unwrap_or(u64::MAX),
1700 component: "document-id rows",
1701 }
1702 })?;
1703 rows.extend_from_slice(matching);
1704 rows.sort_unstable();
1705 Ok(rows)
1706 }
1707
1708 fn query_document_version_index(
1709 &self,
1710 bytes: &[u8],
1711 ) -> Result<Option<Arc<Box<[u32]>>>, crate::lifecycle::StoreError> {
1712 if let Some(cached) = self.document_version_index_cache.get() {
1713 return Ok(Some(Arc::clone(&cached.value)));
1714 }
1715 let mut rows: Vec<u32> = (0..self.meta.row_count).collect();
1716 rows.sort_unstable_by(|&left, &right| {
1717 document_version_entry(bytes, left)
1718 .cmp(document_version_entry(bytes, right))
1719 .then(left.cmp(&right))
1720 });
1721 let index = rows.into_boxed_slice();
1722 let resident_bytes = index
1723 .len()
1724 .checked_mul(std::mem::size_of::<u32>())
1725 .and_then(arc_resident_bytes::<Box<[u32]>>)
1726 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
1727 needed: u64::MAX,
1728 budget: u64::MAX,
1729 component: "snapshot",
1730 })?;
1731 let cached = CachedQueryValue {
1732 value: Arc::new(index),
1733 _accounting: self.account_query_cache(resident_bytes)?,
1734 };
1735 if self.document_version_index_cache.set(cached).is_err() {
1736 }
1738 self.document_version_index_cache
1739 .get()
1740 .map(|cached| Some(Arc::clone(&cached.value)))
1741 .ok_or(crate::lifecycle::StoreError::Synchronization {
1742 component: "document-version index cache",
1743 })
1744 }
1745
1746 pub fn stored_metadata(&self) -> Result<Option<StoredMetadataRows<'_>>, SegmentError> {
1748 let Some(entry) = self
1749 .entries
1750 .iter()
1751 .find(|entry| entry.kind == RegionKind::StoredMetadata.id())
1752 else {
1753 return Ok(None);
1754 };
1755 let region = self.region(RegionKind::StoredMetadata)?;
1756 let declared_rows = region
1757 .get(..4)
1758 .and_then(|value| value.try_into().ok())
1759 .map(u32::from_le_bytes)
1760 .ok_or_else(|| {
1761 SegmentError::Geometry("stored-metadata header is truncated".to_owned())
1762 })?;
1763 let reserved = region
1764 .get(4..8)
1765 .and_then(|value| value.try_into().ok())
1766 .map(u32::from_le_bytes)
1767 .ok_or_else(|| {
1768 SegmentError::Geometry("stored-metadata reserved field is truncated".to_owned())
1769 })?;
1770 if declared_rows != self.meta.row_count || reserved != 0 {
1771 return Err(SegmentError::Geometry(format!(
1772 "stored-metadata rows/reserved {declared_rows}/{reserved}, expected {}/0",
1773 self.meta.row_count
1774 )));
1775 }
1776 let row_count = declared_rows as usize;
1777 let offset_bytes = row_count
1778 .checked_add(1)
1779 .and_then(|count| count.checked_mul(8))
1780 .ok_or_else(|| SegmentError::Geometry("stored-metadata offsets overflow".to_owned()))?;
1781 let offsets_end = 8_usize.checked_add(offset_bytes).ok_or_else(|| {
1782 SegmentError::Geometry("stored-metadata offset end overflow".to_owned())
1783 })?;
1784 let offsets = region.get(8..offsets_end).ok_or_else(|| {
1785 SegmentError::Geometry("stored-metadata offsets are truncated".to_owned())
1786 })?;
1787 let bytes = region.get(offsets_end..).ok_or_else(|| {
1788 SegmentError::Geometry("stored-metadata payload is truncated".to_owned())
1789 })?;
1790 let view = StoredMetadataRows {
1791 offsets,
1792 bytes,
1793 row_count,
1794 };
1795 let first = view
1796 .offsets
1797 .get(..8)
1798 .and_then(|value| value.try_into().ok())
1799 .map(u64::from_le_bytes)
1800 .ok_or_else(|| {
1801 SegmentError::Geometry("stored-metadata first offset is missing".to_owned())
1802 })?;
1803 if first != 0 {
1804 return Err(SegmentError::Geometry(format!(
1805 "stored-metadata first offset is {first}, expected zero"
1806 )));
1807 }
1808 let mut previous = 0_usize;
1809 for row in 0..row_count {
1810 let value = view.row(row).ok_or_else(|| {
1811 SegmentError::Geometry(format!("stored-metadata row {row} is invalid"))
1812 })?;
1813 previous = previous.checked_add(value.len()).ok_or_else(|| {
1814 SegmentError::Geometry("stored-metadata length overflow".to_owned())
1815 })?;
1816 }
1817 if previous != bytes.len() {
1818 return Err(SegmentError::Geometry(format!(
1819 "stored-metadata final offset {previous}, bytes {}",
1820 bytes.len()
1821 )));
1822 }
1823 let _ = entry;
1824 Ok(Some(view))
1825 }
1826
1827 pub fn stored_text(&self) -> Result<Option<StoredTextRows<'_>>, SegmentError> {
1831 let Some(_) = self
1832 .entries
1833 .iter()
1834 .find(|entry| entry.kind == RegionKind::StoredText.id())
1835 else {
1836 return Ok(None);
1837 };
1838 let region = self.region(RegionKind::StoredText)?;
1839 let view = stored_text_view(region, self.meta.row_count)?;
1840 let StoredTextRows {
1841 offsets,
1842 bytes,
1843 row_count,
1844 ..
1845 } = view;
1846 let mut previous = 0_usize;
1847 for row in 0..row_count {
1848 let value = view.row(row).ok_or_else(|| {
1849 SegmentError::Geometry(format!("stored-text row {row} is invalid"))
1850 })?;
1851 let end_index = row.checked_mul(8).ok_or_else(|| {
1852 SegmentError::Geometry("stored-text offset index overflow".to_owned())
1853 })?;
1854 let end = offsets
1855 .get(end_index..end_index + 8)
1856 .and_then(|value| value.try_into().ok())
1857 .map(u64::from_le_bytes)
1858 .and_then(|value| usize::try_from(value).ok())
1859 .ok_or_else(|| {
1860 SegmentError::Geometry(format!("stored-text offset row {row} is invalid"))
1861 })?;
1862 if end < previous || end > bytes.len() {
1863 return Err(SegmentError::Geometry(format!(
1864 "stored-text offset row {row} is outside {previous}..={} bytes",
1865 bytes.len()
1866 )));
1867 }
1868 if value.is_none() && end != previous {
1869 return Err(SegmentError::Geometry(format!(
1870 "stored-text absent row {row} owns bytes {previous}..{end}"
1871 )));
1872 }
1873 previous = end;
1874 }
1875 if previous != bytes.len() {
1876 return Err(SegmentError::Geometry(format!(
1877 "stored-text final offset {previous}, bytes {}",
1878 bytes.len()
1879 )));
1880 }
1881 Ok(Some(view))
1882 }
1883
1884 pub(crate) fn query_stored_text(&self) -> Result<Option<StoredTextRows<'_>>, SegmentError> {
1890 if self.stored_text_validated.get().is_some() || !Self::query_checksums_enabled() {
1891 let Some(entry) = self
1892 .entries
1893 .iter()
1894 .find(|entry| entry.kind == RegionKind::StoredText.id())
1895 else {
1896 return Ok(None);
1897 };
1898 let region = self.region_slice(entry)?;
1899 return stored_text_view(region, self.meta.row_count).map(Some);
1900 }
1901 let view = self.stored_text()?;
1902 if view.is_some() {
1903 let _ = self.stored_text_validated.set(());
1905 }
1906 Ok(view)
1907 }
1908
1909 pub(crate) fn query_stored_metadata(
1912 &self,
1913 ) -> Result<Option<StoredMetadataRows<'_>>, SegmentError> {
1914 if self.stored_metadata_validated.get().is_some() || !Self::query_checksums_enabled() {
1915 let Some(entry) = self
1916 .entries
1917 .iter()
1918 .find(|entry| entry.kind == RegionKind::StoredMetadata.id())
1919 else {
1920 return Ok(None);
1921 };
1922 let region = self.region_slice(entry)?;
1923 return stored_metadata_view(region, self.meta.row_count).map(Some);
1924 }
1925 let view = self.stored_metadata()?;
1926 if view.is_some() {
1927 let _ = self.stored_metadata_validated.set(());
1928 }
1929 Ok(view)
1930 }
1931}
1932
1933fn stored_metadata_view(
1934 region: &[u8],
1935 header_rows: u32,
1936) -> Result<StoredMetadataRows<'_>, SegmentError> {
1937 let declared_rows = region
1938 .get(..4)
1939 .and_then(|value| value.try_into().ok())
1940 .map(u32::from_le_bytes)
1941 .ok_or_else(|| SegmentError::Geometry("stored-metadata header is truncated".to_owned()))?;
1942 let reserved = region
1943 .get(4..8)
1944 .and_then(|value| value.try_into().ok())
1945 .map(u32::from_le_bytes)
1946 .ok_or_else(|| {
1947 SegmentError::Geometry("stored-metadata reserved field is truncated".to_owned())
1948 })?;
1949 if declared_rows != header_rows || reserved != 0 {
1950 return Err(SegmentError::Geometry(format!(
1951 "stored-metadata rows/reserved {declared_rows}/{reserved}, expected {header_rows}/0"
1952 )));
1953 }
1954 let row_count = declared_rows as usize;
1955 let offset_bytes = row_count
1956 .checked_add(1)
1957 .and_then(|count| count.checked_mul(8))
1958 .ok_or_else(|| SegmentError::Geometry("stored-metadata offsets overflow".to_owned()))?;
1959 let offsets_end = 8_usize
1960 .checked_add(offset_bytes)
1961 .ok_or_else(|| SegmentError::Geometry("stored-metadata offset end overflow".to_owned()))?;
1962 let offsets = region.get(8..offsets_end).ok_or_else(|| {
1963 SegmentError::Geometry("stored-metadata offsets are truncated".to_owned())
1964 })?;
1965 let bytes = region
1966 .get(offsets_end..)
1967 .ok_or_else(|| SegmentError::Geometry("stored-metadata payload is truncated".to_owned()))?;
1968 Ok(StoredMetadataRows {
1969 offsets,
1970 bytes,
1971 row_count,
1972 })
1973}
1974
1975fn stored_text_view(region: &[u8], header_rows: u32) -> Result<StoredTextRows<'_>, SegmentError> {
1977 {
1978 let declared_rows = region
1979 .get(..4)
1980 .and_then(|value| value.try_into().ok())
1981 .map(u32::from_le_bytes)
1982 .ok_or_else(|| SegmentError::Geometry("stored-text header is truncated".to_owned()))?;
1983 let reserved = region
1984 .get(4..8)
1985 .and_then(|value| value.try_into().ok())
1986 .map(u32::from_le_bytes)
1987 .ok_or_else(|| {
1988 SegmentError::Geometry("stored-text reserved field is truncated".to_owned())
1989 })?;
1990 if declared_rows != header_rows || reserved != 0 {
1991 return Err(SegmentError::Geometry(format!(
1992 "stored-text rows/reserved {declared_rows}/{reserved}, expected {header_rows}/0"
1993 )));
1994 }
1995 let row_count = declared_rows as usize;
1996 let bitmap_len = row_count.div_ceil(8);
1997 let bitmap_end = 8_usize
1998 .checked_add(bitmap_len)
1999 .ok_or_else(|| SegmentError::Geometry("stored-text bitmap overflow".to_owned()))?;
2000 let offsets_len = row_count
2001 .checked_mul(8)
2002 .ok_or_else(|| SegmentError::Geometry("stored-text offsets overflow".to_owned()))?;
2003 let offsets_end = bitmap_end
2004 .checked_add(offsets_len)
2005 .ok_or_else(|| SegmentError::Geometry("stored-text offset end overflow".to_owned()))?;
2006 let present = region.get(8..bitmap_end).ok_or_else(|| {
2007 SegmentError::Geometry("stored-text presence bitmap is truncated".to_owned())
2008 })?;
2009 let offsets = region.get(bitmap_end..offsets_end).ok_or_else(|| {
2010 SegmentError::Geometry("stored-text offsets are truncated".to_owned())
2011 })?;
2012 let bytes = region
2013 .get(offsets_end..)
2014 .ok_or_else(|| SegmentError::Geometry("stored-text payload is truncated".to_owned()))?;
2015 if !row_count.is_multiple_of(8) {
2016 let used = row_count % 8;
2017 let padding_mask = !((1_u8 << used) - 1);
2018 if present.last().is_some_and(|byte| byte & padding_mask != 0) {
2019 return Err(SegmentError::Geometry(
2020 "stored-text presence padding bits are nonzero".to_owned(),
2021 ));
2022 }
2023 }
2024 Ok(StoredTextRows {
2025 present,
2026 offsets,
2027 bytes,
2028 row_count,
2029 })
2030 }
2031}
2032
2033impl SegmentReader {
2034 pub fn graph_node_blocks(&self) -> Result<GraphNodeBlocks<'_>, SegmentError> {
2036 let blocks = decode_node_blocks(self.region(RegionKind::GraphNodeBlocks)?)?;
2037 if blocks.layout().dims() != self.meta.dims {
2038 return Err(SegmentError::Geometry(format!(
2039 "graph dimensions {}, segment dimensions {}",
2040 blocks.layout().dims(),
2041 self.meta.dims
2042 )));
2043 }
2044 if blocks.node_count() != self.meta.row_count {
2045 return Err(SegmentError::Geometry(format!(
2046 "graph nodes {}, segment rows {}",
2047 blocks.node_count(),
2048 self.meta.row_count
2049 )));
2050 }
2051 Ok(blocks)
2052 }
2053
2054 pub(crate) fn bind_validated_graph_node_blocks(
2055 &self,
2056 descriptor: ValidatedGraphNodeBlocks,
2057 ) -> Result<GraphNodeBlocks<'_>, SegmentError> {
2058 let entry = self.entry(RegionKind::GraphNodeBlocks)?;
2059 let bytes = self.region_slice(entry)?;
2060 descriptor.bind(bytes).map_err(SegmentError::Graph)
2061 }
2062
2063 pub fn validate_all(&self) -> Result<(), SegmentError> {
2065 for entry in &self.entries {
2066 let bytes = self.region_slice(entry)?;
2067 let actual = xxh3_64(bytes);
2068 if actual != entry.checksum {
2069 return Err(FormatError::new(
2070 format!("segment:{}:region-{}", self.meta.id, entry.kind),
2071 FormatCheck::BlockChecksum,
2072 format!("expected {:#018x}, computed {actual:#018x}", entry.checksum),
2073 )
2074 .into());
2075 }
2076 }
2077 let bytes = self.mapping.as_bytes();
2078 let trailer_start = bytes.len().checked_sub(FILE_TRAILER_LEN).ok_or_else(|| {
2079 FormatError::new(
2080 format!("segment: {}", self.meta.id),
2081 FormatCheck::Length,
2082 "missing file trailer",
2083 )
2084 })?;
2085 let expected = read_u64("segment", bytes, trailer_start)?;
2086 let checksummed = bytes.get(..trailer_start).ok_or_else(|| {
2087 FormatError::new("segment", FormatCheck::Length, "invalid trailer position")
2088 })?;
2089 let actual = xxh3_64(checksummed);
2090 if actual != expected {
2091 return Err(FormatError::new(
2092 format!("segment: {}", self.meta.id),
2093 FormatCheck::FileChecksum,
2094 format!("expected {expected:#018x}, computed {actual:#018x}"),
2095 )
2096 .into());
2097 }
2098 Ok(())
2099 }
2100
2101 fn vector_payload(
2102 &self,
2103 kind: RegionKind,
2104 ) -> Result<(super::layout::VectorHeader, &[u8]), SegmentError> {
2105 let region = self.region(kind)?;
2106 let header_bytes = region.get(..VECTOR_HEADER_LEN).ok_or_else(|| {
2107 SegmentError::Geometry(format!("{kind:?} is shorter than vector header"))
2108 })?;
2109 let header = decode_vector_header(header_bytes)?;
2110 if header.dims != self.meta.dims
2111 || header.row_count != self.meta.row_count
2112 || (kind != RegionKind::VectorRescore && header.scheme != self.meta.scheme)
2113 {
2114 return Err(SegmentError::Geometry(format!(
2115 "{kind:?} header scheme/dims/rows {}/{}/{}, segment {}/{}/{}",
2116 header.scheme,
2117 header.dims,
2118 header.row_count,
2119 self.meta.scheme,
2120 self.meta.dims,
2121 self.meta.row_count
2122 )));
2123 }
2124 let payload = region
2125 .get(VECTOR_HEADER_LEN..)
2126 .ok_or_else(|| SegmentError::Geometry(format!("{kind:?} payload offset is invalid")))?;
2127 Ok((header, payload))
2128 }
2129
2130 fn vector_payload_unchecked(
2131 &self,
2132 kind: RegionKind,
2133 ) -> Result<(super::layout::VectorHeader, &[u8]), SegmentError> {
2134 let entry = self.entry(kind)?;
2135 let region = self.region_slice_unaccounted(entry)?;
2136 let header_bytes = region.get(..VECTOR_HEADER_LEN).ok_or_else(|| {
2137 SegmentError::Geometry(format!("{kind:?} is shorter than vector header"))
2138 })?;
2139 let header = decode_vector_header(header_bytes)?;
2140 if header.dims != self.meta.dims
2141 || header.row_count != self.meta.row_count
2142 || (kind != RegionKind::VectorRescore && header.scheme != self.meta.scheme)
2143 {
2144 return Err(SegmentError::Geometry(format!(
2145 "{kind:?} header scheme/dims/rows {}/{}/{}, segment {}/{}/{}",
2146 header.scheme,
2147 header.dims,
2148 header.row_count,
2149 self.meta.scheme,
2150 self.meta.dims,
2151 self.meta.row_count
2152 )));
2153 }
2154 let payload = region
2155 .get(VECTOR_HEADER_LEN..)
2156 .ok_or_else(|| SegmentError::Geometry(format!("{kind:?} payload offset is invalid")))?;
2157 Ok((header, payload))
2158 }
2159
2160 fn validate_rescore_byte_range(&self, start: usize, length: usize) -> Result<(), SegmentError> {
2161 let entry = self.entry(RegionKind::VectorRescore)?;
2162 let region_length = usize::try_from(entry.length).map_err(|_| {
2163 SegmentError::Geometry("rescore region length exceeds usize".to_owned())
2164 })?;
2165 let end = start
2166 .checked_add(length)
2167 .ok_or_else(|| SegmentError::Geometry("rescore byte range overflow".to_owned()))?;
2168 if end > region_length {
2169 return Err(SegmentError::Geometry(format!(
2170 "rescore byte range {start}..{end} exceeds {region_length}"
2171 )));
2172 }
2173 if length == 0 || !Self::query_checksums_enabled() {
2174 return Ok(());
2175 }
2176 let first_chunk = start / CHECKSUM_CHUNK_BYTES;
2177 let last_chunk = end.saturating_sub(1) / CHECKSUM_CHUNK_BYTES;
2178 let mut validated = self.rescore_valid_chunks.lock().map_err(|_| {
2179 SegmentError::Geometry("rescore validation state is poisoned".to_owned())
2180 })?;
2181 for chunk in first_chunk..=last_chunk {
2182 let word = chunk / u64::BITS as usize;
2183 let bit = chunk % u64::BITS as usize;
2184 let mask = 1_u64 << bit;
2185 let state = validated.get_mut(word).ok_or_else(|| {
2186 SegmentError::Geometry(format!("rescore checksum chunk {chunk} is untracked"))
2187 })?;
2188 if *state & mask != 0 {
2189 continue;
2190 }
2191 let chunk_index = u32::try_from(chunk).map_err(|_| {
2192 SegmentError::Geometry("rescore checksum chunk exceeds u32".to_owned())
2193 })?;
2194 self.region_chunk(RegionKind::VectorRescore, chunk_index)?;
2195 *state |= mask;
2196 }
2197 Ok(())
2198 }
2199
2200 fn account_query_cache(
2201 &self,
2202 bytes: usize,
2203 ) -> Result<Option<crate::lifecycle::stats::AccountedCounter>, crate::lifecycle::StoreError>
2204 {
2205 let Some(accounting) = &self.query_accounting else {
2206 return Ok(None);
2207 };
2208 let mut counter = crate::lifecycle::stats::AccountedCounter::new(
2209 accounting,
2210 crate::lifecycle::stats::AllocationComponent::Snapshot,
2211 )?;
2212 counter.set(bytes)?;
2213 Ok(Some(counter))
2214 }
2215
2216 fn entry(&self, kind: RegionKind) -> Result<&RegionEntry, SegmentError> {
2217 self.entries
2218 .iter()
2219 .find(|entry| entry.kind == kind.id())
2220 .ok_or(SegmentError::MissingRegion(kind))
2221 }
2222
2223 fn region_slice(&self, entry: &RegionEntry) -> Result<&[u8], SegmentError> {
2224 let bytes = self.region_slice_unaccounted(entry)?;
2225 account_data_read(bytes.len());
2226 Ok(bytes)
2227 }
2228
2229 fn region_slice_unaccounted(&self, entry: &RegionEntry) -> Result<&[u8], SegmentError> {
2230 let start = usize::try_from(entry.offset)
2231 .map_err(|_| SegmentError::Geometry("region offset exceeds usize".to_owned()))?;
2232 let length = usize::try_from(entry.length)
2233 .map_err(|_| SegmentError::Geometry("region length exceeds usize".to_owned()))?;
2234 let end = start
2235 .checked_add(length)
2236 .ok_or_else(|| SegmentError::Geometry("region end overflow".to_owned()))?;
2237 let bytes = self.mapping.as_bytes().get(start..end).ok_or_else(|| {
2238 SegmentError::Geometry(format!(
2239 "region {} range {start}..{end} exceeds file {}",
2240 entry.kind,
2241 self.mapping.length()
2242 ))
2243 })?;
2244 Ok(bytes)
2245 }
2246
2247 fn chunk_checksum(&self, kind: RegionKind, chunk_index: u32) -> Result<u64, SegmentError> {
2248 let table = self.region(RegionKind::ChecksumTable)?;
2249 let mut cursor = Cursor::new("checksum-table", table);
2250 let count = cursor.u32()?;
2251 let chunk_size = cursor.u32()?;
2252 if chunk_size as usize != CHECKSUM_CHUNK_BYTES {
2253 return Err(SegmentError::Geometry(format!(
2254 "checksum chunk size {chunk_size}, expected {CHECKSUM_CHUNK_BYTES}"
2255 )));
2256 }
2257 let mut found = None;
2258 for _ in 0..count {
2259 let entry_kind = cursor.u16()?;
2260 let reserved = cursor.u16()?;
2261 let entry_chunk = cursor.u32()?;
2262 let checksum = cursor.u64()?;
2263 if reserved != 0 {
2264 return Err(SegmentError::Geometry(
2265 "checksum-table reserved field is non-zero".to_owned(),
2266 ));
2267 }
2268 if entry_kind == kind.id() && entry_chunk == chunk_index {
2269 found = Some(checksum);
2270 }
2271 }
2272 cursor.finish().map_err(SegmentError::Geometry)?;
2273 found.ok_or_else(|| {
2274 SegmentError::Geometry(format!(
2275 "checksum table has no {kind:?} chunk {chunk_index}"
2276 ))
2277 })
2278 }
2279}
2280
2281const DOCUMENT_VERSION_BYTES: usize = 24;
2284
2285fn encode_document_version(version: DocumentVersion) -> [u8; DOCUMENT_VERSION_BYTES] {
2286 let doc_id = version.doc_id().get().to_le_bytes();
2287 let revision = version.revision().get().to_le_bytes();
2288 let mut bytes = [0_u8; DOCUMENT_VERSION_BYTES];
2289 for (slot, byte) in bytes.iter_mut().zip(doc_id.into_iter().chain(revision)) {
2290 *slot = byte;
2291 }
2292 bytes
2293}
2294
2295fn document_version_entry(bytes: &[u8], row: u32) -> &[u8] {
2299 (row as usize)
2300 .checked_mul(DOCUMENT_VERSION_BYTES)
2301 .and_then(|start| bytes.get(start..start.checked_add(DOCUMENT_VERSION_BYTES)?))
2302 .unwrap_or(&[])
2303}
2304
2305fn document_id_entry(bytes: &[u8], row: u32) -> &[u8] {
2306 document_version_entry(bytes, row)
2307 .get(..std::mem::size_of::<u128>())
2308 .unwrap_or(&[])
2309}
2310
2311fn arc_resident_bytes<T>(deep_bytes: usize) -> Option<usize> {
2312 deep_bytes
2313 .checked_add(std::mem::size_of::<T>())?
2314 .checked_add(2 * std::mem::size_of::<usize>())
2315}
2316
2317fn rescore_validation_words(entries: &[RegionEntry]) -> Result<usize, SegmentError> {
2318 let Some(entry) = entries
2319 .iter()
2320 .find(|entry| entry.kind == RegionKind::VectorRescore.id())
2321 else {
2322 return Ok(0);
2323 };
2324 let length = usize::try_from(entry.length)
2325 .map_err(|_| SegmentError::Geometry("rescore region length exceeds usize".to_owned()))?;
2326 let chunks = length.div_ceil(CHECKSUM_CHUNK_BYTES);
2327 Ok(chunks.div_ceil(u64::BITS as usize))
2328}
2329
2330pub fn validate_header_with_vfs(
2332 vfs: &dyn Vfs,
2333 path: &Path,
2334 expected: &SegmentMeta,
2335) -> Result<usize, SegmentError> {
2336 let actual_length = vfs
2337 .open(path)
2338 .map_err(|error| SegmentError::io(path, error))?;
2339 let mut header_bytes = vfs
2340 .read_range(path, 0, FILE_HEADER_LEN)
2341 .map_err(|error| SegmentError::io(path, error))?;
2342 let artifact = path.display().to_string();
2343 #[cfg(any(test, feature = "test-seams"))]
2344 let actual_family = header_bytes
2345 .get(8..10)
2346 .and_then(|family| <[u8; 2]>::try_from(family).ok())
2347 .map(u16::from_le_bytes);
2348 let fixed =
2349 decode_header(&artifact, FormatFamily::Segment, &header_bytes).map_err(|error| {
2350 #[cfg(any(test, feature = "test-seams"))]
2351 crate::lifecycle::record_storage_segment_format_fault(
2352 &error,
2353 actual_family,
2354 expected.id,
2355 );
2356 SegmentError::from(error)
2357 })?;
2358 let header_length = usize::try_from(fixed.header_length)
2359 .map_err(|_| SegmentError::Geometry("segment header length exceeds usize".to_owned()))?;
2360 validate_bounded_header_length(&artifact, header_length, actual_length)?;
2361 let remaining = header_length.saturating_sub(FILE_HEADER_LEN);
2362 if remaining > 0 {
2363 let tail = vfs
2364 .read_range(path, FILE_HEADER_LEN as u64, remaining)
2365 .map_err(|error| SegmentError::io(path, error))?;
2366 header_bytes.extend_from_slice(&tail);
2367 }
2368 let parsed = parse_segment_header(&artifact, &header_bytes, actual_length, expected.id)?;
2369 if !parsed.meta.same_segment_file(expected) {
2370 return Err(SegmentError::Geometry(format!(
2371 "manifest metadata {expected:?}, header metadata {:?}",
2372 parsed.meta
2373 )));
2374 }
2375 Ok(header_length)
2376}
2377
2378pub fn validate_segment_bytes(bytes: &[u8]) -> Result<SegmentMeta, SegmentError> {
2380 let id_bytes: [u8; 16] = bytes
2381 .get(FILE_HEADER_LEN..FILE_HEADER_LEN.saturating_add(16))
2382 .ok_or_else(|| {
2383 FormatError::new(
2384 "segment-bytes",
2385 FormatCheck::Length,
2386 format!("missing segment id in {} bytes", bytes.len()),
2387 )
2388 })?
2389 .try_into()
2390 .map_err(|_| {
2391 FormatError::new("segment-bytes", FormatCheck::Length, "invalid segment id")
2392 })?;
2393 let id = SegmentId::from_bytes(id_bytes);
2394 let actual_length = u64::try_from(bytes.len())
2395 .map_err(|_| SegmentError::Geometry("byte slice length exceeds u64".to_owned()))?;
2396 let parsed = parse_segment_header("segment-bytes", bytes, actual_length, id)?;
2397 for entry in &parsed.entries {
2398 let start = usize::try_from(entry.offset)
2399 .map_err(|_| SegmentError::Geometry("region offset exceeds usize".to_owned()))?;
2400 let length = usize::try_from(entry.length)
2401 .map_err(|_| SegmentError::Geometry("region length exceeds usize".to_owned()))?;
2402 let end = start
2403 .checked_add(length)
2404 .ok_or_else(|| SegmentError::Geometry("region end overflow".to_owned()))?;
2405 let region = bytes.get(start..end).ok_or_else(|| {
2406 FormatError::new(
2407 "segment-bytes",
2408 FormatCheck::BlockLength,
2409 format!("region {} range {start}..{end} is truncated", entry.kind),
2410 )
2411 })?;
2412 let actual = xxh3_64(region);
2413 if actual != entry.checksum {
2414 return Err(FormatError::new(
2415 format!("segment-bytes:region-{}", entry.kind),
2416 FormatCheck::BlockChecksum,
2417 format!("expected {:#018x}, computed {actual:#018x}", entry.checksum),
2418 )
2419 .into());
2420 }
2421 }
2422 let trailer = bytes.len().checked_sub(FILE_TRAILER_LEN).ok_or_else(|| {
2423 FormatError::new("segment-bytes", FormatCheck::Length, "missing file trailer")
2424 })?;
2425 let expected = read_u64("segment-bytes", bytes, trailer)?;
2426 let checksummed = bytes.get(..trailer).ok_or_else(|| {
2427 FormatError::new(
2428 "segment-bytes",
2429 FormatCheck::Length,
2430 "invalid trailer offset",
2431 )
2432 })?;
2433 let actual = xxh3_64(checksummed);
2434 if actual != expected {
2435 return Err(FormatError::new(
2436 "segment-bytes",
2437 FormatCheck::FileChecksum,
2438 format!("expected {expected:#018x}, computed {actual:#018x}"),
2439 )
2440 .into());
2441 }
2442 Ok(parsed.meta)
2443}
2444
2445fn preflight_region_count(
2446 artifact: &str,
2447 bytes: &[u8],
2448 actual_file_length: u64,
2449) -> Result<usize, SegmentError> {
2450 let fixed = decode_header(artifact, FormatFamily::Segment, bytes)?;
2451 if fixed.file_length != actual_file_length {
2452 return Err(FormatError::new(
2453 artifact,
2454 FormatCheck::FileLength,
2455 format!(
2456 "declared {}, actual {actual_file_length}",
2457 fixed.file_length
2458 ),
2459 )
2460 .into());
2461 }
2462 let header_length = usize::try_from(fixed.header_length)
2463 .map_err(|_| SegmentError::Geometry("header length exceeds usize".to_owned()))?;
2464 validate_bounded_header_length(artifact, header_length, actual_file_length)?;
2465 let header_bytes = bytes.get(..header_length).ok_or_else(|| {
2466 FormatError::new(
2467 artifact,
2468 FormatCheck::Length,
2469 format!("declared header {header_length}, available {}", bytes.len()),
2470 )
2471 })?;
2472 Ok(usize::from(read_u16(
2473 artifact,
2474 header_bytes,
2475 FILE_HEADER_LEN.saturating_add(20),
2476 )?))
2477}
2478
2479fn parse_segment_header(
2480 artifact: &str,
2481 bytes: &[u8],
2482 actual_file_length: u64,
2483 expected_id: SegmentId,
2484) -> Result<ParsedHeader, SegmentError> {
2485 #[cfg(any(test, feature = "test-seams"))]
2486 let actual_family = bytes
2487 .get(8..10)
2488 .and_then(|family| <[u8; 2]>::try_from(family).ok())
2489 .map(u16::from_le_bytes);
2490 let fixed = decode_header(artifact, FormatFamily::Segment, bytes).map_err(|error| {
2491 #[cfg(any(test, feature = "test-seams"))]
2492 crate::lifecycle::record_storage_segment_format_fault(&error, actual_family, expected_id);
2493 SegmentError::from(error)
2494 })?;
2495 if fixed.file_length != actual_file_length {
2496 return Err(FormatError::new(
2497 artifact,
2498 FormatCheck::FileLength,
2499 format!(
2500 "declared {}, actual {actual_file_length}",
2501 fixed.file_length
2502 ),
2503 )
2504 .into());
2505 }
2506 let header_length = usize::try_from(fixed.header_length)
2507 .map_err(|_| SegmentError::Geometry("header length exceeds usize".to_owned()))?;
2508 validate_bounded_header_length(artifact, header_length, actual_file_length)?;
2509 let header_bytes = bytes.get(..header_length).ok_or_else(|| {
2510 FormatError::new(
2511 artifact,
2512 FormatCheck::Length,
2513 format!("declared header {header_length}, available {}", bytes.len()),
2514 )
2515 })?;
2516 let id_bytes: [u8; 16] = header_bytes
2517 .get(FILE_HEADER_LEN..FILE_HEADER_LEN.saturating_add(16))
2518 .ok_or_else(|| FormatError::new(artifact, FormatCheck::Length, "missing segment id"))?
2519 .try_into()
2520 .map_err(|_| FormatError::new(artifact, FormatCheck::Length, "invalid segment id"))?;
2521 let actual_id = SegmentId::from_bytes(id_bytes);
2522 if actual_id != expected_id {
2523 #[cfg(any(test, feature = "test-seams"))]
2524 crate::lifecycle::record_storage_segment_identity_fault(artifact, expected_id, actual_id);
2525 return Err(SegmentError::WrongObject {
2526 artifact: artifact.to_owned(),
2527 expected: expected_id,
2528 actual: actual_id,
2529 });
2530 }
2531 let row_count = read_u32(artifact, header_bytes, FILE_HEADER_LEN.saturating_add(16))?;
2532 let region_count = usize::from(read_u16(
2533 artifact,
2534 header_bytes,
2535 FILE_HEADER_LEN.saturating_add(20),
2536 )?);
2537 if read_u16(artifact, header_bytes, FILE_HEADER_LEN.saturating_add(22))? != 0 {
2538 return Err(SegmentError::Geometry(
2539 "segment prefix reserved field is non-zero".to_owned(),
2540 ));
2541 }
2542 let scheme = read_u16(artifact, header_bytes, FILE_HEADER_LEN.saturating_add(24))?;
2543 FormatRegistry::require_scheme(scheme)
2544 .map_err(|error| SegmentError::Geometry(error.to_string()))?;
2545 if read_u16(artifact, header_bytes, FILE_HEADER_LEN.saturating_add(26))? != 0 {
2546 return Err(SegmentError::Geometry(
2547 "segment scheme padding is non-zero".to_owned(),
2548 ));
2549 }
2550 let dims = read_u32(artifact, header_bytes, FILE_HEADER_LEN.saturating_add(28))?;
2551 let expected_header_length = FILE_HEADER_LEN
2552 .checked_add(SEGMENT_PREFIX_LEN)
2553 .and_then(|value| value.checked_add(region_count.saturating_mul(REGION_ENTRY_LEN)))
2554 .and_then(|value| value.checked_add(8))
2555 .ok_or_else(|| SegmentError::Geometry("header length overflow".to_owned()))?;
2556 if header_length != expected_header_length {
2557 return Err(FormatError::new(
2558 artifact,
2559 FormatCheck::HeaderLength,
2560 format!("declared {header_length}, directory implies {expected_header_length}"),
2561 )
2562 .into());
2563 }
2564 let checksum_offset = header_length.saturating_sub(8);
2565 let expected_checksum = read_u64(artifact, header_bytes, checksum_offset)?;
2566 let checksum_bytes = header_bytes.get(..checksum_offset).ok_or_else(|| {
2567 FormatError::new(
2568 artifact,
2569 FormatCheck::Length,
2570 "invalid header checksum offset",
2571 )
2572 })?;
2573 let actual_checksum = xxh3_64(checksum_bytes);
2574 if actual_checksum != expected_checksum {
2575 return Err(FormatError::new(
2576 artifact,
2577 FormatCheck::BlockChecksum,
2578 format!("header expected {expected_checksum:#018x}, computed {actual_checksum:#018x}"),
2579 )
2580 .into());
2581 }
2582
2583 let mut entries: Vec<RegionEntry> = Vec::with_capacity(region_count);
2584 let directory_start = FILE_HEADER_LEN.saturating_add(SEGMENT_PREFIX_LEN);
2585 for position in 0..region_count {
2586 let offset = directory_start
2587 .checked_add(position.saturating_mul(REGION_ENTRY_LEN))
2588 .ok_or_else(|| SegmentError::Geometry("directory offset overflow".to_owned()))?;
2589 let entry = RegionEntry {
2590 kind: read_u16(artifact, header_bytes, offset)?,
2591 version: read_u16(artifact, header_bytes, offset.saturating_add(2))?,
2592 reserved: read_u32(artifact, header_bytes, offset.saturating_add(4))?,
2593 offset: read_u64(artifact, header_bytes, offset.saturating_add(8))?,
2594 length: read_u64(artifact, header_bytes, offset.saturating_add(16))?,
2595 checksum: read_u64(artifact, header_bytes, offset.saturating_add(24))?,
2596 };
2597 if entry.reserved != 0 {
2598 return Err(SegmentError::Geometry(format!(
2599 "region {} reserved field is non-zero",
2600 entry.kind
2601 )));
2602 }
2603 if entries.iter().any(|existing| existing.kind == entry.kind) {
2604 return Err(SegmentError::Geometry(format!(
2605 "duplicate region kind {}",
2606 entry.kind
2607 )));
2608 }
2609 validate_entry(artifact, &entry, header_length, actual_file_length)?;
2610 if let Some(kind) = RegionKind::from_id(entry.kind)
2611 && let Some(family) = kind.family()
2612 {
2613 FormatRegistry::require(family.id(), entry.version)
2614 .map_err(|error| crate::format::frame::registry_error(artifact, error))?;
2615 }
2616 entries.push(entry);
2617 }
2618 validate_non_overlapping(&entries)?;
2619 Ok(ParsedHeader {
2620 meta: SegmentMeta {
2621 id: actual_id,
2622 row_count,
2623 scheme,
2624 dims,
2625 file_size: actual_file_length,
2626 epoch_id: None,
2627 clustering_key_range: super::ClusteringKeyRange::Unstamped,
2628 },
2629 header_length,
2630 entries,
2631 })
2632}
2633
2634fn validate_bounded_header_length(
2635 artifact: &str,
2636 header_length: usize,
2637 actual_file_length: u64,
2638) -> Result<(), SegmentError> {
2639 if !(FILE_HEADER_LEN + SEGMENT_PREFIX_LEN + 8..=MAX_SEGMENT_HEADER_BYTES)
2640 .contains(&header_length)
2641 {
2642 return Err(FormatError::new(
2643 artifact,
2644 FormatCheck::HeaderLength,
2645 format!("header length {header_length} is outside bounded v1 range"),
2646 )
2647 .into());
2648 }
2649 let file_length = usize::try_from(actual_file_length)
2650 .map_err(|_| SegmentError::Geometry("file length exceeds usize".to_owned()))?;
2651 if header_length > file_length.saturating_sub(FILE_TRAILER_LEN) {
2652 return Err(FormatError::new(
2653 artifact,
2654 FormatCheck::HeaderLength,
2655 format!("header {header_length} exceeds file {file_length}"),
2656 )
2657 .into());
2658 }
2659 Ok(())
2660}
2661
2662fn validate_entry(
2663 artifact: &str,
2664 entry: &RegionEntry,
2665 header_length: usize,
2666 file_length: u64,
2667) -> Result<(), SegmentError> {
2668 let offset = usize::try_from(entry.offset)
2669 .map_err(|_| SegmentError::Geometry("region offset exceeds usize".to_owned()))?;
2670 let length = usize::try_from(entry.length)
2671 .map_err(|_| SegmentError::Geometry("region length exceeds usize".to_owned()))?;
2672 let minimum = align_up(header_length, REGION_ALIGNMENT)?;
2673 if offset < minimum || !offset.is_multiple_of(REGION_ALIGNMENT) {
2674 return Err(SegmentError::Geometry(format!(
2675 "region {} offset {offset} is below/aligned from {minimum}",
2676 entry.kind
2677 )));
2678 }
2679 let end = offset
2680 .checked_add(length)
2681 .ok_or_else(|| SegmentError::Geometry("region end overflow".to_owned()))?;
2682 let maximum = usize::try_from(file_length)
2683 .map_err(|_| SegmentError::Geometry("file length exceeds usize".to_owned()))?
2684 .checked_sub(FILE_TRAILER_LEN)
2685 .ok_or_else(|| SegmentError::Geometry("file is shorter than trailer".to_owned()))?;
2686 if end > maximum {
2687 return Err(FormatError::new(
2688 artifact,
2689 FormatCheck::BlockLength,
2690 format!("region {} ends at {end}, maximum {maximum}", entry.kind),
2691 )
2692 .into());
2693 }
2694 Ok(())
2695}
2696
2697fn validate_non_overlapping(entries: &[RegionEntry]) -> Result<(), SegmentError> {
2698 for (position, entry) in entries.iter().enumerate() {
2699 let entry_end = entry
2700 .offset
2701 .checked_add(entry.length)
2702 .ok_or_else(|| SegmentError::Geometry("region u64 end overflow".to_owned()))?;
2703 for other in entries.iter().skip(position.saturating_add(1)) {
2704 let other_end = other
2705 .offset
2706 .checked_add(other.length)
2707 .ok_or_else(|| SegmentError::Geometry("region u64 end overflow".to_owned()))?;
2708 if entry.offset < other_end && other.offset < entry_end {
2709 let (previous, current, previous_end) = if entry.offset <= other.offset {
2710 (entry, other, entry_end)
2711 } else {
2712 (other, entry, other_end)
2713 };
2714 return Err(SegmentError::Geometry(format!(
2715 "region {} begins {} before previous end {previous_end} for region {}",
2716 current.kind, current.offset, previous.kind
2717 )));
2718 }
2719 }
2720 }
2721 Ok(())
2722}
2723
2724fn cast_slice<'a, T>(
2725 bytes: &'a [u8],
2726 count: usize,
2727 artifact: &str,
2728) -> Result<&'a [T], SegmentError> {
2729 #[cfg(target_endian = "big")]
2730 {
2731 let _ = (bytes, count);
2732 return Err(SegmentError::Geometry(format!(
2733 "{artifact} zero-copy requires little-endian target"
2734 )));
2735 }
2736 #[cfg(target_endian = "little")]
2737 {
2738 let expected = count
2739 .checked_mul(std::mem::size_of::<T>())
2740 .ok_or_else(|| SegmentError::Geometry(format!("{artifact} byte length overflow")))?;
2741 if bytes.len() != expected {
2742 return Err(SegmentError::Geometry(format!(
2743 "{artifact} bytes {}, expected {expected}",
2744 bytes.len()
2745 )));
2746 }
2747 if !(bytes.as_ptr() as usize).is_multiple_of(std::mem::align_of::<T>()) {
2748 return Err(SegmentError::Geometry(format!(
2749 "{artifact} address is not {}-byte aligned",
2750 std::mem::align_of::<T>()
2751 )));
2752 }
2753 Ok(unsafe { std::slice::from_raw_parts(bytes.as_ptr().cast::<T>(), count) })
2757 }
2758}