coven_database/
snapshot_write.rs

1//! Audience snapshots stream the counted prefix from one pinned transaction.
2use crate::internal_schema::WRITE_DEVICE_SQL;
3use crate::write_encoding::{counter, decoded, slot_counter};
4use crate::{sqlite::DatabaseConnection, write_schema::WriteSchema, DbError};
5use coven_format::{
6    admin::SnapshotId,
7    merge_fields,
8    snapshot::{section, SnapshotEncoder, SnapshotHeader, SnapshotRecord, SECTIONS},
9    snapshot_rows::{AppliedWrite, SyncedColumn},
10};
11use coven_merge::{Audience, WriteId};
12use std::collections::BTreeSet;
13
14/// A snapshot failed to read, encode or reach its plaintext consumer.
15#[derive(Debug, thiserror::Error)]
16pub enum SnapshotWriteError<E> {
17    /// Reading committed data failed.
18    #[error(transparent)]
19    Database(#[from] DbError),
20    /// A record could not be encoded.
21    #[error(transparent)]
22    Format(#[from] coven_format::Error),
23    /// The consumer refused a frame.
24    #[error("snapshot output: {0}")]
25    Output(E),
26}
27
28impl<E: coven_foundation::task::CarriesBug> coven_foundation::task::CarriesBug
29    for SnapshotWriteError<E>
30{
31    fn bug(&self) -> Option<&coven_foundation::task::Bug> {
32        match self {
33            Self::Database(error) => error.bug(),
34            Self::Format(_) => None,
35            Self::Output(error) => error.bug(),
36        }
37    }
38}
39
40/// Returns the snapshot's breaking version, read in the transaction that
41/// reads its schema version (§16.3).
42pub(crate) fn write<E>(
43    db: &DatabaseConnection,
44    schema: &WriteSchema,
45    id: SnapshotId,
46    begin: impl FnOnce(&SnapshotHeader) -> Result<(), E>,
47    mut emit: impl FnMut(Vec<u8>) -> Result<(), E>,
48) -> Result<u32, SnapshotWriteError<E>> {
49    let covers = crate::sequence::applied(db)?;
50    let audience = id.audience.clone();
51    let removed = crate::published::removals(db, &schema.schema, &schema.rules)?;
52    let mut counts = [0; SECTIONS];
53    let mut columns = BTreeSet::new();
54    crate::snapshot_view::rows(db, schema, &audience, |row| {
55        counts[section::MERGE] += 1;
56        columns.extend(
57            row.state.cells().keys().map(|column| (row.state.row().table.clone(), column.clone())),
58        );
59        Ok::<_, DbError>(())
60    })?;
61    crate::snapshot_view::losses(db, &removed, &audience, |loss| {
62        counts[section::LOSS] += 1;
63        if let coven_format::loss::LossValues::Cell { column, .. } = &loss.values {
64            columns.insert((loss.row.table.clone(), column.clone()));
65        }
66        Ok::<_, DbError>(())
67    })?;
68    counts[section::WRITE] = db.query_row(
69        "SELECT count(*) FROM _coven_writes WHERE slot<=?1",
70        [covers.to_be_bytes()],
71        |r| r.get::<_, i64>(0).map(|n| n as u64),
72    )?;
73    counts[section::COLUMN] = columns.len() as u64;
74    crate::write_sources::visit_evidence(db, &audience, |_| {
75        counts[section::AUTHORED] += 1;
76        Ok::<_, DbError>(())
77    })?;
78    crate::snapshot_view::creations(db, &audience, |_| {
79        counts[section::CREATION] += 1;
80        Ok::<_, DbError>(())
81    })?;
82    let evidence = crate::authorship::evidence(
83        db,
84        schema,
85        &audience,
86        crate::authorship::Author {
87            device: id.device,
88            member: crate::store_state_tables::device_member(db, id.device)?,
89        },
90    )?;
91    counts[section::EVIDENCE] = evidence.len() as u64;
92    if audience == Audience::Store {
93        for (at, kind) in [
94            (section::ADMIN, "admin"),
95            (section::KEY_COPY, "key"),
96            (section::ANNOUNCEMENT, "announcement"),
97        ] {
98            counts[at] = db.query_row(
99                "SELECT count(*) FROM _coven_slot_records WHERE kind=?1 AND slot<=?2",
100                rusqlite::params![kind, covers.to_be_bytes()],
101                |r| r.get::<_, i64>(0).map(|n| n as u64),
102            )?;
103        }
104    }
105    let claims = crate::sequence::claims(db)?;
106    counts[section::CLAIM] = claims.len() as u64;
107    let breaking = crate::raise_capture::breaking_version(db)?;
108    let (mut encoder, header) = SnapshotEncoder::start(SnapshotHeader {
109        id,
110        schema_version: db.schema_version()?,
111        covers,
112        counts,
113    })?;
114    begin(encoder.header()).map_err(SnapshotWriteError::Output)?;
115    emit(header).map_err(SnapshotWriteError::Output)?;
116    let mut record = |record| emit(encoder.record(record)?).map_err(SnapshotWriteError::Output);
117    db.for_each(&format!("SELECT timestamp,number,read_through,slot FROM _coven_writes WHERE slot<=?1 ORDER BY {WRITE_DEVICE_SQL},number"),[covers.to_be_bytes()],|r|record(SnapshotRecord::Write(read_write(r).map_err(DbError::from)?)))?;
118    for (table, column) in columns {
119        record(SnapshotRecord::Column(SyncedColumn { table, column }))?;
120    }
121    crate::snapshot_view::rows(db, schema, &audience, |row| record(SnapshotRecord::Merge(row)))?;
122    crate::snapshot_view::losses(db, &removed, &audience, |loss| {
123        record(SnapshotRecord::Loss(loss))
124    })?;
125    crate::write_sources::visit_evidence(db, &audience, |authored| {
126        record(SnapshotRecord::Authored(authored))
127    })?;
128    crate::snapshot_view::creations(db, &audience, |creation| {
129        record(SnapshotRecord::Creation(creation))
130    })?;
131    for evidence in evidence {
132        record(SnapshotRecord::Evidence(evidence))?;
133    }
134    if audience == Audience::Store {
135        for kind in ["admin", "key", "announcement"] {
136            db.for_each("SELECT slot,record FROM _coven_slot_records WHERE kind=?1 AND slot<=?2 ORDER BY slot",rusqlite::params![kind,covers.to_be_bytes()],|r| {
137                let original=coven_format::floor::OriginalSlot {slot:slot_counter(r.get(0).map_err(DbError::from)?).map_err(DbError::from)?,object:r.get(1).map_err(DbError::from)?};
138                record(match kind {"admin"=>SnapshotRecord::Admin(original),"key"=>SnapshotRecord::KeyCopy(original),_=>SnapshotRecord::Announcement(original)})
139            })?;
140        }
141    }
142    for claim in claims {
143        record(SnapshotRecord::Claim(claim))?;
144    }
145    emit(encoder.finish()?).map_err(SnapshotWriteError::Output)?;
146    Ok(breaking)
147}
148
149pub(crate) fn read_write(r: &rusqlite::Row<'_>) -> rusqlite::Result<AppliedWrite> {
150    let timestamp = decoded(merge_fields::decode_timestamp(&r.get::<_, Vec<u8>>(0)?))?;
151    Ok(AppliedWrite {
152        id: WriteId { device: timestamp.device(), number: counter(r.get(1)?)? },
153        timestamp,
154        read_through: counter(r.get(2)?)?,
155        slot: slot_counter(r.get(3)?)?,
156    })
157}
158
159#[cfg(test)]
160#[path = "snapshot_write_tests.rs"]
161pub(crate) mod tests;
coven_database/snapshot_write.rs

coven_database/
snapshot_write.rs
