krabka_remote_storage/index/
txn.rs

1//! The Kafka transaction-index on-disk layout and the range-overlap test the
2//! fetch path applies to it.
3//!
4//! An entry records one aborted transaction's offset span and the producer
5//! that wrote it, so a fetch can report the aborted transactions that
6//! intersect the offsets it returns.
7
8use {
9    super::{LogOffset, corrupt_index},
10    crate::error::RemoteStorageError,
11    krabka_verified::remote_txn::{RemoteTxnOverlapDecision, remote_txn_overlap_decision},
12    std::mem,
13    zerocopy::{
14        BigEndian, FromBytes, Immutable, KnownLayout, Unaligned,
15        byteorder::{I16, I64},
16    },
17};
18
19/// The only aborted-transaction record version Kafka writes or accepts.
20const SUPPORTED_VERSION: i16 = 0;
21
22/// One 34-byte aborted-transaction index entry.
23///
24/// The entry is a big-endian i16 `version` (always 0), then
25/// `producer_id` i64 BE, `start_offset` i64 BE, `last_offset` i64 BE, and
26/// `last_stable_offset` i64 BE. It mirrors
27/// `krabka_log::txn_index::AbortedTxnRaw`, so the remote-tier copy of a
28/// `.txnindex` file decodes through the same byte layout that wrote the
29/// local index.
30#[derive(Debug, Clone, Copy, FromBytes, KnownLayout, Immutable, Unaligned)]
31#[repr(C)]
32pub struct AbortedTxnIndexEntry {
33    /// Record version. Always 0; a reader refuses anything else.
34    pub version: I16<BigEndian>,
35    /// Producer that wrote, and then aborted, the transaction.
36    pub producer_id: I64<BigEndian>,
37    /// First offset the aborted transaction wrote.
38    pub start_offset: I64<BigEndian>,
39    /// Last offset the aborted transaction wrote.
40    pub last_offset: I64<BigEndian>,
41    /// The last stable offset at the moment this transaction aborted.
42    pub last_stable_offset: I64<BigEndian>,
43}
44
45/// Byte length of one serialized aborted-transaction index entry.
46const TXN_INDEX_ENTRY_LEN: usize = mem::size_of::<AbortedTxnIndexEntry>();
47
48const _: [(); 34] = [(); TXN_INDEX_ENTRY_LEN];
49
50/// Borrows Kafka's transaction-index format as a zero-copy
51/// `&[AbortedTxnIndexEntry]`.
52///
53/// Each entry is 34 bytes: a big-endian i16
54/// version (always 0), then `producer_id`, `start_offset`, `last_offset`,
55/// and `last_stable_offset`, each a big-endian i64. The result borrows from
56/// `bytes`.
57///
58/// # Errors
59///
60/// Returns [`RemoteStorageError::Io`] when the object store returned bytes
61/// that do not form an entry array, or when an entry's version prefix is
62/// not the one supported version.
63pub fn parse_txn_index(bytes: &[u8]) -> Result<&[AbortedTxnIndexEntry], RemoteStorageError> {
64    if !bytes.len().is_multiple_of(TXN_INDEX_ENTRY_LEN) {
65        return Err(corrupt_index("transaction"));
66    }
67    let entries = <[AbortedTxnIndexEntry]>::ref_from_bytes(bytes)
68        .map_err(|_| corrupt_index("transaction"))?;
69    if entries
70        .iter()
71        .any(|entry| entry.version.get() != SUPPORTED_VERSION)
72    {
73        return Err(corrupt_index("transaction"));
74    }
75    if entries.iter().any(|entry| {
76        matches!(
77            remote_txn_overlap_decision(
78                entry.start_offset.get(),
79                entry.last_offset.get(),
80                entry.start_offset.get(),
81                entry.last_offset.get(),
82            ),
83            RemoteTxnOverlapDecision::Invalid
84        )
85    }) {
86        return Err(corrupt_index("transaction"));
87    }
88    Ok(entries)
89}
90
91/// Reports whether an aborted-transaction entry overlaps the inclusive offset
92/// range `[from_offset, to_offset]`.
93///
94/// It mirrors the overlap test in
95/// `TxnIndex::aborted_in_range` against an inclusive range: the entry's
96/// `[start, last]` intersects `[from, to]` if and only if
97/// `start <= to && last >= from`.
98#[must_use]
99pub fn txn_overlaps(
100    entry: &AbortedTxnIndexEntry,
101    from_offset: LogOffset,
102    to_offset: LogOffset,
103) -> bool {
104    matches!(
105        remote_txn_overlap_decision(
106            entry.start_offset.get(),
107            entry.last_offset.get(),
108            from_offset,
109            to_offset,
110        ),
111        RemoteTxnOverlapDecision::Overlap
112    )
113}
114
115#[cfg(test)]
116mod tests {
117    use {
118        super::*,
119        assert2::assert,
120        krabka_ids::{Offset, ProducerId},
121    };
122
123    #[derive(derive_more::From, derive_more::Into, Clone, Copy)]
124    struct TxnIndexVersion(i16);
125
126    #[derive(Clone, Copy, krabka_macros::FieldDefaults)]
127    struct TxnIndexEntrySetup {
128        #[default(TxnIndexVersion(0))]
129        version: TxnIndexVersion,
130        #[default(ProducerId(1000))]
131        producer_id: ProducerId,
132        #[default(Offset(0))]
133        start: Offset,
134        #[default(Offset(4))]
135        last: Offset,
136        #[default(Offset(5))]
137        lso: Offset,
138    }
139
140    #[bon::builder]
141    fn entry_bytes(
142        #[builder(into, default = TxnIndexVersion (0))] version: TxnIndexVersion,
143        #[builder(into, default = ProducerId (1000))] producer_id: ProducerId,
144        #[builder(into, default = Offset (0))] start: Offset,
145        #[builder(into, default = Offset (4))] last: Offset,
146        #[builder(into, default = Offset (5))] lso: Offset,
147    ) -> Vec<u8> {
148        let setup = TxnIndexEntrySetup {
149            version,
150            producer_id,
151            start,
152            last,
153            lso,
154        };
155        let TxnIndexEntrySetup {
156            version,
157            producer_id,
158            start,
159            last,
160            lso,
161        } = setup;
162        let mut buf = Vec::new();
163        buf.extend_from_slice(&version.0.to_be_bytes());
164        buf.extend_from_slice(&producer_id.0.to_be_bytes());
165        buf.extend_from_slice(&start.0.to_be_bytes());
166        buf.extend_from_slice(&last.0.to_be_bytes());
167        buf.extend_from_slice(&lso.0.to_be_bytes());
168        buf
169    }
170
171    #[yare::parameterized(
172        one_entry = { vec![(0_i64, 4_i64, 1000_i64)] },
173        two_entries = { vec![(0_i64, 4_i64, 1000_i64), (10, 14, 2000)] },
174    )]
175    fn parse_txn_index_round_trips_known_entries(specs: Vec<(i64, i64, i64)>) {
176        /// The decoded fields of one aborted-transaction entry.
177        #[derive(Debug, PartialEq, Eq)]
178        struct DecodedTxn {
179            start_offset: i64,
180            last_offset: i64,
181            producer_id: i64,
182            last_stable_offset: i64,
183        }
184        // Mirror TxnIndex::append: 2B version, 8B producer_id, 8B
185        // start_offset, 8B last_offset, 8B last_stable_offset, all BE.
186        let mut buf = Vec::new();
187        for (start, last, pid) in &specs {
188            buf.extend(
189                entry_bytes()
190                    .producer_id(*pid)
191                    .start(*start)
192                    .last(*last)
193                    .lso(*last + 1)
194                    .call(),
195            );
196        }
197        let entries = parse_txn_index(&buf).expect("valid txn index");
198        let decoded: Vec<DecodedTxn> = entries
199            .iter()
200            .map(|e| DecodedTxn {
201                start_offset: e.start_offset.get(),
202                last_offset: e.last_offset.get(),
203                producer_id: e.producer_id.get(),
204                last_stable_offset: e.last_stable_offset.get(),
205            })
206            .collect();
207        assert2::assert!(
208            decoded
209                == specs
210                    .iter()
211                    .map(|(start, last, producer_id)| DecodedTxn {
212                        start_offset: *start,
213                        last_offset: *last,
214                        producer_id: *producer_id,
215                        last_stable_offset: *last + 1,
216                    })
217                    .collect::<Vec<_>>()
218        );
219    }
220
221    #[test]
222    fn parse_txn_index_rejects_trailing_partial_bytes() {
223        let mut buf = entry_bytes().call();
224        // 5 trailing bytes that don't complete a 34-byte entry.
225        buf.extend_from_slice(&[0xAA; 5]);
226        assert!(parse_txn_index(&buf).is_err(), "partial entry is corrupt");
227    }
228
229    #[test]
230    fn parse_txn_index_empty_is_empty() {
231        assert!(parse_txn_index(&[]).expect("empty is valid").is_empty());
232    }
233
234    #[test]
235    fn parse_txn_index_rejects_an_unsupported_version() {
236        let buf = entry_bytes().version(1).call();
237        assert!(
238            parse_txn_index(&buf).is_err(),
239            "an unsupported version prefix is corrupt"
240        );
241    }
242
243    fn sample_aborted_txn_entry(start: i64, last: i64, lso: i64) -> AbortedTxnIndexEntry {
244        AbortedTxnIndexEntry {
245            version: I16::new(0),
246            producer_id: I64::new(1),
247            start_offset: I64::new(start),
248            last_offset: I64::new(last),
249            last_stable_offset: I64::new(lso),
250        }
251    }
252
253    #[yare::parameterized(
254        before_entry = { 0, 9, false },
255        touches_first_offset = { 0, 10, true },
256        inside_entry = { 11, 13, true },
257        touches_last_offset = { 14, 100, true },
258        after_entry = { 15, 100, false },
259        covers_entry = { 0, 100, true },
260    )]
261    fn txn_overlaps_boundaries(start: i64, end: i64, want: bool) {
262        let e = sample_aborted_txn_entry(10, 14, 15);
263        assert2::assert!(txn_overlaps(&e, start, end) == want);
264    }
265
266    #[test]
267    fn txn_overlaps_rejects_inverted_query_and_parser_rejects_inverted_entry() {
268        let valid = sample_aborted_txn_entry(10, 14, 15);
269        let inverted = sample_aborted_txn_entry(14, 10, 11);
270
271        assert!(!txn_overlaps(&valid, 14, 10));
272        assert!(!txn_overlaps(&inverted, 0, 100));
273
274        let bytes = entry_bytes()
275            .producer_id(1)
276            .start(14)
277            .last(10)
278            .lso(11)
279            .call();
280        assert!(parse_txn_index(&bytes).is_err());
281    }
282}
krabka_remote_storage/index/txn.rs

krabka_remote_storage/index/
txn.rs
