krabka_log/
stamp_index.rs

1//! Per-partition `.stampindex` sidecar. A two-byte header, then one
2//! fixed-width record per stamped offset range in the segment:
3//!
4//!   `version`:     i16 (big-endian), once at the start of the file, always
5//!                  [`STAMP_INDEX_VERSION`]
6//!
7//! and per record:
8//!
9//!   `base_offset`: i64 (big-endian)
10//!   `last_offset`: i64 (big-endian)
11//!   `stamp`:       u64 (big-endian)
12//!
13//! The stamp is an additional internal coordinate, a packed
14//! `TimestampSource` reading, stored beside the wire-exact `.log`. Nothing
15//! ever touches the `.log` bytes. The stampindex is internal metadata that is
16//! retained and truncated with its segment. It never leaves the broker on any
17//! client-facing API. This mirrors the `.txnindex` sidecar pattern.
18
19use {
20    crate::{
21        error::LogError,
22        index::{rewrite_retained_entries, rewrite_sidecar_file},
23        io::{IoTarget, LogIo, file_io, write_all},
24    },
25    krabka_ids::Offset,
26    std::{
27        fs::{self, OpenOptions},
28        io::ErrorKind,
29        mem,
30        path::{Path, PathBuf},
31        sync::Arc,
32    },
33    tracing::instrument,
34    zerocopy::{
35        BigEndian, FromBytes, Immutable, IntoBytes, KnownLayout, Unaligned,
36        byteorder::{I64, U64},
37    },
38};
39
40const ENTRY_BYTES: usize = 24;
41
42/// The `.stampindex` format version this build writes and reads.
43///
44/// It is the big-endian `i16` at the front of every file, and part of the 1.x
45/// on-disk contract: a 1.x broker reads every version an earlier 1.x broker
46/// wrote, so a later layout takes a new number and keeps this one readable.
47pub const STAMP_INDEX_VERSION: i16 = 0;
48
49const HEADER_BYTES: usize = mem::size_of::<i16>();
50
51/// The artifact name the version errors carry.
52const ARTIFACT: &str = "stampindex";
53
54/// An inclusive range of offsets, `[first, last]`.
55#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56pub struct OffsetRange {
57    /// The first offset of the range.
58    pub first: Offset,
59    /// The last offset of the range.
60    pub last: Offset,
61}
62
63/// The base offsets and the last offsets of a run of stamp entries, each in
64/// entry order.
65struct StampCoordinates {
66    bases: Vec<i64>,
67    lasts: Vec<i64>,
68}
69
70/// One stamped offset range. The inclusive offsets
71/// `[base_offset, last_offset]` all carry `stamp`.
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
73pub struct StampEntry {
74    pub base_offset: Offset,
75    pub last_offset: Offset,
76    pub stamp: u64,
77}
78
79/// On-disk byte layout of one `StampEntry`. `zerocopy` reinterprets it in
80/// place from the file bytes.
81#[derive(Debug, Clone, Copy, FromBytes, IntoBytes, KnownLayout, Immutable, Unaligned)]
82#[repr(C)]
83struct StampEntryRaw {
84    base_offset: I64<BigEndian>,
85    last_offset: I64<BigEndian>,
86    stamp: U64<BigEndian>,
87}
88
89impl StampEntryRaw {
90    fn new(entry: StampEntry) -> Self {
91        Self {
92            base_offset: I64::new(entry.base_offset.0),
93            last_offset: I64::new(entry.last_offset.0),
94            stamp: U64::new(entry.stamp),
95        }
96    }
97}
98
99const _: [(); ENTRY_BYTES] = [(); mem::size_of::<StampEntryRaw>()];
100
101#[derive(Debug)]
102pub struct StampIndex {
103    path: PathBuf,
104    io: Arc<dyn LogIo>,
105    entries: Vec<StampEntry>,
106}
107
108impl StampIndex {
109    #[krabka_macros::index_open]
110    /// Open or recover a `.stampindex` file at the given path. This method
111    /// reads the entire file into memory at startup. A missing file means
112    /// zero stamped ranges. So does an empty one: the first append creates
113    /// the file and then writes the header with its entry, so a crash between
114    /// the two leaves an empty file that holds nothing to misread.
115    /// # Errors
116    /// Returns [`LogError::MissingFormatVersion`] for the headerless layout a
117    /// broker before 1.0 wrote, [`LogError::UnsupportedFormatVersion`] for a
118    /// version other than `STAMP_INDEX_VERSION`, and an error when log I/O
119    /// fails or the entries are not a whole number of fixed-width records.
120    pub fn open(path: PathBuf) -> Result<Self, LogError> {
121        let mut entries = Vec::new();
122        match fs::read(&path) {
123            Ok(bytes) if bytes.is_empty() => {}
124            Ok(bytes) => {
125                let body = Self::versioned_body(&path, &bytes)?;
126                if !body.len().is_multiple_of(ENTRY_BYTES) {
127                    return Err(LogError::Corrupt(format!(
128                        "stampindex {} has {} entry bytes, not divisible by {}",
129                        path.display(),
130                        body.len(),
131                        ENTRY_BYTES,
132                    )));
133                }
134                let raws = <[StampEntryRaw]>::ref_from_bytes(body).map_err(|_| {
135                    LogError::Corrupt(format!(
136                        "stampindex {} entry bytes do not form whole entries",
137                        path.display(),
138                    ))
139                })?;
140                entries.reserve(raws.len());
141                for raw in raws {
142                    entries.push(StampEntry {
143                        base_offset: Offset(raw.base_offset.get()),
144                        last_offset: Offset(raw.last_offset.get()),
145                        stamp: raw.stamp.get(),
146                    });
147                }
148                entries.sort_unstable_by_key(|entry| {
149                    (entry.base_offset.0, entry.last_offset.0, entry.stamp)
150                });
151                // A write followed by an uncertain sync can be retried and
152                // leave an exact duplicate on disk. Canonicalize that retry,
153                // but reject a duplicate range with a different stamp below.
154                entries.dedup();
155                if !Self::entries_valid(&entries) {
156                    return Err(LogError::Corrupt(format!(
157                        "stampindex {} contains inverted or overlapping ranges",
158                        path.display()
159                    )));
160                }
161            }
162            Err(e) if e.kind() == ErrorKind::NotFound => {}
163            Err(e) => return Err(LogError::Io(e)),
164        }
165        tracing::Span::current().record("entries", entries.len());
166        Ok(Self {
167            path,
168            io: file_io(),
169            entries,
170        })
171    }
172
173    /// Check the header of a non-empty file and return the entry bytes after
174    /// it.
175    ///
176    /// A broker before 1.0 wrote bare 24-byte entries, so a file whose whole
177    /// length is a multiple of the entry width has no header. The headed
178    /// layout is two bytes longer than such a multiple, so the two never
179    /// collide.
180    fn versioned_body<'a>(path: &Path, bytes: &'a [u8]) -> Result<&'a [u8], LogError> {
181        if bytes.len().is_multiple_of(ENTRY_BYTES) {
182            return Err(LogError::MissingFormatVersion {
183                artifact: ARTIFACT,
184                path: path.to_path_buf(),
185            });
186        }
187        let Some((&header, body)) = bytes.split_first_chunk::<HEADER_BYTES>() else {
188            return Err(LogError::Corrupt(format!(
189                "stampindex {} is {} bytes, shorter than its version header",
190                path.display(),
191                bytes.len(),
192            )));
193        };
194        let version = i16::from_be_bytes(header);
195        if version != STAMP_INDEX_VERSION {
196            return Err(LogError::UnsupportedFormatVersion {
197                artifact: ARTIFACT,
198                path: path.to_path_buf(),
199                found: i64::from(version),
200            });
201        }
202        Ok(body)
203    }
204
205    /// Append one stamped-range entry.
206    ///
207    /// Entries need not arrive in offset order. Transactional ranges are
208    /// added when their commit marker lands, and two interleaved transactions
209    /// can commit in either order. The in-memory index is kept in canonical
210    /// offset order, and ranges themselves must not overlap.
211    #[instrument(
212        level = "debug",
213        skip(self),
214        fields(stamp = entry.stamp),
215        err,
216    )]
217    /// # Errors
218    /// Returns an error when appending to or syncing the file fails.
219    pub fn append(&mut self, entry: StampEntry) -> Result<(), LogError> {
220        if let Some(position) = self.entry_position(entry) {
221            if self.entries.get(position) == Some(&entry) {
222                return Ok(());
223            }
224            return Err(LogError::Corrupt(format!(
225                "stamp range {}..={} conflicts with its existing stamp in {}",
226                entry.base_offset,
227                entry.last_offset,
228                self.path.display()
229            )));
230        }
231        let Some(position) = Self::insertion_position(&self.entries, entry) else {
232            if entry.last_offset < entry.base_offset {
233                return Err(LogError::InvalidArgument(format!(
234                    "stamp range {}..={} is inverted",
235                    entry.base_offset, entry.last_offset
236                )));
237            }
238            return Err(LogError::Corrupt(format!(
239                "stamp range {}..={} overlaps an existing range in {}",
240                entry.base_offset,
241                entry.last_offset,
242                self.path.display()
243            )));
244        };
245        let f = OpenOptions::new()
246            .create(true)
247            .append(true)
248            .open(&self.path)
249            .map_err(LogError::Io)?;
250        // A new or empty file gets its header in the same write as its first
251        // entry.
252        let mut bytes = Vec::with_capacity(HEADER_BYTES + ENTRY_BYTES);
253        if f.metadata().map_err(LogError::Io)?.len() == 0 {
254            bytes.extend_from_slice(&STAMP_INDEX_VERSION.to_be_bytes());
255        }
256        bytes.extend_from_slice(StampEntryRaw::new(entry).as_bytes());
257        write_all(&*self.io, IoTarget::StampIndex, &f, &bytes).map_err(LogError::Io)?;
258        self.io
259            .sync_file(IoTarget::StampIndex, &f)
260            .map_err(LogError::Io)?;
261        self.entries.insert(position, entry);
262        Ok(())
263    }
264
265    fn entry_position(&self, entry: StampEntry) -> Option<usize> {
266        Self::exact_range_index(&self.entries, entry.base_offset, entry.last_offset)
267    }
268
269    /// The entries are sorted and disjoint: open validates this invariant,
270    /// insertion checks both adjacent ranges, and other mutations only
271    /// replace stamps or remove ranges. The general verified lookup kernels
272    /// remain the oracle for callers that do not have that invariant.
273    fn exact_range_index(entries: &[StampEntry], first: Offset, last: Offset) -> Option<usize> {
274        if entries.last().is_none_or(|entry| entry.last_offset < first) {
275            return None;
276        }
277        entries
278            .binary_search_by_key(&first, |entry| entry.base_offset)
279            .ok()
280            .filter(|&position| {
281                entries
282                    .get(position)
283                    .is_some_and(|entry| entry.last_offset == last)
284            })
285    }
286
287    fn insertion_position(entries: &[StampEntry], entry: StampEntry) -> Option<usize> {
288        if entry.last_offset < entry.base_offset {
289            return None;
290        }
291        // Increasing, disjoint batches are the ordinary append path. They
292        // need no search or temporary parallel-coordinate allocation.
293        if entries
294            .last()
295            .is_none_or(|last| last.last_offset < entry.base_offset)
296        {
297            return Some(entries.len());
298        }
299        let position = entries.partition_point(|known| known.base_offset < entry.base_offset);
300        if position
301            .checked_sub(1)
302            .and_then(|previous| entries.get(previous))
303            .is_some_and(|previous| previous.last_offset >= entry.base_offset)
304        {
305            return None;
306        }
307        if entries
308            .get(position)
309            .is_some_and(|next| entry.last_offset >= next.base_offset)
310        {
311            return None;
312        }
313        Some(position)
314    }
315
316    /// Insert a committed transactional range or replace the same exact
317    /// range. Replacement upgrades append-time entries written by older
318    /// brokers to the marker-time commit stamp.
319    ///
320    /// # Errors
321    /// Returns an error for a partial overlap or when rewriting the sidecar
322    /// fails.
323    pub fn upsert(&mut self, entry: StampEntry) -> Result<(), LogError> {
324        if let Some(position) = self.entry_position(entry) {
325            if self.entries.get(position) != Some(&entry) {
326                let entries: Vec<StampEntry> = self
327                    .entries
328                    .iter()
329                    .enumerate()
330                    .map(|(at, known)| if at == position { entry } else { *known })
331                    .collect();
332                self.rewrite(&entries)?;
333                self.entries = entries;
334            }
335            return Ok(());
336        }
337        self.append(entry)
338    }
339
340    /// Remove entries that cover offsets at or after `offset` and rewrite the
341    /// sidecar. Log truncation calls this after truncating the segment bytes.
342    ///
343    /// # Errors
344    /// Returns an error when rewriting or syncing the sidecar fails.
345    pub fn truncate_from(&mut self, offset: Offset) -> Result<(), LogError> {
346        let keep = |entry: &StampEntry| entry.last_offset < offset;
347        if let Some(entries) =
348            rewrite_retained_entries(&self.entries, keep, |entries| self.rewrite(entries))?
349        {
350            self.entries = entries;
351        }
352        Ok(())
353    }
354
355    /// Remove exact ranges, if present. Startup uses this to hide
356    /// append-time transactional entries created by older brokers while the
357    /// transaction is still open.
358    ///
359    /// # Errors
360    /// Returns an error when rewriting or syncing the sidecar fails.
361    pub fn remove_ranges(&mut self, ranges: &[OffsetRange]) -> Result<(), LogError> {
362        let mut entries = self.entries.clone();
363        for &OffsetRange { first, last } in ranges {
364            if let Some(position) = Self::exact_range_index(&entries, first, last) {
365                entries.remove(position);
366            }
367        }
368        if entries.len() == self.entries.len() {
369            return Ok(());
370        }
371        self.rewrite(&entries)?;
372        self.entries = entries;
373        Ok(())
374    }
375
376    fn rewrite(&self, entries: &[StampEntry]) -> Result<(), LogError> {
377        let file = rewrite_sidecar_file(&self.path)?;
378        let mut bytes = Vec::with_capacity(HEADER_BYTES + entries.len() * ENTRY_BYTES);
379        bytes.extend_from_slice(&STAMP_INDEX_VERSION.to_be_bytes());
380        for &entry in entries {
381            bytes.extend_from_slice(StampEntryRaw::new(entry).as_bytes());
382        }
383        write_all(&*self.io, IoTarget::StampIndex, &file, &bytes).map_err(LogError::Io)?;
384        self.io
385            .sync_file(IoTarget::StampIndex, &file)
386            .map_err(LogError::Io)
387    }
388
389    /// Route this sidecar's writes and syncs through `io`.
390    pub(crate) fn set_io(&mut self, io: Arc<dyn LogIo>) { self.io = io; }
391
392    #[must_use]
393    pub fn entries(&self) -> &[StampEntry] { &self.entries }
394
395    fn coordinates(entries: &[StampEntry]) -> StampCoordinates {
396        let (bases, lasts) = entries
397            .iter()
398            .map(|entry| (entry.base_offset.0, entry.last_offset.0))
399            .unzip();
400        StampCoordinates { bases, lasts }
401    }
402
403    fn entries_valid(entries: &[StampEntry]) -> bool {
404        let StampCoordinates { bases, lasts } = Self::coordinates(entries);
405        krabka_verified::stamp_ranges_valid(&bases, &lasts)
406    }
407
408    /// The stamp of the entry whose inclusive `[base_offset, last_offset]`
409    /// range contains `offset`, or `None` when no entry covers it.
410    #[must_use]
411    pub fn stamp_for_offset(&self, offset: Offset) -> Option<u64> {
412        let after = self
413            .entries
414            .partition_point(|entry| entry.base_offset <= offset);
415        let entry = self.entries.get(after.checked_sub(1)?)?;
416        (offset <= entry.last_offset).then_some(entry.stamp)
417    }
418}
419
420#[cfg(test)]
421mod tests {
422
423    use {
424        super::*,
425        crate::{
426            index::{IndexFixture, index_fixture},
427            test_support::stamp_entry,
428        },
429        std::{
430            fs::{self, File},
431            io,
432            path::Path,
433            sync::atomic::{AtomicBool, AtomicUsize, Ordering},
434        },
435        tempfile::TempDir,
436    };
437
438    fn index_with(entries: &[StampEntry]) -> IndexFixture<StampIndex> {
439        let IndexFixture {
440            dir,
441            path,
442            mut index,
443        } = index_fixture("00.stampindex", |path| StampIndex::open(path.to_path_buf()));
444        for entry in entries {
445            index.append(*entry).unwrap();
446        }
447        IndexFixture { dir, path, index }
448    }
449
450    fn write_entries(path: &Path, entries: &[StampEntry]) {
451        let mut bytes = STAMP_INDEX_VERSION.to_be_bytes().to_vec();
452        for &entry in entries {
453            bytes.extend_from_slice(StampEntryRaw::new(entry).as_bytes());
454        }
455        fs::write(path, bytes).unwrap();
456    }
457
458    #[test]
459    fn ordered_searches_match_the_general_verified_oracles() {
460        for length in 0..20 {
461            let entries: Vec<_> = (0..length)
462                .map(|number| stamp_entry(number * 5 - 40, number * 5 - 38, 1))
463                .collect();
464            let StampCoordinates { bases, lasts } = StampIndex::coordinates(&entries);
465            assert2::assert!(krabka_verified::stamp_ranges_valid(&bases, &lasts));
466            let IndexFixture {
467                dir: _dir, index, ..
468            } = index_with(&entries);
469            for first in -45..65 {
470                let expected_stamp =
471                    krabka_verified::covering_stamp_range_index(&bases, &lasts, first)
472                        .map(|position| entries[position].stamp);
473                assert2::assert!(index.stamp_for_offset(Offset(first)) == expected_stamp);
474                for last in -45..65 {
475                    let entry = stamp_entry(first, last, 1);
476                    assert2::assert!(
477                        StampIndex::exact_range_index(&entries, Offset(first), Offset(last))
478                            == krabka_verified::exact_stamp_range_index(
479                                &bases, &lasts, first, last
480                            )
481                    );
482                    assert2::assert!(
483                        StampIndex::insertion_position(&entries, entry)
484                            == krabka_verified::stamp_range_insertion_index(
485                                &bases, &lasts, first, last
486                            )
487                    );
488                }
489            }
490        }
491    }
492
493    #[test]
494    fn sequential_and_interleaved_upserts_keep_the_same_v0_bytes() {
495        let IndexFixture {
496            dir: _dir,
497            path,
498            mut index,
499        } = index_with(&[]);
500        let mut expected = Vec::new();
501        for number in 0..100 {
502            let entry = stamp_entry(number * 3, number * 3 + 1, 10);
503            index.upsert(entry).unwrap();
504            expected.push(entry);
505        }
506        // Replace older append-time stamps and insert late transactional
507        // ranges into the gaps without changing canonical ordering.
508        for number in (0..100).rev() {
509            index
510                .upsert(stamp_entry(number * 3, number * 3 + 1, 20))
511                .unwrap();
512            index
513                .upsert(stamp_entry(number * 3 + 2, number * 3 + 2, 30))
514                .unwrap();
515        }
516        expected = expected
517            .into_iter()
518            .flat_map(|mut entry| {
519                entry.stamp = 20;
520                [
521                    entry,
522                    stamp_entry(entry.last_offset.0 + 1, entry.last_offset.0 + 1, 30),
523                ]
524            })
525            .collect();
526        assert2::assert!(index.entries() == expected);
527        assert2::assert!(StampIndex::open(path.clone()).unwrap().entries() == expected);
528        let mut bytes = STAMP_INDEX_VERSION.to_be_bytes().to_vec();
529        for entry in &expected {
530            bytes.extend_from_slice(StampEntryRaw::new(*entry).as_bytes());
531        }
532        // The last middle insert appends physically; a retained-entry rewrite
533        // canonicalizes the file just as the old implementation does.
534        index.truncate_from(Offset(299)).unwrap();
535        expected.pop();
536        bytes.truncate(bytes.len() - ENTRY_BYTES);
537        assert2::assert!(index.entries() == expected);
538        assert2::assert!(fs::read(path).unwrap() == bytes);
539    }
540
541    #[test]
542    fn uncertain_sync_retry_is_canonical_after_reopen() {
543        #[derive(Debug)]
544        struct FailSyncOnce(AtomicBool);
545        impl LogIo for FailSyncOnce {
546            fn sync_file(&self, _target: IoTarget, file: &File) -> io::Result<()> {
547                if self.0.swap(false, Ordering::SeqCst) {
548                    Err(io::Error::other("injected uncertain stamp sync"))
549                } else {
550                    file.sync_data()
551                }
552            }
553        }
554        let IndexFixture {
555            dir: _dir,
556            path,
557            mut index,
558        } = index_with(&[]);
559        let first = stamp_entry(0, 2, 100);
560        index.append(first).unwrap();
561        index.set_io(Arc::new(FailSyncOnce(AtomicBool::new(true))));
562        let second = stamp_entry(5, 8, 200);
563        assert2::assert!(let Err(LogError::Io(_)) = index.upsert(second));
564        assert2::assert!(index.entries() == [first]);
565        index.upsert(second).unwrap();
566        assert2::assert!(StampIndex::open(path).unwrap().entries() == [first, second]);
567    }
568
569    #[test]
570    fn searches_preserve_i64_boundary_ranges() {
571        let entries = [
572            stamp_entry(i64::MIN, i64::MIN + 1, 1),
573            stamp_entry(i64::MAX - 1, i64::MAX, 2),
574        ];
575        let IndexFixture {
576            dir: _dir, index, ..
577        } = index_with(&entries);
578        for (offset, expected) in [(i64::MIN, Some(1)), (0, None), (i64::MAX, Some(2))] {
579            assert2::assert!(index.stamp_for_offset(Offset(offset)) == expected);
580        }
581        assert2::assert!(
582            StampIndex::insertion_position(&entries, stamp_entry(-1, 1, 3)) == Some(1)
583        );
584        assert2::assert!(
585            StampIndex::insertion_position(&entries, stamp_entry(i64::MIN, i64::MAX, 3)).is_none()
586        );
587    }
588
589    #[yare::parameterized(
590        hundred = { 100 },
591        thousand = { 1_000 },
592        ten_thousand = { 10_000 },
593    )]
594    fn sequential_upserts_append_one_fixed_width_entry_each(ranges: i64) {
595        #[derive(Debug, Default)]
596        struct CountSyncs(AtomicUsize);
597        impl LogIo for CountSyncs {
598            fn sync_file(&self, _target: IoTarget, _file: &File) -> io::Result<()> {
599                self.0.fetch_add(1, Ordering::Relaxed);
600                // This scaling regression measures index operations and
601                // exact file bytes, excluding device fsync latency.
602                Ok(())
603            }
604        }
605        let IndexFixture {
606            dir: _dir,
607            path,
608            mut index,
609        } = index_with(&[]);
610        let io = Arc::new(CountSyncs::default());
611        index.set_io(Arc::<CountSyncs>::clone(&io));
612        for number in 0..ranges {
613            index
614                .upsert(stamp_entry(
615                    number * 3,
616                    number * 3 + 1,
617                    u64::try_from(number).unwrap(),
618                ))
619                .unwrap();
620        }
621        let count = usize::try_from(ranges).unwrap();
622        assert2::assert!(index.entries().len() == count);
623        assert2::assert!(io.0.load(Ordering::Relaxed) == count);
624        assert2::assert!(
625            fs::metadata(&path).unwrap().len()
626                == u64::try_from(HEADER_BYTES + ENTRY_BYTES * count).unwrap()
627        );
628        assert2::assert!(StampIndex::open(path).unwrap().entries() == index.entries());
629    }
630
631    /// The two entries the golden fixture holds, in offset order.
632    const GOLDEN_ENTRIES: [StampEntry; 2] = [
633        StampEntry {
634            base_offset: Offset(5),
635            last_offset: Offset(7),
636            stamp: 0x0102_0304_0506_0708,
637        },
638        StampEntry {
639            base_offset: Offset(10),
640            last_offset: Offset(12),
641            stamp: 2_000,
642        },
643    ];
644
645    /// `GOLDEN_ENTRIES` as a 1.0 broker writes them: the version header,
646    /// then each entry's base offset, last offset and stamp, big-endian.
647    #[rustfmt::skip]
648    const GOLDEN_BYTES: [u8; 50] = [
649        0x00, 0x00, // version 0
650        0, 0, 0, 0, 0, 0, 0, 5, // base_offset 5
651        0, 0, 0, 0, 0, 0, 0, 7, // last_offset 7
652        1, 2, 3, 4, 5, 6, 7, 8, // stamp
653        0, 0, 0, 0, 0, 0, 0, 10, // base_offset 10
654        0, 0, 0, 0, 0, 0, 0, 12, // last_offset 12
655        0, 0, 0, 0, 0, 0, 0x07, 0xd0, // stamp 2000
656    ];
657
658    /// The writer lays the file out byte for byte as the 1.x contract fixes
659    /// it, whether the entries arrive by append or by a rewrite.
660    #[test]
661    fn the_writer_produces_the_golden_bytes() {
662        let dir = TempDir::new().unwrap();
663        let appended = dir.path().join("appended.stampindex");
664        let mut idx = StampIndex::open(appended.clone()).unwrap();
665        // Out of order, so the second write lands on a file that already has
666        // its header.
667        idx.append(GOLDEN_ENTRIES[1]).unwrap();
668        idx.append(GOLDEN_ENTRIES[0]).unwrap();
669        // A rewrite lays the entries out in offset order.
670        let rewritten = dir.path().join("rewritten.stampindex");
671        let mut idx = StampIndex::open(rewritten.clone()).unwrap();
672        for entry in GOLDEN_ENTRIES {
673            idx.append(entry).unwrap();
674        }
675        idx.append(StampEntry {
676            base_offset: Offset(20),
677            last_offset: Offset(20),
678            stamp: 9,
679        })
680        .unwrap();
681        idx.truncate_from(Offset(20)).unwrap();
682
683        let mut appended_golden = GOLDEN_BYTES[..2].to_vec();
684        appended_golden.extend_from_slice(&GOLDEN_BYTES[26..]);
685        appended_golden.extend_from_slice(&GOLDEN_BYTES[2..26]);
686        assert2::assert!(fs::read(appended).unwrap() == appended_golden);
687        assert2::assert!(fs::read(rewritten).unwrap() == GOLDEN_BYTES);
688    }
689
690    #[test]
691    fn the_golden_bytes_decode_to_their_entries() {
692        let dir = TempDir::new().unwrap();
693        let path = dir.path().join("golden.stampindex");
694        fs::write(&path, GOLDEN_BYTES).unwrap();
695        assert2::assert!(StampIndex::open(path).unwrap().entries() == GOLDEN_ENTRIES);
696    }
697
698    /// A version other than 0 is refused with the version it found, and a
699    /// headerless file, the layout a broker before 1.0 wrote, is refused as
700    /// one that predates 1.0.
701    #[yare::parameterized(
702        one = { "one", 1_i16 },
703        negative = { "negative", -1_i16 },
704        maximum = { "max", i16::MAX },
705    )]
706    fn open_refuses_an_unknown_version(name: &str, version: i16) {
707        let dir = TempDir::new().unwrap();
708        let path = dir.path().join(format!("{name}.stampindex"));
709        let mut bytes = GOLDEN_BYTES.to_vec();
710        bytes[..2].copy_from_slice(&version.to_be_bytes());
711        fs::write(&path, bytes).unwrap();
712        let error = StampIndex::open(path.clone()).unwrap_err();
713        assert2::assert!(
714            let LogError::UnsupportedFormatVersion {
715                artifact: "stampindex",
716                path: found_path,
717                found,
718            } = error
719        );
720        assert2::assert!((found_path, found) == (path, i64::from(version)));
721    }
722
723    #[yare::parameterized(
724        one_entry = { "one entry", 1 },
725        two_entries = { "two entries", 2 },
726    )]
727    fn open_refuses_a_headerless_file(name: &str, entries: usize) {
728        let dir = TempDir::new().unwrap();
729        let path = dir.path().join(format!("{name}.stampindex"));
730        fs::write(&path, &GOLDEN_BYTES[2..2 + entries * ENTRY_BYTES]).unwrap();
731        let error = StampIndex::open(path.clone()).unwrap_err();
732        assert2::assert!(
733            let LogError::MissingFormatVersion { artifact: "stampindex", path: found_path } =
734                error
735        );
736        assert2::assert!(found_path == path);
737    }
738
739    /// A missing file, an empty one, and one that holds only its header all
740    /// mean zero stamped ranges.
741    #[yare::parameterized(
742        missing = { "missing", None },
743        empty = { "empty", Some(Vec::new()) },
744        header_only = { "header only", Some(GOLDEN_BYTES[..2].to_vec()) },
745    )]
746    fn stamp_empty_file_yields_empty_entries(name: &str, bytes: Option<Vec<u8>>) {
747        let dir = TempDir::new().unwrap();
748        let path = dir.path().join(format!("{name}.stampindex"));
749        if let Some(bytes) = bytes {
750            fs::write(&path, bytes).unwrap();
751        }
752        let idx = StampIndex::open(path).unwrap();
753        assert2::assert!(idx.entries() == &[]);
754    }
755
756    #[test]
757    fn stamp_append_round_trips_through_disk() {
758        let IndexFixture {
759            dir: _dir,
760            path,
761            index: _idx,
762        } = index_with(&[stamp_entry(5, 7, 1_000), stamp_entry(10, 12, 2_000)]);
763
764        let idx2 = StampIndex::open(path).unwrap();
765        assert2::assert!(
766            idx2.entries() == &[stamp_entry(5, 7, 1_000), stamp_entry(10, 12, 2_000),]
767        );
768    }
769
770    /// Only a real `NotFound` means "no index yet". Every other I/O error
771    /// must surface as `LogError::Io`. Here the path is a directory, so the
772    /// read fails with a kind other than `NotFound`. To swallow that error
773    /// would return an empty index over a real failure.
774    #[test]
775    fn open_surfaces_non_notfound_io_error() {
776        let dir = TempDir::new().unwrap();
777        let err = StampIndex::open(dir.path().to_path_buf()).unwrap_err();
778        assert2::assert!(let LogError::Io(_) = err);
779    }
780
781    /// A torn header, or entry bytes that are not a whole number of
782    /// entries, is corrupt.
783    #[yare::parameterized(
784        torn_header = { 1 },
785        partial_first_entry = { HEADER_BYTES + 1 },
786        partial_second_entry = { HEADER_BYTES + ENTRY_BYTES + 1 },
787    )]
788    fn stamp_corrupt_length_is_rejected(len: usize) {
789        let dir = TempDir::new().unwrap();
790        let path = dir.path().join(format!("{len}.stampindex"));
791        fs::write(&path, vec![0_u8; len]).unwrap();
792        let err = StampIndex::open(path).unwrap_err();
793        assert2::assert!(let LogError::Corrupt(_) = err, "length {len}");
794    }
795
796    #[test]
797    fn open_canonicalizes_exact_retries() {
798        let dir = TempDir::new().unwrap();
799        let first = stamp_entry(0, 2, 100);
800        let second = stamp_entry(10, 12, 200);
801        let path = dir.path().join("retries.stampindex");
802        write_entries(&path, &[second, first, second]);
803        assert2::assert!(StampIndex::open(path).unwrap().entries() == [first, second]);
804    }
805
806    #[yare::parameterized(
807        inverted = { "inverted", vec![stamp_entry(7, 6, 1)] },
808        overlap = { "overlap", vec![stamp_entry(0, 4, 1), stamp_entry(4, 8, 2)] },
809        conflicting_retry = {
810            "conflicting-retry",
811            vec![stamp_entry(0, 4, 1), stamp_entry(0, 4, 2)],
812        },
813    )]
814    fn open_rejects_malformed_ranges(name: &str, entries: Vec<StampEntry>) {
815        let dir = TempDir::new().unwrap();
816        let path = dir.path().join(format!("{name}.stampindex"));
817        write_entries(&path, &entries);
818        assert2::assert!(let LogError::Corrupt(_) = StampIndex::open(path).unwrap_err());
819    }
820
821    #[test]
822    fn out_of_order_append_stays_canonical_across_reopen() {
823        let IndexFixture {
824            dir: _dir,
825            path,
826            index: mut idx,
827        } = index_with(&[]);
828        let first = stamp_entry(0, 2, 100);
829        let second = stamp_entry(10, 12, 200);
830        idx.append(second).unwrap();
831        idx.append(first).unwrap();
832        assert2::assert!(idx.entries() == [first, second]);
833        assert2::assert!(StampIndex::open(path).unwrap().entries() == [first, second]);
834    }
835
836    #[test]
837    fn mutation_io_failures_leave_the_in_memory_index_unchanged() {
838        let dir = TempDir::new().unwrap();
839        let path = dir.path().join("00.stampindex");
840        let original = stamp_entry(0, 2, 100);
841        let mut idx = StampIndex::open(path.clone()).unwrap();
842        idx.append(original).unwrap();
843        fs::remove_file(&path).unwrap();
844        fs::create_dir(&path).unwrap();
845
846        let replacement = StampEntry {
847            stamp: 200,
848            ..original
849        };
850        assert2::assert!(let LogError::Io(_) = idx.upsert(replacement).unwrap_err());
851        assert2::assert!(idx.entries() == [original]);
852        assert2::assert!(
853            let LogError::Io(_) = idx.remove_ranges(&[OffsetRange {
854                first: Offset(0),
855                last: Offset(2),
856            }]).unwrap_err()
857        );
858        assert2::assert!(idx.entries() == [original]);
859        assert2::assert!(
860            let LogError::Io(_) = idx
861                .append(stamp_entry(4, 5, 300))
862                .unwrap_err()
863        );
864        assert2::assert!(idx.entries() == [original]);
865    }
866
867    #[yare::parameterized(
868        first_start = { Offset(0), Some(100) },
869        first_end = { Offset(4), Some(100) },
870        second_start = { Offset(10), Some(200) },
871        second_end = { Offset(14), Some(200) },
872        gap = { Offset(5), None },
873        past_end = { Offset(15), None },
874    )]
875    fn stamp_for_offset_finds_covering_range(offset: Offset, expected: Option<u64>) {
876        let IndexFixture {
877            dir: _dir,
878            path: _path,
879            index: idx,
880        } = index_with(&[stamp_entry(0, 4, 100), stamp_entry(10, 14, 200)]);
881
882        // Inclusive endpoints and interior offsets resolve to their range.
883        assert2::assert!(idx.stamp_for_offset(offset) == expected);
884    }
885
886    #[test]
887    fn append_rejects_overlapping_ranges_but_accepts_exact_retry() {
888        let IndexFixture {
889            dir: _dir,
890            path: _path,
891            index: mut idx,
892        } = index_with(&[]);
893        let entry = stamp_entry(5, 7, 100);
894        idx.append(entry).unwrap();
895        idx.append(entry).unwrap();
896        assert2::assert!(idx.entries() == [entry]);
897
898        let error = idx.append(stamp_entry(7, 9, 200)).unwrap_err();
899        assert2::assert!(let LogError::Corrupt(_) = error);
900
901        let error = idx.append(stamp_entry(10, 9, 300)).unwrap_err();
902        assert2::assert!(let LogError::InvalidArgument(_) = error);
903    }
904
905    #[test]
906    fn upsert_replaces_only_an_exact_range() {
907        let IndexFixture {
908            dir: _dir,
909            path,
910            index: mut idx,
911        } = index_with(&[stamp_entry(5, 7, 100)]);
912
913        idx.upsert(stamp_entry(5, 7, 200)).unwrap();
914        assert2::assert!(idx.stamp_for_offset(Offset(6)) == Some(200));
915        assert2::assert!(StampIndex::open(path).unwrap().entries() == idx.entries());
916    }
917
918    #[yare::parameterized(
919        extends_existing_range = { 5, 8 },
920        starts_inside_existing_range = { 4, 7 },
921    )]
922    fn upsert_rejects_non_exact_range(base: i64, last: i64) {
923        let IndexFixture {
924            dir: _dir,
925            path: _path,
926            index: mut idx,
927        } = index_with(&[stamp_entry(5, 7, 100)]);
928        let error = idx.upsert(stamp_entry(base, last, 300)).unwrap_err();
929        assert2::assert!(let LogError::Corrupt(_) = error);
930    }
931
932    #[test]
933    fn truncate_from_removes_tail_entries_on_disk() {
934        let IndexFixture {
935            dir: _dir,
936            path,
937            index: mut idx,
938        } = index_with(&[
939            stamp_entry(0, 2, 100),
940            stamp_entry(3, 6, 103),
941            stamp_entry(10, 12, 110),
942        ]);
943
944        idx.truncate_from(Offset(6)).unwrap();
945        assert2::assert!(StampIndex::open(path).unwrap().entries() == [stamp_entry(0, 2, 100)]);
946    }
947
948    #[test]
949    fn remove_ranges_requires_both_exact_boundaries() {
950        let IndexFixture {
951            dir: _dir,
952            path,
953            index: mut idx,
954        } = index_with(&[stamp_entry(0, 1, 10), stamp_entry(2, 3, 20)]);
955
956        idx.remove_ranges(&[
957            OffsetRange {
958                first: Offset(0),
959                last: Offset(9),
960            },
961            OffsetRange {
962                first: Offset(9),
963                last: Offset(3),
964            },
965        ])
966        .unwrap();
967        assert2::assert!(idx.entries().len() == 2);
968        idx.remove_ranges(&[OffsetRange {
969            first: Offset(2),
970            last: Offset(3),
971        }])
972        .unwrap();
973
974        assert2::assert!(StampIndex::open(path).unwrap().entries() == [stamp_entry(0, 1, 10)]);
975    }
976
977    #[test]
978    fn append_rejects_overlapping_single_offset_range_as_corrupt() {
979        let IndexFixture {
980            dir: _dir,
981            path: _path,
982            index: mut idx,
983        } = index_with(&[stamp_entry(0, 5, 10)]);
984
985        // Overlapping single-offset range (last_offset == base_offset)
986        let err = idx.append(stamp_entry(2, 2, 20)).unwrap_err();
987        assert2::assert!(let LogError::Corrupt(_) = err);
988    }
989
990    #[test]
991    fn set_io_routes_stamp_index_writes() {
992        use std::sync::atomic::{AtomicBool, Ordering};
993        #[derive(Debug)]
994        struct SpyIo(Arc<AtomicBool>);
995        impl LogIo for SpyIo {
996            fn write_at(&self, _t: IoTarget, file: &File, buf: &[u8]) -> io::Result<usize> {
997                use std::io::Write;
998                self.0.store(true, Ordering::SeqCst);
999                (&*file).write(buf)
1000            }
1001        }
1002        let called = Arc::new(AtomicBool::new(false));
1003        let IndexFixture {
1004            dir: _dir,
1005            path: _path,
1006            index: mut idx,
1007        } = index_with(&[]);
1008        idx.set_io(Arc::new(SpyIo(Arc::clone(&called))));
1009        idx.append(stamp_entry(0, 0, 1)).unwrap();
1010        assert2::assert!(called.load(Ordering::SeqCst));
1011    }
1012}
krabka_log/stamp_index.rs

krabka_log/
stamp_index.rs
