coven_database/
download.rs

1//! Authenticated, decoded writes at the sync/database boundary.
2use crate::sqlite::DatabaseConnection;
3use crate::DbError;
4use coven_format::write::WriteHeader;
5#[cfg(test)]
6use coven_format::write::{WritePart, WriteRecord};
7use coven_merge::Audience;
8use std::collections::BTreeSet;
9
10/// A write whose signature and opened parts sync has already authenticated.
11#[derive(Clone, Debug)]
12#[cfg(test)]
13pub(crate) struct DownloadedWrite {
14    /// The authenticated write header.
15    pub header: WriteHeader,
16    /// Every part, including ones the receiving device could not decrypt.
17    pub parts: Vec<DownloadedPart>,
18}
19
20/// Readable and unreadable parts belong to the same atomic slot.
21#[derive(Clone, Debug)]
22#[cfg(test)]
23pub(crate) enum DownloadedPart {
24    /// Authenticated plaintext changes for an audience.
25    Opened(WritePart),
26    /// The device does not have this audience's key.
27    Skipped(Audience),
28}
29
30#[cfg(test)]
31impl From<WriteRecord> for DownloadedWrite {
32    fn from(record: WriteRecord) -> Self {
33        Self {
34            header: record.header,
35            parts: record.parts.into_iter().map(DownloadedPart::Opened).collect(),
36        }
37    }
38}
39
40/// Applying a previously applied write never runs it again.
41#[derive(Clone, Debug, PartialEq, Eq)]
42pub enum ApplyOutcome {
43    /// All opened parts and all merge metadata committed together.
44    Applied,
45    /// This slot had already been consumed, including skipped or lost parts.
46    AlreadyApplied,
47    /// Nothing applied; sync retains the write until this prerequisite changes.
48    Waiting(WriteWait),
49}
50
51/// A prerequisite for applying a downloaded write.
52#[derive(Clone, Debug, PartialEq, Eq)]
53pub enum WriteWait {
54    /// These original row changes are unavailable even if their slots were consumed.
55    Sources(Vec<coven_format::write_sources::ChangeId>),
56    /// The app must support this schema version.
57    SchemaVersion(u32),
58    /// The local breaking migration awaits publication and snapshot reload,
59    /// and an opened part's audience has no capture to judge it under.
60    SchemaPublication(u32),
61    /// The selected boundary has not committed its replacement data.
62    Reload(Audience),
63}
64
65pub(crate) fn source_wait(result: Result<ApplyOutcome, DbError>) -> Result<ApplyOutcome, DbError> {
66    match result {
67        Err(DbError::WriteWaiting(wait)) => Ok(ApplyOutcome::Waiting(wait)),
68        Err(DbError::MissingWriteSources(sources)) => {
69            Ok(ApplyOutcome::Waiting(WriteWait::Sources(sources)))
70        }
71        other => other,
72    }
73}
74
75/// What the write `header`, whose `opened` parts this device reads, waits
76/// for before it applies. Its prefix needs no check: its slot holds its
77/// `read_through` to the slot's `after`, which the caller has applied.
78pub(crate) fn prerequisite(
79    database: &DatabaseConnection,
80    header: &WriteHeader,
81    opened: &[Audience],
82) -> Result<Option<WriteWait>, DbError> {
83    if header.schema_version > database.schema_version()? {
84        return Ok(Some(WriteWait::SchemaVersion(header.schema_version)));
85    }
86    let publication: Option<u32> = database.query_row(
87        "SELECT (SELECT publication FROM _coven_snapshot_schema WHERE EXISTS(SELECT 1 FROM _coven_members))",
88        [], |r| r.get(0),
89    )?;
90    if let Some(publication) = publication.filter(|version| *version > 0) {
91        let applied = crate::write_boundary::WriteBoundary::load(database)?
92            .into_iter()
93            .filter_map(|b| match b.rule {
94                crate::write_boundary::BoundaryRule::Raise(version)
95                    if b.audience == Audience::Store =>
96                {
97                    Some(version)
98                }
99                _ => None,
100            })
101            .max()
102            .unwrap_or(0);
103        // Before its raise is kept, a migrated device judges a received
104        // write under its captures; a part in an audience without one waits.
105        if publication > applied {
106            let captured: BTreeSet<_> = crate::raise_capture::provisional(database)?
107                .into_iter()
108                .map(|(audience, _, _)| audience)
109                .collect();
110            if !opened.iter().all(|audience| captured.contains(audience)) {
111                return Ok(Some(WriteWait::SchemaPublication(publication)));
112            }
113        }
114    }
115    Ok(None)
116}
117
118#[cfg(test)]
119#[path = "download_tests.rs"]
120mod tests;
coven_database/download.rs

coven_database/
download.rs
