coven_merge/
state.rs

1use crate::{ColumnValue, MergeError, Operation, RowId, Timestamp, Write, WriteId, WriteOracle};
2use std::collections::BTreeMap;
3
4/// The winning write and its value in one column (`_coven_cells`).
5#[derive(Clone, Debug, PartialEq, Eq)]
6pub struct Cell<V> {
7    /// The write whose timestamp won.
8    pub write: WriteId,
9    /// The value and parent generations that write set.
10    pub value: ColumnValue<V>,
11}
12
13/// A value is identified by the column and write that set it.
14#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
15pub struct LostKey {
16    /// The synced column's name.
17    pub column: String,
18    /// The write that set the lost value.
19    pub write: WriteId,
20}
21
22/// A value replaced only by writes that had not read it (`_coven_lost`).
23#[derive(Clone, Debug, PartialEq, Eq)]
24pub struct LostValue<V> {
25    /// The incarnation in which the value was set.
26    pub incarnation: u64,
27    /// The value, with its original reference metadata.
28    pub value: ColumnValue<V>,
29    /// The earliest delete, or otherwise the current winning setter.
30    pub replaced_by: WriteId,
31}
32
33/// Whether a row at `generation` is present: a row starts absent at 0 and
34/// each insert or delete advances it, so an odd generation is an
35/// incarnation (§8.3).
36pub const fn generation_present(generation: u64) -> bool {
37    !generation.is_multiple_of(2)
38}
39
40/// One table, key and audience's merged state. Removed rows retain this state;
41/// a removal never advances their generation (§8.4).
42#[derive(Clone, Debug, PartialEq, Eq)]
43pub struct RowState<V> {
44    pub(crate) row: RowId,
45    pub(crate) generation: u64,
46    pub(crate) generations: BTreeMap<u64, WriteId>,
47    pub(crate) cells: BTreeMap<String, Cell<V>>,
48    pub(crate) lost: BTreeMap<LostKey, LostValue<V>>,
49}
50
51impl<V> RowState<V> {
52    /// A row that has never been inserted, at generation zero.
53    pub fn new(row: RowId) -> Self {
54        Self {
55            row,
56            generation: 0,
57            generations: BTreeMap::new(),
58            cells: BTreeMap::new(),
59            lost: BTreeMap::new(),
60        }
61    }
62    /// Load the compact records kept by the database or a snapshot. Validate
63    /// their generation sequence, cell presence, reference metadata and lost
64    /// replacers without replaying write bodies. As with `apply`, the oracle
65    /// is the causally closed applied set; it supplies timestamps and past.
66    pub fn from_parts<O: WriteOracle>(
67        row: RowId,
68        generations: BTreeMap<u64, WriteId>,
69        cells: BTreeMap<String, Cell<V>>,
70        lost: BTreeMap<LostKey, LostValue<V>>,
71        oracle: &O,
72    ) -> Result<Self, O::Error> {
73        let mut generation: u64 = 0;
74        let mut previous_stamp = None;
75        for (g, write) in &generations {
76            let expected = generation.checked_add(1).ok_or(MergeError::GenerationExhausted)?;
77            if *g != expected {
78                return Err(MergeError::GenerationGap(expected).into());
79            }
80            let stamp = timestamp(oracle, *write)?;
81            if previous_stamp.is_some_and(|previous| previous >= stamp) {
82                return Err(MergeError::GenerationTimestamp(*g).into());
83            }
84            previous_stamp = Some(stamp);
85            generation = *g;
86        }
87        if !generation_present(generation) && !cells.is_empty() {
88            return Err(MergeError::DeletedRowHasCells.into());
89        }
90        for (column, cell) in &cells {
91            let stamp = timestamp(oracle, cell.write)?;
92            if previous_stamp.is_some_and(|start| stamp < start) {
93                return Err(MergeError::CellBeforeIncarnation(column.clone()).into());
94            }
95            for parent in cell.value.parents.values() {
96                parent.validate_written(&row)?;
97            }
98        }
99        let state = Self { row, generation, generations, cells, lost };
100        for (key, value) in &state.lost {
101            state.validate_loss(key, value, oracle)?;
102        }
103        Ok(state)
104    }
105
106    /// Validate an independently decoded cell loss against this row's validated
107    /// generations and winning cells. The oracle is the causally closed applied
108    /// set, as for `from_parts`; existing losses do not affect this check.
109    pub fn validate_loss<O: WriteOracle>(
110        &self,
111        key: &LostKey,
112        value: &LostValue<V>,
113        oracle: &O,
114    ) -> Result<(), O::Error> {
115        let invalid = || MergeError::InvalidLostValue(key.clone());
116        let inc = value.incarnation;
117        if !generation_present(inc) || inc > self.generation {
118            return Err(invalid().into());
119        }
120        let start = self.generations.get(&inc).ok_or_else(invalid)?;
121        let setter_stamp = timestamp(oracle, key.write)?;
122        if setter_stamp < timestamp(oracle, *start)? {
123            return Err(invalid().into());
124        }
125        let canonical = if inc < self.generation {
126            let delete = inc.checked_add(1).ok_or(MergeError::GenerationExhausted)?;
127            *self.generations.get(&delete).ok_or_else(invalid)?
128        } else {
129            let cell = self.cells.get(&key.column).ok_or_else(invalid)?;
130            if timestamp(oracle, cell.write)? <= setter_stamp {
131                return Err(invalid().into());
132            }
133            cell.write
134        };
135        if value.replaced_by != canonical || oracle.had_read(canonical, key.write)? {
136            return Err(invalid().into());
137        }
138        for parent in value.value.parents.values() {
139            parent.validate_written(&self.row)?;
140        }
141        Ok(())
142    }
143
144    /// The table, key and audience this state belongs to.
145    pub fn row(&self) -> &RowId {
146        &self.row
147    }
148    /// How many times the row has been created, deleted or re-added.
149    pub fn generation(&self) -> u64 {
150        self.generation
151    }
152    /// Whether the row is present in merged state, before removal rules.
153    pub fn present(&self) -> bool {
154        generation_present(self.generation)
155    }
156    /// The smallest-stamped write that moved the row to each generation.
157    pub fn generations(&self) -> &BTreeMap<u64, WriteId> {
158        &self.generations
159    }
160    /// The winning write and value of every set column.
161    pub fn cells(&self) -> &BTreeMap<String, Cell<V>> {
162        &self.cells
163    }
164    /// Every lost value, its incarnation and what replaced it.
165    pub fn lost(&self) -> &BTreeMap<LostKey, LostValue<V>> {
166        &self.lost
167    }
168
169    /// Carry a row through a schema change. The caller supplies the surviving
170    /// identity and maps each column value, including its reference identities.
171    /// Dropped columns disappear; generations, setters and replacers stay fixed.
172    pub fn map_schema(
173        mut self,
174        row: RowId,
175        mut column: impl FnMut(String, ColumnValue<V>) -> Option<(String, ColumnValue<V>)>,
176    ) -> Self {
177        self.row = row;
178        self.cells = self
179            .cells
180            .into_iter()
181            .filter_map(|(name, cell)| {
182                let (name, value) = column(name, cell.value)?;
183                Some((name, Cell { write: cell.write, value }))
184            })
185            .collect();
186        self.lost = self
187            .lost
188            .into_iter()
189            .filter_map(|(key, lost)| {
190                let (name, value) = column(key.column, lost.value)?;
191                Some((LostKey { column: name, write: key.write }, LostValue { value, ..lost }))
192            })
193            .collect();
194        self
195    }
196
197    /// §9.2: a unique claim is ordered by the latest winning setter of any of
198    /// its columns. Null reference substitutions keep that setter.
199    #[cfg(test)]
200    pub(crate) fn claim_order(
201        &self,
202        columns: &[String],
203        order: impl Fn(WriteId) -> Result<crate::ClaimOrder, MergeError>,
204    ) -> Result<crate::ClaimOrder, MergeError> {
205        let mut latest = None;
206        for column in columns {
207            let cell = self
208                .cells
209                .get(column)
210                .ok_or_else(|| MergeError::MissingClaimColumn(column.clone()))?;
211            let stamp = order(cell.write)?;
212            latest = Some(match latest {
213                Some(old) => std::cmp::max(old, stamp),
214                None => stamp,
215            });
216        }
217        latest.ok_or(MergeError::EmptyClaim)
218    }
219}
220
221/// A change to the row's lost-value records, to commit with its merged state.
222#[derive(Clone, Debug, PartialEq, Eq)]
223pub enum LostChange<V> {
224    /// Create or replace a lost-value record.
225    Put(LostKey, LostValue<V>),
226    /// A replacing write had read the value, so it is no longer lost.
227    Remove(LostKey),
228}
229
230/// The complete new row state and the changes to its lost-value records.
231#[derive(Clone, Debug, PartialEq, Eq)]
232pub struct RowUpdate<V> {
233    /// Commit this state with the write's other row changes.
234    pub state: RowState<V>,
235    /// Changes to `_coven_lost` for this row only.
236    pub lost_changes: Vec<LostChange<V>>,
237}
238
239pub(crate) fn timestamp<O: WriteOracle>(oracle: &O, id: WriteId) -> Result<Timestamp, O::Error> {
240    oracle.timestamp(id)?.ok_or_else(|| MergeError::MissingWrite(id).into())
241}
242
243/// Apply one write to one row (proofs/merge.md B5; Lean `step`). It touches only
244/// the supplied row. All past writes must be applied and the arriving write
245/// must be absent from the oracle. Call for each changed row against the same
246/// pre-write oracle, then commit all results in the database's transaction.
247/// Parity, future generations, missing metadata and overflow are checked here.
248/// A nonzero generation needs a read at least as recent as its earliest
249/// recorded changer. That is the locally checkable part of assumption 3:
250/// the canonical record cannot tell whether another read write also reached
251/// the generation. The authoring database establishes the full witness when
252/// producing a change; test histories validate it explicitly. The input
253/// state is unchanged on every error path.
254pub fn apply<V: Clone + Eq, P: crate::WritePast, O: WriteOracle>(
255    state: &RowState<V>,
256    write: &Write<V, P>,
257    oracle: &O,
258) -> Result<RowUpdate<V>, O::Error> {
259    write.validate_metadata(oracle)?;
260    let change = write
261        .changes
262        .get(&state.row)
263        .ok_or_else(|| MergeError::MissingChange(state.row.clone()))?;
264    change.validate(&state.row)?;
265    if change.generation > state.generation {
266        return Err(MergeError::GenerationAhead { row: state.row.clone() }.into());
267    }
268    if change.generation != 0 {
269        let first = state.generations.get(&change.generation).ok_or_else(|| {
270            MergeError::GenerationNotSeen { row: state.row.clone(), generation: change.generation }
271        })?;
272        let earliest = timestamp(oracle, *first)?;
273        let mut possible_witness = false;
274        for past in write.had_read.frontier() {
275            possible_witness |= timestamp(oracle, *past)? >= earliest;
276        }
277        if !possible_witness {
278            return Err(MergeError::GenerationNotSeen {
279                row: state.row.clone(),
280                generation: change.generation,
281            }
282            .into());
283        }
284    }
285    let incarnation = change.incarnation()?;
286    let columns = change.operation.columns();
287    let deletes = matches!(change.operation, Operation::Delete);
288    let mut next = state.clone();
289    // Existing lost values are updated from the OLD row, exactly as lostStep.
290    for (key, old) in &state.lost {
291        let sets = columns.is_some_and(|c| c.contains_key(&key.column));
292        let replaces = (deletes && change.generation == old.incarnation)
293            || (sets
294                && incarnation == old.incarnation
295                && timestamp(oracle, key.write)? < write.timestamp);
296        if replaces {
297            if write.had_read.contains(&key.write) {
298                next.lost.remove(key);
299            } else {
300                let replacement = if deletes {
301                    if old.incarnation < state.generation
302                        && timestamp(oracle, old.replaced_by)? < write.timestamp
303                    {
304                        old.replaced_by
305                    } else {
306                        write.id
307                    }
308                } else if old.incarnation < state.generation
309                    || timestamp(oracle, old.replaced_by)? > write.timestamp
310                {
311                    old.replaced_by
312                } else {
313                    write.id
314                };
315                next.lost
316                    .insert(key.clone(), LostValue { replaced_by: replacement, ..old.clone() });
317            }
318        }
319    }
320    // Current values become lost only when the arriving replacer hadn't read them.
321    for (column, current) in &state.cells {
322        let replaces = (deletes && change.generation == state.generation)
323            || (incarnation == state.generation
324                && columns.is_some_and(|c| c.contains_key(column))
325                && timestamp(oracle, current.write)? < write.timestamp);
326        if replaces && !write.had_read.contains(&current.write) {
327            next.lost.insert(
328                LostKey { column: column.clone(), write: current.write },
329                LostValue {
330                    incarnation: state.generation,
331                    value: current.value.clone(),
332                    replaced_by: write.id,
333                },
334            );
335        }
336    }
337    // An arriving value loses to an already-applied delete or later setter.
338    if let Some(columns) = columns {
339        for (column, value) in columns {
340            let replacement = if incarnation < state.generation {
341                let deleted_generation =
342                    incarnation.checked_add(1).ok_or(MergeError::GenerationExhausted)?;
343                Some(*state.generations.get(&deleted_generation).ok_or_else(|| {
344                    MergeError::GenerationNotSeen {
345                        row: state.row.clone(),
346                        generation: deleted_generation,
347                    }
348                })?)
349            } else if incarnation == state.generation {
350                match state.cells.get(column) {
351                    Some(current) if timestamp(oracle, current.write)? > write.timestamp => {
352                        Some(current.write)
353                    }
354                    _ => None,
355                }
356            } else {
357                None
358            };
359            if let Some(replaced_by) = replacement {
360                next.lost.insert(
361                    LostKey { column: column.clone(), write: write.id },
362                    LostValue { incarnation, value: value.clone(), replaced_by },
363                );
364            }
365        }
366    }
367    if change.operation.advances() {
368        let generation = change.after()?;
369        let winner = match state.generations.get(&generation) {
370            Some(old) if timestamp(oracle, *old)? < write.timestamp => *old,
371            _ => write.id,
372        };
373        next.generations.insert(generation, winner);
374        if state.generation < generation {
375            next.generation = generation;
376            next.cells.clear();
377            if let Operation::Insert(columns) = &change.operation {
378                for (column, value) in columns {
379                    next.cells
380                        .insert(column.clone(), Cell { write: write.id, value: value.clone() });
381                }
382            }
383        }
384    }
385    if incarnation == state.generation {
386        if let Some(columns) = columns {
387            for (column, value) in columns {
388                let wins = match state.cells.get(column) {
389                    Some(old) => timestamp(oracle, old.write)? < write.timestamp,
390                    None => true,
391                };
392                if wins {
393                    next.cells
394                        .insert(column.clone(), Cell { write: write.id, value: value.clone() });
395                }
396            }
397        }
398    }
399    let mut lost_changes = Vec::new();
400    for key in state.lost.keys() {
401        if !next.lost.contains_key(key) {
402            lost_changes.push(LostChange::Remove(key.clone()));
403        }
404    }
405    for (key, value) in &next.lost {
406        if state.lost.get(key) != Some(value) {
407            lost_changes.push(LostChange::Put(key.clone(), value.clone()));
408        }
409    }
410    Ok(RowUpdate { state: next, lost_changes })
411}
412
413#[cfg(test)]
414#[path = "state_tests.rs"]
415mod tests;
coven_merge/state.rs

coven_merge/
state.rs
