coven_database/
raise_capture.rs

1//! The original queue prefix captured for an unpublished breaking raise.
2//!
3//! A capture fixes eligibility through its marker. Snapshot bytes are made from
4//! the counted base only after that prefix has acquired slots; later local
5//! originals remain outside it.
6use crate::write_encoding::{audience, counter};
7use crate::{sqlite::DatabaseConnection, write_schema::WriteSchema, DbError};
8use coven_format::{admin::SnapshotId, snapshot::SnapshotHeader};
9use coven_foundation::id_source::DeviceId;
10use coven_merge::{Audience, WriteId};
11use rusqlite::params;
12
13/// A breaking raise's fixed original prefix and current counted coverage.
14#[derive(Clone, Debug, PartialEq, Eq)]
15pub struct RaiseCapture {
16    /// Audience raised by this capture.
17    pub audience: Audience,
18    /// Schema version to announce.
19    pub version: u32,
20    /// Last captured original, including the migration marker.
21    pub through: WriteId,
22    /// Applied slot when this capture was prepared.
23    pub covers: u64,
24}
25
26pub(crate) fn capture(
27    db: &DatabaseConnection,
28    _schema: &WriteSchema,
29    marker: WriteId,
30    version: u32,
31) -> Result<bool, DbError> {
32    let state = crate::store_state_tables::read(db)?;
33    db.internal_execute("DELETE FROM _coven_raise_captures", [])?;
34    let Some(member) = state
35        .store
36        .as_ref()
37        .and_then(|_| state.devices.get(&marker.device))
38        .map(|device| device.member)
39    else {
40        return Ok(false);
41    };
42    for selected in state.readable_audiences(&member) {
43        db.internal_execute("INSERT INTO _coven_raise_captures(audience,version,through,covers) VALUES(?1,?2,?3,?4)",params![&selected,version,marker.number.to_be_bytes(),crate::sequence::applied(db)?.to_be_bytes()])?;
44    }
45    crate::pending::end_captured_raise_waits(db, marker.number)?;
46    Ok(true)
47}
48
49/// The creator's raise, as the store it created comes into being. A breaking
50/// migration that ran before the store existed captured nothing, as there
51/// was no shared version to raise; the store's history records the version
52/// its creating app runs only once this device raises it, through the marker
53/// that migration queued (§18.3).
54pub(crate) fn capture_created(
55    db: &DatabaseConnection,
56    schema: &WriteSchema,
57) -> Result<(), DbError> {
58    if breaking_version(db)? == 0 || !provisional(db)?.is_empty() {
59        return Ok(());
60    }
61    let device = db
62        .local_device()
63        .ok_or(DbError::Invariant { what: "a writer connection observes this device" })?;
64    let queued = db.query(
65        "SELECT number FROM _coven_uploads WHERE device=?1 ORDER BY number DESC",
66        [device],
67        |r| counter(r.get::<_, Vec<u8>>(0)?),
68    )?;
69    for number in queued {
70        let marker = WriteId { device, number };
71        let header = crate::upload::queued(db, marker)?
72            .ok_or(DbError::Invariant { what: "a queued write has its header" })?
73            .header;
74        if header.disposition == coven_format::write::WriteDisposition::Migration {
75            let version = db.schema_version()?;
76            if capture(db, schema, marker, version)? {
77                publish_at(db, version)?;
78            }
79            return Ok(());
80        }
81    }
82    Ok(())
83}
84
85/// Raise a circle this device's member just made, when the base its creation
86/// recorded is below this installation's newest breaking migration: it was
87/// prepared before that migration and applied after it, so the migration's
88/// capture didn't include it. The circle has no rows yet, so this device holds
89/// its state at its own version, and its later writes wait for the raise as
90/// a migration's do. A device that starts reading a circle with history holds
91/// none of it at the circle's version, and waits for a device that held it
92/// (§15.5, §18.3).
93pub(crate) fn capture_made(
94    db: &DatabaseConnection,
95    circle: coven_foundation::id_source::CircleId,
96    state: &crate::StoreState,
97    operation: Option<&crate::migration::MigrationOperation>,
98) -> Result<(), DbError> {
99    let audience = Audience::Circle(circle);
100    let device = db
101        .local_device()
102        .ok_or(DbError::Invariant { what: "a writer connection observes this device" })?;
103    let made = state
104        .devices
105        .get(&device)
106        .is_some_and(|local| local.active() && state.reads_circle(circle, &local.member));
107    if !made || state.reached(&audience) >= breaking_version(db)? {
108        return Ok(());
109    }
110    // Every capture holds the same prefix; with none, the prefix is every
111    // write this device has made so far.
112    let through = match captured_through(db)? {
113        Some(through) => through,
114        None => crate::write_metadata::newest_number(db, device)?.unwrap_or(0),
115    };
116    db.internal_execute(
117        "INSERT INTO _coven_raise_captures(audience,version,through,covers) VALUES(?1,?2,?3,?4)",
118        params![
119            &audience,
120            db.schema_version()?,
121            through.to_be_bytes(),
122            crate::sequence::applied(db)?.to_be_bytes()
123        ],
124    )?;
125    crate::pending::end_captured_raise_waits(db, through)?;
126    // The migration's journal may have finished with its own raises; passes
127    // publish captures while one is open.
128    if let Some(operation) = operation {
129        if crate::operation::of_kind(db, crate::operation_kind::SCHEMA_RAISE)?.is_empty() {
130            journal(db, operation, db.schema_version()?)?;
131        }
132    }
133    Ok(())
134}
135
136/// Journal the operation through which passes publish this installation's
137/// raise captures at `version`, at its first step.
138pub(crate) fn journal(
139    db: &DatabaseConnection,
140    operation: &crate::migration::MigrationOperation,
141    version: u32,
142) -> Result<(), DbError> {
143    let new = operation(version)?;
144    let id = crate::operation::insert(db, &new)?;
145    crate::operation::advance(
146        db,
147        &crate::OperationUpdate { id, previous: 0, last_step: 1, data: new.data },
148    )
149}
150
151pub(crate) fn raise_to(
152    db: &DatabaseConnection,
153    _tables: &[crate::SyncedTable],
154    device: DeviceId,
155    version: u32,
156) -> Result<bool, DbError> {
157    if captures(db, device)?.is_empty() {
158        return Ok(false);
159    }
160    // A capture is made through one of this device's writes, which stays.
161    let newest = crate::write_metadata::newest_number(db, device)?
162        .ok_or(DbError::Invariant { what: "a raise capture's device has a write" })?;
163    db.internal_execute(
164        "UPDATE _coven_raise_captures SET version=?1,through=?2,covers=?3",
165        params![version, newest.to_be_bytes(), crate::sequence::applied(db)?.to_be_bytes()],
166    )?;
167    crate::pending::end_captured_raise_waits(db, newest)?;
168    Ok(true)
169}
170
171pub(crate) fn prepare(
172    db: &DatabaseConnection,
173    _schema: &WriteSchema,
174    device: DeviceId,
175    selected: &Audience,
176    version: u32,
177) -> Result<Option<RaiseCapture>, DbError> {
178    db.internal_execute(
179        "UPDATE _coven_raise_captures SET covers=?1 WHERE audience=?2 AND version=?3",
180        params![crate::sequence::applied(db)?.to_be_bytes(), selected, version],
181    )?;
182    Ok(captures(db, device)?
183        .into_iter()
184        .find(|capture| capture.audience == *selected && capture.version == version))
185}
186
187/// This device's first queued write that waits for a raise: one made at or
188/// after its newest breaking migration, with a part in an audience that
189/// hasn't reached that migration's version and can still be raised
190/// ([`crate::StoreState::awaits_raise`]). Every device would refuse that
191/// part until the audience has, so it waits, and the writes behind it with it
192/// (§18.5). A raise's captured prefix goes out before its raise; while one
193/// waits, it holds everything after it already.
194pub(crate) fn first_ahead(
195    db: &DatabaseConnection,
196    device: DeviceId,
197) -> Result<Option<(u64, Audience)>, DbError> {
198    let breaking = breaking_version(db)?;
199    if breaking == 0 || captured_through(db)?.is_some() {
200        return Ok(None);
201    }
202    let state = crate::store_state_tables::read(db)?;
203    let below = |audience: &Audience| state.awaits_raise(audience, breaking);
204    if !below(&Audience::Store) && !state.circles.keys().any(|c| below(&Audience::Circle(*c))) {
205        return Ok(None);
206    }
207    let queued = db.query(
208        "SELECT number FROM _coven_uploads WHERE device=?1 ORDER BY number",
209        [device],
210        |r| counter(r.get::<_, Vec<u8>>(0)?),
211    )?;
212    for number in queued {
213        let write = crate::upload::queued(db, WriteId { device, number })?
214            .ok_or(DbError::Invariant { what: "a queued write has its header" })?;
215        if write.header.schema_version >= breaking {
216            if let Some(part) = write.parts.iter().find(|part| below(&part.audience)) {
217                return Ok(Some((number, part.audience.clone())));
218            }
219        }
220    }
221    Ok(None)
222}
223
224/// Record why the queue stops at a write ahead of an audience's version, as
225/// [`first_ahead`] finds it (§18.5).
226pub(crate) fn record_ahead<P: crate::PendingValue>(
227    db: &DatabaseConnection,
228    device: DeviceId,
229) -> Result<(), DbError> {
230    if let Some((number, audience)) = first_ahead(db, device)? {
231        let write = WriteId { device, number };
232        crate::pending::save(db, &P::raise_wait(write, audience, breaking_version(db)?))?;
233    }
234    Ok(())
235}
236
237/// End every raise wait the installed store state has reached or whose
238/// circle it deleted, in the commit that installs it; a write still below
239/// another audience's version records that wait instead.
240pub(crate) fn end_raise_waits<P: crate::PendingValue>(
241    db: &DatabaseConnection,
242) -> Result<(), DbError> {
243    let state = crate::store_state_tables::read(db)?;
244    for record in crate::pending::read::<P>(db)? {
245        if let Some((write, audience, version)) = record.raise_awaited() {
246            if !state.awaits_raise(audience, version) {
247                crate::pending::clear(db, record.subject())?;
248                record_ahead::<P>(db, write.device)?;
249            }
250        }
251    }
252    Ok(())
253}
254
255pub(crate) fn captured_through(db: &DatabaseConnection) -> Result<Option<u64>, DbError> {
256    db.query_row("SELECT max(through) FROM _coven_raise_captures", [], |r| {
257        r.get::<_, Option<Vec<u8>>>(0)?.map(counter).transpose()
258    })
259}
260
261pub(crate) fn provisional(db: &DatabaseConnection) -> Result<Vec<(Audience, u32, u64)>, DbError> {
262    db.query(
263        "SELECT audience,version,covers FROM _coven_raise_captures ORDER BY audience",
264        [],
265        |r| Ok((audience(r, 0)?, r.get(1)?, counter(r.get(2)?)?)),
266    )
267}
268
269pub(crate) fn captures(
270    db: &DatabaseConnection,
271    device: DeviceId,
272) -> Result<Vec<RaiseCapture>, DbError> {
273    db.query(
274        "SELECT audience,version,through,covers FROM _coven_raise_captures ORDER BY audience",
275        [],
276        |r| {
277            Ok(RaiseCapture {
278                audience: audience(r, 0)?,
279                version: r.get(1)?,
280                through: WriteId { device, number: counter(r.get(2)?)? },
281                covers: counter(r.get(3)?)?,
282            })
283        },
284    )
285}
286
287pub(crate) fn write<E>(
288    db: &DatabaseConnection,
289    schema: &WriteSchema,
290    id: SnapshotId,
291    version: u32,
292    begin: impl FnOnce(&SnapshotHeader) -> Result<(), E>,
293    emit: impl FnMut(Vec<u8>) -> Result<(), E>,
294) -> Result<u32, crate::SnapshotWriteError<E>> {
295    let capture = captures(db, id.device)?
296        .into_iter()
297        .find(|capture| capture.audience == id.audience && capture.version == version)
298        .ok_or(DbError::DamagedDatabase)?;
299    if crate::upload::unconfirmed(db, capture.through)?.is_some() {
300        return Err(DbError::ClaimUnsettled.into());
301    }
302    if capture.covers != crate::sequence::applied(db)? {
303        return Err(DbError::CaptureChanged.into());
304    }
305    crate::snapshot_write::write(db, schema, id, begin, emit)
306}
307
308/// The version of the newest breaking migration this installation ran, zero
309/// if none: received writes are read from it on (§18.5), and a snapshot this
310/// installation writes announces it as its breaking version (§16.3).
311pub(crate) fn breaking_version(db: &DatabaseConnection) -> Result<u32, DbError> {
312    db.query_row("SELECT minimum FROM _coven_snapshot_schema WHERE singleton=1", [], |r| r.get(0))
313}
314
315/// Read received writes from schema version `minimum` on, the version a
316/// breaking migration requires (§18.5).
317pub(crate) fn require(db: &DatabaseConnection, minimum: u32) -> Result<(), DbError> {
318    db.internal_execute(
319        "UPDATE _coven_snapshot_schema SET minimum=?1 WHERE singleton=1",
320        [minimum],
321    )?;
322    Ok(())
323}
324
325/// Wait for this installation's raise to publish at schema `version` before
326/// applying received writes past it (§18.5).
327pub(crate) fn publish_at(db: &DatabaseConnection, version: u32) -> Result<(), DbError> {
328    db.internal_execute(
329        "UPDATE _coven_snapshot_schema SET publication=?1 WHERE singleton=1",
330        [version],
331    )?;
332    Ok(())
333}
334
335/// A snapshot of `audience` loaded at schema `version` covers every capture
336/// of it at that version or before.
337pub(crate) fn loaded(
338    db: &DatabaseConnection,
339    audience: &Audience,
340    version: u32,
341) -> Result<(), DbError> {
342    db.internal_execute(
343        "DELETE FROM _coven_raise_captures WHERE audience=?1 AND version<=?2",
344        params![audience, version],
345    )?;
346    Ok(())
347}
348
349pub(crate) fn discard(
350    db: &DatabaseConnection,
351    audience: &Audience,
352    version: u32,
353) -> Result<(), DbError> {
354    db.internal_execute(
355        "DELETE FROM _coven_raise_captures WHERE audience=?1 AND version=?2",
356        params![audience, version],
357    )?;
358    Ok(())
359}
360
361#[cfg(test)]
362#[path = "raise_capture_tests.rs"]
363pub(crate) mod tests;
coven_database/raise_capture.rs

coven_database/
raise_capture.rs
