coven_database/
write_boundary.rs

1//! Checked entry boundaries, their immutable coverage, and slot selection.
2use crate::sqlite::DatabaseConnection;
3use crate::write_encoding::{audience, counter, damaged};
4use crate::{DbError, StoreState, StoreVersion};
5use coven_format::{
6    admin::SnapshotId, snapshot_rows::ChangeVerdict, value::Slot, write::WriteHeader,
7};
8use coven_foundation::id_source::DeviceId;
9use coven_merge::Audience;
10use rusqlite::params;
11use std::collections::BTreeSet;
12
13/// A checked raise/reset and the snapshot named by that entry. Its selection is
14/// derived from kept entries, independently of whether coverage has been read.
15#[derive(Clone, Debug, PartialEq, Eq)]
16pub enum Boundary {
17    /// A breaking schema raise.
18    SchemaChange(StoreVersion),
19    /// A reset of the named snapshot's audience.
20    Reset {
21        /// The reset's counted slot.
22        entry: Slot,
23        /// Its snapshot, including the audience.
24        snapshot: SnapshotId,
25    },
26}
27
28impl Boundary {
29    /// The checked entry that owns this boundary.
30    pub fn entry(&self) -> Slot {
31        match self {
32            Self::SchemaChange(version) => version.entry,
33            Self::Reset { entry, .. } => *entry,
34        }
35    }
36
37    /// The immutable snapshot named by this boundary.
38    pub fn snapshot(&self) -> &SnapshotId {
39        match self {
40            Self::SchemaChange(version) => &version.snapshot,
41            Self::Reset { snapshot, .. } => snapshot,
42        }
43    }
44
45    /// The no-op rule for a raise: one that doesn't go past the version its
46    /// audience has reached, a circle's base included, changes nothing.
47    pub fn schema_in_place(state: &StoreState, version: u32, snapshot: &SnapshotId) -> bool {
48        version <= state.reached(&snapshot.audience)
49    }
50
51    /// Repeating the current reset does not introduce a new boundary.
52    pub fn reset_in_place(state: &StoreState, snapshot: &SnapshotId) -> bool {
53        state.resets.get(&snapshot.audience) == Some(snapshot)
54    }
55
56    /// Whether applying this checked boundary leaves the applied state unchanged.
57    pub fn already_in_place(&self, state: &StoreState) -> bool {
58        match self {
59            Self::SchemaChange(version) => {
60                Self::schema_in_place(state, version.number, &version.snapshot)
61            }
62            Self::Reset { snapshot, .. } => Self::reset_in_place(state, snapshot),
63        }
64    }
65
66    /// Realize a non-no-op boundary after the preceding prefix authorizes it.
67    pub fn apply(&self, state: &mut StoreState) {
68        match self {
69            Self::SchemaChange(version) => {
70                state.schema.insert(version.snapshot.audience.clone(), version.clone());
71            }
72            Self::Reset { snapshot, .. } => {
73                state.resets.insert(snapshot.audience.clone(), snapshot.clone());
74            }
75        }
76    }
77
78    pub(crate) fn load_selected(database: &DatabaseConnection) -> Result<Vec<Self>, DbError> {
79        Ok(retained(database)?.into_iter().map(|row| row.boundary).collect())
80    }
81}
82
83/// A checked raise or reset and its authenticated snapshot coverage.
84#[derive(Clone, Debug, PartialEq, Eq)]
85pub struct BoundaryCoverage {
86    /// The checked change, including its original slot and snapshot identity.
87    pub boundary: Boundary,
88    /// The slot coverage authenticated by its snapshot prefix.
89    pub included: u64,
90}
91
92/// Keep a checked boundary, once, at its entry's slot.
93pub(crate) fn record(database: &DatabaseConnection, boundary: &Boundary) -> Result<(), DbError> {
94    let (kind, version) = match boundary {
95        Boundary::SchemaChange(version) => ("schema", Some(version.number)),
96        Boundary::Reset { .. } => ("reset", None),
97    };
98    let entry = boundary.entry();
99    let snapshot = boundary.snapshot();
100    database.internal_execute(
101        "INSERT INTO _coven_boundaries(slot,kind,version,audience,snapshot_device,snapshot_number)
102         VALUES(?1,?2,?3,?4,?5,?6) ON CONFLICT(slot) DO NOTHING",
103        params![
104            entry.get().to_be_bytes().as_slice(),
105            kind,
106            version,
107            &snapshot.audience,
108            snapshot.device,
109            snapshot.number.to_be_bytes().as_slice()
110        ],
111    )?;
112    Ok(())
113}
114
115/// Keep a boundary with its authenticated coverage and realize it in the
116/// data view: whether this applied it, which a boundary already applied
117/// does not. Coverage once checked never changes.
118pub(crate) fn record_coverage(
119    database: &DatabaseConnection,
120    coverage: &BoundaryCoverage,
121) -> Result<bool, DbError> {
122    record(database, &coverage.boundary)?;
123    let previous: (Option<Vec<u8>>, bool) = database.query_row(
124        "SELECT included,applied FROM _coven_boundaries WHERE slot=?1",
125        [coverage.boundary.entry().get().to_be_bytes().as_slice()],
126        |row| Ok((row.get(0)?, row.get(1)?)),
127    )?;
128    let included = coverage.included.to_be_bytes().to_vec();
129    if let Some(checked) = previous.0 {
130        if checked != included {
131            return Err(crate::snapshot_error::invalid("checked boundary coverage changed"));
132        }
133    }
134    if previous.1 {
135        return Ok(false);
136    }
137    database.internal_execute(
138        "UPDATE _coven_boundaries SET included=?2,applied=1 WHERE slot=?1",
139        params![coverage.boundary.entry().get().to_be_bytes().as_slice(), included],
140    )?;
141    Ok(true)
142}
143
144/// Install the reload's effective readable boundaries, retaining every
145/// checked coverage fact. The applied bit records the committed data view,
146/// not slot selection. It also distinguishes a repeated reset reload.
147pub(crate) fn replace(
148    database: &DatabaseConnection,
149    boundaries: Vec<BoundaryCoverage>,
150) -> Result<Vec<Slot>, DbError> {
151    let mut ids = BTreeSet::new();
152    for boundary in &boundaries {
153        if !ids.insert(boundary.boundary.entry()) {
154            return Err(crate::snapshot_error::invalid("reload repeats a boundary slot"));
155        }
156    }
157    let previous: BTreeSet<_> = database
158        .query("SELECT slot FROM _coven_boundaries WHERE applied=1 AND kind='reset'", [], |row| {
159            crate::store_state_tables::slot(row, 0)
160        })?
161        .into_iter()
162        .collect();
163    let changed_resets = boundaries
164        .iter()
165        .filter_map(|coverage| match &coverage.boundary {
166            Boundary::Reset { entry, .. } if ids.contains(entry) && !previous.contains(entry) => {
167                Some(*entry)
168            }
169            _ => None,
170        })
171        .collect();
172    database.internal_execute("DELETE FROM _coven_boundaries", [])?;
173    for boundary in &boundaries {
174        record_coverage(database, boundary)?;
175    }
176    Ok(changed_resets)
177}
178
179struct RetainedBoundary {
180    boundary: Boundary,
181    included: Option<u64>,
182    applied: bool,
183}
184
185fn retained(database: &DatabaseConnection) -> Result<Vec<RetainedBoundary>, DbError> {
186    database.query(
187        "SELECT b.kind,b.version,b.audience,b.snapshot_device,b.snapshot_number,b.slot,b.included,b.applied
188         FROM _coven_boundaries b ORDER BY b.slot",
189        [],
190        |row| {
191            let snapshot = SnapshotId {
192                audience: audience(row, 2)?,
193                device: row.get::<_, DeviceId>(3)?,
194                number: counter(row.get(4)?)?,
195            };
196            let entry = crate::store_state_tables::slot(row, 5)?;
197            let boundary = match (row.get::<_, String>(0)?.as_str(), row.get::<_, Option<u32>>(1)?) {
198                ("schema", Some(number)) => Boundary::SchemaChange(StoreVersion { number, snapshot, entry }),
199                ("reset", None) => Boundary::Reset { entry, snapshot },
200                _ => return Err(damaged(0, rusqlite::types::Type::Text)),
201            };
202            let included = row.get::<_, Option<Vec<u8>>>(6)?.map(counter).transpose()?;
203            Ok(RetainedBoundary { boundary, included, applied: row.get(7)? })
204        },
205    )
206}
207
208/// What a realized boundary decides about an uncovered part of its audience.
209#[derive(Clone, Debug, PartialEq, Eq)]
210pub(crate) enum BoundaryRule {
211    /// An older-schema part needs admission through the raise's retained
212    /// breaking steps (§18.6).
213    Raise(u32),
214    /// A reset ignores a part whose write had not read this entry (§19.4).
215    Reset(Slot),
216}
217
218/// The boundary facts realized by the current data projection.
219pub(crate) struct WriteBoundary {
220    pub(crate) audience: Audience,
221    pub(crate) rule: BoundaryRule,
222    pub(crate) included: u64,
223}
224
225impl WriteBoundary {
226    pub(crate) fn load(database: &DatabaseConnection) -> Result<Vec<Self>, DbError> {
227        let mut state = StoreState::default();
228        let mut result = Vec::new();
229        for row in retained(database)? {
230            if row.boundary.already_in_place(&state) {
231                continue;
232            }
233            row.boundary.apply(&mut state);
234            if row.applied {
235                result.push(Self {
236                    audience: row.boundary.snapshot().audience.clone(),
237                    rule: match &row.boundary {
238                        Boundary::SchemaChange(version) => BoundaryRule::Raise(version.number),
239                        Boundary::Reset { entry, .. } => BoundaryRule::Reset(*entry),
240                    },
241                    included: row.included.ok_or(DbError::DamagedDatabase)?,
242                });
243            }
244        }
245        Ok(result)
246    }
247
248    /// The realized boundaries, with each unpublished raise capture judging
249    /// its audience as its kept raise will: a migrated device applies
250    /// received writes before its raise is kept, so a capture taken again
251    /// after its catch-up covers them (§16.3, §18.5). A capture whose
252    /// audience already realized a raise to its version adds nothing.
253    pub(crate) fn judging(database: &DatabaseConnection) -> Result<Vec<Self>, DbError> {
254        let mut boundaries = Self::load(database)?;
255        for (audience, version, included) in crate::raise_capture::provisional(database)? {
256            let realized = boundaries.iter().any(|b| {
257                b.audience == audience
258                    && matches!(b.rule, BoundaryRule::Raise(number) if number >= version)
259            });
260            if !realized {
261                boundaries.push(Self { audience, rule: BoundaryRule::Raise(version), included });
262            }
263        }
264        Ok(boundaries)
265    }
266
267    /// This boundary's verdict on an uncovered part of its audience: a kept
268    /// reset ignores a part whose write had not read the reset entry (§19.4);
269    /// a raise returns a candidate schema exclusion, resolved by the interpreter
270    /// using the retained breaking steps (§18.6).
271    pub(crate) fn excludes(
272        &self,
273        header: &WriteHeader,
274        audience: &Audience,
275        slot: Option<Slot>,
276    ) -> Option<ChangeVerdict> {
277        if *audience != self.audience || slot.is_some_and(|slot| slot.get() <= self.included) {
278            return None;
279        }
280        match &self.rule {
281            BoundaryRule::Raise(version) => {
282                (header.schema_version < *version).then_some(ChangeVerdict::Excluded(*version))
283            }
284            BoundaryRule::Reset(entry) => {
285                (header.read_through < entry.get()).then_some(ChangeVerdict::Ignored(*entry))
286            }
287        }
288    }
289
290    /// Whether the selected boundaries ignore one part (§18.6): its changes
291    /// stay as evidence and are never checked or merged.
292    pub(crate) fn ignore(
293        boundaries: &[Self],
294        header: &WriteHeader,
295        audience: &Audience,
296        slot: Option<Slot>,
297    ) -> bool {
298        matches!(
299            Self::part_verdict(boundaries, header, audience, slot),
300            Some(ChangeVerdict::Ignored(_))
301        )
302    }
303
304    /// The selected boundaries' combined verdict on one part. Reset
305    /// eligibility comes first, so an ignored part never gains a schema loss;
306    /// otherwise the highest excluding raise names the loss.
307    pub(crate) fn part_verdict(
308        boundaries: &[Self],
309        header: &WriteHeader,
310        audience: &Audience,
311        slot: Option<Slot>,
312    ) -> Option<ChangeVerdict> {
313        let verdicts: Vec<_> = boundaries
314            .iter()
315            .filter_map(|boundary| boundary.excludes(header, audience, slot))
316            .collect();
317        verdicts
318            .iter()
319            .find(|verdict| matches!(verdict, ChangeVerdict::Ignored(_)))
320            .cloned()
321            .or_else(|| {
322                verdicts
323                    .iter()
324                    .filter_map(|verdict| match verdict {
325                        ChangeVerdict::Excluded(version) => Some(*version),
326                        _ => None,
327                    })
328                    .max()
329                    .map(ChangeVerdict::Excluded)
330            })
331    }
332}
333#[cfg(test)]
334#[path = "write_boundary_tests.rs"]
335pub(crate) mod tests;
coven_database/write_boundary.rs

coven_database/
write_boundary.rs
