commonware_runtime/utils/buffer/paged/
sealed.rs

1//! Read-only counterpart to [`super::Writer`]: an immutable, page-cache-backed read handle for
2//! a blob whose logical content will no longer change.
3//!
4//! # Sealing
5//!
6//! [`super::Writer::seal`] returns a [`Sealed`] read handle and starts an fsync. Reads observe
7//! flushed bytes immediately, while durability waits for the sync handle.
8//!
9//! # Cheap sharing
10//!
11//! [`Sealed`] is `Clone` and shares its state via `Arc<SealedInner>`. Clones do not coordinate via
12//! any lock; they share the underlying [`Blob`] handle (which provides its own synchronization)
13//! and the page cache.
14
15use super::{CHECKSUM_SIZE, CacheRef, Replay, read::PageReader, view::View};
16use crate::{Blob, Error, IoBuf, IoBufMut, IoBufs, ReadOptions};
17use commonware_utils::Widen;
18use std::{num::NonZeroUsize, sync::Arc};
19
20/// An immutable, page-cache-backed read handle for a [Blob]. The read-only counterpart to
21/// [`super::Writer`].
22pub struct Sealed<B: Blob> {
23    inner: Arc<SealedInner<B>>,
24}
25
26impl<B: Blob> Clone for Sealed<B> {
27    fn clone(&self) -> Self {
28        Self {
29            inner: self.inner.clone(),
30        }
31    }
32}
33
34struct SealedInner<B: Blob> {
35    /// The underlying blob being wrapped.
36    blob: B,
37
38    /// Size of the sealed view, in bytes.
39    size: u64,
40
41    /// Logical bytes of the partial last page, if the blob ends in one. Bytes at offsets
42    /// `[size - partial_page.len(), size)` come from here; bytes below come from full pages on the
43    /// blob (via the page cache).
44    partial_page: Option<IoBuf>,
45
46    /// Reference to the page cache used for reads of full pages.
47    cache_ref: CacheRef,
48
49    /// Page-cache id. [`super::Writer::seal`] preserves the writer id so hot full pages remain
50    /// valid across the transition. [`super::Writer::snapshot`] uses a fresh id because the writer
51    /// can keep mutating its own cache namespace.
52    id: u64,
53}
54
55impl<B: Blob> Sealed<B> {
56    /// Construct a [`Sealed`] from already-validated parts. Invoked by [`super::Writer::seal`].
57    pub(super) fn new(
58        blob: B,
59        size: u64,
60        partial_page: Option<IoBuf>,
61        cache_ref: CacheRef,
62        id: u64,
63    ) -> Self {
64        Self {
65            inner: Arc::new(SealedInner {
66                blob,
67                size,
68                partial_page,
69                cache_ref,
70                id,
71            }),
72        }
73    }
74
75    /// Returns the size of the blob.
76    pub fn size(&self) -> u64 {
77        self.inner.size
78    }
79
80    /// Logical offset at which the partial-page bytes begin. Equal to `size` when there is no
81    /// partial page.
82    fn partial_offset(&self) -> u64 {
83        self.inner.size
84            - self
85                .inner
86                .partial_page
87                .as_ref()
88                .map_or(0, |p| p.len() as u64)
89    }
90
91    /// Returns a borrowed view over this blob.
92    fn view(&self) -> View<'_, B> {
93        View {
94            blob: &self.inner.blob,
95            cache_ref: &self.inner.cache_ref,
96            id: self.inner.id,
97            size: self.inner.size,
98            tail_offset: self.partial_offset(),
99            tail: self
100                .inner
101                .partial_page
102                .as_ref()
103                .map_or(&[][..], |p| p.as_ref()),
104        }
105    }
106
107    /// Read exactly `len` immutable bytes starting at `offset`.
108    pub async fn read_at(&self, offset: u64, len: usize) -> Result<IoBufs, Error> {
109        self.view().read_at(offset, len).await
110    }
111
112    /// Read into `buf` if it can be done synchronously without I/O. Returns `true` only if all
113    /// `buf.len()` bytes were satisfied from the page cache and/or the in-memory tail. When `false`
114    /// is returned, the contents of `buf` are unspecified.
115    pub fn try_read_sync_into(&self, buf: &mut [u8], offset: u64) -> bool {
116        self.view().try_read_sync_into(buf, offset)
117    }
118
119    /// Reads bytes starting at `offset` into `buf`.
120    pub async fn read_into(&self, buf: &mut [u8], offset: u64) -> Result<(), Error> {
121        self.view().read_into(buf, offset).await
122    }
123
124    /// Reads up to `len` bytes starting at `offset`, but only as many as are available.
125    ///
126    /// Returns the buffer (truncated to actual bytes read) and the number of bytes read. Returns
127    /// an error if no bytes are available at the given offset.
128    pub async fn read_up_to(
129        &self,
130        offset: u64,
131        len: usize,
132        bufs: impl Into<IoBufMut> + Send,
133    ) -> Result<(IoBufMut, usize), Error> {
134        self.view().read_up_to(offset, len, bufs).await
135    }
136
137    /// Read multiple fixed-size items at sorted byte offsets into a contiguous caller buffer.
138    ///
139    /// `buf` must be exactly `offsets.len() * item_size` bytes. All offsets must be sorted,
140    /// non-overlapping, and within bounds.
141    ///
142    /// Returns the number of items fully served without a blob read (from the in-memory tail and the
143    /// page cache). The remaining items required at least one blob read.
144    pub async fn read_many_into(
145        &self,
146        buf: &mut [u8],
147        offsets: &[u64],
148        item_size: NonZeroUsize,
149    ) -> Result<usize, Error> {
150        self.view().read_many_into(buf, offsets, item_size).await
151    }
152
153    /// Like [`Self::read_many_into`], but synchronous and cache-only. Returns the indices of
154    /// items that require a blob read. Their slots in `buf` hold unspecified bytes.
155    pub fn try_read_many_sync_into(
156        &self,
157        buf: &mut [u8],
158        offsets: &[u64],
159        item_size: NonZeroUsize,
160    ) -> Vec<usize> {
161        self.view().try_read_many_sync_into(buf, offsets, item_size)
162    }
163
164    /// Like [`Self::try_read_many_sync_into`], but for variable-length `(offset, len)` ranges:
165    /// `buf` holds one slot per range, back to back.
166    pub fn try_read_ranges_sync_into(&self, buf: &mut [u8], ranges: &[(u64, usize)]) -> Vec<usize> {
167        self.view().try_read_ranges_sync_into(buf, ranges)
168    }
169
170    /// Returns a [Replay] for sequentially reading all logical bytes of the sealed view.
171    ///
172    /// Sealed values have no write buffer to flush, so unlike [`super::Writer::replay`] this method
173    /// is not async. Every underlying blob read performed by the returned replay uses
174    /// `read_options`, including refills after seeking.
175    pub fn replay(
176        &self,
177        buffer_size: NonZeroUsize,
178        read_options: ReadOptions,
179    ) -> Result<Replay<B>, Error> {
180        let page_size_nz = self.inner.cache_ref.page_size();
181        let page_size: u64 = page_size_nz.widen();
182        let physical_page_size = page_size
183            .checked_add(CHECKSUM_SIZE)
184            .ok_or(Error::OffsetOverflow)?;
185        let prefetch_pages = (buffer_size.get() / physical_page_size as usize).max(1);
186
187        let partial_len = self
188            .inner
189            .partial_page
190            .as_ref()
191            .map_or(0, |p| p.len() as u64);
192        let full_pages = (self.inner.size - partial_len) / page_size;
193        let pages = full_pages + u64::from(partial_len > 0);
194        let physical_blob_size = physical_page_size
195            .checked_mul(pages)
196            .ok_or(Error::OffsetOverflow)?;
197        let logical_blob_size = self.inner.size;
198
199        let reader = PageReader::new(
200            self.inner.blob.clone(),
201            physical_blob_size,
202            logical_blob_size,
203            prefetch_pages,
204            page_size_nz,
205            read_options,
206        );
207        Ok(Replay::new(reader))
208    }
209
210    /// Page-cache id used for reads. Exposed for tests that verify the id is preserved across
211    /// [`super::Writer::seal`].
212    #[cfg(test)]
213    pub(super) fn cache_id(&self) -> u64 {
214        self.inner.id
215    }
216}
217
218#[cfg(test)]
219mod tests {
220    use super::*;
221    use crate::{
222        Buf, Runner as _, Storage as _,
223        buffer::{paged::Writer, tests::SyncTrackingBlob},
224        deterministic,
225        mocks::{DelayedSyncBlob, next_pending_sync},
226    };
227    use commonware_macros::test_traced;
228    use commonware_utils::{NZU16, NZUsize};
229    use std::num::NonZeroU16;
230
231    const PAGE_SIZE: NonZeroU16 = NZU16!(103); // janky page size to test alignment
232    const BUFFER_SIZE: usize = PAGE_SIZE.get() as usize * 2;
233
234    /// Seal a [Writer] and assert the returned sync handle makes it durable.
235    #[test_traced("DEBUG")]
236    fn test_seal_starts_sync() {
237        let executor = deterministic::Runner::default();
238        executor.start(|context: deterministic::Context| async move {
239            let blob = SyncTrackingBlob::new();
240            let cache_ref =
241                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
242            let mut append = Writer::new(blob.clone(), 0, BUFFER_SIZE, cache_ref)
243                .await
244                .unwrap();
245
246            // Append some data crossing several pages but don't sync.
247            let data: Vec<u8> = (0u8..=255).cycle().take(300).collect();
248            append.append(&data).await.unwrap();
249
250            let (durable_before, _writes_before, full_before, range_before) = blob.snapshot();
251            assert!(
252                durable_before.is_empty(),
253                "no bytes should be durable before the seal's sync"
254            );
255
256            let (sealed, sync) = append.seal().await.unwrap();
257            sync.await.unwrap();
258
259            let (durable_after, _writes_after, full_after, range_after) = blob.snapshot();
260            assert_eq!(full_after, full_before + 1);
261            assert!(
262                !durable_after.is_empty(),
263                "the seal's sync handle must make the appended bytes durable"
264            );
265            assert_eq!(
266                range_after, range_before,
267                "seal must not invoke a range-scoped write"
268            );
269
270            assert_eq!(sealed.size(), 300);
271        });
272    }
273
274    /// Sealing consumes the unique write handle; outstanding readers remain valid and agree
275    /// with the sealed view.
276    #[test_traced("DEBUG")]
277    fn test_seal_succeeds_with_readers() {
278        let executor = deterministic::Runner::default();
279        executor.start(|context: deterministic::Context| async move {
280            let (blob, blob_size) = context.open("test_partition", b"readers").await.unwrap();
281            let cache_ref =
282                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
283            let mut writer = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
284                .await
285                .unwrap();
286            writer.append(b"hello world").await.unwrap();
287
288            // A snapshot captures the buffered bytes as an owned, frozen read handle.
289            let reader = writer.snapshot().await.unwrap();
290            let reader_clone = reader.clone();
291            assert_eq!(reader.size(), 11);
292
293            // Seal succeeds while snapshots exist.
294            let (sealed, sync) = writer.seal().await.unwrap();
295            sync.await.unwrap();
296            assert_eq!(sealed.size(), 11);
297
298            // Both snapshot handles keep reading the frozen state and agree with the sealed view.
299            for r in [&reader, &reader_clone] {
300                assert_eq!(r.size(), 11);
301                let via_reader = r.read_at(0, 11).await.unwrap().coalesce();
302                let via_sealed = sealed.read_at(0, 11).await.unwrap().coalesce();
303                assert_eq!(via_reader.as_ref(), b"hello world");
304                assert_eq!(via_sealed.as_ref(), via_reader.as_ref());
305            }
306        });
307    }
308
309    /// A reader created before sealing reads full pages and the partial page after the seal,
310    /// from both the page cache and the blob.
311    #[test_traced("DEBUG")]
312    fn test_reader_full_pages_after_seal() {
313        let executor = deterministic::Runner::default();
314        executor.start(|context: deterministic::Context| async move {
315            let (blob, blob_size) = context.open("test_partition", b"rdr_pages").await.unwrap();
316            // A single-page cache forces most full-page reads to miss and hit the blob.
317            let cache_ref = super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(1));
318            let mut writer = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
319                .await
320                .unwrap();
321
322            let page_size = PAGE_SIZE.get() as usize;
323            let total = page_size * 3 + 7;
324            let data: Vec<u8> = (0u8..=255).cycle().take(total).collect();
325            writer.append(&data).await.unwrap();
326
327            let reader = writer.snapshot().await.unwrap();
328            let (sealed, sync) = writer.seal().await.unwrap();
329            sync.await.unwrap();
330            assert_eq!(reader.size(), total as u64);
331
332            // Full range, a page-straddling range, and the partial page, each compared
333            // against the sealed view.
334            let cases = [
335                (0u64, total),
336                (page_size as u64 - 3, 6),
337                ((page_size * 3) as u64, 7),
338            ];
339            for (offset, len) in cases {
340                let via_reader = reader.read_at(offset, len).await.unwrap().coalesce();
341                let via_sealed = sealed.read_at(offset, len).await.unwrap().coalesce();
342                assert_eq!(
343                    via_reader.as_ref(),
344                    &data[offset as usize..offset as usize + len]
345                );
346                assert_eq!(via_sealed.as_ref(), via_reader.as_ref());
347            }
348        });
349    }
350
351    /// Sealing preserves the originating [Writer]'s page-cache id.
352    #[test_traced("DEBUG")]
353    fn test_seal_preserves_cache_id() {
354        let executor = deterministic::Runner::default();
355        executor.start(|context: deterministic::Context| async move {
356            let (blob, blob_size) = context.open("test_partition", b"cache_id").await.unwrap();
357            let cache_ref =
358                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
359            let append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
360                .await
361                .unwrap();
362            let append_id = append.cache_id();
363            let (sealed, sync) = append.seal().await.unwrap();
364            sync.await.unwrap();
365            assert_eq!(sealed.cache_id(), append_id);
366        });
367    }
368
369    /// Sealing an empty blob yields an empty sealed view.
370    #[test_traced("DEBUG")]
371    fn test_seal_empty_blob() {
372        let executor = deterministic::Runner::default();
373        executor.start(|context: deterministic::Context| async move {
374            let (blob, blob_size) = context.open("test_partition", b"empty").await.unwrap();
375            let cache_ref =
376                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
377            let append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
378                .await
379                .unwrap();
380            let (sealed, sync) = append.seal().await.unwrap();
381            sync.await.unwrap();
382
383            assert_eq!(sealed.size(), 0);
384
385            // Out-of-bounds reads error.
386            let mut buf = [0u8; 1];
387            let err = sealed.read_into(&mut buf, 0).await.unwrap_err();
388            assert!(matches!(err, Error::BlobInsufficientLength));
389        });
390    }
391
392    /// Sealing a blob whose size is exactly a page-multiple has no partial page.
393    #[test_traced("DEBUG")]
394    fn test_seal_full_pages_only() {
395        let executor = deterministic::Runner::default();
396        executor.start(|context: deterministic::Context| async move {
397            let (blob, blob_size) = context.open("test_partition", b"full").await.unwrap();
398            let cache_ref =
399                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
400            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
401                .await
402                .unwrap();
403
404            // Append exactly two pages.
405            let page_size = PAGE_SIZE.get() as usize;
406            let data: Vec<u8> = (0u8..=255).cycle().take(page_size * 2).collect();
407            append.append(&data).await.unwrap();
408            let (sealed, sync) = append.seal().await.unwrap();
409            sync.await.unwrap();
410
411            assert_eq!(sealed.size(), data.len() as u64);
412
413            // Read everything back.
414            let mut buf = vec![0u8; data.len()];
415            sealed.read_into(&mut buf, 0).await.unwrap();
416            assert_eq!(buf, data);
417        });
418    }
419
420    /// Sealing a blob whose size is smaller than one page yields only a partial page.
421    #[test_traced("DEBUG")]
422    fn test_seal_partial_only() {
423        let executor = deterministic::Runner::default();
424        executor.start(|context: deterministic::Context| async move {
425            let (blob, blob_size) = context.open("test_partition", b"partial").await.unwrap();
426            let cache_ref =
427                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
428            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
429                .await
430                .unwrap();
431
432            // Append fewer than one page of data.
433            let data: Vec<u8> = (0u8..=50).collect();
434            append.append(&data).await.unwrap();
435            let (sealed, sync) = append.seal().await.unwrap();
436            sync.await.unwrap();
437
438            assert_eq!(sealed.size(), data.len() as u64);
439
440            let mut buf = vec![0u8; data.len()];
441            sealed.read_into(&mut buf, 0).await.unwrap();
442            assert_eq!(buf, data);
443        });
444    }
445
446    /// Reads that straddle the partial-page boundary stitch together cache and partial bytes.
447    #[test_traced("DEBUG")]
448    fn test_seal_full_plus_partial_straddle() {
449        let executor = deterministic::Runner::default();
450        executor.start(|context: deterministic::Context| async move {
451            let (blob, blob_size) = context.open("test_partition", b"straddle").await.unwrap();
452            let cache_ref =
453                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
454            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
455                .await
456                .unwrap();
457
458            // One full page + a partial.
459            let page_size = PAGE_SIZE.get() as usize;
460            let total = page_size + 17;
461            let data: Vec<u8> = (0u8..=255).cycle().take(total).collect();
462            append.append(&data).await.unwrap();
463            let (sealed, sync) = append.seal().await.unwrap();
464            sync.await.unwrap();
465
466            assert_eq!(sealed.size(), total as u64);
467
468            // Straddle read: 5 bytes before the boundary and 10 after.
469            let off = (page_size - 5) as u64;
470            let len = 15usize;
471            let mut buf = vec![0u8; len];
472            sealed.read_into(&mut buf, off).await.unwrap();
473            assert_eq!(buf, data[page_size - 5..page_size - 5 + len]);
474
475            // Read fully within partial.
476            let off = page_size as u64;
477            let mut buf = vec![0u8; 10];
478            sealed.read_into(&mut buf, off).await.unwrap();
479            assert_eq!(buf, data[page_size..page_size + 10]);
480
481            // Read fully within first full page.
482            let mut buf = vec![0u8; 20];
483            sealed.read_into(&mut buf, 0).await.unwrap();
484            assert_eq!(buf, data[..20]);
485        });
486    }
487
488    /// `Sealed::read_at` exposes the same data as `read_into`.
489    #[test_traced("DEBUG")]
490    fn test_sealed_read_at() {
491        let executor = deterministic::Runner::default();
492        executor.start(|context: deterministic::Context| async move {
493            let (blob, blob_size) = context.open("test_partition", b"read_at").await.unwrap();
494            let cache_ref =
495                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
496            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
497                .await
498                .unwrap();
499
500            let data: Vec<u8> = (0u8..=255).cycle().take(250).collect();
501            append.append(&data).await.unwrap();
502            let (sealed, sync) = append.seal().await.unwrap();
503            sync.await.unwrap();
504
505            let bufs = sealed.read_at(0, data.len()).await.unwrap();
506            let coalesced = bufs.coalesce();
507            assert_eq!(coalesced.as_ref(), data.as_slice());
508        });
509    }
510
511    /// `Sealed::read_many_into` returns items at sorted, possibly straddling, offsets.
512    #[test_traced("DEBUG")]
513    fn test_sealed_read_many_into() {
514        let executor = deterministic::Runner::default();
515        executor.start(|context: deterministic::Context| async move {
516            let (blob, blob_size) = context.open("test_partition", b"rmany").await.unwrap();
517            let cache_ref =
518                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
519            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
520                .await
521                .unwrap();
522
523            // Two pages worth so reads exercise both cache and partial.
524            let page_size = PAGE_SIZE.get() as usize;
525            let total = page_size + 50;
526            let data: Vec<u8> = (0u8..=255).cycle().take(total).collect();
527            append.append(&data).await.unwrap();
528            let (sealed, sync) = append.seal().await.unwrap();
529            sync.await.unwrap();
530
531            // 4-byte items at three positions: pure cache, straddling boundary, pure partial.
532            let offsets = [0u64, (page_size - 2) as u64, (page_size + 10) as u64];
533            let item_size = 4usize;
534            let mut out = vec![0u8; offsets.len() * item_size];
535            sealed
536                .read_many_into(&mut out, &offsets, NZUsize!(item_size))
537                .await
538                .unwrap();
539
540            for (i, &off) in offsets.iter().enumerate() {
541                assert_eq!(
542                    &out[i * item_size..(i + 1) * item_size],
543                    &data[off as usize..off as usize + item_size],
544                );
545            }
546        });
547    }
548
549    /// `Sealed::try_read_many_sync_into` serves cached pages and the in-memory tail, and maps
550    /// missed ranges (including straddling prefixes) back to item indices.
551    #[test_traced("DEBUG")]
552    fn test_sealed_try_read_many_sync_into() {
553        let executor = deterministic::Runner::default();
554        executor.start(|context: deterministic::Context| async move {
555            let (blob, blob_size) = context.open("test_partition", b"rmany_sync").await.unwrap();
556            // Capacity of one page makes hit/miss behavior deterministic: the cache holds
557            // exactly the last page touched.
558            let cache_ref = super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(1));
559            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
560                .await
561                .unwrap();
562
563            // Two full pages plus a partial tail page held in memory by the sealed view.
564            let page_size = PAGE_SIZE.get() as usize;
565            let total = page_size * 2 + 50;
566            let data: Vec<u8> = (0u8..=255).cycle().take(total).collect();
567            append.append(&data).await.unwrap();
568            let (sealed, sync) = append.seal().await.unwrap();
569            sync.await.unwrap();
570
571            // Items: page 0, page 1, straddling page 1 and the tail, pure tail.
572            let offsets = [
573                0u64,
574                page_size as u64,
575                (page_size * 2 - 2) as u64,
576                (page_size * 2 + 10) as u64,
577            ];
578            let item_size = 4usize;
579            let check = |out: &[u8], indices: &[usize]| {
580                for &i in indices {
581                    let off = offsets[i] as usize;
582                    assert_eq!(
583                        &out[i * item_size..(i + 1) * item_size],
584                        &data[off..off + item_size],
585                    );
586                }
587            };
588
589            // With only page 0 cached, items touching page 1 are misses. The tail item is
590            // served from the sealed view's in-memory bytes.
591            sealed.read_at(0, page_size).await.unwrap();
592            let mut out = vec![0u8; offsets.len() * item_size];
593            let misses = sealed.try_read_many_sync_into(&mut out, &offsets, NZUsize!(item_size));
594            assert_eq!(misses, vec![1, 2]);
595            check(&out, &[0, 3]);
596
597            // With only page 1 cached, item 0 becomes the miss and the straddler is served.
598            sealed.read_at(page_size as u64, page_size).await.unwrap();
599            let mut out = vec![0u8; offsets.len() * item_size];
600            let misses = sealed.try_read_many_sync_into(&mut out, &offsets, NZUsize!(item_size));
601            assert_eq!(misses, vec![0]);
602            check(&out, &[1, 2, 3]);
603        });
604    }
605
606    /// `Sealed::try_read_ranges_sync_into` serves cached pages and the in-memory tail for
607    /// variable-length ranges, and maps missed ranges back to range indices, including when a
608    /// zero-length range shares its offset with the missed range that follows it.
609    #[test_traced("DEBUG")]
610    fn test_sealed_try_read_ranges_sync_into() {
611        let executor = deterministic::Runner::default();
612        executor.start(|context: deterministic::Context| async move {
613            let (blob, blob_size) = context
614                .open("test_partition", b"rranges_sync")
615                .await
616                .unwrap();
617            // Capacity of one page makes hit/miss behavior deterministic: the cache holds
618            // exactly the last page touched.
619            let cache_ref = super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(1));
620            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
621                .await
622                .unwrap();
623
624            // Two full pages plus a partial tail page held in memory by the sealed view.
625            let page_size = PAGE_SIZE.get() as usize;
626            let total = page_size * 2 + 50;
627            let data: Vec<u8> = (0u8..=255).cycle().take(total).collect();
628            append.append(&data).await.unwrap();
629            let (sealed, sync) = append.seal().await.unwrap();
630            sync.await.unwrap();
631
632            // Ranges: page 0, a zero-length range sharing its offset with the page 1 range
633            // that follows it, page 1, straddling page 1 and the tail, pure tail.
634            let ranges = [
635                (0u64, 3usize),
636                (page_size as u64 + 5, 0),
637                (page_size as u64 + 5, 7),
638                ((page_size * 2 - 2) as u64, 4),
639                ((page_size * 2 + 10) as u64, 4),
640            ];
641            let total_len: usize = ranges.iter().map(|&(_, len)| len).sum();
642            let check = |out: &[u8], indices: &[usize]| {
643                let mut start = 0;
644                for (i, &(off, len)) in ranges.iter().enumerate() {
645                    if indices.contains(&i) {
646                        let off = off as usize;
647                        assert_eq!(&out[start..start + len], &data[off..off + len]);
648                    }
649                    start += len;
650                }
651            };
652
653            // With only page 0 cached, the page 1 range and the straddler's prefix miss. The
654            // zero-length range never misses. The tail range is served in memory.
655            sealed.read_at(0, page_size).await.unwrap();
656            let mut out = vec![0u8; total_len];
657            let misses = sealed.try_read_ranges_sync_into(&mut out, &ranges);
658            assert_eq!(misses, vec![2, 3]);
659            check(&out, &[0, 4]);
660
661            // With only page 1 cached, range 0 becomes the miss and the rest are served.
662            sealed.read_at(page_size as u64, page_size).await.unwrap();
663            let mut out = vec![0u8; total_len];
664            let misses = sealed.try_read_ranges_sync_into(&mut out, &ranges);
665            assert_eq!(misses, vec![0]);
666            check(&out, &[1, 2, 3, 4]);
667        });
668    }
669
670    /// `Sealed::read_many_into` falls back to blob reads for full-page cache misses.
671    #[test_traced("DEBUG")]
672    fn test_sealed_read_many_into_cache_miss() {
673        let executor = deterministic::Runner::default();
674        executor.start(|context: deterministic::Context| async move {
675            let (blob, blob_size) = context.open("test_partition", b"rmany_miss").await.unwrap();
676            let cache_ref = super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(1));
677            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
678                .await
679                .unwrap();
680
681            let page_size = PAGE_SIZE.get() as usize;
682            let data: Vec<u8> = (0u8..=255).cycle().take(page_size * 2).collect();
683            append.append(&data).await.unwrap();
684            let (sealed, sync) = append.seal().await.unwrap();
685            sync.await.unwrap();
686
687            let offsets = [0u64, page_size as u64];
688            let item_size = 4usize;
689            let mut out = vec![0u8; offsets.len() * item_size];
690            sealed
691                .read_many_into(&mut out, &offsets, NZUsize!(item_size))
692                .await
693                .unwrap();
694
695            for (i, &off) in offsets.iter().enumerate() {
696                assert_eq!(
697                    &out[i * item_size..(i + 1) * item_size],
698                    &data[off as usize..off as usize + item_size],
699                );
700            }
701        });
702    }
703
704    #[test_traced("DEBUG")]
705    #[should_panic(expected = "ranges must be sorted and non-overlapping")]
706    fn test_sealed_read_many_into_rejects_unsorted_offsets() {
707        let executor = deterministic::Runner::default();
708        executor.start(|context: deterministic::Context| async move {
709            let (blob, blob_size) = context.open("test_partition", b"rmany_bad").await.unwrap();
710            let cache_ref =
711                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
712            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
713                .await
714                .unwrap();
715            append.append(&[7; 32]).await.unwrap();
716            let (sealed, sync) = append.seal().await.unwrap();
717            sync.await.unwrap();
718
719            let mut out = vec![0u8; 8];
720            let _ = sealed.read_many_into(&mut out, &[8, 4], NZUsize!(4)).await;
721        });
722    }
723
724    /// `Sealed::read_many_into` validates all caller-provided offsets before reading.
725    #[test_traced("DEBUG")]
726    fn test_sealed_read_many_into_rejects_invalid_offsets() {
727        let executor = deterministic::Runner::default();
728        executor.start(|context: deterministic::Context| async move {
729            let (blob, blob_size) = context.open("test_partition", b"rmany_bad").await.unwrap();
730            let cache_ref =
731                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
732            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
733                .await
734                .unwrap();
735            append.append(&[7; 32]).await.unwrap();
736            let (sealed, sync) = append.seal().await.unwrap();
737            sync.await.unwrap();
738
739            let mut out = vec![0u8; 8];
740            let err = sealed
741                .read_many_into(&mut out, &[u64::MAX - 1, 8], NZUsize!(4))
742                .await
743                .unwrap_err();
744            assert!(matches!(err, Error::OffsetOverflow));
745
746            let err = sealed
747                .read_many_into(&mut out, &[28, 32], NZUsize!(4))
748                .await
749                .unwrap_err();
750            assert!(matches!(err, Error::BlobInsufficientLength));
751        });
752    }
753
754    /// `try_read_sync_into` succeeds when bytes come purely from the in-memory partial page.
755    #[test_traced("DEBUG")]
756    fn test_sealed_try_read_sync_partial() {
757        let executor = deterministic::Runner::default();
758        executor.start(|context: deterministic::Context| async move {
759            let (blob, blob_size) = context
760                .open("test_partition", b"trs_partial")
761                .await
762                .unwrap();
763            let cache_ref =
764                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
765            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
766                .await
767                .unwrap();
768
769            let page_size = PAGE_SIZE.get() as usize;
770            let total = page_size + 30;
771            let data: Vec<u8> = (0u8..=255).cycle().take(total).collect();
772            append.append(&data).await.unwrap();
773            let (sealed, sync) = append.seal().await.unwrap();
774            sync.await.unwrap();
775
776            // Read fully within partial.
777            let mut buf = vec![0u8; 10];
778            assert!(sealed.try_read_sync_into(&mut buf, page_size as u64));
779            assert_eq!(buf, data[page_size..page_size + 10]);
780
781            // Out of bounds returns false.
782            let mut buf = vec![0u8; 10];
783            assert!(!sealed.try_read_sync_into(&mut buf, total as u64));
784        });
785    }
786
787    /// `try_read_sync_into` can stitch a cached full-page prefix to in-memory partial bytes.
788    #[test_traced("DEBUG")]
789    fn test_sealed_try_read_sync_straddles_cached_and_partial() {
790        let executor = deterministic::Runner::default();
791        executor.start(|context: deterministic::Context| async move {
792            let (blob, blob_size) = context
793                .open("test_partition", b"trs_straddle")
794                .await
795                .unwrap();
796            let cache_ref =
797                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
798            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
799                .await
800                .unwrap();
801
802            let page_size = PAGE_SIZE.get() as usize;
803            let total = page_size + 30;
804            let data: Vec<u8> = (0u8..=255).cycle().take(total).collect();
805            append.append(&data).await.unwrap();
806            let (sealed, sync) = append.seal().await.unwrap();
807            sync.await.unwrap();
808
809            let mut buf = vec![0u8; 12];
810            assert!(sealed.try_read_sync_into(&mut buf, (page_size - 4) as u64));
811            assert_eq!(buf, data[page_size - 4..page_size + 8]);
812        });
813    }
814
815    /// Synchronous reads past the sealed size are rejected.
816    #[test_traced("DEBUG")]
817    fn test_sealed_try_read_sync_out_of_bounds() {
818        let executor = deterministic::Runner::default();
819        executor.start(|context: deterministic::Context| async move {
820            let (blob, blob_size) = context.open("test_partition", b"trs_fail").await.unwrap();
821            let cache_ref =
822                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
823            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
824                .await
825                .unwrap();
826
827            let page_size = PAGE_SIZE.get() as usize;
828            let data: Vec<u8> = (0u8..=255).cycle().take(page_size + 5).collect();
829            append.append(&data).await.unwrap();
830            let (sealed, sync) = append.seal().await.unwrap();
831            sync.await.unwrap();
832
833            let mut buf = vec![9u8; 10];
834            assert!(!sealed.try_read_sync_into(&mut buf, data.len() as u64));
835        });
836    }
837
838    /// `Sealed::replay` streams all logical bytes including the partial page.
839    #[test_traced("DEBUG")]
840    fn test_sealed_replay() {
841        let executor = deterministic::Runner::default();
842        executor.start(|context: deterministic::Context| async move {
843            let (blob, blob_size) = context.open("test_partition", b"replay").await.unwrap();
844            let cache_ref =
845                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
846            let mut append = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
847                .await
848                .unwrap();
849
850            // Two pages + a partial, synced so the bytes are on disk before sealing.
851            let page_size = PAGE_SIZE.get() as usize;
852            let total = page_size * 2 + 25;
853            let data: Vec<u8> = (0u8..=255).cycle().take(total).collect();
854            append.append(&data).await.unwrap();
855            append.sync().await.unwrap();
856            let (sealed, sync) = append.seal().await.unwrap();
857            sync.await.unwrap();
858
859            let mut replay = sealed
860                .replay(NZUsize!(BUFFER_SIZE), ReadOptions::default())
861                .unwrap();
862            assert_eq!(replay.blob_size(), total as u64);
863
864            // Drain all logical bytes.
865            let mut out = Vec::with_capacity(total);
866            while replay.ensure(1).await.unwrap() {
867                let chunk = replay.chunk();
868                let copy_len = chunk.len();
869                out.extend_from_slice(chunk);
870                replay.advance(copy_len);
871            }
872            assert_eq!(out, data);
873        });
874    }
875
876    /// Replaying a snapshot must stop at the snapshot's logical boundary, even if the live writer
877    /// later extends the same physical page.
878    #[test_traced("DEBUG")]
879    fn test_snapshot_replay_stays_frozen_after_writer_growth() {
880        let executor = deterministic::Runner::default();
881        executor.start(|context: deterministic::Context| async move {
882            let (blob, blob_size) = context
883                .open("test_partition", b"snapshot_replay_growth")
884                .await
885                .unwrap();
886            let cache_ref =
887                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
888            let mut writer = Writer::new(blob, blob_size, BUFFER_SIZE, cache_ref)
889                .await
890                .unwrap();
891
892            let page_size = PAGE_SIZE.get() as usize;
893            let mut original = vec![0xAA; page_size];
894            original.extend_from_slice(b"old");
895            writer.append(&original).await.unwrap();
896            writer.sync().await.unwrap();
897
898            let snapshot = writer.snapshot().await.unwrap();
899            let snapshot_bytes = snapshot
900                .read_at(0, snapshot.size() as usize)
901                .await
902                .unwrap()
903                .coalesce();
904            let mut replay = snapshot
905                .replay(NZUsize!(BUFFER_SIZE), ReadOptions::default())
906                .unwrap();
907            assert_eq!(replay.blob_size(), original.len() as u64);
908
909            writer.append(b"newtail").await.unwrap();
910            writer.sync().await.unwrap();
911
912            let mut out = Vec::new();
913            while replay.ensure(1).await.unwrap() {
914                let chunk = replay.chunk();
915                let copy_len = chunk.len();
916                out.extend_from_slice(chunk);
917                replay.advance(copy_len);
918            }
919
920            assert_eq!(out.as_slice(), snapshot_bytes.as_ref());
921            assert_eq!(out, original);
922        });
923    }
924
925    /// `Sealed::replay` works without a prior `Append::sync` because `Append::seal` writes bytes
926    /// to the blob before starting its sync.
927    #[test_traced("DEBUG")]
928    fn test_seal_replay_without_sync() {
929        let executor = deterministic::Runner::default();
930        executor.start(|context: deterministic::Context| async move {
931            let blob = SyncTrackingBlob::new();
932            let cache_ref =
933                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
934            let mut append = Writer::new(blob.clone(), 0, BUFFER_SIZE, cache_ref)
935                .await
936                .unwrap();
937
938            let page_size = PAGE_SIZE.get() as usize;
939            let total = page_size * 2 + 25;
940            let data: Vec<u8> = (0u8..=255).cycle().take(total).collect();
941            append.append(&data).await.unwrap();
942            let (sealed, sync) = append.seal().await.unwrap();
943
944            let (_durable, _writes, full_syncs, range_syncs) = blob.snapshot();
945            assert_eq!(full_syncs, 1);
946            assert_eq!(range_syncs, 0);
947            sync.await.unwrap();
948
949            let mut replay = sealed
950                .replay(NZUsize!(BUFFER_SIZE), ReadOptions::default())
951                .unwrap();
952            assert_eq!(replay.blob_size(), total as u64);
953
954            let mut out = Vec::with_capacity(total);
955            while replay.ensure(1).await.unwrap() {
956                let chunk = replay.chunk();
957                let copy_len = chunk.len();
958                out.extend_from_slice(chunk);
959                replay.advance(copy_len);
960            }
961            assert_eq!(out, data);
962        });
963    }
964
965    /// Reads and replay through `Sealed` observe flushed bytes while the seal's sync handle is
966    /// still pending; they never wait for durability.
967    #[test_traced("DEBUG")]
968    fn test_sealed_reads_while_seal_sync_pending() {
969        let executor = deterministic::Runner::default();
970        executor.start(|context: deterministic::Context| async move {
971            let inner = SyncTrackingBlob::new();
972            let (blob, pending) = DelayedSyncBlob::new(inner);
973            let cache_ref =
974                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
975            let mut append = Writer::new(blob, 0, BUFFER_SIZE, cache_ref).await.unwrap();
976
977            let page_size = PAGE_SIZE.get() as usize;
978            let total = page_size * 2 + 50;
979            let data: Vec<u8> = (0u8..=255).cycle().take(total).collect();
980            append.append(&data).await.unwrap();
981
982            // Seal parks its sync; leave it parked while reading.
983            let (sealed, sync) = append.seal().await.unwrap();
984            assert_eq!(pending.lock().len(), 1, "the seal sync should be parked");
985
986            let read = sealed.read_at(0, total).await.unwrap().coalesce();
987            assert_eq!(read.as_ref(), &data[..]);
988
989            let mut replay = sealed
990                .replay(NZUsize!(BUFFER_SIZE), ReadOptions::default())
991                .unwrap();
992            assert_eq!(replay.blob_size(), total as u64);
993            let mut replayed = Vec::new();
994            while replay.ensure(1).await.unwrap() {
995                let n = {
996                    let chunk = replay.chunk();
997                    replayed.extend_from_slice(chunk);
998                    chunk.len()
999                };
1000                replay.advance(n);
1001            }
1002            assert_eq!(replayed, data);
1003            assert_eq!(
1004                pending.lock().len(),
1005                1,
1006                "reads must not consume the pending sync"
1007            );
1008
1009            // Release the sync and confirm the handle completes.
1010            next_pending_sync(&pending).release.send(Ok(())).unwrap();
1011            sync.await.unwrap();
1012        });
1013    }
1014
1015    /// Sealing a recovered, already-synced partial page must not rewrite it.
1016    #[test_traced("DEBUG")]
1017    fn test_seal_recovered_synced_partial_page_no_write() {
1018        let executor = deterministic::Runner::default();
1019        executor.start(|context: deterministic::Context| async move {
1020            let blob = SyncTrackingBlob::new();
1021            let cache_ref =
1022                super::CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(BUFFER_SIZE));
1023            let data: Vec<u8> = (0u8..=255)
1024                .cycle()
1025                .take(PAGE_SIZE.get() as usize - 17)
1026                .collect();
1027
1028            {
1029                let mut writer = Writer::new(blob.clone(), 0, BUFFER_SIZE, cache_ref.clone())
1030                    .await
1031                    .unwrap();
1032                writer.append(&data).await.unwrap();
1033                writer.sync().await.unwrap();
1034            }
1035
1036            let (_, writes, full_syncs, range_syncs) = blob.snapshot();
1037            let mut recovered = Writer::new(blob.clone(), blob.size(), BUFFER_SIZE, cache_ref)
1038                .await
1039                .unwrap();
1040            assert_eq!(recovered.size(), data.len() as u64);
1041
1042            recovered.sync().await.unwrap();
1043            let (_, writes_after_sync, full_after_sync, range_after_sync) = blob.snapshot();
1044            assert_eq!(
1045                writes_after_sync, writes,
1046                "syncing an unchanged recovered partial page must not rewrite it"
1047            );
1048            assert_eq!(full_after_sync, full_syncs + 1);
1049            assert_eq!(range_after_sync, range_syncs);
1050
1051            let (sealed, sync) = recovered.seal().await.unwrap();
1052            sync.await.unwrap();
1053            let (_, writes_after_seal, full_after_seal, range_after_seal) = blob.snapshot();
1054            assert_eq!(
1055                writes_after_seal, writes_after_sync,
1056                "sealing an unchanged recovered partial page must not rewrite it"
1057            );
1058            assert_eq!(full_after_seal, full_after_sync);
1059            assert_eq!(range_after_seal, range_after_sync);
1060
1061            let read = sealed.read_at(0, data.len()).await.unwrap().coalesce();
1062            assert_eq!(read.as_ref(), data.as_slice());
1063        });
1064    }
1065}
commonware_runtime/utils/buffer/paged/sealed.rs

commonware_runtime/utils/buffer/paged/
sealed.rs
