coven_merge/
input.rs

1use crate::{MergeError, Timestamp};
2use coven_foundation::id_source::{CircleId, DeviceId};
3use std::collections::{BTreeMap, BTreeSet};
4
5/// A foreign key's columns in declaration order (Appendix B.5).
6#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
7pub struct ConstraintColumns(
8    /// Ordered column names, retaining composite-constraint boundaries.
9    pub Vec<String>,
10);
11
12impl<const N: usize> From<[&str; N]> for ConstraintColumns {
13    fn from(columns: [&str; N]) -> Self {
14        Self(columns.into_iter().map(str::to_owned).collect())
15    }
16}
17
18/// A unique constraint's ordered terms and optional partial predicate (Appendix B.5).
19/// Column terms use their names; expression terms retain their SQL text.
20#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
21pub struct UniqueConstraint {
22    /// Column names or expression text, in declaration order.
23    pub terms: Vec<String>,
24    /// The WHERE expression as written for a partial unique index.
25    pub partial: Option<String>,
26}
27
28impl<const N: usize> From<[&str; N]> for UniqueConstraint {
29    fn from(terms: [&str; N]) -> Self {
30        Self { terms: terms.into_iter().map(str::to_owned).collect(), partial: None }
31    }
32}
33
34/// A foreign key's stable identity, independent of SQLite's positional id (Appendix B.5).
35#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
36pub struct ForeignKey {
37    /// Referencing columns in declaration order.
38    pub columns: ConstraintColumns,
39    /// Referenced table.
40    pub parent: String,
41    /// Referenced columns, paired in order with `columns`.
42    pub parent_columns: ConstraintColumns,
43}
44
45impl ForeignKey {
46    /// Name both sides of a declared reference, including implicit primary keys.
47    pub fn new(
48        columns: impl Into<ConstraintColumns>,
49        parent: impl Into<String>,
50        parent_columns: impl Into<ConstraintColumns>,
51    ) -> Self {
52        Self {
53            columns: columns.into(),
54            parent: parent.into(),
55            parent_columns: parent_columns.into(),
56        }
57    }
58}
59
60/// One device's numbered write (§4).
61#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
62pub struct WriteId {
63    /// The install's 64-bit device id.
64    pub device: DeviceId,
65    /// The device's original write number.
66    pub number: u64,
67}
68
69/// A positive position in the store's shared sequence (§5). Slot 0 names
70/// nothing: a prefix applied through 0 has applied no slot, and that count
71/// stays a plain `u64`.
72#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
73pub struct Slot(std::num::NonZeroU64);
74
75impl Slot {
76    /// The first slot of a sequence.
77    pub const FIRST: Self = Self(std::num::NonZeroU64::MIN);
78
79    /// The slot numbered `number`, or `None` for 0.
80    pub fn new(number: u64) -> Option<Self> {
81        std::num::NonZeroU64::new(number).map(Self)
82    }
83
84    /// The slot's number, at least 1.
85    pub fn get(self) -> u64 {
86        self.0.get()
87    }
88}
89
90/// The slots from a first one through a last number, in order: empty when
91/// the first lies past the last. The last stays a plain number, since 0
92/// (nothing applied) ends an empty range.
93#[derive(Clone, Copy, Debug)]
94pub struct SlotRange {
95    next: Option<Slot>,
96    last: u64,
97}
98
99impl SlotRange {
100    /// The slots from `first` through slot number `last`.
101    pub fn new(first: Slot, last: u64) -> Self {
102        Self { next: Some(first).filter(|first| first.get() <= last), last }
103    }
104}
105
106impl Iterator for SlotRange {
107    type Item = Slot;
108
109    fn next(&mut self) -> Option<Slot> {
110        let slot = self.next?;
111        self.next = slot.get().checked_add(1).filter(|next| *next <= self.last).and_then(Slot::new);
112        Some(slot)
113    }
114}
115
116impl From<std::num::NonZeroU64> for Slot {
117    fn from(number: std::num::NonZeroU64) -> Self {
118        Self(number)
119    }
120}
121
122impl From<Slot> for std::num::NonZeroU64 {
123    fn from(slot: Slot) -> Self {
124        slot.0
125    }
126}
127
128/// Test support: the slot numbered `number`, which must be positive.
129#[cfg(any(test, feature = "test-utils"))]
130pub fn slot(number: u64) -> Slot {
131    Slot::new(number).expect("a test slot number is positive")
132}
133
134/// Every synced row reaches the store or one circle (§15).
135#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
136pub enum Audience {
137    /// Every member can read the row.
138    Store,
139    /// Only members of this circle can read the row.
140    Circle(CircleId),
141}
142
143/// A row is one table, primary key and audience, with generations of its own.
144#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
145pub struct RowId {
146    /// The synced table's name.
147    pub table: String,
148    /// The primary key, encoded by the caller so byte order is key order.
149    pub key: Vec<u8>,
150    /// The store or circle that can read the row.
151    pub audience: Audience,
152}
153
154/// The parent and odd incarnation a reference names (§9.1).
155#[derive(Clone, Debug, PartialEq, Eq)]
156pub struct Parent {
157    /// The referenced row, including its audience.
158    pub row: RowId,
159    /// The parent's generation.
160    pub generation: u64,
161}
162
163impl Parent {
164    /// Validate a written reference from `child`: it must name an odd
165    /// incarnation in the child's audience or the store (§9.1, §15.3).
166    pub fn validate_written(&self, child: &RowId) -> Result<(), MergeError> {
167        if !crate::generation_present(self.generation) {
168            return Err(MergeError::ParentGeneration(self.generation));
169        }
170        validate_audience(child, &self.row)
171    }
172}
173
174/// A column's value and the parent generations recorded with it.
175/// The metadata follows the winning setter, including for values kept as lost.
176#[derive(Clone, Debug, PartialEq, Eq)]
177pub struct ColumnValue<V> {
178    /// The app's value; the merge does not interpret SQL values.
179    pub value: V,
180    /// Foreign-key names and their parents for this column's written value.
181    /// Each cell retains the complete parent identity recorded by its setter.
182    pub parents: BTreeMap<ForeignKey, Parent>,
183}
184
185/// One row's insert, update or delete. A delete cannot set columns.
186#[derive(Clone, Debug, PartialEq, Eq)]
187pub enum Operation<V> {
188    /// Start the next incarnation with these columns.
189    Insert(BTreeMap<String, ColumnValue<V>>),
190    /// Set these columns in the existing incarnation.
191    Update(BTreeMap<String, ColumnValue<V>>),
192    /// End the existing incarnation.
193    Delete,
194}
195
196impl<V> Operation<V> {
197    pub(crate) fn columns(&self) -> Option<&BTreeMap<String, ColumnValue<V>>> {
198        match self {
199            Self::Insert(c) | Self::Update(c) => Some(c),
200            Self::Delete => None,
201        }
202    }
203    pub(crate) fn advances(&self) -> bool {
204        !matches!(self, Self::Update(_))
205    }
206}
207
208/// A row change carries the generation on the authoring device (§8.3).
209#[derive(Clone, Debug, PartialEq, Eq)]
210pub struct Change<V> {
211    /// Generation at which the change was made.
212    pub generation: u64,
213    /// The change and the values it sets.
214    pub operation: Operation<V>,
215}
216
217impl<V> Change<V> {
218    /// Validate generation parity, advancement without overflow, and every
219    /// written parent's generation and audience for this row (§8.3–§9.1, §15).
220    /// Values and their encodings are opaque here. The authoring database
221    /// establishes generation witnesses; [`crate::apply`] checks the received
222    /// change against applied metadata.
223    pub fn validate(&self, row: &RowId) -> Result<(), MergeError> {
224        self.incarnation()?;
225        self.after()?;
226        if let Some(columns) = self.operation.columns() {
227            for value in columns.values() {
228                for parent in value.parents.values() {
229                    parent.validate_written(row)?;
230                }
231            }
232        }
233        Ok(())
234    }
235
236    /// The row's generation after this change: an insert or a delete advances
237    /// it, an update keeps it (§8.3).
238    pub fn after(&self) -> Result<u64, MergeError> {
239        if self.operation.advances() {
240            self.generation.checked_add(1).ok_or(MergeError::GenerationExhausted)
241        } else {
242            Ok(self.generation)
243        }
244    }
245
246    /// The incarnation this change belongs to: generation + 1 for an insert,
247    /// generation for an update or delete. Invalid parity is a typed error.
248    pub fn incarnation(&self) -> Result<u64, MergeError> {
249        if matches!(self.operation, Operation::Insert(_))
250            == crate::generation_present(self.generation)
251        {
252            return Err(MergeError::GenerationParity(self.generation));
253        }
254        if matches!(self.operation, Operation::Insert(_)) {
255            self.generation.checked_add(1).ok_or(MergeError::GenerationExhausted)
256        } else {
257            Ok(self.generation)
258        }
259    }
260}
261
262/// A decoded write's merge inputs, independent of its storage encoding.
263#[derive(Clone, Debug, PartialEq, Eq)]
264pub struct Write<V, P = BTreeSet<WriteId>> {
265    /// Device and original write number.
266    pub id: WriteId,
267    /// Unique total-order timestamp (§7.2).
268    pub timestamp: Timestamp,
269    /// The writes in the author's `read_through` prefix and its own earlier
270    /// writes (§7.1). Consuming a write does not imply retaining its row changes.
271    /// The sequence resolves this relation from the prefix and own predecessors.
272    pub had_read: P,
273    /// At most one change for each table, key and audience.
274    pub changes: BTreeMap<RowId, Change<V>>,
275}
276
277impl<V, P: WritePast> Write<V, P> {
278    /// Check the write's identity and causal timestamp against the applied set.
279    /// This also validates writes with no row changes available to the receiver.
280    pub fn validate_metadata<O: WriteOracle>(&self, oracle: &O) -> Result<(), O::Error> {
281        if oracle.timestamp(self.id)?.is_some() {
282            return Err(MergeError::DuplicateWrite(self.id).into());
283        }
284        if self.timestamp.device() != self.id.device {
285            return Err(MergeError::TimestampDevice(self.id).into());
286        }
287        for past in self.had_read.frontier() {
288            if crate::state::timestamp(oracle, *past)? >= self.timestamp {
289                return Err(MergeError::CausalTimestamp(self.id).into());
290            }
291        }
292        Ok(())
293    }
294}
295
296/// A causally closed authored past. Its frontier bounds the timestamps of
297/// every covered write; sequence order itself is not timestamp order.
298pub trait WritePast {
299    /// Whether the writer had consumed this write's position.
300    fn contains(&self, write: &WriteId) -> bool;
301    /// Writes whose causal pasts cover this entire past, including themselves.
302    fn frontier(&self) -> impl Iterator<Item = &WriteId>;
303}
304
305impl WritePast for BTreeSet<WriteId> {
306    fn contains(&self, write: &WriteId) -> bool {
307        self.contains(write)
308    }
309    fn frontier(&self) -> impl Iterator<Item = &WriteId> {
310        self.iter()
311    }
312}
313
314/// Access to applied write metadata. The arriving write is not yet in this
315/// view. Implementations must report missing metadata, never invent it, and
316/// a failure to read it is their own error, never a missing write.
317/// No applied row-change history is required by the incremental step.
318pub trait WriteOracle {
319    /// Reading the metadata failed, or a merge check on it did.
320    type Error: From<MergeError>;
321    /// The timestamp of an applied write, or `None` if it is not applied.
322    fn timestamp(&self, write: WriteId) -> Result<Option<Timestamp>, Self::Error>;
323    /// Whether an applied write had read another write.
324    fn had_read(&self, reader: WriteId, earlier: WriteId) -> Result<bool, Self::Error>;
325}
326
327pub(crate) fn validate_audience(row: &RowId, parent: &RowId) -> Result<(), MergeError> {
328    if parent.audience != Audience::Store && parent.audience != row.audience {
329        return Err(MergeError::ReferenceAudience(row.clone()));
330    }
331    Ok(())
332}
333
334#[cfg(test)]
335#[path = "input_tests.rs"]
336mod tests;
coven_merge/input.rs

coven_merge/
input.rs
