coven_database/
staged_damage.rs

1//! Telling damage to a snapshot's staged records on this device from the
2//! snapshot's own (§19.1): the digest of what a staging holds, and, for
3//! tests, damage to it.
4
5#[cfg(any(test, feature = "test-utils"))]
6use super::table;
7use super::{SnapshotStaging, StagedTables};
8use crate::snapshot_error::SnapshotError;
9use crate::sqlite::DatabaseConnection;
10use crate::DbError;
11
12/// A digest of one staging's records as its tables hold them. Staging is a
13/// function of the snapshot's bytes, so two stagings of the same snapshot
14/// hold the same records, and differ only where this device's disk changed
15/// one (§19.1).
16#[derive(Clone, Copy, Debug, PartialEq, Eq)]
17pub struct StagedDigest([u8; 32]);
18
19/// Read back every record `tables` hold and digest them, each table's rows
20/// in the order of their encodings, so the digest is of their values alone.
21fn digest(db: &DatabaseConnection, tables: &StagedTables) -> Result<StagedDigest, DbError> {
22    let mut hasher = coven_crypto::ContentHasher::new();
23    let mut table = |name: &str, select: &str, params: &[&dyn rusqlite::ToSql]| {
24        let mut rows = Vec::new();
25        db.for_each(select, params, |row| {
26            rows.push(encoded(row)?);
27            Ok::<_, DbError>(())
28        })?;
29        rows.sort_unstable();
30        hasher.update(&(name.len() as u64).to_be_bytes());
31        hasher.update(name.as_bytes());
32        hasher.update(&(rows.len() as u64).to_be_bytes());
33        for row in rows {
34            hasher.update(&row);
35        }
36        Ok::<_, DbError>(())
37    };
38    // A table is named for its staging; the digest names it without.
39    let staging = format!("_coven_staged_{}_", tables.id);
40    for name in &tables.tables {
41        table(&name.replacen(&staging, "", 1), &format!("SELECT * FROM {name}"), &[])?;
42    }
43    table(
44        "_coven_snapshot_writes",
45        "SELECT device,number,timestamp,read_through,slot FROM temp._coven_snapshot_writes WHERE source=?1",
46        &[&tables.id],
47    )?;
48    Ok(StagedDigest(*hasher.finish().as_bytes()))
49}
50
51/// One row's values, each tagged with its type and text or blob lengths
52/// written out, so no two rows share an encoding.
53fn encoded(row: &rusqlite::Row<'_>) -> Result<Vec<u8>, DbError> {
54    use rusqlite::types::ValueRef;
55    let mut bytes = Vec::new();
56    for i in 0..row.as_ref().column_count() {
57        match row.get_ref(i)? {
58            ValueRef::Null => bytes.push(0),
59            ValueRef::Integer(value) => {
60                bytes.push(1);
61                bytes.extend_from_slice(&value.to_be_bytes());
62            }
63            ValueRef::Real(value) => {
64                bytes.push(2);
65                bytes.extend_from_slice(&value.to_bits().to_be_bytes());
66            }
67            ValueRef::Text(value) | ValueRef::Blob(value) => {
68                bytes.push(3);
69                bytes.extend_from_slice(&(value.len() as u64).to_be_bytes());
70                bytes.extend_from_slice(value);
71            }
72        }
73    }
74    Ok(bytes)
75}
76
77/// A failure of a staged snapshot's body carries the digest of its staging
78/// as the failed read left it, for its caller to tell damage on this device
79/// from the snapshot's own (§19.1). When the staging can't be read back,
80/// that failure is the one reported: the input database failed.
81pub(super) fn with_staged_digest(
82    db: &DatabaseConnection,
83    stagings: &[SnapshotStaging],
84    error: DbError,
85) -> DbError {
86    let DbError::Snapshot(SnapshotError::Body { snapshot, error, staged: None }) = error else {
87        return error;
88    };
89    let tables = stagings
90        .iter()
91        .find(|staging| staging.expected == snapshot)
92        .and_then(|staging| staging.tables.as_ref());
93    match tables.map(|tables| digest(db, tables)) {
94        Some(Ok(staged)) => SnapshotError::Body { snapshot, error, staged: Some(staged) }.into(),
95        Some(Err(failure)) => failure,
96        None => SnapshotError::Body { snapshot, error, staged: None }.into(),
97    }
98}
99
100/// Damage a test makes to a staging, as damage on this device's disk would
101/// (§19.1).
102#[cfg(any(test, feature = "test-utils"))]
103#[derive(Clone, Copy, Debug, PartialEq, Eq)]
104pub enum StagedDamage {
105    /// Its first staged record is cut to one byte, which a load or file
106    /// check refuses.
107    Record,
108    /// Its staged writes claim to have read through their own slots, which
109    /// finishing the staging refuses.
110    Writes,
111}
112
113/// Make `damage` to the staging `staging`.
114#[cfg(any(test, feature = "test-utils"))]
115pub(super) fn damage(
116    db: &DatabaseConnection,
117    staging: i64,
118    damage: StagedDamage,
119) -> Result<(), DbError> {
120    let changed = match damage {
121        StagedDamage::Record => {
122            let records = table(staging, "records");
123            db.internal_execute(
124                &format!(
125                    "UPDATE {records} SET record=substr(record,1,1) WHERE rowid=(SELECT min(rowid) FROM {records})"
126                ),
127                [],
128            )?
129        }
130        StagedDamage::Writes => db.internal_execute(
131            "UPDATE temp._coven_snapshot_writes SET read_through=slot WHERE source=?1",
132            [staging],
133        )?,
134    };
135    match changed {
136        0 => {
137            Err(DbError::Invariant { what: "a test damages a staging that holds what it changes" })
138        }
139        _ => Ok(()),
140    }
141}
142
143/// The disk error a test's request fails a staging removal with.
144#[cfg(any(test, feature = "test-utils"))]
145pub(super) fn removal_failure() -> DbError {
146    DbError::Sqlite(rusqlite::Error::SqliteFailure(
147        rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_IOERR),
148        Some("injected staging removal failure".into()),
149    ))
150}
151
152impl crate::Database {
153    /// Fail the next removal of the stagings a failed load or file check
154    /// ends, as a disk error would; the stagings keep their records until
155    /// they are dropped (§16.2).
156    #[cfg(any(test, feature = "test-utils"))]
157    pub async fn fail_next_staging_removal(&self) -> Result<(), DbError> {
158        self.call(|inner| {
159            inner.inputs.fail_removal.store(true, std::sync::atomic::Ordering::SeqCst);
160            Ok::<_, DbError>(())
161        })
162        .await
163    }
164}
165
166impl SnapshotStaging {
167    /// The digest of the records it holds now (§19.1).
168    pub async fn digest(&self) -> Result<StagedDigest, DbError> {
169        self.id()?;
170        let tables = self.tables.as_ref().map(|tables| (tables.id, tables.tables.clone()));
171        let (id, tables) = tables.expect("a staging with an id holds its tables");
172        self.database
173            .call(move |inner| inner.inputs.write(|db| digest(db, &StagedTables { id, tables })))
174            .await
175    }
176}
coven_database/staged_damage.rs

coven_database/
staged_damage.rs
