coven_database/
sequence_apply.rs

1//! Atomic application of the next shared slot and its provisional queue overlay.
2
3use crate::download_stream::PreparedWrite;
4use crate::{sqlite::DatabaseConnection, write_schema::WriteSchema, DbError, StoreState};
5use coven_format::{admin::StoreChange, floor::OriginalSlot, sequence::SlotHeader};
6use coven_merge::Slot;
7use std::collections::BTreeSet;
8
9/// An authenticated slot whose floor check completed before installation.
10pub struct IncomingSlot<R> {
11    /// Its exact sequence identity.
12    pub slot: Slot,
13    /// The authenticated device, claim, author and preceding prefix.
14    pub header: SlotHeader,
15    /// SHA-256 of every complete sealed slot byte, including its signature.
16    pub digest: [u8; 32],
17    /// Opened content and the policy decisions made against the preceding prefix.
18    pub content: SlotContent<R>,
19}
20
21/// One slot kind; readable write parts are applied together.
22pub enum SlotContent<R> {
23    /// Original writes in increasing author write-number order.
24    Writes {
25        /// The complete sealed slot's length, signature included.
26        size: u64,
27        /// The slot's inner writes.
28        writes: Vec<R>,
29    },
30    /// An original admin object and its checked effect, absent for a refusal.
31    Admin {
32        /// Exact signed evidence retained by store snapshots.
33        original: OriginalSlot,
34        /// The effect authorized by the preceding prefix.
35        effect: Option<AdminEffect>,
36    },
37    /// An authenticated key-copy object, including refused copies.
38    KeyCopy(OriginalSlot),
39    /// An authenticated snapshot announcement. Its original is kept only when
40    /// it is accepted: no rule reads a refused one (§16.1).
41    Announcement {
42        /// Exact signed evidence retained by store snapshots.
43        original: OriginalSlot,
44        /// The announcement as this device accepted it against the preceding
45        /// prefix; absent for a refused one, and for one a reload stages again
46        /// at or below its store snapshot's coverage, whose originals the
47        /// snapshot or this device already holds.
48        accepted: Option<coven_format::sequence::SnapshotAnnouncement>,
49    },
50}
51
52/// Membership and required reload work installed with an authorized admin slot.
53pub struct AdminEffect {
54    /// The checked original change.
55    pub change: StoreChange,
56    /// Membership, keys and audience state after the change.
57    pub state: StoreState,
58    /// The activated snapshot for a raise or reset.
59    pub boundary: Option<crate::Boundary>,
60    /// Audiences requiring replacement data. `apply_slot` records their wait
61    /// without consuming the slot; `load_snapshots` installs it with their data.
62    pub reload: Vec<coven_merge::Audience>,
63    /// Provider work recorded atomically with this state.
64    pub provider_access: Vec<crate::ProviderAccessChange>,
65}
66
67pub(crate) fn retain(
68    db: &DatabaseConnection,
69    kind: &str,
70    original: &OriginalSlot,
71) -> Result<(), DbError> {
72    db.internal_execute("INSERT INTO _coven_slot_records(slot,kind,record) VALUES(?1,?2,?3) ON CONFLICT(slot) DO NOTHING",
73        rusqlite::params![original.slot.get().to_be_bytes().as_slice(),kind,original.object])?;
74    Ok(())
75}
76
77/// Forget the original slots of announcements no rule reads any more
78/// (§16.8). Each is a kept announcement's: one missing means the history the
79/// caller holds and these records disagree.
80pub(crate) fn forget_announcements(
81    db: &DatabaseConnection,
82    slots: &BTreeSet<Slot>,
83) -> Result<(), DbError> {
84    for slot in slots {
85        let deleted = db.internal_execute(
86            "DELETE FROM _coven_slot_records WHERE slot=?1 AND kind='announcement'",
87            [slot.get().to_be_bytes()],
88        )?;
89        if deleted != 1 {
90            return Err(DbError::DamagedDatabase);
91        }
92    }
93    Ok(())
94}
95
96pub(crate) fn apply<P: crate::PendingValue>(
97    db: &mut DatabaseConnection,
98    before: &DatabaseConnection,
99    schema: &WriteSchema,
100    incoming: IncomingSlot<PreparedWrite<'_>>,
101    writers: &crate::Writers,
102    files: &crate::file_write::WriteFiles<'_>,
103    raises: Option<&crate::migration::MigrationOperation>,
104    authenticate: impl FnOnce() -> Result<(), DbError>,
105) -> Result<crate::ApplyOutcome, DbError> {
106    crate::download::source_wait(db.stream_transaction(|db| {
107        let files = &crate::file_write::FileWrite::new(db, files);
108        // Applying a slot runs no migration and no DDL.
109        let _constants = db.keep_constants();
110        let applied = crate::sequence::applied(db)?;
111        if incoming.slot.get() <= applied {
112            return Ok(crate::ApplyOutcome::AlreadyApplied);
113        }
114        if applied.checked_add(1) != Some(incoming.slot.get()) || incoming.header.after > applied {
115            return Err(DbError::SequenceGap { applied, received: incoming.slot });
116        }
117        if let SlotContent::Admin { effect: Some(effect), .. } = &incoming.content {
118            if let Some(audience) = effect.reload.first() {
119                authenticate()?;
120                for required in &effect.reload {
121                    crate::reload_gate::require_reload::<P>(db, required)?;
122                }
123                return Ok(crate::ApplyOutcome::Waiting(crate::WriteWait::Reload(audience.clone())));
124            }
125        }
126        crate::download_stage::begin(db)?;
127        // An admin slot with an effect changes membership, cut-offs or
128        // boundaries, and with them how every queued write applies: every
129        // kept row is put back and the whole queue folds again. Any other
130        // slot puts back only the kept rows its writes reach (§6.2).
131        let whole = matches!(&incoming.content, SlotContent::Admin { effect: Some(_), .. });
132        let mut touched = if whole {
133            crate::published::restore(db, schema)?
134        } else {
135            BTreeSet::new()
136        };
137        let mut reach = crate::queued_writes::Reach::new();
138        crate::pending::clear_slot(db, incoming.slot)?;
139        match incoming.content {
140            SlotContent::Writes { size, writes } => stage_writes::<P>(
141                db,
142                schema,
143                incoming.slot,
144                &incoming.header,
145                size,
146                writes,
147                Staging::Next(&mut reach),
148            )?,
149            SlotContent::Admin { original, effect } => {
150                retain(db, "admin", &original)?;
151                if let Some(effect) = effect {
152                    let old = crate::store_state_tables::deleted_circles(db)?;
153                    let created = !crate::store_state_tables::created(db)?
154                        && effect.state.store.is_some();
155                    crate::sequence::cutoff(db, incoming.slot, &effect.change)?;
156                    crate::store_state_tables::replace(db, &effect.state)?;
157                    if created {
158                        crate::raise_capture::capture_created(db, schema)?;
159                    }
160                    if let coven_format::admin::StoreChange::CreateCircle { circle, .. } =
161                        &effect.change
162                    {
163                        crate::raise_capture::capture_made(db, *circle, &effect.state, raises)?;
164                    }
165                    crate::provider_access::adopt::<P>(db, &effect.provider_access)?;
166                    for circle in effect.state.circle_deletions.keys().filter(|circle| !old.contains_key(circle)) {
167                        touched.extend(db.query(
168                            "SELECT DISTINCT table_name,key,audience FROM _coven_rows WHERE audience=?1",
169                            [coven_merge::Audience::Circle(*circle)],
170                            crate::row_queries::read_identity,
171                        )?);
172                    }
173                    if let Some(boundary) = effect.boundary {
174                        crate::write_boundary::record(db, &boundary)?;
175                    }
176                    crate::raise_capture::end_raise_waits::<P>(db)?;
177                }
178            }
179            SlotContent::KeyCopy(original) => retain(db, "key", &original)?,
180            SlotContent::Announcement { original, accepted } => {
181                if let Some(announcement) = accepted {
182                    retain(db, "announcement", &original)?;
183                    snapshot_announced(db, &announcement)?;
184                }
185            }
186        }
187        authenticate()?;
188        crate::sequence::note_claim(db, &incoming.header, incoming.slot, incoming.digest)?;
189        crate::sequence::set_applied(db, incoming.slot.get())?;
190        crate::forget_records::forget(db, writers)?;
191        let refold = if whole {
192            crate::queued_writes::Refold::Whole
193        } else {
194            touched.extend(reach.restored().iter().cloned());
195            crate::queued_writes::Refold::Reached(&reach)
196        };
197        crate::queued_writes::overlay(db, schema, &mut touched, refold)?;
198        let deleted = crate::store_state_tables::deleted_circles(db)?;
199        crate::download_stage::materialize(db, before, schema, &deleted, touched, files)?;
200        files.before_commit()?;
201        crate::operation::end_on_stop(db)?;
202        Ok(crate::ApplyOutcome::Applied)
203    }))
204}
205
206/// `announcement` was accepted. An ordinary snapshot's coverage only moves
207/// forward and is always usable, so a new device replays for its audience
208/// only the slots after it, and those at or below it never count toward that
209/// audience's next snapshot again (§16.2): their records of the audience's
210/// parts go, and each slot's size once no audience's part records it.
211fn snapshot_announced(
212    db: &DatabaseConnection,
213    announcement: &coven_format::sequence::SnapshotAnnouncement,
214) -> Result<(), DbError> {
215    if announcement.purpose != coven_format::sequence::Purpose::Ordinary {
216        return Ok(());
217    }
218    let covers = announcement.covers.to_be_bytes();
219    db.internal_execute(
220        "DELETE FROM _coven_write_slot_audiences WHERE audience=?1 AND slot<=?2",
221        rusqlite::params![announcement.snapshot.audience, covers],
222    )?;
223    db.internal_execute(
224        "DELETE FROM _coven_write_slots WHERE slot<=?1
225         AND NOT EXISTS(SELECT 1 FROM _coven_write_slot_audiences a WHERE a.slot=_coven_write_slots.slot)",
226        [covers],
227    )?;
228    Ok(())
229}
230
231/// How a slot's writes are staged: as the next slot, or replayed by a reload.
232pub(crate) enum Staging<'a> {
233    /// The next slot, beside the queue: each write puts back the kept rows
234    /// it reaches before it merges (§6.2).
235    Next(&'a mut crate::queued_writes::Reach),
236    /// A reload's, into only the parts newer than their snapshots. The
237    /// reload put back every kept row first.
238    Reload(&'a crate::snapshot_coverage::SnapshotCoverage),
239}
240
241/// Ordinary application and reloads use the same ordering, cut-off and original
242/// validation.
243pub(crate) fn stage_writes<P: crate::PendingValue>(
244    db: &DatabaseConnection,
245    schema: &WriteSchema,
246    slot: Slot,
247    header: &SlotHeader,
248    size: u64,
249    writes: Vec<PreparedWrite<'_>>,
250    mut staging: Staging<'_>,
251) -> Result<(), DbError> {
252    // A slot's object never changes, so a reload that stages it again
253    // finds the same size.
254    let stored = crate::sqlite::integer("writes slot size", size)?;
255    db.internal_execute(
256        "INSERT INTO _coven_write_slots(slot,size) VALUES(?1,?2) ON CONFLICT(slot) DO NOTHING",
257        rusqlite::params![slot.get().to_be_bytes(), stored],
258    )?;
259    let mut previous = None;
260    for write in writes {
261        header.check_write(&write.header().header)?;
262        if previous.is_some_and(|number| number >= write.header().header.position.number) {
263            return Err(crate::snapshot_error::invalid(
264                "slot writes are not in original number order",
265            ));
266        }
267        previous = Some(write.header().header.position.number);
268        for part in &write.header().parts {
269            db.internal_execute(
270                "INSERT INTO _coven_write_slot_audiences(audience,slot) VALUES(?1,?2) ON CONFLICT DO NOTHING",
271                rusqlite::params![part.audience, slot.get().to_be_bytes()],
272            )?;
273        }
274        let cutoff =
275            crate::sequence::refusal(db, slot, header.device, header.author, write.signer())?;
276        if let Some((cutoff, cause)) = cutoff {
277            db.internal_execute("INSERT INTO _coven_write_refusals(slot,cutoff) VALUES(?1,?2) ON CONFLICT(slot) DO NOTHING",
278                rusqlite::params![slot.get().to_be_bytes(), cutoff.get().to_be_bytes()])?;
279            crate::pending::save(db, &P::writes_refused(slot, cutoff, cause))?;
280        }
281        if let Staging::Reload(coverage) = &staging {
282            for audience in write.unopened_audiences() {
283                if coverage.needs(audience, slot) {
284                    return Err(crate::snapshot_error::invalid(
285                        "reload lacks a readable audience part",
286                    ));
287                }
288            }
289        }
290        match &mut staging {
291            Staging::Next(reach) => crate::download_stream::stage_selected(
292                db,
293                schema,
294                write,
295                slot,
296                cutoff.is_some(),
297                |_| true,
298                Some(reach),
299            )?,
300            Staging::Reload(coverage) => crate::download_stream::stage_selected(
301                db,
302                schema,
303                write,
304                slot,
305                cutoff.is_some(),
306                |audience| coverage.needs(audience, slot),
307                None,
308            )?,
309        }
310    }
311    Ok(())
312}
313
314impl<R: std::io::Read> IncomingSlot<crate::DownloadedWriteStream<R>> {
315    pub(crate) fn prepare<'a>(
316        self,
317        inputs: &crate::input_preparation::PreparedInputs<'a>,
318    ) -> Result<IncomingSlot<PreparedWrite<'a>>, DbError> {
319        let content = match self.content {
320            SlotContent::Writes { size, writes } => SlotContent::Writes {
321                size,
322                writes: writes
323                    .into_iter()
324                    .map(|write| write.prepare(inputs))
325                    .collect::<Result<_, _>>()?,
326            },
327            SlotContent::Admin { original, effect } => SlotContent::Admin { original, effect },
328            SlotContent::KeyCopy(original) => SlotContent::KeyCopy(original),
329            SlotContent::Announcement { original, accepted } => {
330                SlotContent::Announcement { original, accepted }
331            }
332        };
333        Ok(IncomingSlot { slot: self.slot, header: self.header, digest: self.digest, content })
334    }
335}
336
337#[cfg(test)]
338#[path = "sequence_apply_tests.rs"]
339pub(crate) mod tests;
coven_database/sequence_apply.rs

coven_database/
sequence_apply.rs
