krabka_log/
txn_index.rs

1//! Per-segment `.txnindex` file. One fixed-width, version-prefixed record per
2//! aborted transaction in the segment:
3//!
4//!   `version`:             i16 (big-endian), always 0
5//!   `producer_id`:         i64 (big-endian)
6//!   `first_offset`:        i64 (big-endian)
7//!   `last_offset`:         i64 (big-endian)
8//!   `last_stable_offset`:  i64 (big-endian)
9//!
10//! The byte layout matches Apache Kafka's `TransactionIndex` (a version
11//! prefix on every record, derived from `AbortedTxn.json`'s field order), so
12//! `kafka-dump-log --offsets-decoder` can dump it.
13
14use {
15    crate::{
16        error::LogError,
17        index::{read_sidecar, rewrite_retained_entries, rewrite_sidecar_file},
18    },
19    krabka_ids::{Offset, ProducerId},
20    std::{collections::HashSet, fs::OpenOptions, io::Write, mem, path::PathBuf},
21    tracing::instrument,
22    zerocopy::{
23        BigEndian, FromBytes, Immutable, IntoBytes, KnownLayout, Unaligned,
24        byteorder::{I16, I64},
25    },
26};
27
28const ENTRY_BYTES: usize = 34;
29
30/// The only aborted-transaction record version krabka writes or accepts.
31const SUPPORTED_VERSION: i16 = 0;
32
33#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
34pub struct AbortedTxn {
35    pub start_offset: Offset,
36    pub last_offset: Offset,
37    pub producer_id: ProducerId,
38    /// The last stable offset at the moment this transaction aborted, that
39    /// is, Kafka's `ProducerStateManager.lastStableOffset(completedTxn)`.
40    pub last_stable_offset: Offset,
41}
42
43impl AbortedTxn {
44    /// Whether the transaction's records overlap the offsets `[start, end)`:
45    /// the test [`TxnIndex::aborted_in_range`] applies, so a caller that
46    /// collected the transactions of a wider range can narrow them to the
47    /// bytes it served without going back to the index.
48    #[must_use]
49    pub fn overlaps(&self, start: Offset, end: Offset) -> bool {
50        krabka_verified::aborted_transaction_overlaps(
51            self.start_offset.0,
52            self.last_offset.0,
53            start.0,
54            end.0,
55        )
56    }
57}
58
59/// On-disk byte layout of one `AbortedTxn` entry. `zerocopy` reinterprets it
60/// in place from the file bytes.
61#[derive(Debug, Clone, Copy, FromBytes, IntoBytes, KnownLayout, Immutable, Unaligned)]
62#[repr(C)]
63struct AbortedTxnRaw {
64    version: I16<BigEndian>,
65    producer_id: I64<BigEndian>,
66    first_offset: I64<BigEndian>,
67    last_offset: I64<BigEndian>,
68    last_stable_offset: I64<BigEndian>,
69}
70
71const _: [(); ENTRY_BYTES] = [(); mem::size_of::<AbortedTxnRaw>()];
72
73impl AbortedTxnRaw {
74    fn new(entry: AbortedTxn) -> Self {
75        Self {
76            version: I16::new(SUPPORTED_VERSION),
77            producer_id: I64::new(entry.producer_id.0),
78            first_offset: I64::new(entry.start_offset.0),
79            last_offset: I64::new(entry.last_offset.0),
80            last_stable_offset: I64::new(entry.last_stable_offset.0),
81        }
82    }
83}
84
85#[derive(Debug)]
86pub struct TxnIndex {
87    path: PathBuf,
88    entries: Vec<AbortedTxn>,
89    /// Exact identities already represented by `entries`. Keep the vector for
90    /// Kafka's first-occurrence iteration order and this set for retry lookup.
91    identities: HashSet<AbortedTxn>,
92    /// Whether the file exists: it did at open, or a write created it. A
93    /// segment with no aborted transaction usually has none.
94    on_disk: bool,
95}
96
97impl TxnIndex {
98    #[krabka_macros::index_open]
99    /// Open or recover a `.txnindex` file at the given path. This method
100    /// reads the entire file into memory at startup. An empty file or a
101    /// missing file is acceptable and means zero aborted transactions.
102    /// # Errors
103    /// Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.
104    pub fn open(path: PathBuf) -> Result<Self, LogError> {
105        let read = read_sidecar::<AbortedTxnRaw, _>(
106            &path,
107            "txnindex",
108            "length is a multiple of ENTRY_BYTES and AbortedTxnRaw is Unaligned",
109            |raws| {
110                let mut entries = Vec::with_capacity(raws.len());
111                let mut identities = HashSet::with_capacity(raws.len());
112                for raw in raws {
113                    if raw.version.get() != SUPPORTED_VERSION {
114                        return Err(LogError::Corrupt(format!(
115                            "txnindex {} contains an unsupported record version {}",
116                            path.display(),
117                            raw.version.get(),
118                        )));
119                    }
120                    let entry = AbortedTxn {
121                        start_offset: Offset(raw.first_offset.get()),
122                        last_offset: Offset(raw.last_offset.get()),
123                        producer_id: ProducerId(raw.producer_id.get()),
124                        last_stable_offset: Offset(raw.last_stable_offset.get()),
125                    };
126                    if !Self::entry_valid(entry) {
127                        return Err(LogError::Corrupt(format!(
128                            "txnindex {} contains an invalid aborted interval",
129                            path.display()
130                        )));
131                    }
132                    if identities.insert(entry) {
133                        entries.push(entry);
134                    }
135                }
136                Ok((entries, identities))
137            },
138        )?;
139        // A missing file is an index with no entries.
140        let on_disk = read.is_some();
141        let (entries, identities) = read.unwrap_or_default();
142        tracing::Span::current().record("entries", entries.len());
143        Ok(Self {
144            path,
145            entries,
146            identities,
147            on_disk,
148        })
149    }
150
151    /// Append one aborted-txn entry.
152    #[instrument(
153        level = "debug",
154        skip(self),
155        fields(producer_id = entry.producer_id.0),
156        err,
157    )]
158    /// # Errors
159    /// Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.
160    pub fn append(&mut self, entry: AbortedTxn) -> Result<(), LogError> {
161        if !Self::entry_valid(entry) {
162            return Err(LogError::InvalidArgument(format!(
163                "invalid aborted transaction {}..={} for producer {}",
164                entry.start_offset, entry.last_offset, entry.producer_id
165            )));
166        }
167        if self.identities.contains(&entry) {
168            return Ok(());
169        }
170        let mut f = OpenOptions::new()
171            .create(true)
172            .append(true)
173            .open(&self.path)
174            .map_err(LogError::Io)?;
175        self.on_disk = true;
176        let raw = AbortedTxnRaw::new(entry);
177        f.write_all(raw.as_bytes()).map_err(LogError::Io)?;
178        f.sync_data().map_err(LogError::Io)?;
179        self.entries.push(entry);
180        self.identities.insert(entry);
181        Ok(())
182    }
183
184    /// Remove entries whose inclusive end reaches the truncated suffix and
185    /// rewrite the sidecar before it is reused as the active transaction index.
186    ///
187    /// # Errors
188    ///
189    /// Returns an I/O error if the transaction-index sidecar cannot be
190    /// rewritten or synchronized.
191    pub fn truncate_from(&mut self, offset: Offset) -> Result<(), LogError> {
192        if let Some(entries) = rewrite_retained_entries(
193            &self.entries,
194            |entry| entry.last_offset < offset,
195            |entries| {
196                let mut file = rewrite_sidecar_file(&self.path)?;
197                for entry in entries {
198                    let raw = AbortedTxnRaw::new(*entry);
199                    file.write_all(raw.as_bytes()).map_err(LogError::Io)?;
200                }
201                file.sync_data().map_err(LogError::Io)
202            },
203        )? {
204            self.entries = entries;
205            self.identities = self.entries.iter().copied().collect();
206        }
207        Ok(())
208    }
209
210    #[must_use]
211    pub fn entries(&self) -> &[AbortedTxn] { &self.entries }
212
213    /// Whether the index has a file on disk.
214    #[must_use]
215    pub fn on_disk(&self) -> bool { self.on_disk }
216
217    fn entry_valid(entry: AbortedTxn) -> bool {
218        krabka_verified::aborted_transaction_interval(
219            Some(entry.start_offset.0),
220            entry.last_offset.0,
221            entry.producer_id.get(),
222        ) == Some((entry.start_offset.0, entry.last_offset.0))
223    }
224
225    /// Kafka's `TransactionIndex.collectAbortedTxns`: push every aborted
226    /// transaction whose offset range overlaps `[start, end)` onto `out`, in
227    /// index order, and return whether the search is complete.
228    ///
229    /// An entry's `last_stable_offset` is the first offset of any transaction
230    /// still open when it aborted. Once that reaches `end`, every transaction
231    /// that overlaps the range had completed by then, and its entry is no
232    /// later than this one, so the search stops here and skips the rest of
233    /// this index and every later segment's.
234    pub fn collect_aborted(&self, start: Offset, end: Offset, out: &mut Vec<AbortedTxn>) -> bool {
235        for entry in &self.entries {
236            if krabka_verified::aborted_transaction_overlaps(
237                entry.start_offset.0,
238                entry.last_offset.0,
239                start.0,
240                end.0,
241            ) {
242                out.push(*entry);
243            }
244            if entry.last_stable_offset >= end {
245                return true;
246            }
247        }
248        false
249    }
250
251    /// Aborted transactions whose offset range overlaps `[start, end)`.
252    pub fn aborted_in_range(
253        &self,
254        start: Offset,
255        end: Offset,
256    ) -> impl Iterator<Item = &AbortedTxn> {
257        self.entries.iter().filter(move |entry| {
258            krabka_verified::aborted_transaction_overlaps(
259                entry.start_offset.0,
260                entry.last_offset.0,
261                start.0,
262                end.0,
263            )
264        })
265    }
266}
267
268#[cfg(test)]
269mod tests {
270
271    use {
272        super::*,
273        crate::{index::IndexFixture, test_support::aborted_txn},
274        std::{fs, iter, path::Path},
275        tempfile::TempDir,
276    };
277
278    fn populated_index(name: &str, entries: &[AbortedTxn]) -> IndexFixture<TxnIndex> {
279        let dir = TempDir::new().unwrap();
280        let path = dir.path().join(name);
281        let mut index = TxnIndex::open(path.clone()).unwrap();
282        for &entry in entries {
283            index.append(entry).unwrap();
284        }
285        IndexFixture { dir, path, index }
286    }
287
288    /// The same three-offset entry seeds retry, mutation and disk-round-trip cases.
289    fn retry_entry() -> AbortedTxn { aborted_txn().bounds(Offset(5)..=Offset(7)).stable(8).call() }
290
291    #[derive(Clone, Copy, krabka_macros::FieldDefaults)]
292    struct RangedIndexSetup<'a> {
293        #[default("00.txnindex")]
294        name: &'a str,
295    }
296
297    /// One transaction covers offsets 10 through 20 for fetch-boundary checks.
298    #[bon::builder]
299    fn ranged_index(#[builder(default = "00.txnindex")] name: &'_ str) -> IndexFixture<TxnIndex> {
300        let setup = RangedIndexSetup { name };
301        populated_index(setup.name, &[aborted_txn()
302            .producer(1)
303            .bounds(Offset(10)..=Offset(20))
304            .stable(21)
305            .call()])
306    }
307
308    /// A malformed entry and the name of what is wrong with it.
309    struct MalformedEntry {
310        name: &'static str,
311        entry: AbortedTxn,
312    }
313
314    /// The two independently specified malformed entries are refused both at append and open.
315    fn malformed_entries() -> [MalformedEntry; 2] {
316        [
317            MalformedEntry {
318                name: "inverted",
319                entry: aborted_txn()
320                    .producer(1)
321                    .bounds(Offset(8)..=Offset(7))
322                    .stable(8)
323                    .call(),
324            },
325            MalformedEntry {
326                name: "negative-producer",
327                entry: aborted_txn()
328                    .producer(-1)
329                    .bounds(Offset(7)..=Offset(8))
330                    .stable(9)
331                    .call(),
332            },
333        ]
334    }
335
336    fn write_entry(path: &Path, entry: AbortedTxn) {
337        let raw = AbortedTxnRaw::new(entry);
338        fs::write(path, raw.as_bytes()).unwrap();
339    }
340
341    /// The exact 34-byte wire layout of one aborted-transaction record:
342    /// a big-endian i16 version of 0, then `producer_id`, `first_offset`,
343    /// `last_offset`, and `last_stable_offset`, each a big-endian i64.
344    #[test]
345    fn append_writes_the_exact_kafka_wire_layout() {
346        let entry = aborted_txn()
347            .bounds(Offset(5)..=Offset(9))
348            .stable(10)
349            .call();
350        let IndexFixture {
351            dir: _dir,
352            path,
353            index: _index,
354        } = populated_index("00.txnindex", &[entry]);
355
356        let bytes = fs::read(&path).unwrap();
357        let mut expected = Vec::new();
358        expected.extend_from_slice(&0_i16.to_be_bytes()); // version
359        expected.extend_from_slice(&1000_i64.to_be_bytes()); // producer_id
360        expected.extend_from_slice(&5_i64.to_be_bytes()); // first_offset
361        expected.extend_from_slice(&9_i64.to_be_bytes()); // last_offset
362        expected.extend_from_slice(&10_i64.to_be_bytes()); // last_stable_offset
363        assert2::assert!(bytes.len() == ENTRY_BYTES);
364        assert2::assert!(bytes == expected);
365    }
366
367    /// Three aborted transactions produce a file whose length is exactly
368    /// `3 * 34` and whose decoded entries round-trip exactly.
369    #[test]
370    fn three_entries_round_trip_at_the_exact_kafka_entry_size() {
371        let dir = TempDir::new().unwrap();
372        let path = dir.path().join("00.txnindex");
373        let entries = [
374            aborted_txn().call(),
375            aborted_txn()
376                .producer(2000)
377                .bounds(Offset(4)..=Offset(6))
378                .stable(7)
379                .call(),
380            aborted_txn()
381                .producer(3000)
382                .bounds(Offset(8)..=Offset(10))
383                .stable(11)
384                .call(),
385        ];
386        let mut index = TxnIndex::open(path.clone()).unwrap();
387        for entry in entries {
388            index.append(entry).unwrap();
389        }
390
391        let bytes = fs::read(&path).unwrap();
392        assert2::assert!(bytes.len() == 3 * ENTRY_BYTES);
393
394        let reopened = TxnIndex::open(path).unwrap();
395        assert2::assert!(reopened.entries() == entries);
396    }
397
398    /// A record whose version prefix is not 0 is refused on open.
399    #[test]
400    fn an_unsupported_version_is_refused_on_open() {
401        let dir = TempDir::new().unwrap();
402        let path = dir.path().join("00.txnindex");
403        let mut bytes = Vec::new();
404        bytes.extend_from_slice(&1_i16.to_be_bytes()); // unsupported version
405        bytes.extend_from_slice(&1000_i64.to_be_bytes());
406        bytes.extend_from_slice(&0_i64.to_be_bytes());
407        bytes.extend_from_slice(&3_i64.to_be_bytes());
408        bytes.extend_from_slice(&4_i64.to_be_bytes());
409        fs::write(&path, bytes).unwrap();
410        assert2::assert!(let LogError::Corrupt(_) = TxnIndex::open(path).unwrap_err());
411    }
412
413    /// The range end is exclusive: an aborted transaction that begins exactly
414    /// where the fetch ends is not in it.
415    ///
416    /// A consumer uses this list to skip aborted records inside the batch range
417    /// it was handed. Including one that starts past the end makes it skip
418    /// records it was never sent.
419    #[test]
420    fn an_aborted_txn_starting_at_the_range_end_is_outside_it() {
421        let IndexFixture {
422            dir: _dir,
423            path: _path,
424            index,
425        } = ranged_index().name("00000000000000000000.txnindex").call();
426
427        let found =
428            |start: i64, end: i64| index.aborted_in_range(Offset(start), Offset(end)).count();
429        assert2::check!(found(0, 10) == 0, "a fetch ending where the txn begins");
430        assert2::check!(found(0, 11) == 1, "a fetch reaching one past its start");
431        assert2::check!(found(20, 30) == 1, "a fetch starting on its last offset");
432        assert2::check!(found(21, 30) == 0, "a fetch starting past its last offset");
433    }
434
435    #[test]
436    fn empty_file_yields_empty_entries() {
437        let dir = TempDir::new().unwrap();
438        let path = dir.path().join("00.txnindex");
439        let idx = TxnIndex::open(path).unwrap();
440        assert2::assert!(idx.entries() == &[]);
441    }
442
443    #[test]
444    fn malformed_intervals_fail_closed_at_append_and_open() {
445        let dir = TempDir::new().unwrap();
446        let path = dir.path().join("00.txnindex");
447        let mut index = TxnIndex::open(path).unwrap();
448        for MalformedEntry { entry, .. } in malformed_entries() {
449            assert2::assert!(let LogError::InvalidArgument(_) = index.append(entry).unwrap_err());
450        }
451        assert2::assert!(index.entries().is_empty());
452
453        let partial = dir.path().join("partial.txnindex");
454        fs::write(&partial, [0_u8]).unwrap();
455        assert2::assert!(let LogError::Corrupt(_) = TxnIndex::open(partial).unwrap_err());
456
457        for MalformedEntry { name, entry } in malformed_entries() {
458            let path = dir.path().join(format!("{name}.txnindex"));
459            write_entry(&path, entry);
460            assert2::assert!(let LogError::Corrupt(_) = TxnIndex::open(path).unwrap_err());
461        }
462    }
463
464    #[test]
465    fn exact_aborted_interval_retry_is_idempotent() {
466        let entry = retry_entry();
467        let IndexFixture {
468            dir: _dir,
469            path,
470            mut index,
471        } = populated_index("00.txnindex", &[entry]);
472        index.append(entry).unwrap();
473        assert2::assert!(index.entries() == [entry]);
474
475        let raw = AbortedTxnRaw::new(entry);
476        let mut bytes = raw.as_bytes().to_vec();
477        bytes.extend_from_slice(raw.as_bytes());
478        fs::write(&path, bytes).unwrap();
479        assert2::assert!(TxnIndex::open(path).unwrap().entries() == [entry]);
480    }
481
482    /// A retry can be separated from its first write by other records (for
483    /// example after an uncertain sync and restart). Identity includes every
484    /// persisted field, while iteration retains each identity's first place.
485    #[test]
486    fn reopen_deduplicates_non_adjacent_retries_by_all_fields_in_first_seen_order() {
487        let dir = TempDir::new().unwrap();
488        let path = dir.path().join("00.txnindex");
489        let first = retry_entry();
490        let changed_producer = AbortedTxn {
491            producer_id: ProducerId(first.producer_id.0 + 1),
492            ..first
493        };
494        let changed_start = AbortedTxn {
495            start_offset: Offset(first.start_offset.0 + 1),
496            ..first
497        };
498        let changed_last = AbortedTxn {
499            last_offset: Offset(first.last_offset.0 + 1),
500            ..first
501        };
502        let changed_lso = AbortedTxn {
503            last_stable_offset: Offset(first.last_stable_offset.0 + 1),
504            ..first
505        };
506        let written = [
507            first,
508            changed_producer,
509            changed_start,
510            changed_last,
511            changed_lso,
512            first,
513        ];
514        let bytes = written
515            .iter()
516            .flat_map(|&entry| AbortedTxnRaw::new(entry).as_bytes().to_vec())
517            .collect::<Vec<_>>();
518        fs::write(&path, &bytes).unwrap();
519
520        let mut reopened = TxnIndex::open(path.clone()).unwrap();
521        assert2::assert!(reopened.entries() == &written[..5]);
522
523        // Retrying after reopen neither writes nor reorders the first entry.
524        reopened.append(first).unwrap();
525        assert2::check!(reopened.entries() == &written[..5]);
526        assert2::check!(fs::read(path).unwrap() == bytes);
527    }
528
529    #[test]
530    fn mutation_io_failures_leave_the_in_memory_index_unchanged() {
531        let original = retry_entry();
532        let IndexFixture {
533            dir: _dir,
534            path,
535            mut index,
536        } = populated_index("00.txnindex", &[original]);
537        fs::remove_file(&path).unwrap();
538        fs::create_dir(&path).unwrap();
539
540        assert2::assert!(
541            let LogError::Io(_) = index
542                .append(aborted_txn().bounds(Offset(10)..=Offset(12)).stable(13).call())
543                .unwrap_err()
544        );
545        assert2::assert!(index.entries() == [original]);
546        assert2::assert!(let LogError::Io(_) = index.truncate_from(Offset(0)).unwrap_err());
547        assert2::assert!(index.entries() == [original]);
548    }
549
550    #[test]
551    fn empty_or_inverted_fetch_range_matches_no_aborted_interval() {
552        let IndexFixture {
553            dir: _dir,
554            path: _path,
555            index,
556        } = ranged_index().call();
557        assert2::assert!(index.aborted_in_range(Offset(10), Offset(10)).count() == 0);
558        assert2::assert!(index.aborted_in_range(Offset(20), Offset(10)).count() == 0);
559    }
560
561    #[test]
562    fn append_round_trips_through_disk() {
563        let dir = TempDir::new().unwrap();
564        let path = dir.path().join("00.txnindex");
565        let mut idx = TxnIndex::open(path.clone()).unwrap();
566        idx.append(retry_entry()).unwrap();
567        idx.append(
568            aborted_txn()
569                .bounds(Offset(10)..=Offset(12))
570                .stable(13)
571                .call(),
572        )
573        .unwrap();
574
575        let idx2 = TxnIndex::open(path).unwrap();
576        assert2::assert!(
577            idx2.entries()
578                == &[
579                    AbortedTxn {
580                        start_offset: Offset(5),
581                        last_offset: Offset(7),
582                        producer_id: ProducerId(1000),
583                        last_stable_offset: Offset(8)
584                    },
585                    AbortedTxn {
586                        start_offset: Offset(10),
587                        last_offset: Offset(12),
588                        producer_id: ProducerId(1000),
589                        last_stable_offset: Offset(13)
590                    },
591                ]
592        );
593    }
594
595    #[test]
596    fn truncate_from_removes_overlapping_tail_entries_and_is_retryable() {
597        let dir = TempDir::new().unwrap();
598        let path = dir.path().join("00.txnindex");
599        let mut index = TxnIndex::open(path.clone()).unwrap();
600        for (start, last) in [(0, 4), (5, 9)] {
601            index
602                .append(AbortedTxn {
603                    start_offset: Offset(start),
604                    last_offset: Offset(last),
605                    producer_id: ProducerId(1),
606                    last_stable_offset: Offset(last + 1),
607                })
608                .unwrap();
609        }
610
611        for _ in 0..2 {
612            index.truncate_from(Offset(5)).unwrap();
613            assert2::assert!(index.entries().len() == 1);
614            assert2::assert!(index.entries()[0].last_offset == Offset(4));
615        }
616        let reopened = TxnIndex::open(path).unwrap();
617        assert2::assert!(reopened.entries() == index.entries());
618
619        index.truncate_from(Offset(4)).unwrap();
620        assert2::assert!(index.entries().is_empty());
621    }
622
623    #[test]
624    fn truncate_removes_membership_so_the_same_entry_can_be_reappended() {
625        let entry = retry_entry();
626        let IndexFixture {
627            dir: _dir,
628            path,
629            mut index,
630        } = populated_index("00.txnindex", &[entry]);
631
632        index.truncate_from(entry.last_offset).unwrap();
633        assert2::assert!(index.entries().is_empty());
634        index.append(entry).unwrap();
635        assert2::check!(index.entries() == [entry]);
636        assert2::check!(TxnIndex::open(path).unwrap().entries() == [entry]);
637    }
638
639    #[test]
640    fn aborted_in_range_overlaps() {
641        let dir = TempDir::new().unwrap();
642        let path = dir.path().join("00.txnindex");
643        let mut idx = TxnIndex::open(path).unwrap();
644        idx.append(
645            aborted_txn()
646                .producer(1)
647                .bounds(Offset(0)..=Offset(4))
648                .stable(5)
649                .call(),
650        )
651        .unwrap();
652        idx.append(
653            aborted_txn()
654                .producer(2)
655                .bounds(Offset(10)..=Offset(14))
656                .stable(15)
657                .call(),
658        )
659        .unwrap();
660
661        let in_3_to_12 = idx
662            .aborted_in_range(Offset(3), Offset(12))
663            .copied()
664            .collect::<Vec<_>>();
665        let in_5_to_9 = idx
666            .aborted_in_range(Offset(5), Offset(9))
667            .copied()
668            .collect::<Vec<_>>();
669        assert2::assert!(
670            in_3_to_12
671                == vec![
672                    aborted_txn()
673                        .producer(1)
674                        .bounds(Offset(0)..=Offset(4))
675                        .stable(5)
676                        .call(),
677                    aborted_txn()
678                        .producer(2)
679                        .bounds(Offset(10)..=Offset(14))
680                        .stable(15)
681                        .call(),
682                ]
683        );
684        assert2::assert!(in_5_to_9 == Vec::new());
685    }
686
687    /// No file before anything writes one.
688    fn write_nothing(_: &mut TxnIndex) {}
689
690    /// The first entry creates the file.
691    fn append_one(index: &mut TxnIndex) { index.append(retry_entry()).unwrap(); }
692
693    /// Truncating every entry rewrites the file empty.
694    fn append_then_empty(index: &mut TxnIndex) {
695        append_one(index);
696        index.truncate_from(Offset(0)).unwrap();
697    }
698
699    /// Whether an index says it has a file is whether the file exists, and a
700    /// reopen agrees.
701    #[yare::parameterized(
702        untouched = { write_nothing, false },
703        appended = { append_one, true },
704        appended_then_emptied = { append_then_empty, true },
705    )]
706    fn an_index_knows_whether_its_file_exists(write: fn(&mut TxnIndex), exists: bool) {
707        let dir = TempDir::new().unwrap();
708        let path = dir.path().join("00.txnindex");
709        let mut index = TxnIndex::open(path.clone()).unwrap();
710
711        write(&mut index);
712
713        assert2::check!(index.on_disk() == exists);
714        assert2::check!(path.exists() == exists);
715        assert2::check!(TxnIndex::open(path).unwrap().on_disk() == exists);
716    }
717
718    /// Kafka's `TransactionIndex.collectAbortedTxns`: an entry that overlaps
719    /// the range is taken, and the first entry whose last stable offset
720    /// reaches the range's end completes the search, after it is taken.
721    #[yare::parameterized(
722        nothing_completes = { 0, 30, &[0, 1, 2], false },
723        the_first_entry_completes = { 0, 5, &[0], true },
724        a_later_entry_completes = { 0, 12, &[0, 1], true },
725        the_completing_entry_lies_before_the_range = { 10, 11, &[], true },
726        an_empty_range = { 10, 10, &[], true },
727    )]
728    fn collect_aborted_stops_at_the_first_complete_entry(
729        start: i64,
730        end: i64,
731        taken: &[usize],
732        complete: bool,
733    ) {
734        let entries = [
735            aborted_txn()
736                .producer(1)
737                .bounds(Offset(0)..=Offset(4))
738                .stable(5)
739                .call(),
740            aborted_txn()
741                .producer(2)
742                .bounds(Offset(6)..=Offset(9))
743                .stable(12)
744                .call(),
745            aborted_txn()
746                .producer(3)
747                .bounds(Offset(3)..=Offset(20))
748                .stable(21)
749                .call(),
750        ];
751        let IndexFixture {
752            dir: _dir,
753            path: _path,
754            index,
755        } = populated_index("00.txnindex", &entries);
756        let mut out = vec![entries[2]];
757
758        let actual = index.collect_aborted(Offset(start), Offset(end), &mut out);
759
760        let expected: Vec<AbortedTxn> = iter::once(entries[2])
761            .chain(taken.iter().map(|&at| entries[at]))
762            .collect();
763        assert2::check!((actual, out) == (complete, expected));
764    }
765}
krabka_log/txn_index.rs

krabka_log/
txn_index.rs
