coven_merge/
history.rs

1use crate::{MergeError, RowId, Timestamp, Write, WriteId, WriteOracle};
2use std::collections::BTreeMap;
3#[cfg(test)]
4use {
5    crate::{Cell, LostKey, LostValue, Operation, RowState},
6    std::collections::BTreeSet,
7};
8
9/// A finite, causally closed set of writes satisfying the merge proof's four
10/// assumptions. It is also an in-memory oracle for an applied set.
11/// Test support: requires `test-utils` outside this crate's unit tests.
12#[derive(Clone, Debug)]
13pub struct History<V> {
14    writes: BTreeMap<WriteId, Write<V>>,
15}
16
17impl<V> History<V> {
18    /// Validate a set without imposing an arrival order. Check timestamp
19    /// uniqueness, timestamp causality, generation witnesses and parity.
20    pub fn new(writes: impl IntoIterator<Item = Write<V>>) -> Result<Self, MergeError> {
21        let mut by_id = BTreeMap::new();
22        let mut stamps = BTreeMap::new();
23        for write in writes {
24            if write.timestamp.device() != write.id.device {
25                return Err(MergeError::TimestampDevice(write.id));
26            }
27            if by_id.contains_key(&write.id) {
28                return Err(MergeError::DuplicateWrite(write.id));
29            }
30            if let Some(other) = stamps.insert(write.timestamp, write.id) {
31                return Err(MergeError::DuplicateTimestamp(other, write.id));
32            }
33            for (row, change) in &write.changes {
34                change.validate(row)?;
35            }
36            by_id.insert(write.id, write);
37        }
38        let history = Self { writes: by_id };
39        for write in history.writes.values() {
40            for past in &write.had_read {
41                let stamp = history
42                    .writes
43                    .get(past)
44                    .map(|w| w.timestamp)
45                    .ok_or(MergeError::MissingWrite(*past))?;
46                if stamp >= write.timestamp {
47                    return Err(MergeError::CausalTimestamp(write.id));
48                }
49            }
50            for (row, change) in &write.changes {
51                if change.generation != 0 {
52                    let mut seen = false;
53                    for past in &write.had_read {
54                        seen |= history.moved_to(*past, row)? == Some(change.generation);
55                    }
56                    if !seen {
57                        return Err(MergeError::GenerationNotSeen {
58                            row: row.clone(),
59                            generation: change.generation,
60                        });
61                    }
62                }
63            }
64        }
65        Ok(history)
66    }
67
68    fn moved_to(&self, write: WriteId, row: &RowId) -> Result<Option<u64>, MergeError> {
69        let write = self.writes.get(&write).ok_or(MergeError::MissingWrite(write))?;
70        match write.changes.get(row) {
71            Some(change) if change.operation.advances() => Ok(Some(change.after()?)),
72            _ => Ok(None),
73        }
74    }
75
76    /// Writes by their device and number; this is a set, not an arrival order.
77    pub fn writes(&self) -> &BTreeMap<WriteId, Write<V>> {
78        &self.writes
79    }
80}
81
82impl<V> WriteOracle for History<V> {
83    type Error = MergeError;
84    fn timestamp(&self, write: WriteId) -> Result<Option<Timestamp>, MergeError> {
85        Ok(self.writes.get(&write).map(|w| w.timestamp))
86    }
87    fn had_read(&self, reader: WriteId, earlier: WriteId) -> Result<bool, MergeError> {
88        let write = self.writes.get(&reader).ok_or(MergeError::MissingWrite(reader))?;
89        if !self.writes.contains_key(&earlier) {
90            return Err(MergeError::MissingWrite(earlier));
91        }
92        Ok(write.had_read.contains(&earlier))
93    }
94}
95
96/// The merged state as a function of a set alone (proofs/merge.md B4; Lean
97/// `IsSpec`). This does not call `apply` or sort writes into a causal order.
98/// Each component is selected directly from all setters and deletes.
99#[cfg(test)]
100pub(crate) fn from_writes<V: Clone>(
101    history: &History<V>,
102) -> Result<BTreeMap<RowId, RowState<V>>, MergeError> {
103    let rows: BTreeSet<_> =
104        history.writes.values().flat_map(|w| w.changes.keys().cloned()).collect();
105    let mut result = BTreeMap::new();
106    for row in rows {
107        let changes: Vec<_> =
108            history.writes.values().filter_map(|w| w.changes.get(&row).map(|c| (w, c))).collect();
109        let mut state = RowState::new(row.clone());
110        for (write, change) in &changes {
111            if change.operation.advances() {
112                let generation = change.after()?;
113                state.generation = state.generation.max(generation);
114                let earlier = match state.generations.get(&generation) {
115                    Some(old) => write.timestamp < crate::state::timestamp(history, *old)?,
116                    None => true,
117                };
118                if earlier {
119                    state.generations.insert(generation, write.id);
120                }
121            }
122        }
123        for (write, change) in &changes {
124            if let Some(columns) = change.operation.columns() {
125                let incarnation = change.incarnation()?;
126                for (column, value) in columns {
127                    if incarnation == state.generation {
128                        let wins = match state.cells.get(column) {
129                            Some(old) => {
130                                write.timestamp > crate::state::timestamp(history, old.write)?
131                            }
132                            None => true,
133                        };
134                        if wins {
135                            state.cells.insert(
136                                column.clone(),
137                                Cell { write: write.id, value: value.clone() },
138                            );
139                        }
140                    }
141                    let mut replacers = Vec::new();
142                    let mut deletes = Vec::new();
143                    for (other, ch) in &changes {
144                        if matches!(ch.operation, Operation::Delete) && ch.generation == incarnation
145                        {
146                            deletes.push(*other);
147                            replacers.push(*other);
148                        } else if ch.incarnation()? == incarnation
149                            && ch.operation.columns().is_some_and(|c| c.contains_key(column))
150                            && write.timestamp < other.timestamp
151                        {
152                            replacers.push(*other);
153                        }
154                    }
155                    if !replacers.is_empty()
156                        && replacers.iter().all(|w| !w.had_read.contains(&write.id))
157                    {
158                        let replacement = deletes
159                            .iter()
160                            .min_by_key(|w| w.timestamp)
161                            .or_else(|| replacers.iter().max_by_key(|w| w.timestamp));
162                        if let Some(replacement) = replacement {
163                            state.lost.insert(
164                                LostKey { column: column.clone(), write: write.id },
165                                LostValue {
166                                    incarnation,
167                                    value: value.clone(),
168                                    replaced_by: replacement.id,
169                                },
170                            );
171                        }
172                    }
173                }
174            }
175        }
176        result.insert(row, state);
177    }
178    Ok(result)
179}
180
181#[cfg(test)]
182#[path = "history_tests.rs"]
183mod tests;
coven_merge/history.rs

coven_merge/
history.rs
