coven_format/
snapshot_rows.rs

1//! Snapshot merge rows carry app values and generations; losses travel separately.
2
3use crate::error::{require, Error, Rule as FormatRule};
4use crate::value::{positive, Slot, Value};
5use crate::wire::{wire_struct, Decoder, Encoder, Wire};
6use coven_merge::{
7    Change, ColumnValue, Operation, RowId, RowState, Timestamp, WriteId, WriteOracle,
8};
9use std::collections::BTreeMap;
10
11/// Applied write metadata needed to reconstruct merge's state without write bodies.
12#[derive(Clone, Debug, PartialEq, Eq)]
13pub struct AppliedWrite {
14    /// The applied write.
15    pub id: WriteId,
16    /// Slot containing this published write.
17    pub slot: Slot,
18    /// Its timestamp.
19    pub timestamp: Timestamp,
20    /// Last applied slot when authored; own earlier writes are implicit.
21    pub read_through: u64,
22}
23wire_struct!(AppliedWrite, id, slot, timestamp, read_through);
24impl AppliedWrite {
25    pub(crate) fn validate(&self) -> Result<(), Error> {
26        positive(self.id.number)?;
27        require(
28            self.timestamp.device() == self.id.device,
29            "applied write timestamp",
30            FormatRule::TimestampDevice,
31        )?;
32        require(
33            self.read_through < self.slot.get(),
34            "applied write read-through",
35            FormatRule::Coverage,
36        )
37    }
38}
39
40/// A synced column, named independently of device-local numeric ids.
41#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
42pub struct SyncedColumn {
43    /// The table.
44    pub table: String,
45    /// The column.
46    pub column: String,
47}
48wire_struct!(SyncedColumn, table => crate::wire::get_name, column => crate::wire::get_name);
49
50/// A row's merge state, retained even while removal rules hide it from the app.
51#[derive(Clone, Debug, PartialEq, Eq)]
52pub struct MergeRow {
53    /// Generations and winning cells only; losses occupy the loss section.
54    pub state: RowState<Value>,
55}
56impl MergeRow {
57    pub(crate) fn put(&self, out: &mut Encoder) -> Result<(), Error> {
58        self.state.row().put(out)?;
59        self.state.generations().put(out)?;
60        self.state.cells().put(out)
61    }
62    pub(crate) fn get<O: WriteOracle>(input: &mut Decoder<'_>, oracle: &O) -> Result<Self, O::Error>
63    where
64        O::Error: From<Error>,
65    {
66        let state = RowState::from_parts(
67            Wire::get(input)?,
68            Wire::get(input)?,
69            crate::wire::get_name_map(input)?,
70            BTreeMap::new(),
71            oracle,
72        )?;
73        Ok(Self { state })
74    }
75    pub(crate) fn validate(&self) -> Result<(), Error> {
76        crate::merge_wire::state(&self.state)?;
77        require(
78            self.state.lost().is_empty(),
79            "merge losses belong in the loss section",
80            FormatRule::Required,
81        )?;
82        Ok(())
83    }
84}
85
86/// How a device interpreted one original row change (§9.5, §14.1, §18.6, §19.4).
87/// It decides whether the change survives as a source for later changes,
88/// independently of dismissible losses and consumed slots.
89#[derive(Clone, Debug, PartialEq, Eq)]
90pub enum ChangeVerdict {
91    /// Never interpreted here: a loaded snapshot already carries its effects.
92    Unjudged,
93    /// It entered the merge.
94    Applied,
95    /// A breaking change to this schema version excluded its part (§18.6).
96    Excluded(u32),
97    /// An earlier removal or credential replacement cut off this write.
98    Refused,
99    /// It used data that does not survive: these sources, in role order.
100    Dependent(Vec<crate::write_sources::ChangeId>),
101    /// A kept reset ignored its part: it had not read that entry (§19.4).
102    Ignored(Slot),
103    /// It lacked permission: it directly changed another member's owned row,
104    /// or was not the exact foreign-key effect of authorized changes (§9.6).
105    NotAuthorized(crate::loss::AuthorizationFailure),
106    /// It entered the merge as the exact foreign-key effect of these changes
107    /// in its own write: each action's root and intermediate causes, in
108    /// execution order (§9.6).
109    Derived(Vec<crate::write_sources::ChangeId>),
110}
111impl ChangeVerdict {
112    pub(crate) fn validate(&self) -> Result<(), Error> {
113        match self {
114            Self::Excluded(version) => {
115                require(*version > 0, "breaking schema version", FormatRule::Required)
116            }
117            Self::Ignored(_) => Ok(()),
118            Self::Dependent(sources) | Self::Derived(sources) => {
119                require(!sources.is_empty(), "verdict sources", FormatRule::Required)?;
120                sources.iter().try_for_each(|s| positive(s.write.number))
121            }
122            Self::Unjudged | Self::Applied | Self::Refused | Self::NotAuthorized(_) => Ok(()),
123        }
124    }
125
126    /// Whether the change survives as data for later changes.
127    pub fn retained(&self) -> bool {
128        matches!(self, Self::Unjudged | Self::Applied | Self::Derived(_))
129    }
130}
131impl Wire for ChangeVerdict {
132    fn put(&self, out: &mut Encoder) -> Result<(), Error> {
133        match self {
134            Self::Unjudged => 0u8.put(out),
135            Self::Applied => 1u8.put(out),
136            Self::Excluded(version) => {
137                2u8.put(out)?;
138                version.put(out)
139            }
140            Self::Refused => 3u8.put(out),
141            Self::Dependent(sources) => {
142                4u8.put(out)?;
143                sources.put(out)
144            }
145            Self::Ignored(entry) => {
146                5u8.put(out)?;
147                entry.put(out)
148            }
149            Self::NotAuthorized(reason) => {
150                6u8.put(out)?;
151                reason.put(out)
152            }
153            Self::Derived(causes) => {
154                7u8.put(out)?;
155                causes.put(out)
156            }
157        }
158    }
159    fn get(input: &mut Decoder<'_>) -> Result<Self, Error> {
160        match u8::get(input)? {
161            0 => Ok(Self::Unjudged),
162            1 => Ok(Self::Applied),
163            2 => Ok(Self::Excluded(Wire::get(input)?)),
164            3 => Ok(Self::Refused),
165            4 => Ok(Self::Dependent(Wire::get(input)?)),
166            5 => Ok(Self::Ignored(Wire::get(input)?)),
167            6 => Ok(Self::NotAuthorized(Wire::get(input)?)),
168            7 => Ok(Self::Derived(Wire::get(input)?)),
169            tag => Err(Error::UnknownTag { field: "change verdict", tag }),
170        }
171    }
172}
173
174/// A value an accepted change wrote, as its kept record holds it (D7.3): in
175/// full when the column was a reference under the change's schema, since a
176/// later write's foreign-key actions read it and the parents it named; else
177/// only its digest, which tells whether a later delete shows the same value.
178#[derive(Clone, Debug, PartialEq, Eq)]
179pub enum KeptValue {
180    /// The value itself.
181    Value(Value),
182    /// [`coven_crypto::kept_value_digest`] of the value's encoding.
183    Digest([u8; 32]),
184}
185impl KeptValue {
186    /// The digest a kept record holds for `value`.
187    pub fn digest(value: &Value) -> Result<[u8; 32], Error> {
188        let mut out = Encoder::new();
189        value.put(&mut out)?;
190        Ok(coven_crypto::kept_value_digest(&out.bytes))
191    }
192
193    /// Whether `value` is the value this one keeps.
194    pub fn matches(&self, value: &Value) -> Result<bool, Error> {
195        Ok(match self {
196            Self::Value(kept) => kept == value,
197            Self::Digest(digest) => Self::digest(value)? == *digest,
198        })
199    }
200}
201
202/// A column a kept change set: its value and parents in full, or a digest,
203/// which names no parents.
204impl Wire for ColumnValue<KeptValue> {
205    fn put(&self, out: &mut Encoder) -> Result<(), Error> {
206        match &self.value {
207            KeptValue::Value(value) => {
208                0u8.put(out)?;
209                value.put(out)?;
210                self.parents.put(out)
211            }
212            KeptValue::Digest(digest) => {
213                1u8.put(out)?;
214                digest.put(out)
215            }
216        }
217    }
218    fn get(input: &mut Decoder<'_>) -> Result<Self, Error> {
219        match u8::get(input)? {
220            0 => {
221                Ok(Self { value: KeptValue::Value(Wire::get(input)?), parents: Wire::get(input)? })
222            }
223            1 => Ok(Self { value: KeptValue::Digest(Wire::get(input)?), parents: BTreeMap::new() }),
224            tag => Err(Error::UnknownTag { field: "kept value", tag }),
225        }
226    }
227}
228
229/// One row's record of an accepted write part, as a device keeps it once the
230/// write is judged (D7.3): what later writes are judged against and what
231/// reopens the part's leaves, without the values it replaced, the sources it
232/// built on or its SQL operations.
233#[derive(Clone, Debug, PartialEq, Eq)]
234pub enum KeptRecord {
235    /// A net change.
236    Change(KeptChange),
237    /// Operations with no net change: only their count, the leaves they take.
238    Execution {
239        /// The row.
240        row: RowId,
241        /// How many operations the record listed, at least one.
242        steps: u32,
243    },
244}
245
246/// A net change's kept record.
247#[derive(Clone, Debug, PartialEq, Eq)]
248pub struct KeptChange {
249    /// Table, encoded primary key and audience.
250    pub row: RowId,
251    /// The generation before the change, and the columns it set: each in full
252    /// when it was a reference under the write's schema, else its digest.
253    pub change: Change<KeptValue>,
254    /// For an audience copy, the original creation it copies (D5.7).
255    pub creator: Option<crate::write_sources::ChangeId>,
256    /// How many operations the record listed.
257    pub steps: u32,
258}
259
260impl Wire for Change<KeptValue> {
261    fn put(&self, out: &mut Encoder) -> Result<(), Error> {
262        self.generation.put(out)?;
263        match &self.operation {
264            Operation::Insert(columns) => {
265                0u8.put(out)?;
266                columns.put(out)
267            }
268            Operation::Update(columns) => {
269                1u8.put(out)?;
270                columns.put(out)
271            }
272            Operation::Delete => 2u8.put(out),
273        }
274    }
275    fn get(input: &mut Decoder<'_>) -> Result<Self, Error> {
276        let generation = Wire::get(input)?;
277        let operation = match u8::get(input)? {
278            0 => Operation::Insert(crate::wire::get_name_map(input)?),
279            1 => Operation::Update(crate::wire::get_name_map(input)?),
280            2 => Operation::Delete,
281            tag => return Err(Error::UnknownTag { field: "row operation", tag }),
282        };
283        Ok(Self { generation, operation })
284    }
285}
286
287impl Wire for KeptRecord {
288    fn put(&self, out: &mut Encoder) -> Result<(), Error> {
289        match self {
290            Self::Change(change) => {
291                0u8.put(out)?;
292                change.row.put(out)?;
293                change.change.put(out)?;
294                change.creator.put(out)?;
295                change.steps.put(out)
296            }
297            Self::Execution { row, steps } => {
298                1u8.put(out)?;
299                row.put(out)?;
300                steps.put(out)
301            }
302        }
303    }
304    fn get(input: &mut Decoder<'_>) -> Result<Self, Error> {
305        match u8::get(input)? {
306            0 => Ok(Self::Change(KeptChange {
307                row: Wire::get(input)?,
308                change: Wire::get(input)?,
309                creator: Wire::get(input)?,
310                steps: Wire::get(input)?,
311            })),
312            1 => Ok(Self::Execution { row: Wire::get(input)?, steps: Wire::get(input)? }),
313            tag => Err(Error::UnknownTag { field: "kept record", tag }),
314        }
315    }
316}
317
318impl KeptRecord {
319    /// The kept record of `record`, whose `reference` says which of its
320    /// columns were references under its write's schema.
321    pub fn keep(
322        record: crate::write::PartRecord<'_>,
323        reference: impl Fn(&str) -> bool,
324    ) -> Result<Self, Error> {
325        // A record's steps are a bounded collection, so their count fits.
326        let steps = |execution: &crate::write::Execution| {
327            crate::error::bound(execution.steps.len(), crate::wire::MAX_ITEMS, "collection")
328                .map(|()| execution.steps.len() as u32)
329        };
330        Ok(match record {
331            crate::write::PartRecord::Change(change) => {
332                let kept = |columns: &BTreeMap<String, ColumnValue<Value>>| {
333                    columns
334                        .iter()
335                        .map(|(name, cell)| {
336                            let kept = if reference(name) {
337                                ColumnValue {
338                                    value: KeptValue::Value(cell.value.clone()),
339                                    parents: cell.parents.clone(),
340                                }
341                            } else {
342                                ColumnValue {
343                                    value: KeptValue::Digest(KeptValue::digest(&cell.value)?),
344                                    parents: BTreeMap::new(),
345                                }
346                            };
347                            Ok((name.clone(), kept))
348                        })
349                        .collect::<Result<BTreeMap<_, _>, Error>>()
350                };
351                let operation = match &change.change.operation {
352                    Operation::Insert(columns) => Operation::Insert(kept(columns)?),
353                    Operation::Update(columns) => Operation::Update(kept(columns)?),
354                    Operation::Delete => Operation::Delete,
355                };
356                Self::Change(KeptChange {
357                    row: change.row.clone(),
358                    change: Change { generation: change.change.generation, operation },
359                    creator: change.creator.clone(),
360                    steps: steps(&change.execution)?,
361                })
362            }
363            crate::write::PartRecord::Execution(execution) => {
364                Self::Execution { row: execution.row.clone(), steps: steps(&execution.execution)? }
365            }
366        })
367    }
368
369    /// Whether this record keeps in full exactly the columns `reference`
370    /// names, as [`KeptRecord::keep`] would.
371    pub fn conforms(&self, reference: impl Fn(&str) -> bool) -> bool {
372        match self {
373            Self::Change(change) => match &change.change.operation {
374                Operation::Insert(columns) | Operation::Update(columns) => {
375                    columns.iter().all(|(name, cell)| {
376                        matches!(cell.value, KeptValue::Value(_)) == reference(name)
377                    })
378                }
379                Operation::Delete => true,
380            },
381            Self::Execution { .. } => true,
382        }
383    }
384
385    /// The changed row.
386    pub fn row(&self) -> &RowId {
387        match self {
388            Self::Change(change) => &change.row,
389            Self::Execution { row, .. } => row,
390        }
391    }
392
393    /// The record's bytes as a device keeps them.
394    pub fn encode(&self) -> Result<Vec<u8>, Error> {
395        self.validate()?;
396        let mut out = Encoder::new();
397        self.put(&mut out)?;
398        Ok(out.bytes)
399    }
400
401    /// Read back kept bytes, checking them again.
402    pub fn decode(bytes: &[u8]) -> Result<Self, Error> {
403        let mut input = Decoder::new(bytes)?;
404        let record = Self::get(&mut input)?;
405        input.finish()?;
406        record.validate()?;
407        Ok(record)
408    }
409
410    pub(crate) fn validate(&self) -> Result<(), Error> {
411        crate::value::row(self.row())?;
412        match self {
413            Self::Execution { steps, .. } => {
414                require(*steps > 0, "execution steps", FormatRule::Required)
415            }
416            Self::Change(change) => {
417                change.change.validate(&change.row).map_err(Error::Merge)?;
418                if let Operation::Insert(columns) | Operation::Update(columns) =
419                    &change.change.operation
420                {
421                    require(!columns.is_empty(), "changed columns", FormatRule::Required)?;
422                    for (name, cell) in columns {
423                        crate::value::name(name)?;
424                        if let KeptValue::Value(value) = &cell.value {
425                            value.validate()?;
426                        }
427                        crate::merge_wire::parents(&cell.parents)?;
428                    }
429                }
430                if let Some(creator) = &change.creator {
431                    // Only a generation-zero insert in another audience copies a row.
432                    require(
433                        matches!(change.change.operation, Operation::Insert(_))
434                            && change.change.generation == 0
435                            && creator.row.table == change.row.table
436                            && creator.row.key == change.row.key
437                            && creator.row.audience != change.row.audience,
438                        "copied creation",
439                        FormatRule::ColumnOperation,
440                    )?;
441                    positive(creator.write.number)?;
442                }
443                Ok(())
444            }
445        }
446    }
447}
448
449/// One accepted write part's shared details in a snapshot's section 4 (D7.3),
450/// before its changes: the authored header. A part may hold fewer changes
451/// than its write made, the rest forgotten, so a row it doesn't hold proves
452/// nothing about the write.
453#[derive(Clone, Debug, PartialEq, Eq)]
454pub struct AuthoredPart {
455    /// The original write header, before any conversion.
456    pub header: crate::write::WriteHeader,
457}
458wire_struct!(AuthoredPart, header);
459
460/// One kept change of the part before it in section 4, its verdict by the
461/// snapshot's author, and the slot at which it stopped being current (D7.3).
462#[derive(Clone, Debug, PartialEq, Eq)]
463pub struct AuthoredChange {
464    /// The change's kept record.
465    pub record: KeptRecord,
466    /// Its interpretation by the snapshot's author.
467    pub verdict: ChangeVerdict,
468    /// The slot at which it stopped being current, once it has, at or below
469    /// the snapshot's coverage (coven.md §16.7).
470    pub stop: Option<Slot>,
471}
472wire_struct!(AuthoredChange, record, verdict, stop);
473impl AuthoredChange {
474    /// The change's identity in the part `header` heads.
475    pub fn id(&self, header: &crate::write::WriteHeader) -> crate::write_sources::ChangeId {
476        crate::write_sources::ChangeId { write: header.position, row: self.record.row().clone() }
477    }
478
479    pub(crate) fn validate(&self) -> Result<(), Error> {
480        self.record.validate()?;
481        self.verdict.validate()
482    }
483
484    /// The checks that need the part's header: a dependency names this row or
485    /// its own write, and a derivation's causes are its own write's changes.
486    pub(crate) fn validate_in(&self, header: &crate::write::WriteHeader) -> Result<(), Error> {
487        match &self.verdict {
488            ChangeVerdict::Dependent(sources) => {
489                for source in sources {
490                    require(
491                        crate::loss::dependency_source(self.record.row(), header.position, source),
492                        "dependency source row",
493                        FormatRule::Required,
494                    )?;
495                }
496                Ok(())
497            }
498            ChangeVerdict::Derived(causes) => require(
499                causes.iter().all(|cause| cause.write == header.position),
500                "derivation cause write",
501                FormatRule::Required,
502            ),
503            _ => Ok(()),
504        }
505    }
506}
507
508/// A record of section 4: a part's shared details, or one of its changes.
509#[derive(Clone, Debug, PartialEq, Eq)]
510pub enum AuthoredRecord {
511    /// A part, before its changes.
512    Part(AuthoredPart),
513    /// A change of the part before it.
514    Change(AuthoredChange),
515}
516impl Wire for AuthoredRecord {
517    fn put(&self, out: &mut Encoder) -> Result<(), Error> {
518        match self {
519            Self::Part(part) => {
520                0u8.put(out)?;
521                part.put(out)
522            }
523            Self::Change(change) => {
524                1u8.put(out)?;
525                change.put(out)
526            }
527        }
528    }
529    fn get(input: &mut Decoder<'_>) -> Result<Self, Error> {
530        match u8::get(input)? {
531            0 => Ok(Self::Part(Wire::get(input)?)),
532            1 => Ok(Self::Change(Wire::get(input)?)),
533            tag => Err(Error::UnknownTag { field: "authored record", tag }),
534        }
535    }
536}
coven_format/snapshot_rows.rs

coven_format/
snapshot_rows.rs
