coven_database/
pending.rs

1//! Typed persistence for sync-owned pending values, without depending on providers.
2
3use crate::{sqlite::DatabaseConnection, DbError, OperationId};
4use coven_foundation::id_source::{DeviceId, FileId, ObjectId};
5use serde::{de::DeserializeOwned, Serialize};
6
7/// Which of this device's work on a file's stored bytes a record is about.
8#[derive(Clone, Copy, Debug, PartialEq, Eq)]
9pub enum FileWork {
10    /// Its upload of them, and its reads of their staged original.
11    Upload,
12    /// Its downloads of them, for the app's pins, whose failures alone are
13    /// recorded. A download's record lasts only while a file over the bytes
14    /// does (§19.1).
15    Download,
16}
17
18/// Sync supplies its complete typed record; the database owns its atomic lifetime.
19/// The associated subject is the stable key, independent of the reason.
20pub trait PendingValue: Serialize + DeserializeOwned + Send + 'static {
21    /// Report a batch consumed after a preceding removal or key replacement.
22    fn writes_refused(
23        slot: coven_merge::Slot,
24        cutoff: coven_merge::Slot,
25        cause: crate::WriteCutoff,
26    ) -> Self;
27    /// Report an audience whose selected boundary has no committed data input.
28    fn reload(audience: coven_merge::Audience) -> Self;
29    /// Report this device's queued write held until `audience` is raised to
30    /// `version` (§18.5).
31    fn raise_wait(
32        write: coven_merge::WriteId,
33        audience: coven_merge::Audience,
34        version: u32,
35    ) -> Self;
36    /// The held write, and the audience and version it waits for a raise to,
37    /// if this record is such a wait: a store state that reaches it ends the
38    /// record as it installs.
39    fn raise_awaited(&self) -> Option<(coven_merge::WriteId, &coven_merge::Audience, u32)>;
40    /// Report one account's unresolved intended sharing change.
41    fn provider_access(operation: OperationId, account: String, grant: bool) -> Self;
42    /// Stable identity, ordered as the app presents it.
43    type Subject: Serialize + Ord;
44    /// The work that cannot advance.
45    fn subject(&self) -> &Self::Subject;
46    /// The journal row, if this record describes an operation.
47    fn operation(&self) -> Option<OperationId>;
48    /// The stored object this record describes, by its uploader and id,
49    /// and which work on it, if any: how coven's views join a record to
50    /// the rows whose file it is.
51    fn object(&self) -> Option<(FileWork, DeviceId, ObjectId)>;
52    /// The move this record describes, if any: cancelling the move ends it.
53    fn move_of(&self) -> Option<crate::MoveId>;
54    /// The file whose pin coven keeps this record describes, if any: the
55    /// record ends with the pin request.
56    fn pin_of(&self) -> Option<FileId>;
57    /// Sequence slot whose progress clears this observation.
58    fn slot(&self) -> Option<coven_merge::Slot>;
59    /// Whether a slot coverage including this object without applying it, as a
60    /// loaded snapshot's do, also end this observation: true for a write that
61    /// waits, false for a verdict on its bytes, which loading never clears (§19.1).
62    fn ends_when_covered(&self) -> bool;
63    /// The build of coven that judges these records, which every record
64    /// saved names: one draws its own at each compilation, so a change to the
65    /// code that judges gives another (§19.1).
66    const BUILD: u128;
67    /// Whether automatic work can retry, given the build that recorded this
68    /// record and the app's installed schema version.
69    fn can_retry(&self, recorded_build: u128, schema_version: u32) -> bool;
70    /// The text of the error behind this record, which `_coven_pending`
71    /// keeps beside it in the same write: none when no error was met in
72    /// this run, or when an app call received it.
73    fn cause(&self) -> Option<&str>;
74    /// This record as read back, with the cause its row keeps.
75    fn with_cause(self, cause: Option<String>) -> Self;
76}
77
78/// A subject as `_coven_pending.subject` keys its record: the subject's
79/// JSON, which sync's types give one spelling.
80#[derive(Clone, Debug, PartialEq, Eq)]
81pub(crate) struct SubjectKey(Vec<u8>);
82
83impl SubjectKey {
84    pub(crate) fn of<S: Serialize>(subject: &S) -> Result<Self, DbError> {
85        serde_json::to_vec(subject).map(Self).map_err(DbError::StateEncoding)
86    }
87}
88
89impl rusqlite::ToSql for SubjectKey {
90    fn to_sql(&self) -> rusqlite::Result<rusqlite::types::ToSqlOutput<'_>> {
91        self.0.to_sql()
92    }
93}
94
95impl rusqlite::types::FromSql for SubjectKey {
96    fn column_result(value: rusqlite::types::ValueRef<'_>) -> rusqlite::types::FromSqlResult<Self> {
97        Vec::column_result(value).map(Self)
98    }
99}
100
101const SELECT: &str =
102    "SELECT subject,value,operation,slot_key,work,device,object,build,move,pin,cause,id,raise_write FROM _coven_pending";
103
104struct StoredRecord {
105    /// Its place in the order the records arose.
106    position: i64,
107    subject: SubjectKey,
108    value: Vec<u8>,
109    operation: Option<i64>,
110    slot_key: Option<Vec<u8>>,
111    object: Option<(String, DeviceId, ObjectId)>,
112    /// The build of coven that recorded it.
113    build: u128,
114    moved: Option<i64>,
115    pin: Option<String>,
116    raise_write: Option<Vec<u8>>,
117    cause: Option<String>,
118}
119impl StoredRecord {
120    fn read(row: &rusqlite::Row<'_>) -> rusqlite::Result<Self> {
121        Ok(Self {
122            subject: row.get(0)?,
123            value: row.get(1)?,
124            operation: row.get(2)?,
125            slot_key: row.get(3)?,
126            object: match (
127                row.get::<_, Option<String>>(4)?,
128                row.get::<_, Option<DeviceId>>(5)?,
129                row.get::<_, Option<ObjectId>>(6)?,
130            ) {
131                (Some(work), Some(device), Some(object)) => Some((work, device, object)),
132                (None, None, None) => None,
133                _ => return Err(crate::write_encoding::damaged(4, rusqlite::types::Type::Text)),
134            },
135            build: u128::from_be_bytes(row.get(7)?),
136            moved: row.get(8)?,
137            pin: row.get(9)?,
138            cause: row.get(10)?,
139            position: row.get(11)?,
140            raise_write: row.get(12)?,
141        })
142    }
143    fn decode<P: PendingValue>(&self) -> Result<P, DbError> {
144        let record: P = serde_json::from_slice(&self.value).map_err(DbError::StateEncoding)?;
145        if SubjectKey::of(record.subject())? != self.subject
146            || record.operation().map(|id| id.0) != self.operation
147            || record.move_of().map(|id| id.0) != self.moved
148            || record.pin_of().map(|file| file.to_string()) != self.pin
149            || stored_raise_write(&record) != self.raise_write
150            || stored_slot_key(&record) != self.slot_key
151            || stored_object(&record)
152                != self
153                    .object
154                    .as_ref()
155                    .map(|(work, device, object)| (work.as_str(), *device, *object))
156        {
157            return Err(DbError::DamagedDatabase);
158        }
159        Ok(record.with_cause(self.cause.clone()))
160    }
161}
162
163/// Every record, in the order the records arose, as `coven_pending` lists
164/// them by position.
165pub(crate) fn read<P: PendingValue>(db: &DatabaseConnection) -> Result<Vec<P>, DbError> {
166    Ok(positioned(db)?.into_iter().map(|(_, record)| record).collect())
167}
168
169/// Every record with its place, in the order the records arose, as
170/// `coven_pending.position` names it.
171pub(crate) fn positioned<P: PendingValue>(
172    db: &DatabaseConnection,
173) -> Result<Vec<(i64, P)>, DbError> {
174    db.query(&format!("{SELECT} ORDER BY id"), [], StoredRecord::read)?
175        .iter()
176        .map(|stored| Ok((stored.position, stored.decode::<P>()?)))
177        .collect()
178}
179
180pub(crate) fn save<P: PendingValue>(db: &DatabaseConnection, record: &P) -> Result<(), DbError> {
181    db.pending_change(|| {
182        let subject = SubjectKey::of(record.subject())?;
183        let value = serde_json::to_vec(record).map_err(DbError::StateEncoding)?;
184        let (work, device, object) = match stored_object(record) {
185            Some((work, device, object)) => (Some(work), Some(device), Some(object)),
186            None => (None, None, None),
187        };
188        // A download of bytes no file over them names any more records
189        // nothing: no pin or read can try them again. The cause is the
190        // record's, written with it.
191        db.internal_execute(
192            "INSERT INTO _coven_pending(subject,value,operation,slot_key,work,device,object,build,move,pin,cause,raise_write)
193             SELECT ?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12
194             WHERE ?5 IS NOT 'download' OR EXISTS(SELECT 1 FROM _coven_stored_files AS s WHERE s.device=?6 AND s.object=?7)
195             ON CONFLICT(subject) DO UPDATE SET value=excluded.value,operation=excluded.operation,slot_key=excluded.slot_key,work=excluded.work,device=excluded.device,object=excluded.object,build=excluded.build,move=excluded.move,pin=excluded.pin,cause=excluded.cause,raise_write=excluded.raise_write
196             WHERE value!=excluded.value OR build!=excluded.build OR cause IS NOT excluded.cause",
197            rusqlite::params![subject, value, record.operation().map(|id| id.0), stored_slot_key(record), work, device, object, P::BUILD.to_be_bytes(), record.move_of(), record.pin_of(), record.cause(), stored_raise_write(record)],
198        )?;
199        Ok(())
200    })
201}
202
203/// Opening clears every record's cause: a cause names an error the run that
204/// wrote it met, and a later run has met none of its own yet (§19.1). It
205/// runs before any app query observes the database.
206pub(crate) fn forget_causes(db: &DatabaseConnection) -> Result<(), DbError> {
207    db.internal_execute("UPDATE _coven_pending SET cause=NULL WHERE cause IS NOT NULL", [])?;
208    Ok(())
209}
210
211pub(crate) fn clear<S: Serialize>(db: &DatabaseConnection, subject: &S) -> Result<(), DbError> {
212    clear_key(db, &SubjectKey::of(subject)?)
213}
214
215/// Clear the record of a subject already encoded for its key.
216pub(crate) fn clear_key(db: &DatabaseConnection, subject: &SubjectKey) -> Result<(), DbError> {
217    db.pending_change(|| {
218        db.internal_execute("DELETE FROM _coven_pending WHERE subject=?1", [subject])?;
219        Ok(())
220    })
221}
222
223/// Clear the record of the move `id`, whatever it says.
224pub(crate) fn clear_move(db: &DatabaseConnection, id: crate::MoveId) -> Result<(), DbError> {
225    db.pending_change(|| {
226        db.internal_execute("DELETE FROM _coven_pending WHERE move=?1", [id])?;
227        Ok(())
228    })
229}
230
231/// An operation's retry wait lives only while its condition does: advancing,
232/// finishing or an explicit retry ends both (§6.7).
233pub(crate) fn clear_operation(db: &DatabaseConnection, id: OperationId) -> Result<(), DbError> {
234    crate::retry::clear(db, crate::RetrySubject::Operation(id))?;
235    db.pending_change(|| {
236        db.internal_execute("DELETE FROM _coven_pending WHERE operation=?1", [id.0])?;
237        Ok(())
238    })
239}
240
241pub(crate) fn operation<P: PendingValue>(
242    db: &DatabaseConnection,
243    id: OperationId,
244) -> Result<Option<P>, DbError> {
245    db.query(&format!("{SELECT} WHERE operation=?1"), [id.0], StoredRecord::read)?
246        .pop()
247        .map(|r| r.decode())
248        .transpose()
249}
250
251/// Whether a subject has a local record.
252pub(crate) fn recorded<S: Serialize>(
253    db: &DatabaseConnection,
254    subject: &S,
255) -> Result<bool, DbError> {
256    db.query_row(
257        "SELECT EXISTS(SELECT 1 FROM _coven_pending WHERE subject=?1)",
258        [SubjectKey::of(subject)?],
259        |r| r.get(0),
260    )
261}
262
263/// A subject's stored record, if it has one.
264fn stored<S: Serialize>(
265    db: &DatabaseConnection,
266    subject: &S,
267) -> Result<Option<StoredRecord>, DbError> {
268    Ok(db
269        .query(
270            &format!("{SELECT} WHERE subject=?1"),
271            [SubjectKey::of(subject)?],
272            StoredRecord::read,
273        )?
274        .pop())
275}
276
277/// The current local record for a subject, if any.
278pub(crate) fn find<P: PendingValue>(
279    db: &DatabaseConnection,
280    subject: &P::Subject,
281) -> Result<Option<P>, DbError> {
282    stored(db, subject)?.map(|stored| stored.decode()).transpose()
283}
284
285pub(crate) fn retry<P: PendingValue>(
286    db: &DatabaseConnection,
287    subject: &P::Subject,
288) -> Result<bool, DbError> {
289    match stored(db, subject)? {
290        None => Ok(true),
291        Some(stored) => Ok(stored.decode::<P>()?.can_retry(stored.build, db.schema_version()?)),
292    }
293}
294
295/// Make every record one `build` recorded, as a store opened by another
296/// build of coven finds the records an earlier one wrote.
297#[cfg(any(test, feature = "test-utils"))]
298pub(crate) fn set_build(db: &DatabaseConnection, build: u128) -> Result<(), DbError> {
299    db.pending_change(|| {
300        db.internal_execute("UPDATE _coven_pending SET build=?1", [build.to_be_bytes()])?;
301        Ok(())
302    })
303}
304
305fn slot_key(slot: coven_merge::Slot, covered: bool) -> Vec<u8> {
306    let mut key = vec![u8::from(covered)];
307    key.extend(slot.get().to_be_bytes());
308    key
309}
310const COVERED_SLOT: u8 = 1;
311
312/// The record's stored object and work on it, as `_coven_pending` holds them.
313fn stored_object<P: PendingValue>(record: &P) -> Option<(&'static str, DeviceId, ObjectId)> {
314    record.object().map(|(work, device, object)| {
315        let work = match work {
316            FileWork::Upload => "upload",
317            FileWork::Download => "download",
318        };
319        (work, device, object)
320    })
321}
322
323fn stored_slot_key<P: PendingValue>(record: &P) -> Option<Vec<u8>> {
324    Some(slot_key(record.slot()?, record.ends_when_covered()))
325}
326
327/// The number of the own write a raise wait holds, as `_coven_pending`
328/// keys it.
329fn stored_raise_write<P: PendingValue>(record: &P) -> Option<Vec<u8>> {
330    record.raise_awaited().map(|(write, _, _)| write.number.to_be_bytes().to_vec())
331}
332
333/// End the raise wait of every own write through `through`, in the commit
334/// that captures those writes into a raise's prefix: they go out before the
335/// raise, and wait for none (§18.5).
336pub(crate) fn end_captured_raise_waits(
337    db: &DatabaseConnection,
338    through: u64,
339) -> Result<(), DbError> {
340    db.pending_change(|| {
341        db.internal_execute(
342            "DELETE FROM _coven_pending WHERE raise_write IS NOT NULL AND raise_write<=?1",
343            [through.to_be_bytes().as_slice()],
344        )?;
345        Ok(())
346    })
347}
348
349/// An applied reset clears data-related observations permanently (§19.4).
350pub(crate) fn reset(db: &DatabaseConnection) -> Result<(), DbError> {
351    db.pending_change(|| {
352        db.internal_execute("DELETE FROM _coven_pending AS p WHERE NOT EXISTS(SELECT 1 FROM _coven_provider_access a WHERE a.operation=p.operation)",[])?;
353        Ok(())
354    })
355}
356
357/// Kept progress removes only this installation's condition, in its transaction.
358pub(crate) fn clear_slot(db: &DatabaseConnection, slot: coven_merge::Slot) -> Result<(), DbError> {
359    db.pending_change(|| {
360        for covered in [false, true] {
361            db.internal_execute(
362                "DELETE FROM _coven_pending WHERE slot_key=?1",
363                [slot_key(slot, covered)],
364            )?;
365        }
366        Ok(())
367    })
368}
369
370/// Snapshot coverage that passes a waiting write ends that
371/// wait in the same transaction. Verdicts on a write's bytes remain (§19.1).
372pub(crate) fn clear_covered_slots(db: &DatabaseConnection) -> Result<(), DbError> {
373    let waiting = db.query(
374        "SELECT slot_key FROM _coven_pending WHERE slot_key>=?1 AND slot_key<?2",
375        ([COVERED_SLOT].as_slice(), [COVERED_SLOT + 1].as_slice()),
376        |row| row.get::<_, Vec<u8>>(0),
377    )?;
378    if waiting.is_empty() {
379        return Ok(());
380    }
381    let applied = crate::sequence::applied(db)?;
382    db.pending_change(|| {
383        for key in waiting {
384            let slot = u64::from_be_bytes(key[1..].try_into().expect("eight-byte slot"));
385            if slot <= applied {
386                db.internal_execute("DELETE FROM _coven_pending WHERE slot_key=?1", [key])?;
387            }
388        }
389        Ok(())
390    })
391}
392
393#[cfg(test)]
394#[path = "pending_tests.rs"]
395pub(crate) mod tests;
coven_database/pending.rs

coven_database/
pending.rs
