coven_format/
snapshot_stream.rs

1//! Feed opened snapshot chunks to the frame decoder without retaining rows.
2
3use crate::chunks::FrameDecoder;
4use crate::error::Error;
5use crate::sealed_snapshot::SnapshotObjectPrefix;
6use crate::snapshot::{SnapshotDecoder, SnapshotHeader, SnapshotRecord, WriteSlots};
7
8/// Incrementally reconstruct frames and check the snapshot's end marker.
9/// Call [`Self::next_record`] repeatedly on each opened chunk, updating the
10/// oracle after each applied-write record before asking for the next record.
11pub struct SnapshotChunkDecoder {
12    prefix: SnapshotObjectPrefix,
13    frames: FrameDecoder,
14    snapshot: Option<SnapshotDecoder>,
15    frame: Vec<u8>,
16}
17impl SnapshotChunkDecoder {
18    /// Supply the authenticated prefix, including the snapshot’s coverage.
19    pub fn new(prefix: SnapshotObjectPrefix) -> Self {
20        Self { prefix, frames: FrameDecoder::new(None), snapshot: None, frame: Vec::new() }
21    }
22
23    /// Header metadata becomes available as soon as its frame is complete.
24    pub fn header(&self) -> Option<&SnapshotHeader> {
25        self.snapshot.as_ref().map(SnapshotDecoder::header)
26    }
27
28    /// Consume bytes through the next complete record, or all available bytes
29    /// if its frame is unfinished. `None` means this input has no further record.
30    /// Discard the decoder after an error; commit staged rows only after `finish`.
31    pub fn next_record<O: WriteSlots>(
32        &mut self,
33        bytes: &mut &[u8],
34        oracle: &O,
35    ) -> Result<Option<SnapshotRecord>, O::Error>
36    where
37        O::Error: From<Error>,
38    {
39        while let Some(frame) = self.frames.next(bytes)? {
40            if let Some(snapshot) = &mut self.snapshot {
41                if let Some(record) = snapshot.frame(&frame, oracle)? {
42                    self.frame = frame;
43                    return Ok(Some(record));
44                }
45            } else {
46                let snapshot = SnapshotDecoder::start(&frame, &self.prefix)?;
47                self.snapshot = Some(snapshot);
48            }
49        }
50        Ok(None)
51    }
52
53    /// Like [`Self::next_record`], before the store's historical signing keys
54    /// are known: only original slots and last claims are decoded, and every
55    /// other record is counted and passed over ([`SnapshotDecoder::history_frame`]).
56    /// A loader still decodes every record with [`Self::next_record`].
57    pub fn next_history(&mut self, bytes: &mut &[u8]) -> Result<Option<SnapshotRecord>, Error> {
58        while let Some(frame) = self.frames.next(bytes)? {
59            if let Some(snapshot) = &mut self.snapshot {
60                if let Some(record) = snapshot.history_frame(&frame)? {
61                    return Ok(Some(record));
62                }
63            } else {
64                self.snapshot = Some(SnapshotDecoder::start(&frame, &self.prefix)?);
65            }
66        }
67        Ok(None)
68    }
69
70    /// The exact bytes of the frame the last returned record came from, for a
71    /// loader that stages the record and reads it back with
72    /// [`SnapshotRecord::decode_checked`].
73    pub fn frame(&self) -> &[u8] {
74        &self.frame
75    }
76
77    /// Refuse EOF without the complete header, declared records and end marker.
78    pub fn finish(&self) -> Result<(), Error> {
79        self.frames.finish()?;
80        self.snapshot.as_ref().ok_or(Error::Truncated)?.finish()
81    }
82}
83
84#[cfg(test)]
85#[path = "snapshot_stream_tests.rs"]
86mod tests;
coven_format/snapshot_stream.rs

coven_format/
snapshot_stream.rs
