coven_format/
snapshot.rs

1//! Snapshots stream a header, eleven ordered sections and an end marker (§16.1).
2//! The caller stages records and supplies the applied write metadata to merge.
3
4use crate::admin::SnapshotId;
5use crate::encode_frame_with;
6use crate::error::{require, Error, Rule};
7use crate::kinds;
8use crate::loss::{Loss, LossCause, LossIdentity};
9use crate::snapshot_rows::*;
10use crate::value::{name, positive, Slot};
11use crate::wire::{decode_frame, decode_frame_of, Decoder, Encoder, Wire};
12use coven_merge::{RowId, WriteId, WriteOracle};
13
14/// Snapshot metadata assembled from its plaintext header and sealed prefix.
15#[derive(Clone, Debug, PartialEq, Eq)]
16pub struct SnapshotHeader {
17    /// Writing device, snapshot number and audience.
18    pub id: SnapshotId,
19    /// The app schema version.
20    pub schema_version: u32,
21    /// Last included slot from the signed sealed prefix.
22    pub covers: u64,
23    /// Counts: applied writes, columns, merged rows, losses, authored
24    /// changes, creations, write evidence, three original-slot sections and
25    /// per-device last included claims.
26    pub counts: [u64; SECTIONS],
27}
28/// The number of record sections (D7).
29pub const SECTIONS: usize = 11;
30
31/// Each record section's place in the stream, and its count's in the
32/// header (D7).
33pub mod section {
34    /// Applied writes.
35    pub const WRITE: usize = 0;
36    /// Columns.
37    pub const COLUMN: usize = 1;
38    /// Merged rows.
39    pub const MERGE: usize = 2;
40    /// Losses.
41    pub const LOSS: usize = 3;
42    /// Authored changes.
43    pub const AUTHORED: usize = 4;
44    /// Creations.
45    pub const CREATION: usize = 5;
46    /// Write evidence.
47    pub const EVIDENCE: usize = 6;
48    /// Original admin slots, the first of the three sections a history
49    /// read decodes.
50    pub const ADMIN: usize = 7;
51    /// Original key-copy slots.
52    pub const KEY_COPY: usize = 8;
53    /// Original announcement slots.
54    pub const ANNOUNCEMENT: usize = 9;
55    /// Each device's last included claim.
56    pub const CLAIM: usize = 10;
57}
58impl Wire for [u64; SECTIONS] {
59    fn put(&self, out: &mut Encoder) -> Result<(), Error> {
60        for n in self {
61            n.put(out)?;
62        }
63        Ok(())
64    }
65    fn get(input: &mut Decoder<'_>) -> Result<Self, Error> {
66        let mut counts = [0; SECTIONS];
67        for count in &mut counts {
68            *count = Wire::get(input)?;
69        }
70        Ok(counts)
71    }
72}
73impl SnapshotHeader {
74    /// The merge rows the snapshot holds.
75    pub fn merge_rows(&self) -> u64 {
76        self.counts[section::MERGE]
77    }
78
79    pub(crate) fn validate(&self) -> Result<(), Error> {
80        positive(self.id.number)?;
81        Ok(())
82    }
83}
84
85/// The section a snapshot's frame holds a record of; none for its header and
86/// end frames.
87#[cfg(any(test, feature = "test-utils"))]
88pub fn frame_section(frame: &[u8]) -> Option<usize> {
89    (frame[0] == kinds::SNAPSHOT_RECORD).then(|| usize::from(frame[crate::FRAME_PREFIX_LEN]))
90}
91
92/// One bounded record in a snapshot, using the owners' logical identities.
93#[derive(Clone, Debug, PartialEq, Eq)]
94pub enum SnapshotRecord<E = crate::authorship::WriteEvidence> {
95    /// Applied write metadata, before any row needing it is decoded.
96    Write(AppliedWrite),
97    /// A synced column.
98    Column(SyncedColumn),
99    /// One row's generations and winning cells, which are its values; losses
100    /// travel separately.
101    Merge(MergeRow),
102    /// A displaced cell, removed row or excluded write's row.
103    Loss(Loss),
104    /// An accepted write part's shared details, or one of its kept changes
105    /// with its verdict, kept as source evidence.
106    Authored(AuthoredRecord),
107    /// A row's selected original creation (§9.6).
108    Creation(crate::authorship::Creation),
109    /// Another member's write header, roots and openings in this audience.
110    Evidence(E),
111    /// Original signed admin slots, including creation; store snapshots only.
112    Admin(crate::floor::OriginalSlot),
113    /// Original signed key-copy slots; store snapshots only.
114    KeyCopy(crate::floor::OriginalSlot),
115    /// Original signed announcement slots; store snapshots only.
116    Announcement(crate::floor::OriginalSlot),
117    /// Last claim included for each known device, including refused claims.
118    Claim(LastClaim),
119}
120
121/// A device's last included claim and its complete-slot digest.
122#[derive(Clone, Debug, PartialEq, Eq)]
123pub struct LastClaim {
124    /// The known device.
125    pub device: coven_foundation::id_source::DeviceId,
126    /// Last included claim across all slot kinds.
127    pub number: u64,
128    /// SHA-256 of that slot's complete bytes; absent exactly when number is zero.
129    pub digest: Option<[u8; 32]>,
130}
131crate::wire::wire_struct!(LastClaim, device, number, digest);
132
133/// Published-write locations from the snapshot's staged metadata and history.
134pub trait WriteSlots: WriteOracle {
135    /// Containing slot, or absence when no authenticated metadata establishes it.
136    fn slot(&self, write: WriteId) -> Result<Option<Slot>, Self::Error>;
137    /// Whether the metadata holds a write of `write`'s device numbered at or
138    /// above it. A device's writes land in the order it numbered them, so
139    /// `write` landed at or before that write's slot (coven.md §16.7).
140    fn landed(&self, write: WriteId) -> Result<bool, Self::Error>;
141}
142impl SnapshotRecord {
143    /// Retain decoded records while replacing wire evidence with its checked form.
144    pub fn try_map_evidence<E, X>(
145        self,
146        prepare: impl FnOnce(crate::authorship::WriteEvidence) -> Result<E, X>,
147    ) -> Result<SnapshotRecord<E>, X> {
148        Ok(match self {
149            Self::Write(value) => SnapshotRecord::Write(value),
150            Self::Column(value) => SnapshotRecord::Column(value),
151            Self::Merge(value) => SnapshotRecord::Merge(value),
152            Self::Loss(value) => SnapshotRecord::Loss(value),
153            Self::Authored(value) => SnapshotRecord::Authored(value),
154            Self::Creation(value) => SnapshotRecord::Creation(value),
155            Self::Admin(value) => SnapshotRecord::Admin(value),
156            Self::KeyCopy(value) => SnapshotRecord::KeyCopy(value),
157            Self::Announcement(value) => SnapshotRecord::Announcement(value),
158            Self::Claim(value) => SnapshotRecord::Claim(value),
159            Self::Evidence(value) => SnapshotRecord::Evidence(prepare(value)?),
160        })
161    }
162
163    pub(crate) fn put(&self, out: &mut Encoder) -> Result<(), Error> {
164        u8::try_from(self.section()).expect("eleven sections").put(out)?;
165        match self {
166            Self::Write(v) => v.put(out),
167            Self::Column(v) => v.put(out),
168            Self::Merge(v) => v.put(out),
169            Self::Loss(v) => v.put(out),
170            Self::Authored(v) => v.put(out),
171            Self::Creation(v) => v.put(out),
172            Self::Evidence(v) => v.put(out),
173            Self::Admin(v) | Self::KeyCopy(v) | Self::Announcement(v) => v.put(out),
174            Self::Claim(v) => v.put(out),
175        }
176    }
177    pub(crate) fn get<O: WriteOracle>(input: &mut Decoder<'_>, oracle: &O) -> Result<Self, O::Error>
178    where
179        O::Error: From<Error>,
180    {
181        #[cfg(any(test, feature = "test-utils"))]
182        coven_foundation::test_work::snapshot_record_decode();
183        Self::read(input, oracle)
184    }
185
186    /// Read back a record frame that [`SnapshotDecoder::frame`] already
187    /// checked, as a loader staged it. Only the record's own fields are read;
188    /// its order, coverage and the stream it came from are not checked again.
189    /// A merge row rebuilds its state with `oracle`, as stored rows do.
190    pub fn decode_checked<O: WriteOracle>(frame: &[u8], oracle: &O) -> Result<Self, O::Error>
191    where
192        O::Error: From<Error>,
193    {
194        let mut input = decode_frame_of(frame, kinds::SNAPSHOT_RECORD, "snapshot record kind")?;
195        let record = Self::read(&mut input, oracle)?;
196        input.finish()?;
197        Ok(record)
198    }
199
200    fn read<O: WriteOracle>(input: &mut Decoder<'_>, oracle: &O) -> Result<Self, O::Error>
201    where
202        O::Error: From<Error>,
203    {
204        let tag = u8::get(input)?;
205        match usize::from(tag) {
206            section::WRITE => Ok(Self::Write(Wire::get(input)?)),
207            section::COLUMN => Ok(Self::Column(Wire::get(input)?)),
208            section::MERGE => Ok(Self::Merge(MergeRow::get(input, oracle)?)),
209            section::LOSS => Ok(Self::Loss(Wire::get(input)?)),
210            section::AUTHORED => Ok(Self::Authored(Wire::get(input)?)),
211            section::CREATION => Ok(Self::Creation(Wire::get(input)?)),
212            section::EVIDENCE => Ok(Self::Evidence(Wire::get(input)?)),
213            section::ADMIN => Ok(Self::Admin(Wire::get(input)?)),
214            section::KEY_COPY => Ok(Self::KeyCopy(Wire::get(input)?)),
215            section::ANNOUNCEMENT => Ok(Self::Announcement(Wire::get(input)?)),
216            section::CLAIM => Ok(Self::Claim(Wire::get(input)?)),
217            _ => Err(Error::UnknownTag { field: "snapshot section", tag }.into()),
218        }
219    }
220    pub(crate) fn validate(&self) -> Result<(), Error> {
221        match self {
222            Self::Write(v) => v.validate(),
223            Self::Column(v) => {
224                name(&v.table)?;
225                name(&v.column)
226            }
227            Self::Merge(v) => v.validate(),
228            Self::Loss(v) => v.validate(),
229            Self::Authored(AuthoredRecord::Part(v)) => v.header.validate(),
230            Self::Authored(AuthoredRecord::Change(v)) => v.validate(),
231            Self::Creation(v) => v.validate(),
232            Self::Evidence(v) => v.validate(),
233            // An admission is an admin entry with its member's copies (D7.6).
234            Self::Admin(v) => v.validate_kind(&[0, 4]),
235            Self::KeyCopy(v) => v.validate_kind(&[2]),
236            Self::Announcement(v) => v.validate_kind(&[3]),
237            Self::Claim(v) => {
238                require((v.number == 0) == v.digest.is_none(), "last claim digest", Rule::Required)
239            }
240        }
241    }
242    /// The section the record belongs to.
243    pub fn section(&self) -> usize {
244        match self {
245            Self::Write(_) => section::WRITE,
246            Self::Column(_) => section::COLUMN,
247            Self::Merge(_) => section::MERGE,
248            Self::Loss(_) => section::LOSS,
249            Self::Authored(_) => section::AUTHORED,
250            Self::Creation(_) => section::CREATION,
251            Self::Evidence(_) => section::EVIDENCE,
252            Self::Admin(_) => section::ADMIN,
253            Self::KeyCopy(_) => section::KEY_COPY,
254            Self::Announcement(_) => section::ANNOUNCEMENT,
255            Self::Claim(_) => section::CLAIM,
256        }
257    }
258    /// The record's place in its section's order. A kept change is ordered
259    /// within `part`, the part record before it.
260    fn key(&self, part: Option<&crate::write::WriteHeader>) -> Result<RecordKey, Error> {
261        Ok(match self {
262            Self::Write(v) => RecordKey::Write(v.id),
263            Self::Column(v) => RecordKey::Column(v.clone()),
264            Self::Merge(v) => RecordKey::Row(v.state.row().clone()),
265            Self::Loss(v) => RecordKey::Loss(v.key()),
266            Self::Authored(AuthoredRecord::Part(v)) => RecordKey::Authored(v.header.position, None),
267            Self::Authored(AuthoredRecord::Change(v)) => {
268                let part = part.ok_or(Error::Invalid {
269                    field: "authored change part",
270                    rule: Rule::SnapshotSequence,
271                })?;
272                RecordKey::Authored(part.position, Some(v.record.row().clone()))
273            }
274            Self::Creation(v) => RecordKey::Row(v.row.clone()),
275            Self::Evidence(v) => RecordKey::Write(v.write),
276            Self::Admin(v) | Self::KeyCopy(v) | Self::Announcement(v) => RecordKey::Slot(v.slot),
277            Self::Claim(v) => RecordKey::Device(v.device),
278        })
279    }
280    fn check_header(&self, h: &SnapshotHeader) -> Result<(), Error> {
281        let audience =
282            |r: &RowId| require(r.audience == h.id.audience, "snapshot audience", Rule::Audience);
283        match self {
284            Self::Write(v) => {
285                require(v.slot.get() <= h.covers, "snapshot write slot", Rule::Coverage)
286            }
287            Self::Column(_) | Self::Evidence(_) | Self::Claim(_) => Ok(()),
288            Self::Loss(v) => {
289                audience(&v.row)?;
290                if let LossCause::Excluded { version, .. } = v.cause {
291                    require(version <= h.schema_version, "loss cause", Rule::Coverage)?;
292                }
293                if let LossCause::Rules(rules) = &v.cause {
294                    for rule in rules {
295                        if let coven_merge::Rule::DeletedCircle(entry) = rule {
296                            require(entry.get() <= h.covers, "loss cause", Rule::Coverage)?;
297                        }
298                    }
299                }
300                Ok(())
301            }
302            Self::Merge(v) => audience(v.state.row()),
303            Self::Creation(v) => audience(&v.row),
304            Self::Authored(AuthoredRecord::Part(v)) => {
305                require(v.header.read_through <= h.covers, "authored read-through", Rule::Coverage)
306            }
307            Self::Authored(AuthoredRecord::Change(v)) => {
308                audience(v.record.row())?;
309                require(
310                    v.stop.is_none_or(|stop| stop.get() <= h.covers),
311                    "authored stop slot",
312                    Rule::Coverage,
313                )?;
314                match &v.verdict {
315                    ChangeVerdict::Excluded(version) => {
316                        require(*version <= h.schema_version, "change verdict", Rule::Coverage)
317                    }
318                    ChangeVerdict::Ignored(entry) => {
319                        require(entry.get() <= h.covers, "change verdict", Rule::Coverage)
320                    }
321                    _ => Ok(()),
322                }
323            }
324            Self::Admin(v) | Self::KeyCopy(v) | Self::Announcement(v) => require(
325                h.id.audience == coven_merge::Audience::Store && v.slot.get() <= h.covers,
326                "snapshot historical slot",
327                Rule::Coverage,
328            ),
329        }
330    }
331
332    /// Check references against authenticated published-write locations. Callers
333    /// populate these from section 0; unknown evidence cannot establish coverage.
334    /// A write whose slot or timestamp a loader reads must have its own entry
335    /// at or below `covers`. A write named only as an identity must have
336    /// landed: section 0 holds each device's last write through coverage,
337    /// but not the entries no check reads (coven.md §16.7).
338    pub fn validate_coverage<S: WriteSlots>(&self, covers: u64, slots: &S) -> Result<(), S::Error>
339    where
340        S::Error: From<Error>,
341    {
342        let (timed, identities) = self.named_writes();
343        for write in timed {
344            let covered = slots.slot(write)?.is_some_and(|slot| slot.get() <= covers);
345            require(covered, "snapshot write coverage", Rule::Coverage)?;
346        }
347        for write in identities {
348            require(slots.landed(write)?, "snapshot write coverage", Rule::Coverage)?;
349        }
350        Ok(())
351    }
352
353    /// The writes this record names, each of which the snapshot must cover:
354    /// those whose timestamp or slot a loader and later merges read, and
355    /// those it names only as identities.
356    fn named_writes(&self) -> (Vec<WriteId>, Vec<WriteId>) {
357        match self {
358            Self::Column(_)
359            | Self::Write(_)
360            | Self::Admin(_)
361            | Self::KeyCopy(_)
362            | Self::Announcement(_)
363            | Self::Claim(_) => (Vec::new(), Vec::new()),
364            // An active lost value follows merge, which reads its setter's
365            // and replacer's timestamps; a frozen one keeps them as names.
366            Self::Loss(v) if v.frozen => (Vec::new(), v.named_writes()),
367            Self::Loss(v) => (v.named_writes(), Vec::new()),
368            Self::Merge(v) => (
369                v.state
370                    .generations()
371                    .values()
372                    .copied()
373                    .chain(v.state.cells().values().map(|cell| cell.write))
374                    .collect(),
375                Vec::new(),
376            ),
377            Self::Creation(v) => (Vec::new(), vec![v.write, v.creator.write]),
378            Self::Evidence(v) => (Vec::new(), vec![v.write]),
379            Self::Authored(AuthoredRecord::Part(v)) => (Vec::new(), vec![v.header.position]),
380            Self::Authored(AuthoredRecord::Change(v)) => {
381                let verdict = match &v.verdict {
382                    ChangeVerdict::Dependent(sources) | ChangeVerdict::Derived(sources) => {
383                        sources.as_slice()
384                    }
385                    _ => &[],
386                };
387                let creator = match &v.record {
388                    KeptRecord::Change(change) => change.creator.as_ref(),
389                    KeptRecord::Execution { .. } => None,
390                };
391                (
392                    Vec::new(),
393                    creator.into_iter().chain(verdict).map(|source| source.write).collect(),
394                )
395            }
396        }
397    }
398}
399
400#[derive(PartialEq, Eq, PartialOrd, Ord)]
401enum RecordKey {
402    Row(RowId),
403    Write(WriteId),
404    Column(SyncedColumn),
405    Loss((RowId, u64, LossIdentity, bool, u8, Option<WriteId>)),
406    /// A part by its write, then each of its changes by row.
407    Authored(WriteId, Option<RowId>),
408    Slot(Slot),
409    Device(coven_foundation::id_source::DeviceId),
410}
411struct SnapshotStreamState {
412    header: SnapshotHeader,
413    remaining: [u64; SECTIONS],
414    previous: Option<(usize, RecordKey)>,
415    /// Section 4's latest part, and whether a change of it has followed.
416    part: Option<(crate::write::WriteHeader, bool)>,
417    ended: bool,
418}
419impl SnapshotStreamState {
420    fn new(header: SnapshotHeader) -> Self {
421        Self { remaining: header.counts, header, previous: None, part: None, ended: false }
422    }
423    /// Every part in section 4 is followed by at least one of its changes.
424    fn close_part(&self) -> Result<(), Error> {
425        require(
426            self.part.as_ref().is_none_or(|(_, changed)| *changed),
427            "authored part without changes",
428            Rule::SnapshotSequence,
429        )
430    }
431    fn accept(&mut self, record: &SnapshotRecord) -> Result<(), Error> {
432        record.check_header(&self.header)?;
433        let section = record.section();
434        require(
435            !self.ended && self.remaining.iter().position(|n| *n > 0) == Some(section),
436            "snapshot section/count",
437            Rule::SnapshotSequence,
438        )?;
439        if let SnapshotRecord::Authored(AuthoredRecord::Change(change)) = record {
440            if let Some((header, _)) = &self.part {
441                change.validate_in(header)?;
442            }
443        }
444        if matches!(record, SnapshotRecord::Authored(AuthoredRecord::Part(_)))
445            || section != section::AUTHORED
446        {
447            self.close_part()?;
448        }
449        let key = record.key(self.part.as_ref().map(|(header, _)| header))?;
450        if let Some((previous_section, previous_key)) = &self.previous {
451            require(
452                *previous_section != section || *previous_key < key,
453                "snapshot record order",
454                Rule::Order,
455            )?;
456        }
457        self.remaining[section] -= 1;
458        self.previous = Some((section, key));
459        match record {
460            SnapshotRecord::Authored(AuthoredRecord::Part(part)) => {
461                self.part = Some((part.header.clone(), false));
462            }
463            SnapshotRecord::Authored(AuthoredRecord::Change(_)) => {
464                if let Some((_, changed)) = &mut self.part {
465                    *changed = true;
466                }
467            }
468            _ => self.part = None,
469        }
470        Ok(())
471    }
472    fn end(&mut self) -> Result<(), Error> {
473        self.close_part()?;
474        require(
475            !self.ended && self.remaining == [0; SECTIONS],
476            "snapshot end",
477            Rule::SnapshotSequence,
478        )?;
479        self.ended = true;
480        Ok(())
481    }
482}
483
484/// Encodes bounded frames, retaining only the header, counts and previous identity.
485pub struct SnapshotEncoder(SnapshotStreamState);
486impl SnapshotEncoder {
487    /// Metadata used to construct the sealed prefix before emitting frames.
488    pub fn header(&self) -> &SnapshotHeader {
489        &self.0.header
490    }
491    /// Start a snapshot and return its header frame.
492    pub fn start(header: SnapshotHeader) -> Result<(Self, Vec<u8>), Error> {
493        header.validate()?;
494        let bytes = encode_frame_with(kinds::SNAPSHOT_HEADER, |out| {
495            header.id.put(out)?;
496            header.schema_version.put(out)?;
497            header.counts.put(out)
498        })?;
499        Ok((Self(SnapshotStreamState::new(header)), bytes))
500    }
501    /// Validate and encode a record. Failure leaves the cursor unchanged.
502    pub fn record(&mut self, record: SnapshotRecord) -> Result<Vec<u8>, Error> {
503        record.validate()?;
504        let bytes = encode_frame_with(kinds::SNAPSHOT_RECORD, |out| record.put(out))?;
505        self.0.accept(&record)?;
506        Ok(bytes)
507    }
508    /// Emit the required end marker after all declared records.
509    pub fn finish(&mut self) -> Result<Vec<u8>, Error> {
510        let bytes = encode_frame_with(kinds::SNAPSHOT_END, |_| Ok(()))?;
511        self.0.end()?;
512        Ok(bytes)
513    }
514}
515
516/// Decodes a snapshot without retaining its rows or write history. The caller
517/// stages records and commits only after `finish` succeeds. Its oracle must
518/// describe the causally closed applied writes recorded by this snapshot.
519/// Populate that oracle from section 1. Loss records add no oracle metadata;
520/// the consumer validates active losses against merge rows. Only the header, counts and
521/// previous record identities are retained between frames. The database consumer
522/// checks cross-section completeness, SQL schema rules and app-row visibility;
523/// merge's removal computation checks which removal rules actually hold.
524pub struct SnapshotDecoder(SnapshotStreamState);
525impl SnapshotDecoder {
526    /// Extract original slots and claim settlement evidence before the store's
527    /// historical signing keys are known. Other sections are counted and skipped;
528    /// this is not row validation. A loader must subsequently decode the entire
529    /// authenticated snapshot with `frame` before installing its data.
530    pub fn history_frame(&mut self, bytes: &[u8]) -> Result<Option<SnapshotRecord>, Error> {
531        let (kind, mut input) = decode_frame(bytes)?;
532        if kind == kinds::SNAPSHOT_END {
533            input.finish()?;
534            self.0.end()?;
535            return Ok(None);
536        }
537        require(kind == kinds::SNAPSHOT_RECORD, "snapshot record kind", Rule::Kind)?;
538        let tag = u8::get(&mut input)?;
539        let at = usize::from(tag);
540        if at < section::ADMIN {
541            require(
542                !self.0.ended && self.0.remaining.iter().position(|n| *n > 0) == Some(at),
543                "snapshot section/count",
544                Rule::SnapshotSequence,
545            )?;
546            self.0.remaining[at] -= 1;
547            return Ok(None);
548        }
549        let record = match at {
550            section::ADMIN => SnapshotRecord::Admin(Wire::get(&mut input)?),
551            section::KEY_COPY => SnapshotRecord::KeyCopy(Wire::get(&mut input)?),
552            section::ANNOUNCEMENT => SnapshotRecord::Announcement(Wire::get(&mut input)?),
553            section::CLAIM => SnapshotRecord::Claim(Wire::get(&mut input)?),
554            _ => return Err(Error::UnknownTag { field: "snapshot section", tag }),
555        };
556        input.finish()?;
557        record.validate()?;
558        self.0.accept(&record)?;
559        Ok(Some(record))
560    }
561
562    /// Read the initial header frame using coverage from the authenticated sealed prefix.
563    pub fn start(
564        frame: &[u8],
565        prefix: &crate::sealed_snapshot::SnapshotObjectPrefix,
566    ) -> Result<Self, Error> {
567        let mut input = decode_frame_of(frame, kinds::SNAPSHOT_HEADER, "snapshot header kind")?;
568        let header = SnapshotHeader {
569            id: Wire::get(&mut input)?,
570            schema_version: Wire::get(&mut input)?,
571            counts: Wire::get(&mut input)?,
572            covers: prefix.covers,
573        };
574        require(header.id.audience == prefix.audience, "snapshot prefix audience", Rule::Audience)?;
575        require(
576            match prefix.purpose {
577                crate::sequence::Purpose::Raise(version) => version == header.schema_version,
578                _ => true,
579            },
580            "snapshot raise version",
581            Rule::RaiseVersion,
582        )?;
583        input.finish()?;
584        header.validate()?;
585        Ok(Self(SnapshotStreamState::new(header)))
586    }
587    /// Identity, schema and coverage of this snapshot.
588    pub fn header(&self) -> &SnapshotHeader {
589        &self.0.header
590    }
591    /// Decode a record or end marker. Merge checks row invariants using the
592    /// supplied oracle. Every error leaves counts and order unchanged.
593    pub fn frame<O: WriteSlots>(
594        &mut self,
595        bytes: &[u8],
596        oracle: &O,
597    ) -> Result<Option<SnapshotRecord>, O::Error>
598    where
599        O::Error: From<Error>,
600    {
601        let (kind, mut input) = decode_frame(bytes)?;
602        if kind == kinds::SNAPSHOT_END {
603            input.finish()?;
604            self.0.end()?;
605            return Ok(None);
606        }
607        require(kind == kinds::SNAPSHOT_RECORD, "snapshot record kind", Rule::Kind)?;
608        let record = SnapshotRecord::get(&mut input, oracle)?;
609        input.finish()?;
610        record.validate()?;
611        record.validate_coverage(self.0.header.covers, oracle)?;
612        self.0.accept(&record)?;
613        Ok(Some(record))
614    }
615    /// Refuse truncation, including EOF between records without an end marker.
616    pub fn finish(&self) -> Result<(), Error> {
617        if self.0.ended {
618            Ok(())
619        } else {
620            Err(Error::Truncated)
621        }
622    }
623}
624
625#[cfg(test)]
626#[path = "snapshot_tests.rs"]
627mod tests;
coven_format/snapshot.rs

coven_format/
snapshot.rs
