krabka_log/segment/
read_raw_desc.rs

1//! The zero-copy descriptor form of the verbatim read, for the `sendfile`
2//! fetch path.
3//!
4//! The walk mirrors `read_raw` byte for byte but returns a file-region
5//! descriptor instead of an owned buffer, so the broker sends the run straight
6//! out of the page cache. The whole module is gated on the SENDFILE alias by
7//! its declaration in the parent, which is why nothing here carries a `cfg`.
8
9use {
10    super::{
11        RawSegmentDesc, Segment,
12        io::SourceIo,
13        read_raw::{RawBatch, RawReadSource, select_raw_range},
14        read_source::ReadSource,
15    },
16    crate::{config::DEFAULT_READ_AHEAD_MAX, error::LogError},
17    krabka_ids::Offset,
18    krabka_protocol::records::{FileRegion, HEADER_LEN},
19    krabka_units::prelude::{ByteSize, ByteSizeExt},
20    std::sync::{Arc, atomic::Ordering},
21    tracing::instrument,
22};
23
24/// Only small batches benefit from reading ahead to amortize header lookups.
25const SMALL_BATCH_THRESHOLD: usize = 16 * 1024;
26
27/// Maximum bytes prefetched after the scanner identifies a small batch.
28const HEADER_SCAN_WINDOW: usize = 256 * 1024;
29
30/// The zero-copy descriptor read of a [`ReadSource`]. Provided for every
31/// source, so it runs the same over a live segment under the log lock and
32/// over a view of one after the lock is released.
33pub trait RawDescSource: ReadSource {
34    /// [`Segment::read_raw_desc_with_policy`] for a fetch that must not walk
35    /// bytes outside the page cache. `None` means the run is not cached, and
36    /// the fetch takes the byte-copy path instead.
37    ///
38    /// The walk reads each batch header with a small synchronous read under
39    /// the log lock. On cached bytes those reads cost microseconds, and
40    /// `sendfile` then sends the run without a copy. On bytes only on the
41    /// disk -- a consumer catching up on an old backlog, or a segment that
42    /// wrote through `O_DIRECT` -- each one waits for the disk while the
43    /// producer's appends wait for the lock, and one large read of the
44    /// window is faster. The page at the far end of the window decides as
45    /// well as the first: the previous fetch's readahead hint usually leaves
46    /// the start cached while the rest is still on its way from the disk, so
47    /// a window whose first or last byte is not cached is read in one read
48    /// by the byte-copy path. A window with no bytes in it never blocks.
49    ///
50    /// The sparse-index lookup that finds the run's start position runs
51    /// once, and both the page-cache probe and the description use it. Off
52    /// an exact index entry that lookup scans header windows, so a second
53    /// lookup would read them again under the log lock.
54    ///
55    /// # Errors
56    /// Returns an error when log I/O fails, or a record or index is corrupt.
57    fn read_raw_desc_if_cached(
58        &self,
59        fetch_offset: Offset,
60        limit_offset: Offset,
61        max_size: ByteSize,
62        read_ahead_max: ByteSize,
63    ) -> Result<Option<RawSegmentDesc>, LogError> {
64        self.describe_with_reader(
65            fetch_offset,
66            limit_offset,
67            max_size,
68            read_ahead_max,
69            Probe::PageCache,
70            |pos, buf, len| self.read_log_range(pos, buf, len),
71        )
72    }
73
74    /// Whether the page cache holds both the first and the last byte of the
75    /// window a fetch from `start_pos` would walk.
76    fn run_is_cached(&self, start_pos: u64, max_size: ByteSize) -> bool {
77        let window = (max_size.bytes_u64().max(HEADER_LEN as u64))
78            .min(self.log_size().saturating_sub(start_pos));
79        // No bytes from the start: an empty segment, which compaction leaves
80        // behind, or a run that starts at the end of the file. The walk reads
81        // nothing there, so it cannot wait on the disk. Asking the page cache
82        // would read at the end of the file, which answers as not cached, and
83        // stop the walk before the later segments that hold the run.
84        if window == 0 {
85            return true;
86        }
87        let last = start_pos + window.saturating_sub(1);
88        self.byte_is_cached(start_pos) && self.byte_is_cached(last)
89    }
90
91    /// Cache capability only; successful probes and misses can change as
92    /// pages enter and leave the cache and must always be checked again.
93    fn byte_is_cached(&self, offset: u64) -> bool {
94        if self.cache_probe_unsupported().load(Ordering::Relaxed) {
95            return true;
96        }
97        self.io().is_cached(self.file(), offset).unwrap_or_else(|| {
98            self.cache_probe_unsupported()
99                .store(true, Ordering::Relaxed);
100            true
101        })
102    }
103
104    /// [`Segment::read_raw_desc`] under the log's configured read policy: the
105    /// readahead hint reaches at most `read_ahead_max` past the read's own
106    /// window.
107    ///
108    /// # Errors
109    /// Returns an error when log I/O fails, a record or index is corrupt, or
110    /// the requested offset violates the segment state.
111    fn read_raw_desc_with_policy(
112        &self,
113        fetch_offset: Offset,
114        limit_offset: Offset,
115        max_size: ByteSize,
116        read_ahead_max: ByteSize,
117    ) -> Result<RawSegmentDesc, LogError> {
118        self.describe_with_reader(
119            fetch_offset,
120            limit_offset,
121            max_size,
122            read_ahead_max,
123            Probe::None,
124            |pos, buf, len| self.read_log_range(pos, buf, len),
125        )
126        .map(|desc| desc.unwrap_or_else(RawSegmentDesc::empty))
127    }
128
129    /// The descriptor walk behind both public forms. `None` means the run
130    /// is not described at all: the segment wrote through `O_DIRECT`, or
131    /// `probe` found the run outside the page cache.
132    ///
133    /// The sparse-index lookup for the start position runs once, before the
134    /// probe, and the walk starts from the position it found. The positional
135    /// reads stay injectable so tests can measure the synchronous I/O of the
136    /// whole fetch, the lookup included, as well as the file region it
137    /// selects.
138    fn describe_with_reader(
139        &self,
140        fetch_offset: Offset,
141        limit_offset: Offset,
142        max_size: ByteSize,
143        read_ahead_max: ByteSize,
144        probe: Probe,
145        mut read_range: impl FnMut(u64, &mut Vec<u8>, usize) -> Result<(), LogError>,
146    ) -> Result<Option<RawSegmentDesc>, LogError> {
147        // The bytes of a segment that wrote through `O_DIRECT` are not in
148        // the page cache: the newest are in its tail cache, and the rest are
149        // only on the disk, sealed or not. Walking their batch headers here
150        // would be one small synchronous disk read per batch under the log
151        // lock, and `sendfile` would then read the run from the disk again.
152        // Describing nothing hands the fetch to the byte-copy path, which
153        // takes the newest bytes from memory and the rest in one large read:
154        // an empty description is what the broker falls back on, and a
155        // segment that contributes nothing is one the log's walk passes over.
156        if self.wrote_direct() {
157            return Ok(None);
158        }
159        let Some(start_pos) = self.raw_start_position_with(
160            fetch_offset,
161            limit_offset,
162            "read_raw_desc",
163            &mut read_range,
164        )?
165        else {
166            return Ok(Some(RawSegmentDesc::empty()));
167        };
168        if probe == Probe::PageCache && !self.run_is_cached(start_pos, max_size) {
169            return Ok(None);
170        }
171        // Scan bounded windows to find headers; keep the selected payload file-backed.
172        let max_bytes = max_size.bytes_usize();
173        let window =
174            (max_bytes.max(HEADER_LEN) as u64).min(self.log_size().saturating_sub(start_pos));
175        self.advise_read(start_pos, window, read_ahead_max);
176        let window = usize::try_from(window)
177            .map_err(|_| LogError::Corrupt("read_raw_desc window too large".into()))?;
178        let mut header_buf = Vec::with_capacity(HEADER_LEN.min(window));
179        let mut header_buf_start = 0usize;
180        let mut header_buf_end = 0usize;
181        // Until a small batch justifies a buffered scan, touch only its header.
182        // Larger reads at sparse batch boundaries pull cold payload pages into
183        // this synchronous walk, delaying sendfile while the log is locked.
184        let mut scan_window = HEADER_LEN;
185        let Some(range) = select_raw_range(fetch_offset, limit_offset, max_bytes, window, |pos| {
186            if pos < header_buf_start || pos + HEADER_LEN > header_buf_end {
187                header_buf.clear();
188                header_buf_start = pos;
189                let read_len = (window - pos).min(scan_window);
190                read_range(start_pos + pos as u64, &mut header_buf, read_len)?;
191                header_buf_end = header_buf_start + header_buf.len();
192            }
193            if pos + HEADER_LEN > header_buf_end {
194                return Ok(None);
195            }
196            let start = pos - header_buf_start;
197            let Some(header) = header_buf.get(start..start + HEADER_LEN) else {
198                return Ok(None);
199            };
200            let batch = RawBatch::decode(header)?;
201            scan_window = if batch.len <= SMALL_BATCH_THRESHOLD {
202                HEADER_SCAN_WINDOW
203            } else {
204                HEADER_LEN
205            };
206            Ok(Some(batch))
207        })?
208        else {
209            return Ok(Some(RawSegmentDesc::empty()));
210        };
211        if start_pos + range.positions.end as u64 > self.log_size() {
212            return Ok(Some(RawSegmentDesc::empty()));
213        }
214        Ok(Some(RawSegmentDesc {
215            start_offset: range.start_offset,
216            last_offset: range.last_offset,
217            region: Some(FileRegion {
218                file: Arc::clone(self.file()),
219                offset: start_pos + range.positions.start as u64,
220                len: range.positions.len(),
221            }),
222        }))
223    }
224}
225
226impl<S: ReadSource + ?Sized> RawDescSource for S {}
227
228impl Segment {
229    /// Descriptor variant of [`Segment::read_raw`] for the zero-copy
230    /// `sendfile` fetch path.
231    ///
232    /// This method runs the **same** boundary walk and selects the identical
233    /// `[start_pos+range_start, start_pos+range_end)` byte range that
234    /// `read_raw` would have sliced. It returns a [`krabka_protocol::records::FileRegion`] descriptor
235    /// instead of a `pread` of the payload into an owned `Bytes`.
236    ///
237    /// The walk reads bounded windows to amortize header lookups and uses the
238    /// header's `batch_length` to find boundaries. It returns a file region
239    /// instead of building an owned copy of the selected payload. The region
240    /// is byte-identical to the `bytes` of `read_raw` for the same
241    /// `(fetch_offset, limit_offset, max_bytes)`.
242    #[instrument(
243        level = "debug",
244        skip(self),
245        fields(base_offset = self.base_offset.0),
246        err,
247    )]
248    /// # Errors
249    /// Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.
250    pub fn read_raw_desc(
251        &self,
252        fetch_offset: Offset,
253        limit_offset: Offset,
254        max_size: ByteSize,
255    ) -> Result<RawSegmentDesc, LogError> {
256        self.read_raw_desc_with_policy(fetch_offset, limit_offset, max_size, DEFAULT_READ_AHEAD_MAX)
257    }
258}
259
260/// Whether the descriptor walk first asks the page cache about its run.
261#[derive(Clone, Copy, PartialEq, Eq)]
262pub enum Probe {
263    /// Walk the run wherever its bytes are.
264    None,
265    /// Describe nothing when the run's first or last byte is not cached.
266    PageCache,
267}
268
269#[cfg(test)]
270mod tests {
271    use {
272        super::{super::header_walk::initial_header_walk_window, *},
273        crate::{
274            config::{DEFAULT_INDEX_INTERVAL, DEFAULT_TIMESTAMP_SCAN_WINDOW},
275            io::LogIo,
276            segment::test_support::{
277                DENSE_INDEX, TempSegment, recording_advice, test_batch_at, test_segment,
278            },
279            test_support::AdviceCall,
280        },
281        bytes::Bytes,
282        krabka_protocol::records::{Record, RecordBatch},
283        krabka_units::prelude::{bytes, mebibytes},
284        std::{fs::File, sync::Mutex},
285    };
286
287    /// Every positional read a fetch makes, as `(position, length)`.
288    type Reads = Vec<(u64, usize)>;
289
290    /// A page cache that holds nothing, so every probe reports a miss.
291    #[derive(Debug)]
292    struct NothingCached;
293
294    impl LogIo for NothingCached {
295        fn is_cached(&self, _file: &File, _offset: u64) -> Option<bool> { Some(false) }
296    }
297
298    #[derive(Debug)]
299    struct ChangingCache(Mutex<Vec<Option<bool>>>);
300
301    impl LogIo for ChangingCache {
302        fn is_cached(&self, _file: &File, _offset: u64) -> Option<bool> {
303            self.0.lock().unwrap().remove(0)
304        }
305    }
306
307    #[test]
308    fn cache_misses_and_hits_are_rechecked_but_unsupported_probes_stop() {
309        let TempSegment {
310            dir: _dir,
311            segment: mut seg,
312        } = test_segment();
313        seg.io = Arc::new(ChangingCache(Mutex::new(vec![
314            Some(false),
315            Some(true),
316            Some(false),
317            None,
318        ])));
319        assert2::assert!(!seg.byte_is_cached(0));
320        assert2::assert!(seg.byte_is_cached(0));
321        assert2::assert!(!seg.byte_is_cached(0));
322        assert2::assert!(seg.byte_is_cached(0));
323        // No fifth I/O response exists: unsupported capabilities suppress
324        // subsequent probes, including probes of a different byte.
325        assert2::assert!(seg.byte_is_cached(123));
326        seg.set_io(Arc::new(NothingCached));
327        assert2::assert!(!seg.byte_is_cached(0));
328        let TempSegment {
329            dir: _other_dir,
330            segment: mut other,
331        } = test_segment();
332        other.io = Arc::new(NothingCached);
333        assert2::assert!(!other.byte_is_cached(0));
334    }
335
336    /// Run the sendfile-path read of `seg` from `fetch_offset` and record
337    /// every positional read it makes, the sparse-index lookup included.
338    fn counted_read(
339        seg: &Segment,
340        fetch_offset: Offset,
341        limit_offset: Offset,
342    ) -> (Option<RawSegmentDesc>, Reads) {
343        let mut reads = Vec::new();
344        let desc = seg
345            .describe_with_reader(
346                fetch_offset,
347                limit_offset,
348                mebibytes(1),
349                DEFAULT_READ_AHEAD_MAX,
350                Probe::PageCache,
351                |pos, buf, len| {
352                    reads.push((pos, len));
353                    seg.read_log_range(pos, buf, len)
354                },
355            )
356            .unwrap();
357        (desc, reads)
358    }
359
360    /// The byte position the sparse index gives as a floor for `offset`.
361    fn index_floor(seg: &Segment, offset: Offset) -> u64 {
362        let rel = u32::try_from(offset.0 - seg.base_offset.0).unwrap();
363        seg.offset_index
364            .floor_entry(rel)
365            .map_or(0, |entry| u64::from(entry.position))
366    }
367
368    /// The sendfile path resolves a run's start position through the sparse
369    /// index once, and both the page-cache probe and the description use it.
370    /// An indexed successor needs only the preceding header; other sparse
371    /// lookups with no later index entry scan one index interval. The description then reads its own headers.
372    /// The region always equals the bytes `read_raw` returns for the same fetch.
373    #[test]
374    fn the_sendfile_read_looks_up_its_start_position_once() {
375        struct Case {
376            name: &'static str,
377            /// `(offset, payload bytes)` of each appended batch.
378            batches: Vec<(i64, usize)>,
379            fetch: i64,
380            /// The lookup's expected first read, absent on an exact index hit.
381            lookup_bytes: Option<usize>,
382        }
383        let lookup_window = initial_header_walk_window();
384        let sparse = DEFAULT_INDEX_INTERVAL;
385        let cases = [
386            // The `descriptor_large_batch` benchmark: one unindexed 1 MiB
387            // batch, fetched from its first offset.
388            Case {
389                name: "first batch, one large batch",
390                batches: vec![(0, 1024 * 1024)],
391                fetch: 0,
392                lookup_bytes: Some(lookup_window),
393            },
394            Case {
395                name: "unindexed small batch behind an index entry",
396                batches: (0..2000).map(|off| (off, 100)).collect(),
397                fetch: 1001,
398                lookup_bytes: Some(HEADER_LEN),
399            },
400            Case {
401                name: "exact index entry",
402                batches: (0..64).map(|off| (off, 32 * 1024)).collect(),
403                fetch: 37,
404                lookup_bytes: None,
405            },
406            Case {
407                name: "fetch inside a compacted gap",
408                batches: vec![(0, 100), (5, 100), (10, 100)],
409                fetch: 3,
410                lookup_bytes: Some(lookup_window),
411            },
412            Case {
413                name: "compacted gap after an indexed batch",
414                batches: vec![(0, 32 * 1024), (5, 32 * 1024), (10, 32 * 1024)],
415                fetch: 6,
416                lookup_bytes: Some(HEADER_LEN),
417            },
418        ];
419        for case in cases {
420            let TempSegment {
421                dir: _dir,
422                segment: mut seg,
423            } = test_segment();
424            for &(off, payload) in &case.batches {
425                let mut batch = test_batch_at(Offset(off));
426                batch.records[0].value = Some(Bytes::from(vec![0; payload]));
427                seg.append(&batch, sparse).unwrap();
428            }
429            let limit = Offset(case.batches.last().unwrap().0 + 1);
430            let fetch = Offset(case.fetch);
431            let raw = seg.read_raw(fetch, limit, mebibytes(1)).unwrap();
432            let start_pos = seg
433                .raw_start_position(fetch, limit, "test")
434                .unwrap()
435                .unwrap();
436            let floor = index_floor(&seg, fetch);
437            let mut expected = Vec::new();
438            if let Some(bytes) = case.lookup_bytes {
439                expected.push((floor, bytes));
440            }
441            expected.push((start_pos, HEADER_LEN));
442
443            let (desc, reads) = counted_read(&seg, fetch, limit);
444
445            let desc = desc.expect("a freshly written segment is in the page cache");
446            let region = desc.region.unwrap();
447            let name = case.name;
448            assert2::check!(reads[..expected.len()] == expected[..], "{name}");
449            assert2::check!(
450                reads.iter().filter(|read| read.1 == lookup_window).count()
451                    == usize::from(case.lookup_bytes == Some(lookup_window)),
452                "{name}"
453            );
454            assert2::check!(desc.start_offset == raw.start_offset, "{name}");
455            assert2::check!(desc.last_offset == raw.last_offset, "{name}");
456            assert2::check!(region_bytes(&region) == raw.bytes.to_vec(), "{name}");
457        }
458    }
459
460    /// A follower asks for the offset immediately after the index floor's
461    /// last offset. Neither locating nor describing its batch should copy a
462    /// payload window, including when compaction left a gap after the floor.
463    #[yare::parameterized(
464        one_kibibyte_records = { 63, 1024, 0 },
465        one_hundred_byte_records = { 600, 100, 0 },
466        one_kibibyte_records_after_a_gap = { 63, 1024, 17 },
467        one_hundred_byte_records_after_a_gap = { 600, 100, 17 },
468    )]
469    fn an_indexed_successor_reads_headers_only(records: i32, payload: usize, gap: i64) {
470        let TempSegment {
471            dir: _dir,
472            segment: mut seg,
473        } = test_segment();
474        let next = i64::from(2 * records);
475        let mut expected = Vec::new();
476        for base in [0, i64::from(records), next + gap] {
477            let batch = RecordBatch {
478                base_offset: base,
479                last_offset_delta: records - 1,
480                records: (0..records)
481                    .map(|offset_delta| Record {
482                        offset_delta,
483                        value: Some(Bytes::from(vec![0; payload])),
484                        ..Record::default()
485                    })
486                    .collect(),
487                ..RecordBatch::default()
488            };
489            seg.append(&batch, DEFAULT_INDEX_INTERVAL).unwrap();
490            if base == next + gap {
491                batch.encode(&mut expected).unwrap();
492            }
493        }
494        let last = next + gap + i64::from(records) - 1;
495
496        let (desc, reads) = counted_read(&seg, Offset(next), Offset(last + 1));
497
498        let desc = desc.unwrap();
499        assert2::assert!(desc.start_offset == Offset(next + gap));
500        assert2::assert!(desc.last_offset == Offset(last));
501        assert2::assert!(region_bytes(&desc.region.unwrap()) == expected);
502        assert2::assert!(reads.iter().all(|&(_, bytes)| bytes <= HEADER_LEN));
503        assert2::assert!(reads.iter().map(|&(_, bytes)| bytes).sum::<usize>() <= 2 * HEADER_LEN);
504    }
505
506    #[test]
507    fn an_inconsistent_index_cannot_skip_the_batch_containing_the_fetch() {
508        let TempSegment {
509            dir: _dir,
510            segment: mut seg,
511        } = test_segment();
512        let batch = RecordBatch {
513            last_offset_delta: 188,
514            records: (0..189)
515                .map(|offset_delta| Record {
516                    offset_delta,
517                    ..Record::default()
518                })
519                .collect(),
520            ..RecordBatch::default()
521        };
522        seg.append(&batch, DEFAULT_INDEX_INTERVAL).unwrap();
523        // This entry claims that the batch at position zero ends at 125,
524        // but its header says 188. Offset 126 is still inside that batch.
525        seg.offset_index.append(125, 0).unwrap();
526        let mut expected = Vec::new();
527        batch.encode(&mut expected).unwrap();
528
529        let (desc, _) = counted_read(&seg, Offset(126), Offset(189));
530
531        let desc = desc.unwrap();
532        assert2::assert!(desc.start_offset == Offset(0));
533        assert2::assert!(desc.last_offset == Offset(188));
534        assert2::assert!(region_bytes(&desc.region.unwrap()) == expected);
535    }
536
537    /// A run outside the page cache is not described, and the probe that
538    /// says so costs only the start-position lookup. A segment that wrote
539    /// through `O_DIRECT` is not described either, and is not read at all.
540    #[test]
541    fn the_sendfile_read_falls_back_without_describing_uncached_or_direct_runs() {
542        let TempSegment {
543            dir: _dir,
544            segment: mut seg,
545        } = test_segment();
546        let mut batch = test_batch_at(Offset(0));
547        batch.records[0].value = Some(Bytes::from(vec![0; 1024 * 1024]));
548        seg.append(&batch, DEFAULT_INDEX_INTERVAL).unwrap();
549        seg.set_io(Arc::new(NothingCached));
550
551        let (desc, reads) = counted_read(&seg, Offset(0), Offset(1));
552        assert2::check!(desc.is_none());
553        assert2::check!(reads == vec![(0, initial_header_walk_window())]);
554
555        seg.wrote_direct = true;
556        let (desc, reads) = counted_read(&seg, Offset(0), Offset(1));
557        assert2::check!(desc.is_none());
558        assert2::check!(reads.is_empty());
559    }
560
561    // `pread` a `FileRegion` into a fresh `Vec`. These are the bytes that the
562    // broker's sendfile would transmit, and that its TLS pread-fallback would
563    // copy.
564    krabka_macros::file_region_fixture!(region_bytes);
565
566    #[test]
567    fn descriptor_scan_reads_only_headers_for_large_batches_and_buffers_small_batches() {
568        let large = 1024 * 1024;
569        let cases = [
570            (vec![large; 10], vec![HEADER_LEN; 10]),
571            (vec![1; 4000], vec![HEADER_LEN, HEADER_SCAN_WINDOW]),
572            (vec![1, large, large, 1, large, large], vec![
573                HEADER_LEN,
574                HEADER_SCAN_WINDOW,
575                HEADER_LEN,
576                HEADER_LEN,
577                HEADER_SCAN_WINDOW,
578                HEADER_LEN,
579            ]),
580        ];
581        for (payload_sizes, mut expected_reads) in cases {
582            let TempSegment {
583                dir: _dir,
584                segment: mut seg,
585            } = test_segment();
586            let mut batch_len = 0;
587            let mut first_batch_len = None;
588            for (off, size) in payload_sizes.iter().enumerate() {
589                let mut batch = test_batch_at(Offset(i64::try_from(off).unwrap()));
590                batch.records[0].value = Some(Bytes::from(vec![0; *size]));
591                batch_len = batch.encoded_len();
592                first_batch_len.get_or_insert(batch_len);
593                seg.append(&batch, DENSE_INDEX).unwrap();
594            }
595            // The small batches cross one scan window; the final read is
596            // clamped to the bytes remaining in the segment.
597            if payload_sizes.len() == 4000 {
598                let buffered_headers = (HEADER_SCAN_WINDOW - HEADER_LEN) / batch_len + 1;
599                let next_read = (1 + buffered_headers) * batch_len;
600                expected_reads.push(usize::try_from(seg.log_size).unwrap() - next_read);
601            }
602            // The first batch has no index entry, so the start-position
603            // lookup reads up to the second batch's entry, capped at one
604            // header-scan window, before the walk begins.
605            expected_reads.insert(
606                0,
607                (first_batch_len.unwrap() + HEADER_LEN)
608                    .min(DEFAULT_TIMESTAMP_SCAN_WINDOW.bytes_usize()),
609            );
610            let limit = Offset(i64::try_from(payload_sizes.len()).unwrap());
611            let mut reads = Vec::new();
612            let desc = seg
613                .describe_with_reader(
614                    Offset(0),
615                    limit,
616                    mebibytes(16),
617                    DEFAULT_READ_AHEAD_MAX,
618                    Probe::None,
619                    |pos, buf, len| {
620                        reads.push((pos, len));
621                        seg.read_log_range(pos, buf, len)
622                    },
623                )
624                .unwrap()
625                .expect("a buffered segment is always described");
626            assert2::assert!(
627                reads.iter().map(|(_, len)| *len).collect::<Vec<_>>() == expected_reads
628            );
629            let raw = seg.read_raw(Offset(0), limit, mebibytes(16)).unwrap();
630            let region = desc.region.unwrap();
631            assert2::assert!(desc.start_offset == raw.start_offset);
632            assert2::assert!(desc.last_offset == raw.last_offset);
633            assert2::assert!(region_bytes(&region) == raw.bytes.to_vec());
634        }
635    }
636
637    /// The load-bearing Increment-D/E invariant: the `read_raw_desc` region
638    /// maps to exactly the bytes that `read_raw` would have returned, for the
639    /// same `(fetch_offset, limit_offset, max_bytes)`. This test covers
640    /// single-batch reads, multi-batch reads, mid-stream start offsets, the
641    /// limit clamp, and the one-batch-over-budget anti-stall rule.
642    #[test]
643    fn read_raw_desc_region_equals_read_raw_bytes() {
644        let TempSegment {
645            dir,
646            segment: mut seg,
647        } = test_segment();
648        for off in 0..5i64 {
649            seg.append(&test_batch_at(Offset(off)), DENSE_INDEX)
650                .unwrap();
651        }
652        let batch_len = u32::try_from(test_batch_at(Offset(0)).encoded_len()).unwrap();
653        let cases = [
654            ("all batches", 0i64, 5i64, mebibytes(10)),
655            ("limit clamp", 0, 3, mebibytes(10)),
656            ("mid-stream start", 2, 5, mebibytes(10)),
657            ("one-batch anti-stall", 0, 5, bytes(1)),
658            ("mid-stream anti-stall", 2, 5, bytes(1)),
659            ("zero-byte anti-stall", 0, 5, bytes(0)),
660            ("window ends inside first batch", 0, 5, bytes(batch_len - 1)),
661            ("window ends after first batch", 0, 5, bytes(batch_len)),
662            (
663                "window holds a second header",
664                0,
665                5,
666                bytes(batch_len + u32::try_from(HEADER_LEN).unwrap()),
667            ),
668            ("last batch", 4, 5, mebibytes(10)),
669            ("past last batch", 5, 9, mebibytes(10)),
670            ("at limit", 0, 0, mebibytes(10)),
671            ("past limit", 2, 1, mebibytes(10)),
672        ];
673        for (_name, fo, lo, mb) in cases {
674            let raw = seg.read_raw(Offset(fo), Offset(lo), mb).unwrap();
675            let desc = seg.read_raw_desc(Offset(fo), Offset(lo), mb).unwrap();
676            assert2::assert!(desc.start_offset == raw.start_offset);
677            assert2::assert!(desc.last_offset == raw.last_offset);
678            match &desc.region {
679                Some(region) => {
680                    assert2::assert!(region.len == raw.bytes.len());
681                    assert2::assert!(region_bytes(region) == raw.bytes.to_vec());
682                }
683                None => assert2::assert!(raw.bytes.is_empty()),
684            }
685        }
686        drop(dir);
687    }
688
689    /// A truncated trailing batch, where the byte budget cuts mid-batch, must
690    /// produce a region whose bytes equal the clipped output of `read_raw`. A
691    /// sendfile of a clipped range is wire-valid, because the consumer drops
692    /// the partial batch.
693    #[test]
694    fn read_raw_desc_matches_read_raw_when_budget_clips_run() {
695        let TempSegment {
696            dir,
697            segment: mut seg,
698        } = test_segment();
699        // Several batches; a mid-size budget will include some but not all.
700        for off in 0..6i64 {
701            seg.append(&test_batch_at(Offset(off)), DENSE_INDEX)
702                .unwrap();
703        }
704        // Budget that admits ~2-3 batches (each batch is small but > a few bytes).
705        let raw = seg.read_raw(Offset(0), Offset(6), bytes(80)).unwrap();
706        let desc = seg.read_raw_desc(Offset(0), Offset(6), bytes(80)).unwrap();
707        let region = desc.region.expect("non-empty");
708        assert2::assert!(desc.start_offset == raw.start_offset);
709        assert2::assert!(desc.last_offset == raw.last_offset);
710        assert2::assert!(region.len == raw.bytes.len());
711        assert2::assert!(region_bytes(&region) == raw.bytes.to_vec());
712        drop(dir);
713    }
714
715    /// The descriptor read asks the kernel for the region it describes, and
716    /// the window after it, in one hint given before its header walk. The
717    /// walk then reads headers the hint already brought in, rather than
718    /// taking one disk read per batch on a cold segment.
719    #[test]
720    fn read_raw_desc_advises_its_region_and_the_next_window_in_one_hint() {
721        let TempSegment {
722            dir: _dir,
723            segment: mut seg,
724        } = test_segment();
725        for off in 0..20i64 {
726            seg.append(&test_batch_at(Offset(off)), DENSE_INDEX)
727                .unwrap();
728        }
729        let advice = recording_advice(&mut seg);
730        let batch_len = u64::try_from(test_batch_at(Offset(0)).encoded_len()).unwrap();
731        let budget = 3 * batch_len;
732
733        let desc = seg
734            .read_raw_desc(Offset(4), Offset(20), bytes(u32::try_from(budget).unwrap()))
735            .unwrap();
736        let region = desc.region.expect("a read inside the segment");
737        assert2::check!(region.len == usize::try_from(budget).unwrap());
738        assert2::check!(
739            advice.take()
740                == vec![AdviceCall {
741                    offset: region.offset,
742                    len: 2 * budget
743                }]
744        );
745    }
746}
krabka_log/segment/read_raw_desc.rs

krabka_log/segment/
read_raw_desc.rs
