coven_database/
snapshot_objects.rs

1use crate::blob_records::live_blob_row;
2use crate::blob_records::load_activated_registration_on;
3use crate::blob_records::validate_live_blob_locator;
4use crate::blob_records::validate_stored_locator_on;
5use crate::blob_records::validate_stored_row_binding_on;
6use crate::remote_object_records::load_remote_object_on;
7use crate::PreparedSnapshotBlob;
8
9use super::*;
10
11pub(crate) fn validate_snapshot_object_owners_on(
12    conn: &Connection,
13    root: &coven_protocol::store_commit::StoreRootRef,
14    meta: &SnapshotMeta,
15) -> Result<(), DbError> {
16    let registration = load_activated_registration_on(conn, root, &meta.author_registration)?;
17    let expected = coven_protocol::remote_object::SnapshotObjectOwner {
18        activation: registration
19            .store_snapshot_activation(&meta.author_registration)
20            .map_err(DbError::from)?
21            .activation_id(),
22        generation: meta.generation,
23    };
24    if meta.successor.activation != expected.activation {
25        return Err(DbError::Message(
26            "verified snapshot successor differs from its author stream activation".to_string(),
27        ));
28    }
29    validate_snapshot_object_owner_records_on(conn, &expected)
30}
31
32pub async fn verify_snapshot_blob_spools(
33    blobs: &[PreparedSnapshotBlob],
34    label: &str,
35) -> Result<(), DbError> {
36    for blob in blobs {
37        if let Some(spool_path) = &blob.spool_path {
38            {
39                let (size, digest) = coven_foundation::local_file::file_facts(spool_path)
40                    .await
41                    .map_err(|error| {
42                        DbError::context(format!("{label} snapshot blob spool"), error)
43                    })?;
44                blob.remote
45                    .object()
46                    .verify_stored_facts(
47                        spool_path,
48                        size,
49                        coven_protocol::store_commit::ObjectHash::from_digest(digest),
50                    )
51                    .map_err(|error| {
52                        DbError::context(format!("{label} snapshot blob spool"), error)
53                    })?;
54            }
55        }
56    }
57    Ok(())
58}
59
60pub fn validate_snapshot_author(
61    author: &StoreDeviceRegistrationRef,
62    local: &StoreDeviceRegistrationRef,
63    label: &str,
64) -> Result<(), DbError> {
65    if author == local {
66        Ok(())
67    } else {
68        Err(DbError::Message(format!(
69            "staged {label} snapshot author differs from local activation"
70        )))
71    }
72}
73
74pub fn validate_snapshot_image(
75    image: &SnapshotImageRef,
76    prepared: &PreparedExactObject,
77    plaintext_hash: ObjectHash,
78    stored_hash: ObjectHash,
79    stored_size: u64,
80    expected_slot: String,
81    label: &str,
82) -> Result<(), DbError> {
83    if image.object == *prepared.reference()
84        && plaintext_hash == image.image_hash
85        && stored_hash == image.object.stored_hash()
86        && stored_size == image.object.stored_size()
87        && image.object.slot().logical_key() == expected_slot
88    {
89        Ok(())
90    } else {
91        Err(DbError::Message(format!(
92            "staged {label} snapshot image differs from its exact reference"
93        )))
94    }
95}
96
97pub(crate) fn validate_snapshot_blob_plans_on(
98    conn: &Connection,
99    gates: &Gates,
100    synced_tables: &[SyncedTable],
101    owner: &coven_protocol::remote_object::SnapshotObjectOwner,
102    blobs: &[PreparedSnapshotBlob],
103) -> Result<(), DbError> {
104    for blob in blobs {
105        blob.remote
106            .validate()
107            .map_err(|error| DbError::context("snapshot remote blob", error))?;
108        let owners = blob.remote.snapshot_owners().collect::<Vec<_>>();
109        if owners != [owner] {
110            return Err(DbError::Message(
111                "snapshot blob owner differs from the verified snapshot stream activation"
112                    .to_string(),
113            ));
114        }
115        if blob.bindings.is_empty()
116            || blob.bindings.iter().any(|binding| {
117                binding.blob().object() != blob.remote.object()
118                    || binding.blob().locator().audience() != blob.authority.remote_audience()
119            })
120            || blob
121                .spool_path
122                .as_ref()
123                .is_some_and(|path| !path.is_absolute())
124        {
125            return Err(DbError::Message(
126                "snapshot blob plan has inconsistent exact references".to_string(),
127            ));
128        }
129        for binding in &blob.bindings {
130            let table = synced_tables
131                .iter()
132                .find(|table| table.name() == binding.table())
133                .ok_or_else(|| {
134                    DbError::Message(format!(
135                        "snapshot blob names undeclared table {:?}",
136                        binding.table()
137                    ))
138                })?;
139            let declaration = table.blob().ok_or_else(|| {
140                DbError::Message(format!(
141                    "snapshot blob names table {:?} without a blob declaration",
142                    table.name()
143                ))
144            })?;
145            let row = live_blob_row(conn, table.name(), binding.row_id(), declaration)?
146                .ok_or_else(|| {
147                    DbError::Message(format!(
148                        "snapshot blob row {:?}/{:?} is absent",
149                        table.name(),
150                        binding.row_id()
151                    ))
152                })?;
153            let audience = gate::live_row_audience(conn, gates, table.name(), binding.row_id())
154                .map_err(DbError::from)?;
155            let audience = RemoteAudience::try_from(audience).map_err(DbError::from)?;
156            validate_live_blob_locator(
157                binding.table(),
158                binding.row_id(),
159                binding.column(),
160                binding.row_stamp(),
161                binding.blob(),
162                declaration,
163                &row,
164                &audience,
165            )?;
166        }
167    }
168    Ok(())
169}
170
171pub(crate) fn persist_snapshot_image_on(
172    conn: &Connection,
173    store_dir: &coven_foundation::store_dir::StoreDir,
174    image: &SnapshotImageRef,
175    owner: coven_protocol::remote_object::SnapshotObjectOwner,
176    label: &str,
177) -> Result<(), DbError> {
178    let image = RemoteObjectRecord::snapshot_activated_image(image, owner)
179        .map_err(|error| DbError::context(format!("{label} ownership"), error))?;
180    persist_exact_remote_object_on(conn, store_dir, &image, label)
181}
182
183/// Record the generation's ownership of the membership rollup it published.
184///
185/// Merged rather than inserted: a rollup is content-addressed over the
186/// membership frontier, so a generation published while membership has not
187/// changed names the object an earlier generation already owns. Both
188/// generations own it, and it is reclaimable only once neither does.
189pub(crate) fn persist_membership_rollup_on(
190    conn: &Connection,
191    store_dir: &coven_foundation::store_dir::StoreDir,
192    rollup: &coven_protocol::store_commit::MembershipRollupRef,
193    owner: coven_protocol::remote_object::SnapshotObjectOwner,
194    label: &str,
195) -> Result<(), DbError> {
196    let object_id = coven_protocol::remote_object::remote_object_id(&rollup.object);
197    let exists: bool = conn
198        .query_row(
199            "SELECT EXISTS(SELECT 1 FROM remote_objects WHERE object_id = ?1)",
200            [object_id.to_string()],
201            |row| row.get(0),
202        )
203        .map_err(DbError::from)?;
204    if exists {
205        let mut remote = load_remote_object_on(conn, object_id)?;
206        remote
207            .merge_snapshot_ownership(rollup, owner)
208            .map_err(|error| DbError::context(format!("{label} ownership"), error))?;
209        return update_remote_object_on(conn, object_id, &remote);
210    }
211    let rollup = RemoteObjectRecord::snapshot_activated_membership_rollup(rollup, owner)
212        .map_err(|error| DbError::context(format!("{label} ownership"), error))?;
213    persist_exact_remote_object_on(conn, store_dir, &rollup, label)
214}
215
216pub fn snapshot_generation_as_i64(generation: u64, label: &str) -> Result<i64, DbError> {
217    i64::try_from(generation)
218        .map_err(|_| DbError::Message(format!("{label} generation exceeds SQLite INTEGER")))
219}
220
221pub(crate) fn validate_snapshot_object_owner_records_on(
222    conn: &Connection,
223    expected: &coven_protocol::remote_object::SnapshotObjectOwner,
224) -> Result<(), DbError> {
225    let mut statement = conn
226        .prepare("SELECT object_id FROM remote_objects ORDER BY object_id")
227        .map_err(DbError::from)?;
228    let object_ids = statement
229        .query_map([], |row| row.get::<_, String>(0))
230        .map_err(DbError::from)?
231        .collect::<Result<Vec<_>, _>>()
232        .map_err(DbError::from)?;
233    drop(statement);
234    for object_id in object_ids {
235        let parsed = object_id.parse().map_err(|error| {
236            DbError::context(format!("snapshot remote object id {object_id:?}"), error)
237        })?;
238        let remote = load_remote_object_on(conn, parsed)?;
239        for owner in remote.snapshot_owners() {
240            if owner.activation != expected.activation || owner.generation > expected.generation {
241                return Err(DbError::Message(format!(
242                    "snapshot remote object {object_id} belongs to another stream or a later generation"
243                )));
244            }
245        }
246    }
247    Ok(())
248}
249
250pub(crate) fn install_snapshot_blob_plan_on(
251    conn: &Connection,
252    blob: &PreparedSnapshotBlob,
253) -> Result<(), DbError> {
254    let object_id = blob.remote.object_id();
255    let exists: bool = conn
256        .query_row(
257            "SELECT EXISTS(SELECT 1 FROM remote_objects WHERE object_id = ?1)",
258            [object_id.to_string()],
259            |row| row.get(0),
260        )
261        .map_err(DbError::from)?;
262    let merged = if exists {
263        let mut existing = load_remote_object_on(conn, object_id)?;
264        for owner in blob.remote.snapshot_owners() {
265            existing
266                .merge_snapshot_owner(blob.bindings[0].blob(), owner.clone())
267                .map_err(|error| DbError::context("merge snapshot blob owner", error))?;
268        }
269        existing
270    } else {
271        blob.remote.clone()
272    };
273    let encoded = serde_json::to_string(&merged)
274        .map_err(|error| DbError::context("serialize snapshot blob", error))?;
275    conn.execute(
276        "INSERT INTO remote_objects (object_id, state) VALUES (?1, ?2)
277         ON CONFLICT(object_id) DO UPDATE SET state = excluded.state",
278        rusqlite::params![object_id.to_string(), encoded],
279    )
280    .map_err(DbError::from)?;
281    conn.execute(
282        "INSERT INTO blob_locators (remote_object_id, locator_hash) VALUES (?1, ?2)
283         ON CONFLICT(remote_object_id) DO NOTHING",
284        rusqlite::params![
285            object_id.to_string(),
286            blob.bindings[0].blob().locator().locator_hash().to_string(),
287        ],
288    )
289    .map_err(DbError::from)?;
290    validate_stored_locator_on(conn, blob.bindings[0].blob())?;
291    let authority = serde_json::to_string(&blob.authority)
292        .map_err(|error| DbError::context("serialize snapshot blob authority", error))?;
293    for binding in &blob.bindings {
294        conn.execute(
295            "INSERT INTO row_blob_locators
296         (table_name, row_id, column_name, row_stamp, audience_authority, remote_object_id)
297         VALUES (?1, ?2, ?3, ?4, ?5, ?6)
298         ON CONFLICT(table_name, row_id, column_name, row_stamp) DO NOTHING",
299            rusqlite::params![
300                binding.table(),
301                binding.row_id(),
302                binding.column(),
303                binding.row_stamp(),
304                authority,
305                object_id.to_string(),
306            ],
307        )
308        .map_err(DbError::from)?;
309        validate_stored_row_binding_on(conn, binding, &blob.authority, object_id)?;
310    }
311    Ok(())
312}
313
314pub(crate) fn install_snapshot_blob_plans_on(
315    conn: &Connection,
316    blobs: &[PreparedSnapshotBlob],
317) -> Result<(), DbError> {
318    for blob in blobs {
319        install_snapshot_blob_plan_on(conn, blob)?;
320        if let Some(path) = &blob.spool_path {
321            conn.execute(
322                "INSERT INTO snapshot_blob_spool_cleanup (path) VALUES (?1)
323                 ON CONFLICT(path) DO NOTHING",
324                [path.to_string_lossy().as_ref()],
325            )
326            .map_err(DbError::from)?;
327        }
328    }
329    Ok(())
330}
coven_database/snapshot_objects.rs

coven_database/
snapshot_objects.rs
