coven_database/
operation.rs

1//! Durable operation records. Sync owns kinds and interprets their data (§11).
2
3use crate::operation_kind::INVITE;
4use crate::{sqlite::DatabaseConnection, DbError};
5
6/// The local, never reused identity of an operation.
7#[derive(
8    Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
9)]
10pub struct OperationId(pub i64);
11
12/// A journal row read as one committed value.
13#[derive(Clone)]
14pub struct OperationRecord {
15    /// Its local identity.
16    pub id: OperationId,
17    /// The kind's stable name, defined by its sync implementation.
18    pub kind: String,
19    /// Last completed step; zero precedes the first step.
20    pub last_step: u32,
21    /// The kind's encoded state, including fixed identities and causal views.
22    pub data: Vec<u8>,
23    /// The initiating app method or `coven`.
24    pub started_by: String,
25}
26
27impl OperationRecord {
28    /// The step past this row, recording `data`, guarded by this row's step
29    /// against a stale caller. An operation's steps fitting u32 is an
30    /// invariant.
31    pub fn next_step(&self, data: Vec<u8>) -> Result<OperationUpdate, crate::DbError> {
32        Ok(OperationUpdate {
33            id: self.id,
34            previous: self.last_step,
35            last_step: self
36                .last_step
37                .checked_add(1)
38                .ok_or(crate::DbError::Invariant { what: "an operation's steps fit u32" })?,
39            data,
40        })
41    }
42}
43
44impl std::fmt::Debug for OperationRecord {
45    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
46        f.debug_struct("OperationRecord")
47            .field("id", &self.id)
48            .field("kind", &self.kind)
49            .field("last_step", &self.last_step)
50            .field("data", &"[REDACTED]")
51            .field("started_by", &self.started_by)
52            .finish()
53    }
54}
55
56/// The initial state committed before any external effect.
57pub struct NewOperation {
58    /// Stable kind name.
59    pub kind: String,
60    /// Encoded initial state.
61    pub data: Vec<u8>,
62    /// App method name or `coven`.
63    pub started_by: String,
64    /// App writes are refused with `DbError::MemberKeysChanging` from its
65    /// insertion until it finishes: a self-replacement of member keys
66    /// confirms its queue and installs its binding meanwhile (§12.3, §11).
67    pub pauses_writes: bool,
68}
69
70/// A step transition, guarded against stale callers.
71pub struct OperationUpdate {
72    /// The operation being advanced.
73    pub id: OperationId,
74    /// Step from which this transition was made.
75    pub previous: u32,
76    /// The completed step.
77    pub last_step: u32,
78    /// Encoded state after the step.
79    pub data: Vec<u8>,
80}
81
82pub(crate) fn insert(db: &DatabaseConnection, new: &NewOperation) -> Result<OperationId, DbError> {
83    let id = db.query_row("INSERT INTO _coven_operations(kind,last_step,data,started_by) VALUES(?1,0,?2,?3) RETURNING id",
84        (&new.kind, &new.data, &new.started_by), |r| Ok(OperationId(r.get(0)?)))?;
85    if new.pauses_writes {
86        // Finishing or discarding the operation deletes its row and with it
87        // the pause, in that same transaction.
88        db.internal_execute(
89            "INSERT INTO _coven_write_pause(paused,operation) VALUES(1,?1)",
90            [id.0],
91        )?;
92    }
93    Ok(id)
94}
95
96/// Journal an operation the app's write asks for, of `kind` as sync names
97/// it: listed in `coven_operations` from this write's commit until the app
98/// acknowledges its outcome.
99pub(crate) fn insert_for_app(
100    db: &DatabaseConnection,
101    new: &NewOperation,
102    kind: &str,
103) -> Result<OperationId, DbError> {
104    let id = insert(db, new)?;
105    db.internal_execute(
106        "INSERT INTO _coven_app_operations(operation,kind) VALUES(?1,?2)",
107        (id.0, kind),
108    )?;
109    Ok(id)
110}
111
112/// List the outcome of an operation the app's write didn't journal, as one
113/// it did: in `coven_operations` until the app acknowledges it. An invite
114/// ends this way when a refusal only a person can fix ends it (§13.2).
115pub(crate) fn list_outcome(
116    db: &DatabaseConnection,
117    id: OperationId,
118    kind: &str,
119    outcome: &crate::OperationOutcome,
120) -> Result<(), DbError> {
121    db.internal_execute(
122        "INSERT INTO _coven_app_operations(operation,kind,outcome) VALUES(?1,?2,?3)",
123        (id.0, kind, outcome.encode()),
124    )?;
125    Ok(())
126}
127
128/// End the app's record of a finished operation: it leaves
129/// `coven_operations`. One still running fails with `OperationRunning`;
130/// one this device doesn't list, never journaled by the app or
131/// acknowledged already, with `OperationUnknown`.
132pub(crate) fn acknowledge(db: &DatabaseConnection, id: OperationId) -> Result<(), DbError> {
133    let finished: Option<bool> = db
134        .query(
135            "SELECT outcome IS NOT NULL FROM _coven_app_operations WHERE operation=?1",
136            [id.0],
137            |row| row.get(0),
138        )?
139        .pop();
140    match finished {
141        None => Err(DbError::OperationUnknown(id)),
142        Some(false) => Err(DbError::OperationRunning(id)),
143        Some(true) => {
144            db.internal_execute("DELETE FROM _coven_app_operations WHERE operation=?1", [id.0])?;
145            Ok(())
146        }
147    }
148}
149
150/// Whether a self-replacement pauses this device's writes (§12.3).
151pub(crate) fn writes_paused(db: &DatabaseConnection) -> Result<bool, DbError> {
152    db.query_row("SELECT EXISTS(SELECT 1 FROM _coven_write_pause WHERE paused=1)", [], |r| r.get(0))
153}
154
155/// Refuse an app write while a self-replacement pauses writes (§12.3).
156pub(crate) fn check_write_pause(db: &DatabaseConnection) -> Result<(), DbError> {
157    if writes_paused(db)? {
158        return Err(DbError::MemberKeysChanging);
159    }
160    Ok(())
161}
162
163/// Reserve the next number of this device's snapshot sequence. Every
164/// preparation takes a fresh one, so an abandoned attempt's path is never
165/// reused, and §19.3's identity check compares stored snapshots with the
166/// greatest number reserved, published or not.
167pub(crate) fn reserve_snapshot(db: &DatabaseConnection) -> Result<std::num::NonZeroU64, DbError> {
168    let number = db.query_row(
169        "UPDATE _coven_snapshot_reservation SET number=number+1 WHERE singleton=1 RETURNING number",
170        [],
171        |r| r.get::<_, i64>(0),
172    )?;
173    u64::try_from(number).ok().and_then(std::num::NonZeroU64::new).ok_or(DbError::DamagedDatabase)
174}
175
176pub(crate) fn read(db: &DatabaseConnection) -> Result<Vec<OperationRecord>, DbError> {
177    db.query(
178        "SELECT id,kind,last_step,data,started_by FROM _coven_operations ORDER BY id",
179        [],
180        record,
181    )
182}
183
184pub(crate) fn of_kind(
185    db: &DatabaseConnection,
186    kind: &str,
187) -> Result<Vec<OperationRecord>, DbError> {
188    db.query(
189        "SELECT id,kind,last_step,data,started_by FROM _coven_operations WHERE kind=?1 ORDER BY id",
190        [kind],
191        record,
192    )
193}
194
195pub(crate) fn find(
196    db: &DatabaseConnection,
197    id: OperationId,
198) -> Result<Option<OperationRecord>, DbError> {
199    Ok(db
200        .query(
201            "SELECT id,kind,last_step,data,started_by FROM _coven_operations WHERE id=?1",
202            [id.0],
203            record,
204        )?
205        .pop())
206}
207
208pub(crate) fn first(
209    db: &DatabaseConnection,
210    kind: &str,
211) -> Result<Option<OperationRecord>, DbError> {
212    Ok(db.query(
213        "SELECT id,kind,last_step,data,started_by FROM _coven_operations WHERE kind=?1 ORDER BY id LIMIT 1",
214        [kind],
215        record,
216    )?.pop())
217}
218
219fn record(row: &rusqlite::Row<'_>) -> rusqlite::Result<OperationRecord> {
220    Ok(OperationRecord {
221        id: OperationId(row.get(0)?),
222        kind: row.get(1)?,
223        last_step: row.get(2)?,
224        data: row.get(3)?,
225        started_by: row.get(4)?,
226    })
227}
228
229pub(crate) fn advance(db: &DatabaseConnection, update: &OperationUpdate) -> Result<(), DbError> {
230    let changed = db.internal_execute(
231        "UPDATE _coven_operations SET last_step=?1,data=?2 WHERE id=?3 AND last_step=?4",
232        (&update.last_step, &update.data, update.id.0, update.previous),
233    )?;
234    if changed != 1 {
235        return Err(DbError::OperationChanged(update.id));
236    }
237    crate::pending::clear_operation(db, update.id)
238}
239
240/// Put the operation `id` at `step`, keeping its data, unguarded by the step
241/// it was at: a provider-access request moves between its steps as the
242/// account's row says (§14.2).
243pub(crate) fn set_step(db: &DatabaseConnection, id: OperationId, step: u32) -> Result<(), DbError> {
244    db.internal_execute(
245        "UPDATE _coven_operations SET last_step=?2 WHERE id=?1 AND last_step!=?2",
246        (id.0, step),
247    )?;
248    Ok(())
249}
250
251/// End the running operation `id` with `outcome`, which `coven_operations`
252/// lists for an operation the app's write journaled. An outcome is final:
253/// an operation that no longer runs fails with `OperationChanged` and keeps
254/// the outcome it ended with (§11.4). The outcome is written first, as the
255/// journal row's end requires; the caller's transaction undoes it when the
256/// operation had already ended.
257pub(crate) fn finish_with_outcome(
258    db: &DatabaseConnection,
259    id: OperationId,
260    outcome: &crate::OperationOutcome,
261) -> Result<(), DbError> {
262    db.internal_execute(
263        "UPDATE _coven_app_operations SET outcome=?2 WHERE operation=?1",
264        (id.0, outcome.encode()),
265    )?;
266    if finish(db, id)? != 1 {
267        return Err(DbError::OperationChanged(id));
268    }
269    Ok(())
270}
271
272/// Delete the row and its condition, returning how many rows were deleted.
273/// An operation whose claim is still open never ends: nothing would finish
274/// that claim, and the device, one claim at a time, would claim no slot
275/// again (§5.1).
276pub(crate) fn finish(db: &DatabaseConnection, id: OperationId) -> Result<usize, DbError> {
277    if crate::claim::held_by(db, id)? {
278        return Err(DbError::Invariant { what: "an operation ends after its claim" });
279    }
280    crate::pending::clear_operation(db, id)?;
281    db.internal_execute("DELETE FROM _coven_operations WHERE id=?1", [id.0])
282}
283
284/// End every operation an app's write journaled that still runs, and every
285/// invite this device made, once the applied state this transaction commits
286/// stops this installation for good: it removes or replaces this device,
287/// removes its member, or binds its member to a pair it doesn't hold (§12.3,
288/// §14.1). None of them would ever run (`end_for_good`).
289pub(crate) fn end_on_stop(db: &DatabaseConnection) -> Result<(), DbError> {
290    let (running, invites) = journaled_for_good(db)?;
291    if running.is_empty() && invites.is_empty() {
292        return Ok(());
293    }
294    match stopped(db)? {
295        Some(stop) => end(db, stop, running, invites),
296        None => Ok(()),
297    }
298}
299
300/// End every operation an app's write journaled that still runs, and every
301/// invite this device made, for an installation stopped for good by `stop`.
302/// The operation whose claim the applied slots record as landed, the removal
303/// itself when this device claimed it, ends landed at its slot; every other
304/// ends stopped, naming the stop: nothing refused it. An invite ends with the operations it started, its
305/// approval's binding and its access record: their journal rows, open claim
306/// and pending records go, and it leaves `coven_invites` (§13.2).
307pub(crate) fn end_for_good(
308    db: &DatabaseConnection,
309    stop: crate::DeviceStop,
310) -> Result<(), DbError> {
311    let (running, invites) = journaled_for_good(db)?;
312    end(db, stop, running, invites)
313}
314
315/// The app's operations still running, and the journal rows of invites and
316/// of the operations invites started.
317fn journaled_for_good(
318    db: &DatabaseConnection,
319) -> Result<(Vec<OperationId>, Vec<OperationId>), DbError> {
320    let running = db.query(
321        "SELECT operation FROM _coven_app_operations WHERE outcome IS NULL ORDER BY operation",
322        [],
323        |r| Ok(OperationId(r.get(0)?)),
324    )?;
325    let invites = db.query(
326        "SELECT id FROM _coven_operations WHERE kind=?1 OR started_by=?1 ORDER BY id",
327        [INVITE],
328        |r| Ok(OperationId(r.get(0)?)),
329    )?;
330    Ok((running, invites))
331}
332
333fn end(
334    db: &DatabaseConnection,
335    stop: crate::DeviceStop,
336    running: Vec<OperationId>,
337    invites: Vec<OperationId>,
338) -> Result<(), DbError> {
339    let landed = crate::claim::landed_operation(db)?;
340    for id in running {
341        let outcome = match landed {
342            Some((entry, slot)) if entry == id => crate::OperationOutcome::Landed(slot),
343            _ => crate::OperationOutcome::Stopped(stop),
344        };
345        crate::claim::drop_operation(db, id)?;
346        finish_with_outcome(db, id, &outcome)?;
347    }
348    for id in invites {
349        crate::claim::drop_operation(db, id)?;
350        finish(db, id)?;
351    }
352    Ok(())
353}
354
355/// How the applied state stops this installation for good, if it does. One
356/// whose device isn't registered yet hasn't started.
357pub(crate) fn stopped(db: &DatabaseConnection) -> Result<Option<crate::DeviceStop>, DbError> {
358    let held: bool =
359        db.query_row("SELECT EXISTS(SELECT 1 FROM _coven_held_keys)", [], |r| r.get(0))?;
360    if !held {
361        return Err(DbError::Invariant {
362            what: "an installation whose writes journal operations records the pairs it holds",
363        });
364    }
365    let found: Option<(String, bool, bool)> = db
366        .query(
367            "SELECT d.state,m.removed,h.signing IS NOT NULL FROM _coven_devices AS d
368             JOIN _coven_members AS m ON m.member=d.member
369             LEFT JOIN _coven_held_keys AS h ON h.signing=m.signing AND h.sealing=m.sealing
370             WHERE d.device=_coven_device()",
371            [],
372            |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
373        )?
374        .pop();
375    Ok(found.and_then(|(state, removed, held)| match (state.as_str(), removed, held) {
376        ("replaced", ..) => Some(crate::DeviceStop::Replaced),
377        ("active", false, true) => None,
378        ("active", false, false) => Some(crate::DeviceStop::MemberKeysReplaced),
379        _ => Some(crate::DeviceStop::Removed),
380    }))
381}
382
383/// Record member pairs this installation holds, or is about to keep.
384pub(crate) fn hold_keys(
385    db: &DatabaseConnection,
386    keys: &[coven_crypto::MemberPublicKeys],
387) -> Result<(), DbError> {
388    for pair in keys {
389        db.internal_execute(
390            "INSERT INTO _coven_held_keys(signing,sealing) VALUES(?1,?2) ON CONFLICT DO NOTHING",
391            (pair.signing.as_bytes().as_slice(), pair.sealing.as_bytes().as_slice()),
392        )?;
393    }
394    Ok(())
395}
396
397#[cfg(test)]
398#[path = "operation_tests.rs"]
399pub(crate) mod tests;
coven_database/operation.rs

coven_database/
operation.rs
