coven_database/store/store_session/
snapshot_image.rs

1use std::path::{Path, PathBuf};
2
3use rusqlite::{Connection, OptionalExtension};
4use tracing::info;
5
6use crate::*;
7use coven_protocol::synced_schema::SyncedTable;
8
9use super::*;
10
11pub struct CreatedSnapshot {
12    db_image: SnapshotDatabaseImage,
13    blobs: Vec<SnapshotBlobFact>,
14}
15
16impl CreatedSnapshot {
17    pub fn new(db_image: SnapshotDatabaseImage, blobs: Vec<SnapshotBlobFact>) -> Self {
18        Self { db_image, blobs }
19    }
20
21    pub fn blobs(&self) -> &[SnapshotBlobFact] {
22        &self.blobs
23    }
24
25    pub async fn read_image(&self) -> Result<Vec<u8>, SnapshotImageError> {
26        self.db_image.read().await
27    }
28
29    pub fn into_parts(self) -> (SnapshotDatabaseImage, Vec<SnapshotBlobFact>) {
30        (self.db_image, self.blobs)
31    }
32
33    #[cfg(any(test, feature = "test-utils"))]
34    pub fn image_path_for_test(&self) -> &Path {
35        self.db_image.path()
36    }
37}
38
39#[derive(Debug, Clone)]
40pub struct SnapshotBlobFact {
41    pub fact: crate::StoreWriteBlobFact,
42    pub audience: RemoteAudience,
43}
44
45#[derive(Debug, thiserror::Error)]
46pub enum SnapshotImageError {
47    #[error("IO error: {0}")]
48    Io(#[from] std::io::Error),
49    #[error("no synced tables registered; refusing to emit an all-cleared snapshot")]
50    NoSyncedTables,
51    #[error("failed to scope snapshot down to shareable data: {0}")]
52    Projection(String),
53    #[error("snapshot database: {0}")]
54    Database(#[source] Box<DbError>),
55    #[error("snapshot gate: {0}")]
56    Gate(#[from] crate::GateError),
57    #[error("snapshot row routing key: {0}")]
58    RowRoutingKey(#[from] coven_protocol::circle::RowRoutingKeyError),
59    #[error("snapshot SQLite: {0}")]
60    Sqlite(#[from] rusqlite::Error),
61    #[error("snapshot remote object: {0}")]
62    RemoteObject(#[from] coven_protocol::remote_object::RemoteObjectRecordError),
63    #[error("snapshot blob declarations: {0}")]
64    BlobDecl(#[from] crate::BlobDeclError),
65    #[error("snapshot routing contract: {0}")]
66    RoutingContract(#[from] crate::SyncRoutingContractError),
67    #[error("snapshot projection {operation}: {source}")]
68    ProjectionSqlite {
69        operation: String,
70        #[source]
71        source: rusqlite::Error,
72    },
73    #[error("snapshot projection {operation}: {source}")]
74    ProjectionDatabase {
75        operation: String,
76        #[source]
77        source: Box<DbError>,
78    },
79    #[error("snapshot projection {operation}: {source}")]
80    ProjectionIo {
81        operation: String,
82        #[source]
83        source: std::io::Error,
84    },
85    #[error("snapshot projection {operation}: {source}")]
86    ProjectionPayloadStore {
87        operation: String,
88        #[source]
89        source: crate::PayloadStoreError,
90    },
91    #[error("snapshot blob {namespace}/{id} plaintext hash: {source}")]
92    BlobHash {
93        namespace: String,
94        id: String,
95        #[source]
96        source: coven_foundation::object_hash::InvalidObjectHash,
97    },
98    #[error(
99        "could not remove staged snapshot database {path}: {cleanup}",
100        path = .path.display()
101    )]
102    Cleanup { path: PathBuf, cleanup: String },
103    #[error(
104        "snapshot operation failed and staged database {path} could not be removed: {cleanup} \
105         (operation error: {cause})",
106        path = .path.display()
107    )]
108    CleanupAfterFailure {
109        path: PathBuf,
110        cleanup: String,
111        cause: Box<SnapshotImageError>,
112    },
113}
114
115impl From<DbError> for SnapshotImageError {
116    fn from(error: DbError) -> Self {
117        Self::Database(Box::new(error))
118    }
119}
120
121#[derive(Debug)]
122pub enum SnapshotImageOperationError<E> {
123    Operation(E),
124    Cleanup {
125        path: PathBuf,
126        cleanup: String,
127    },
128    CleanupAfterFailure {
129        path: PathBuf,
130        cleanup: String,
131        cause: E,
132    },
133}
134
135/// One uncommitted SQLite image and its sidecar files.
136///
137/// The path remains armed until the operation commits or consumes the image.
138/// Failures report cleanup failure instead of leaving a plaintext image behind.
139#[derive(Debug)]
140pub struct SnapshotDatabaseImage {
141    path: PathBuf,
142    armed: bool,
143}
144
145impl SnapshotDatabaseImage {
146    pub fn prepare(path: PathBuf) -> Result<Self, SnapshotImageError> {
147        let mut staged = Self { path, armed: true };
148        if let Err(cleanup) = staged.remove_files() {
149            staged.armed = false;
150            return Err(SnapshotImageError::Cleanup {
151                path: staged.path.clone(),
152                cleanup: cleanup.to_string(),
153            });
154        }
155        Ok(staged)
156    }
157
158    pub fn create(path: PathBuf, plaintext: &[u8]) -> Result<Self, SnapshotImageError> {
159        if let Some(parent) = path.parent() {
160            std::fs::create_dir_all(parent)?;
161        }
162        Self { path, armed: false }.write_new(plaintext)
163    }
164
165    pub fn replace(path: PathBuf, plaintext: &[u8]) -> Result<Self, SnapshotImageError> {
166        Self::prepare(path)?.write_new(plaintext)
167    }
168
169    pub(super) fn prepare_snapshot(temp_dir: &Path) -> Result<Self, SnapshotImageError> {
170        Self::prepare(temp_dir.join("snapshot.db"))
171    }
172
173    #[allow(clippy::too_many_arguments)]
174    pub(super) fn capture_on(
175        self,
176        connection: &Connection,
177        store_dir: &coven_foundation::store_dir::StoreDir,
178        mut authority: VerifiedStoreAuthority,
179        root: &coven_protocol::store_commit::StoreRootRef,
180        tables: &[SyncedTable],
181        routing_encryption: Option<&coven_keys::encryption::EncryptionService>,
182        audience: &coven_protocol::circle::Audience,
183    ) -> Result<CreatedSnapshot, SnapshotImageError> {
184        if tables.is_empty() {
185            return self.finish(Err(SnapshotImageError::NoSyncedTables));
186        }
187        let gates = match crate::Gates::from_tables(connection, tables) {
188            Ok(gates) => gates,
189            Err(error) => {
190                return self.finish(Err(SnapshotImageError::from(error)));
191            }
192        };
193        let routing_key = if gates.has_scoped_graph() {
194            let encryption = match routing_encryption {
195                Some(encryption) => encryption,
196                None => {
197                    return self.finish(Err(SnapshotImageError::Projection(
198                        "scoped snapshot creation requires Store routing encryption".to_string(),
199                    )));
200                }
201            };
202            match coven_protocol::circle::derive_row_routing_key(encryption, root.store_root_hash) {
203                Ok(routing_key) => Some(routing_key),
204                Err(error) => {
205                    return self.finish(Err(SnapshotImageError::from(error)));
206                }
207            }
208        } else {
209            None
210        };
211
212        let source_image = match crate::connection_io::serialize_database_image(connection) {
213            Ok(image) => image,
214            Err(error) => {
215                return self.finish(Err(SnapshotImageError::from(error)));
216            }
217        };
218        let mut snapshot = match Connection::open_in_memory().map_err(DbError::from) {
219            Ok(snapshot) => snapshot,
220            Err(error) => {
221                return self.finish(Err(SnapshotImageError::from(error)));
222            }
223        };
224        if let Err(error) =
225            crate::connection_io::deserialize_database_image_into(&mut snapshot, &source_image)
226        {
227            return self.finish(Err(SnapshotImageError::from(error)));
228        }
229        if let Err(error) = Self::project(
230            &mut snapshot,
231            store_dir,
232            &mut authority,
233            root,
234            tables,
235            routing_key.as_ref(),
236            audience,
237        ) {
238            return self.finish(Err(error));
239        }
240        let blobs = match Self::blob_facts(connection, &snapshot, tables) {
241            Ok(blobs) => blobs,
242            Err(error) => return self.finish(Err(error)),
243        };
244        if matches!(audience, coven_protocol::circle::Audience::Circle(_)) {
245            if let Err(error) = Self::strip_circle_transport_state(&mut snapshot) {
246                return self.finish(Err(error));
247            }
248        }
249
250        let image = match crate::connection_io::serialize_database_image(&snapshot) {
251            Ok(image) => image,
252            Err(error) => {
253                return self.finish(Err(SnapshotImageError::from(error)));
254            }
255        };
256        drop(snapshot);
257        let snapshot = self.write_new(&image)?;
258
259        let plaintext_size = match std::fs::metadata(snapshot.path()) {
260            Ok(metadata) => metadata.len(),
261            Err(error) => return snapshot.finish(Err(SnapshotImageError::Io(error))),
262        };
263        info!(plaintext_size, "created snapshot");
264        Ok(CreatedSnapshot::new(snapshot, blobs))
265    }
266
267    fn write_new(mut self, plaintext: &[u8]) -> Result<Self, SnapshotImageError> {
268        let mut file = match std::fs::OpenOptions::new()
269            .write(true)
270            .create_new(true)
271            .open(&self.path)
272        {
273            Ok(file) => file,
274            Err(error) => {
275                self.armed = false;
276                return Err(SnapshotImageError::Io(error));
277            }
278        };
279        self.armed = true;
280        if let Err(error) = std::io::Write::write_all(&mut file, plaintext) {
281            drop(file);
282            return self.finish(Err(SnapshotImageError::Io(error)));
283        }
284        drop(file);
285        Ok(self)
286    }
287
288    /// Read the exact image's replay state without installing it in a live database.
289    /// The replication owner must authenticate acceptance and membership before admission.
290    pub fn read_replay_baseline(
291        plaintext: &[u8],
292        snapshot: PublishedStoreSnapshot,
293        genesis: &coven_protocol::store_commit::ResolvedStoreDeviceState,
294    ) -> Result<InstalledReplayBaseline, SnapshotImageError> {
295        if coven_protocol::store_commit::ObjectHash::digest(plaintext)
296            != snapshot.meta.image.image_hash
297        {
298            return Err(SnapshotImageError::Projection(
299                "snapshot image differs from its signed hash".into(),
300            ));
301        }
302        snapshot
303            .meta
304            .history_summary
305            .validate_snapshot_baseline()
306            .map_err(DbError::from)?;
307        let mut connection = Connection::open_in_memory()?;
308        crate::connection_io::deserialize_database_image_into(&mut connection, plaintext)?;
309        let coverage = snapshot.meta.coverage.clone();
310        let (reference, state) = if coverage.commits().is_empty() {
311            genesis.validate_canonical().map_err(DbError::from)?;
312            (
313                coven_protocol::store_commit::StoreDeviceStateRef::from_resolved(
314                    coverage.clone(),
315                    genesis,
316                )
317                .map_err(DbError::from)?,
318                genesis.clone(),
319            )
320        } else {
321            crate::store::store_device_state::store_device_state_for_history_cut_on(
322                &connection,
323                &coven_protocol::store_commit::StoreHistoryCut(coverage.commits().clone()),
324            )?
325        };
326        if state != snapshot.meta.state.devices
327            || reference != snapshot.meta.history_summary.post_state
328        {
329            return Err(SnapshotImageError::Projection(
330                "snapshot image device state differs from its signed cut".into(),
331            ));
332        }
333        let states = crate::store::store_device_state::load_covered_store_device_snapshots_on(
334            &connection,
335            &coverage,
336        )?;
337        if states.keys().any(|reference| {
338            snapshot
339                .meta
340                .history_summary
341                .causal_cut
342                .get(&reference.coord)
343                != Some(reference)
344        }) {
345            return Err(SnapshotImageError::Projection(
346                "snapshot image carries a device state outside its exact accepted history".into(),
347            ));
348        }
349        Ok(InstalledReplayBaseline::from_snapshot(snapshot, states))
350    }
351
352    pub fn path(&self) -> &Path {
353        &self.path
354    }
355
356    /// The caller authenticates this snapshot's accepted publication. Its image
357    /// carries exact Store blob provenance after source packages are retired.
358    pub fn contains_reclaimable_store_blob(
359        plaintext: &[u8],
360        snapshot: &coven_protocol::store_commit::SnapshotMeta,
361        stored: &coven_protocol::blob::locator::StoredBlobRef,
362    ) -> Result<bool, SnapshotImageError> {
363        if ObjectHash::digest(plaintext) != snapshot.image.image_hash {
364            return Err(SnapshotImageError::Projection(
365                "snapshot blob inventory differs from its signed image hash".into(),
366            ));
367        }
368        if stored.locator().audience() != RemoteAudience::Store {
369            return Err(SnapshotImageError::Projection(
370                "Store snapshot inventory cannot authorize a Circle blob".into(),
371            ));
372        }
373        let mut connection = Connection::open_in_memory()?;
374        crate::connection_io::deserialize_database_image_into(&mut connection, plaintext)?;
375        let id = coven_protocol::remote_object::remote_object_id(stored.object());
376        let exists: bool = connection.query_row(
377            "SELECT EXISTS(SELECT 1 FROM blob_locators WHERE remote_object_id = ?1)",
378            [id.to_string()],
379            |row| row.get(0),
380        )?;
381        if !exists {
382            return Ok(false);
383        }
384        crate::blob_records::validate_stored_locator_on(&connection, stored)?;
385        let remote = crate::remote_object_records::load_remote_object_on(&connection, id)?;
386        remote.validate_reclaimable_stored_blob(stored)?;
387        let owners = remote.stored_blob_commit_owners();
388        if owners.is_empty()
389            || owners
390                .iter()
391                .any(|owner| snapshot.history_summary.causal_cut.get(&owner.coord) != Some(owner))
392        {
393            return Err(SnapshotImageError::Projection(
394                "snapshot blob inventory has no exact accepted publication owner".into(),
395            ));
396        }
397        let live: bool = connection.query_row(
398            "SELECT EXISTS(SELECT 1 FROM row_blob_locators WHERE remote_object_id = ?1)",
399            [id.to_string()],
400            |row| row.get(0),
401        )?;
402        Ok(!live
403            && remote.snapshot_owners().next().is_none()
404            && remote.retained_replay_owners().next().is_none())
405    }
406
407    pub async fn read(&self) -> Result<Vec<u8>, SnapshotImageError> {
408        tokio::fs::read(&self.path)
409            .await
410            .map_err(|error| SnapshotImageError::ProjectionIo {
411                operation: format!("read staged snapshot database {}", self.path.display()),
412                source: error,
413            })
414    }
415
416    pub fn read_and_discard(self) -> Result<Vec<u8>, SnapshotImageError> {
417        let outcome = std::fs::read(&self.path).map_err(SnapshotImageError::Io);
418        self.finish(outcome)
419    }
420
421    pub fn canonicalize(mut self) -> Result<Self, SnapshotImageError> {
422        match std::fs::canonicalize(&self.path) {
423            Ok(path) => {
424                self.path = path;
425                Ok(self)
426            }
427            Err(error) => self.finish(Err(SnapshotImageError::Io(error))),
428        }
429    }
430
431    pub fn finish<T>(
432        self,
433        outcome: Result<T, SnapshotImageError>,
434    ) -> Result<T, SnapshotImageError> {
435        match self.finish_operation(outcome) {
436            Ok(value) => Ok(value),
437            Err(SnapshotImageOperationError::Operation(cause)) => Err(cause),
438            Err(SnapshotImageOperationError::Cleanup { path, cleanup }) => {
439                Err(SnapshotImageError::Cleanup { path, cleanup })
440            }
441            Err(SnapshotImageOperationError::CleanupAfterFailure {
442                path,
443                cleanup,
444                cause,
445            }) => Err(SnapshotImageError::CleanupAfterFailure {
446                path,
447                cleanup,
448                cause: Box::new(cause),
449            }),
450        }
451    }
452
453    pub fn finish_operation<T, E>(
454        mut self,
455        outcome: Result<T, E>,
456    ) -> Result<T, SnapshotImageOperationError<E>> {
457        let cleanup = self.remove_files();
458        self.armed = false;
459        match (outcome, cleanup) {
460            (Ok(value), Ok(())) => Ok(value),
461            (Err(cause), Ok(())) => Err(SnapshotImageOperationError::Operation(cause)),
462            (Ok(_), Err(cleanup)) => Err(SnapshotImageOperationError::Cleanup {
463                path: self.path.clone(),
464                cleanup: cleanup.to_string(),
465            }),
466            (Err(cause), Err(cleanup)) => Err(SnapshotImageOperationError::CleanupAfterFailure {
467                path: self.path.clone(),
468                cleanup: cleanup.to_string(),
469                cause,
470            }),
471        }
472    }
473
474    pub fn commit(mut self) -> PathBuf {
475        self.armed = false;
476        std::mem::take(&mut self.path)
477    }
478
479    fn project(
480        connection: &mut Connection,
481        store_dir: &coven_foundation::store_dir::StoreDir,
482        authority: &mut VerifiedStoreAuthority,
483        root: &coven_protocol::store_commit::StoreRootRef,
484        synced: &[SyncedTable],
485        routing_key: Option<&coven_protocol::circle::RowRoutingKey>,
486        audience: &coven_protocol::circle::Audience,
487    ) -> Result<(), SnapshotImageError> {
488        let gates =
489            crate::Gates::from_tables(connection, synced).map_err(SnapshotImageError::from)?;
490        if gates.has_scoped_graph() && routing_key.is_none() {
491            return Err(SnapshotImageError::Projection(
492                "scoped snapshot projection requires a row-routing key".to_string(),
493            ));
494        }
495        let transaction = connection
496            .unchecked_transaction()
497            .map_err(SnapshotImageError::from)?;
498        transaction
499            .pragma_update(None, "defer_foreign_keys", "ON")
500            .map_err(SnapshotImageError::from)?;
501        let coverage =
502            crate::store::materialized_commit_index::materialized_frontier_on(&transaction, None)
503                .map_err(SnapshotImageError::from)?;
504        let cleared_materialization_tables = ["materialized_commits"];
505        for table in cleared_materialization_tables {
506            transaction
507                .execute_batch(&format!("DELETE FROM {}", crate::quote_ident(table)))
508                .map_err(|error| SnapshotImageError::ProjectionSqlite {
509                    operation: format!("clear {table}"),
510                    source: error,
511                })?;
512        }
513        if matches!(audience, coven_protocol::circle::Audience::Store) {
514            let records =
515                crate::store::store_session::StoreTransaction::new(&transaction, store_dir);
516            records
517                .project_shared_snapshot_replay_inputs(authority, root)
518                .map_err(SnapshotImageError::from)?;
519            records
520                .retain_snapshot_replay_inputs(
521                    authority,
522                    root,
523                    &coven_protocol::store_commit::CommitFrontier::from_refs(coverage.clone())
524                        .map_err(DbError::from)?,
525                )
526                .map_err(SnapshotImageError::from)?;
527            records
528                .retain_snapshot_device_states(authority, root, coverage)
529                .map_err(SnapshotImageError::from)?;
530        }
531        let preserved_non_synced_tables = match audience {
532            coven_protocol::circle::Audience::Store => SNAPSHOT_PRESERVED_NON_SYNCED_TABLES,
533            coven_protocol::circle::Audience::Circle(_) => CIRCLE_IMAGE_PRESERVED_NON_SYNCED_TABLES,
534            coven_protocol::circle::Audience::Local => {
535                return Err(SnapshotImageError::Projection(
536                    "Local rows cannot enter a snapshot".to_string(),
537                ));
538            }
539        };
540        for table in crate::user_table_names(connection).map_err(|error| {
541            SnapshotImageError::ProjectionSqlite {
542                operation: "list user tables".to_string(),
543                source: error,
544            }
545        })? {
546            if synced.iter().any(|synced| synced.name() == table)
547                || preserved_non_synced_tables.contains(&table.as_str())
548                || cleared_materialization_tables.contains(&table.as_str())
549            {
550                continue;
551            }
552            transaction
553                .execute_batch(&format!("DELETE FROM {}", crate::quote_ident(&table)))
554                .map_err(|error| SnapshotImageError::ProjectionSqlite {
555                    operation: format!("clear {table}"),
556                    source: error,
557                })?;
558        }
559
560        match audience {
561            coven_protocol::circle::Audience::Store => gates
562                .delete_gated_false(&transaction)
563                .map_err(SnapshotImageError::from)?,
564            coven_protocol::circle::Audience::Circle(_) => {
565                crate::retain_snapshot_audience_rows(&transaction, &gates, audience)
566                    .map_err(SnapshotImageError::from)?;
567            }
568            coven_protocol::circle::Audience::Local => {
569                return Err(SnapshotImageError::Projection(
570                    "Local rows cannot enter a snapshot".to_string(),
571                ));
572            }
573        }
574        if let Some(routing_key) = routing_key {
575            crate::prune_private_routes_without_rows(&transaction, &gates)
576                .map_err(SnapshotImageError::from)?;
577            crate::validate_snapshot_routing_state(&transaction, &gates, routing_key, audience)
578                .map_err(SnapshotImageError::from)?;
579        }
580
581        scope_authenticated_blob_graph(&transaction, synced, audience)?;
582        transaction.commit().map_err(SnapshotImageError::from)?;
583        if matches!(audience, coven_protocol::circle::Audience::Store) {
584            connection.execute_batch("VACUUM").map_err(|error| {
585                SnapshotImageError::ProjectionSqlite {
586                    operation: "vacuum".to_string(),
587                    source: error,
588                }
589            })?;
590        }
591        Ok(())
592    }
593
594    fn strip_circle_transport_state(connection: &mut Connection) -> Result<(), SnapshotImageError> {
595        connection
596            .pragma_update(None, "foreign_keys", "ON")
597            .map_err(SnapshotImageError::from)?;
598        let transaction = connection.transaction().map_err(SnapshotImageError::from)?;
599        // These rows are the copy's, describing the spool of the device that
600        // built it, so they are deleted without releasing any payload claim.
601        transaction
602            .execute_batch(
603                "DELETE FROM row_blob_locators;
604                 DELETE FROM blob_locators;
605                 DELETE FROM retained_replay_objects;
606                 DELETE FROM remote_objects;
607                 DELETE FROM retained_merge_materializations;",
608            )
609            .map_err(|error| SnapshotImageError::ProjectionSqlite {
610                operation: "strip Circle snapshot transport state".to_string(),
611                source: error,
612            })?;
613        transaction.commit().map_err(SnapshotImageError::from)?;
614        connection.execute_batch("VACUUM").map_err(|error| {
615            SnapshotImageError::ProjectionSqlite {
616                operation: "vacuum Circle snapshot transport projection".to_string(),
617                source: error,
618            }
619        })?;
620        Ok(())
621    }
622
623    pub fn install_blob_graph(
624        self,
625        owner: &coven_protocol::remote_object::SnapshotObjectOwner,
626        blobs: &[crate::PreparedSnapshotBlob],
627        pending_store_snapshots: &BTreeSet<coven_protocol::objects::ObjectSlot>,
628    ) -> Result<Self, SnapshotImageError> {
629        let result = (|| {
630            let source = std::fs::read(self.path()).map_err(SnapshotImageError::Io)?;
631            let mut connection = Connection::open_in_memory()
632                .map_err(DbError::from)
633                .map_err(SnapshotImageError::from)?;
634            crate::connection_io::deserialize_database_image_into(&mut connection, &source)
635                .map_err(SnapshotImageError::from)?;
636            connection
637                .pragma_update(None, "foreign_keys", "ON")
638                .map_err(SnapshotImageError::from)?;
639            let transaction = connection.transaction().map_err(SnapshotImageError::from)?;
640            for blob in blobs {
641                blob.remote.validate().map_err(SnapshotImageError::from)?;
642                if blob.remote.snapshot_owners().collect::<Vec<_>>() != [owner]
643                    || blob.bindings.is_empty()
644                    || blob
645                        .bindings
646                        .iter()
647                        .any(|binding| binding.blob().object() != blob.remote.object())
648                {
649                    return Err(SnapshotImageError::Projection(
650                        "snapshot blob binding differs from its remote object".to_string(),
651                    ));
652                }
653                crate::install_snapshot_blob_plan_on(&transaction, blob).map_err(|error| {
654                    SnapshotImageError::ProjectionDatabase {
655                        operation: "install snapshot blob".to_string(),
656                        source: Box::new(error),
657                    }
658                })?;
659            }
660            crate::snapshot_objects::replace_snapshot_object_owners_on(
661                &transaction,
662                owner,
663                blobs,
664                pending_store_snapshots,
665            )?;
666            transaction.commit().map_err(SnapshotImageError::from)?;
667            connection.execute_batch("VACUUM").map_err(|error| {
668                SnapshotImageError::ProjectionSqlite {
669                    operation: "vacuum snapshot closure".to_string(),
670                    source: error,
671                }
672            })?;
673            let image = crate::connection_io::serialize_database_image(&connection)
674                .map_err(SnapshotImageError::from)?;
675            connection
676                .close()
677                .map_err(|(_, error)| SnapshotImageError::ProjectionSqlite {
678                    operation: "close snapshot closure image".to_string(),
679                    source: error,
680                })?;
681            let mut file = std::fs::OpenOptions::new()
682                .write(true)
683                .truncate(true)
684                .open(self.path())
685                .map_err(SnapshotImageError::Io)?;
686            std::io::Write::write_all(&mut file, &image).map_err(SnapshotImageError::Io)?;
687            Ok(())
688        })();
689        match result {
690            Ok(()) => Ok(self),
691            Err(error) => self.finish(Err(error)),
692        }
693    }
694
695    fn blob_facts(
696        live: &Connection,
697        snapshot: &Connection,
698        tables: &[SyncedTable],
699    ) -> Result<Vec<SnapshotBlobFact>, SnapshotImageError> {
700        let declarations =
701            crate::BlobDecls::from_tables(snapshot, tables).map_err(SnapshotImageError::from)?;
702        let publications = declarations
703            .publication_blobs_in_db(snapshot)
704            .map_err(SnapshotImageError::from)?;
705        let gates = crate::Gates::from_tables(live, tables).map_err(SnapshotImageError::from)?;
706        let mut facts = Vec::with_capacity(publications.len());
707        for publication in publications {
708            let plaintext_hash = publication.plaintext_hash.parse().map_err(|error| {
709                SnapshotImageError::BlobHash {
710                    namespace: publication.blob.namespace.clone(),
711                    id: publication.blob.id.clone(),
712                    source: error,
713                }
714            })?;
715            let external_path =
716                if publication.blob.provenance == coven_protocol::blob::Provenance::UserProvided {
717                    live.query_row(
718                        "SELECT path FROM local_blob_refs
719                         WHERE table_name = ?1 AND row_id = ?2 AND column_name = ?3
720                           AND row_stamp = ?4 AND namespace = ?5 AND blob_id = ?6",
721                        rusqlite::params![
722                            publication.table,
723                            publication.row_id,
724                            publication.column,
725                            publication.row_stamp,
726                            publication.blob.namespace,
727                            publication.blob.id,
728                        ],
729                        |row| row.get::<_, String>(0),
730                    )
731                    .optional()
732                    .map_err(SnapshotImageError::from)?
733                    .map(PathBuf::from)
734                } else {
735                    None
736                };
737            let previous = crate::previous_row_blob_for_write_on(
738                snapshot,
739                &publication.table,
740                &publication.row_id,
741                &publication.row_stamp,
742                &publication.column,
743                &publication.blob,
744                publication.plaintext_size,
745                plaintext_hash,
746            )
747            .map_err(SnapshotImageError::from)?;
748            let audience = match crate::live_row_audience(
749                live,
750                &gates,
751                &publication.table,
752                &publication.row_id,
753            )
754            .map_err(SnapshotImageError::from)?
755            {
756                coven_protocol::circle::Audience::Store => RemoteAudience::Store,
757                coven_protocol::circle::Audience::Circle(circle_id) => {
758                    RemoteAudience::Circle(circle_id)
759                }
760                coven_protocol::circle::Audience::Local => {
761                    return Err(SnapshotImageError::Projection(format!(
762                        "scoped snapshot retains local blob row {:?}/{:?}",
763                        publication.table, publication.row_id
764                    )));
765                }
766            };
767            facts.push(SnapshotBlobFact {
768                fact: crate::StoreWriteBlobFact {
769                    table: publication.table,
770                    row_id: publication.row_id,
771                    row_stamp: publication.row_stamp,
772                    column: publication.column,
773                    blob: publication.blob,
774                    plaintext_size: publication.plaintext_size,
775                    plaintext_hash,
776                    external_path,
777                    previous,
778                    audience_move: None,
779                },
780                audience,
781            });
782        }
783        Ok(facts)
784    }
785
786    fn remove_files(&self) -> std::io::Result<()> {
787        for candidate in [
788            self.path.clone(),
789            PathBuf::from(format!("{}-wal", self.path.display())),
790            PathBuf::from(format!("{}-shm", self.path.display())),
791        ] {
792            match std::fs::remove_file(candidate) {
793                Ok(()) => {}
794                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
795                Err(error) => return Err(error),
796            }
797        }
798        Ok(())
799    }
800}
801
802impl Drop for SnapshotDatabaseImage {
803    fn drop(&mut self) {
804        if !self.armed {
805            return;
806        }
807        if let Err(error) = self.remove_files() {
808            tracing::warn!(
809                path = %self.path.display(),
810                %error,
811                "could not remove abandoned staged snapshot database"
812            );
813        }
814    }
815}
816
817pub(super) fn snapshot_image_db_error(error: SnapshotImageError) -> DbError {
818    DbError::from(error)
819}
820
821const SNAPSHOT_PRESERVED_NON_SYNCED_TABLES: &[&str] = &[
822    "_coven_audience",
823    "_coven_row_routes",
824    "remote_objects",
825    "blob_locators",
826    "row_blob_locators",
827    "store_device_registration_activations",
828    "store_device_state_snapshots",
829    "store_device_states",
830    "store_author_exclusion_activations",
831    "store_publication_current",
832    "store_publication_entries",
833    // Successor Circle heads locate their signed activation through this index.
834    // Keep it in both the shared image and the recipient's replay baseline.
835    "stream_activations",
836    "circle_control_activations",
837    "circle_access_cache",
838    "circle_bootstrap_coverage",
839    "circle_current_state",
840    "retained_merge_materializations",
841    "retained_replay_objects",
842];
843
844const CIRCLE_IMAGE_PRESERVED_NON_SYNCED_TABLES: &[&str] = &[
845    "_coven_audience",
846    "_coven_row_routes",
847    "remote_objects",
848    "blob_locators",
849    "row_blob_locators",
850    "retained_merge_materializations",
851    "retained_replay_objects",
852];
853
854fn scope_authenticated_blob_graph(
855    connection: &Connection,
856    synced: &[SyncedTable],
857    audience: &coven_protocol::circle::Audience,
858) -> Result<(), SnapshotImageError> {
859    connection
860        .execute_batch(
861            "CREATE TEMP TABLE snapshot_live_blob_bindings (
862                 table_name TEXT NOT NULL,
863                 row_id TEXT NOT NULL,
864                 column_name TEXT NOT NULL,
865                 row_stamp TEXT NOT NULL,
866                 PRIMARY KEY (table_name, row_id, column_name, row_stamp)
867             ) STRICT;",
868        )
869        .map_err(|error| SnapshotImageError::ProjectionSqlite {
870            operation: "create blob scope".to_string(),
871            source: error,
872        })?;
873    for table in synced {
874        let Some(declaration) = table.blob() else {
875            continue;
876        };
877        connection
878            .execute(
879                &format!(
880                    "INSERT INTO snapshot_live_blob_bindings
881                     (table_name, row_id, column_name, row_stamp)
882                     SELECT ?1, id, ?2, _updated_at FROM {}
883                     WHERE {} IS NOT NULL",
884                    crate::quote_ident(table.name()),
885                    crate::quote_ident(&declaration.id_column),
886                ),
887                rusqlite::params![table.name(), &declaration.id_column],
888            )
889            .map_err(|error| SnapshotImageError::ProjectionSqlite {
890                operation: format!("collect live blob bindings for {:?}", table.name()),
891                source: error,
892            })?;
893    }
894    // As above: the projection prunes the copy's rows, never this device's
895    // payload claims.
896    connection.execute_batch(
897        "DELETE FROM row_blob_locators
898             WHERE NOT EXISTS (
899                 SELECT 1 FROM snapshot_live_blob_bindings AS live
900                 WHERE live.table_name = row_blob_locators.table_name
901                   AND live.row_id = row_blob_locators.row_id
902                   AND live.column_name = row_blob_locators.column_name
903                   AND live.row_stamp = row_blob_locators.row_stamp
904             );",
905    )?;
906    // Accepted Store blobs stay in the encrypted inventory until their exact
907    // deletion receipt. Their source packages may already have been retired.
908    // Circle images carry only their live row bindings and strip transport
909    // state after collecting the bootstrap closure.
910    let mut statement = connection.prepare(
911        "SELECT remote_object_id FROM blob_locators
912         WHERE NOT EXISTS (SELECT 1 FROM row_blob_locators AS binding
913                           WHERE binding.remote_object_id = blob_locators.remote_object_id)",
914    )?;
915    let orphan_ids = statement
916        .query_map([], |row| row.get::<_, String>(0))?
917        .collect::<Result<Vec<_>, _>>()?;
918    drop(statement);
919    for object_id in orphan_ids {
920        let id = object_id
921            .parse()
922            .map_err(|error| DbError::context("snapshot inventory object id", error))?;
923        let remote = crate::remote_object_records::load_remote_object_on(connection, id)?;
924        let keep = if matches!(audience, coven_protocol::circle::Audience::Store)
925            && remote.is_activated_stored_blob()
926        {
927            let locator =
928                crate::blob_records::carried_blob_locator(&remote, "snapshot blob inventory")?;
929            locator.audience() == RemoteAudience::Store
930                && !remote.stored_blob_commit_owners().is_empty()
931        } else {
932            false
933        };
934        if !keep {
935            connection.execute(
936                "DELETE FROM blob_locators WHERE remote_object_id = ?1",
937                [&object_id],
938            )?;
939        }
940    }
941    connection
942        .execute_batch(
943            "
944             DELETE FROM remote_objects
945             WHERE NOT EXISTS (
946                 SELECT 1 FROM blob_locators AS locator
947                 WHERE locator.remote_object_id = remote_objects.object_id
948             ) AND NOT EXISTS (
949                 SELECT 1 FROM retained_replay_objects AS retained
950                 WHERE retained.object_id = remote_objects.object_id
951             );
952             DROP TABLE snapshot_live_blob_bindings;",
953        )
954        .map_err(|error| SnapshotImageError::ProjectionSqlite {
955            operation: "scope blob ownership graph".to_string(),
956            source: error,
957        })?;
958    Ok(())
959}
960
961pub(super) fn verify_circle_bootstrap_image(
962    image: &[u8],
963    reference: &coven_protocol::circle::CircleBootstrapRef,
964    circle_id: coven_protocol::circle::CircleId,
965    tables: &[SyncedTable],
966    routing_key: Option<&coven_protocol::circle::RowRoutingKey>,
967) -> Result<(), SnapshotImageError> {
968    if coven_protocol::store_commit::ObjectHash::digest(image) != reference.image.image_hash {
969        return Err(SnapshotImageError::Projection(
970            "Circle bootstrap image differs from its signed hash".to_string(),
971        ));
972    }
973    let mut connection = Connection::open_in_memory()
974        .map_err(DbError::from)
975        .map_err(SnapshotImageError::from)?;
976    crate::connection_io::deserialize_database_image_into(&mut connection, image)
977        .map_err(SnapshotImageError::from)?;
978    verify_circle_bootstrap_connection(&connection, reference, circle_id, tables, routing_key)
979}
980
981pub(crate) fn verify_circle_bootstrap_connection(
982    connection: &Connection,
983    reference: &coven_protocol::circle::CircleBootstrapRef,
984    circle_id: coven_protocol::circle::CircleId,
985    tables: &[SyncedTable],
986    routing_key: Option<&coven_protocol::circle::RowRoutingKey>,
987) -> Result<(), SnapshotImageError> {
988    connection
989        .pragma_update(None, "foreign_keys", "ON")
990        .map_err(SnapshotImageError::from)?;
991    let schema_version: u32 = connection
992        .pragma_query_value(None, "user_version", |row| row.get(0))
993        .map_err(SnapshotImageError::from)?;
994    if schema_version != reference.schema_version {
995        return Err(SnapshotImageError::Projection(format!(
996            "Circle bootstrap schema is {schema_version}, expected {}",
997            reference.schema_version
998        )));
999    }
1000    let routing_contract = crate::SyncRoutingContract::from_connection(connection, tables)
1001        .map_err(SnapshotImageError::from)?;
1002    if routing_contract.hash() != reference.sync_routing_hash {
1003        return Err(SnapshotImageError::Projection(
1004            "Circle bootstrap routing contract differs from its signed hash".to_string(),
1005        ));
1006    }
1007    let gates = crate::Gates::from_tables(connection, tables).map_err(SnapshotImageError::from)?;
1008    if gates.has_scoped_graph() {
1009        let routing_key = routing_key.ok_or_else(|| {
1010            SnapshotImageError::Projection(
1011                "scoped Circle bootstrap verification requires Store routing authentication"
1012                    .to_string(),
1013            )
1014        })?;
1015        crate::validate_snapshot_routing_state(
1016            connection,
1017            &gates,
1018            routing_key,
1019            &coven_protocol::circle::Audience::Circle(circle_id),
1020        )
1021        .map_err(SnapshotImageError::from)?;
1022    }
1023    let declarations =
1024        crate::BlobDecls::from_tables(connection, tables).map_err(SnapshotImageError::from)?;
1025    let rows = declarations
1026        .publication_blobs_in_db(connection)
1027        .map_err(SnapshotImageError::from)?;
1028    if rows.len() != reference.blobs.len() {
1029        return Err(SnapshotImageError::Projection(
1030            "Circle bootstrap blob closure does not exactly cover its image rows".to_string(),
1031        ));
1032    }
1033    for row in &rows {
1034        let mut matching = reference.blobs.iter().filter(|binding| {
1035            row.table == binding.table()
1036                && row.row_id == binding.row_id()
1037                && row.row_stamp == binding.row_stamp()
1038                && row.column == binding.column()
1039        });
1040        let binding = matching.next().ok_or_else(|| {
1041            SnapshotImageError::Projection(
1042                "Circle bootstrap image row has no exact signed blob binding".to_string(),
1043            )
1044        })?;
1045        if matching.next().is_some()
1046            || &row.blob != binding.blob()
1047            || row.plaintext_size != binding.plaintext_size()
1048            || row.plaintext_hash != binding.plaintext_hash().to_string()
1049            || !matches!(
1050                binding.authority(),
1051                coven_protocol::blob::RowBlobAuthority::Remote(
1052                    coven_protocol::audience_package::PackageAudience::Circle {
1053                        circle_id: binding_circle,
1054                        ..
1055                    }
1056                ) if *binding_circle == circle_id
1057            )
1058            || binding.stored().is_none()
1059        {
1060            return Err(SnapshotImageError::Projection(
1061                "Circle bootstrap blob closure differs from an exact image row".to_string(),
1062            ));
1063        }
1064    }
1065    for table in crate::user_table_names(connection).map_err(SnapshotImageError::from)? {
1066        if tables.iter().any(|synced| synced.name() == table)
1067            || matches!(table.as_str(), "_coven_audience" | "_coven_row_routes")
1068        {
1069            continue;
1070        }
1071        let count: i64 = connection
1072            .query_row(
1073                &format!("SELECT COUNT(*) FROM {}", crate::quote_ident(&table)),
1074                [],
1075                |row| row.get(0),
1076            )
1077            .map_err(SnapshotImageError::from)?;
1078        if count != 0 {
1079            return Err(SnapshotImageError::Projection(format!(
1080                "Circle bootstrap retains non-projection table {table:?}"
1081            )));
1082        }
1083    }
1084    Ok(())
1085}
1086
1087#[cfg(test)]
1088#[path = "snapshot_image_tests.rs"]
1089mod tests;
coven_database/store/store_session/snapshot_image.rs

coven_database/store/store_session/
snapshot_image.rs
