coven_database/store/store_session/
received_snapshot.rs

1use super::*;
2
3impl StoreSession<'_> {
4    pub(crate) fn prepare_received_snapshot_circles(
5        &mut self,
6        selection: &crate::StagedCircleRestore,
7        receiver_wall_ms: u64,
8    ) -> Result<(), DbError> {
9        let root = self.required_root_authority()?;
10        let transaction = self.conn.unchecked_transaction()?;
11        StoreTransaction::new(&transaction, self.store_dir).restore_device_join_snapshot_circles(
12            &root,
13            selection,
14            self.synced_tables,
15            self.blob_decls,
16            receiver_wall_ms,
17        )?;
18        transaction.commit()?;
19        self.verified_store_authority
20            .forget_superseded_replay_baseline();
21        Ok(())
22    }
23}
24
25/// A closed, migrated snapshot database and the directory owning its payloads.
26/// Installation opens the database only within the receiving database operation.
27pub struct PreparedStoreSnapshot {
28    directory: SnapshotPreparationDirectory,
29}
30
31impl PreparedStoreSnapshot {
32    pub(crate) fn seal(core: crate::DatabaseCore) -> Result<Self, DbError> {
33        Ok(Self {
34            directory: core.close_snapshot()?,
35        })
36    }
37
38    fn read<T>(
39        &self,
40        read: impl FnOnce(
41            &rusqlite::Connection,
42            &coven_foundation::store_dir::StoreDir,
43        ) -> Result<T, DbError>,
44    ) -> Result<T, DbError> {
45        let directory = coven_foundation::store_dir::StoreDir::new_ephemeral(&self.directory.path);
46        // Read through SQLite so committed WAL pages remain part of the source.
47        // The preparation owns the database, its sidecars, and payloads together.
48        let source = rusqlite::Connection::open_with_flags(
49            directory.db_path(),
50            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
51        )?;
52        read(&source, &directory)
53    }
54
55    pub(crate) fn install_on(
56        self,
57        receiver: &mut crate::store::StoreSession<'_>,
58        expected: crate::StorePublicationBoundary,
59        materializations: Vec<crate::PreparedMergeMaterialization>,
60        accepted: crate::AcceptedStorePublicationInterval,
61        mut snapshots: Vec<coven_protocol::store_commit::RetainedReplaySnapshotAuthority>,
62        membership: coven_protocol::membership::LocalStoreMembership,
63        routing_encryption: Option<&coven_keys::encryption::EncryptionService>,
64        routing_key: Option<coven_protocol::circle::RowRoutingKey>,
65        receiver_wall_ms: u64,
66    ) -> Result<
67        (
68            crate::store::AppliedMergeMaterialization,
69            Vec<coven_protocol::store_commit::StoreBatchCommitRef>,
70        ),
71        DbError,
72    > {
73        let outcome = self.read(|source, source_dir| {
74            let source_records = StoreRecords::new(source, source_dir);
75            let source_version: u32 =
76                source.pragma_query_value(None, "user_version", |row| row.get(0))?;
77            let source_routing = crate::database_open::load_coven_metadata(source)?;
78            if source_version != receiver.schema_version
79                || source_routing.hash() != receiver.sync_routing_hash
80            {
81                return Err(DbError::Message(
82                    "prepared checkpoint differs from the receiver schema".into(),
83                ));
84            }
85            let root = receiver.required_root_authority()?;
86            let mut source_authority = VerifiedStoreAuthority::default();
87            if source_authority.required_root_authority_on(source_records)? != root {
88                return Err(DbError::Message(
89                    "prepared checkpoint belongs to another Store".into(),
90                ));
91            }
92            let baseline =
93                retained_replay::load_replay_baseline_on(source_records)?.ok_or_else(|| {
94                    DbError::Message("prepared checkpoint has no replay baseline".into())
95                })?;
96            let crate::RetainedReplayAuthority::InstalledSnapshot(checkpoint) = &baseline.authority
97            else {
98                return Err(DbError::Message(
99                    "prepared checkpoint has genesis authority".into(),
100                ));
101            };
102            let snapshot = accepted
103                .interval()
104                .current()
105                .latest_snapshot()
106                .cloned()
107                .ok_or_else(|| {
108                    DbError::Message("received interval has no accepted checkpoint".into())
109                })?;
110            if checkpoint.snapshot != snapshot.snapshot {
111                return Err(DbError::Message(
112                    "prepared image is not the received current checkpoint".into(),
113                ));
114            }
115            snapshots.push(checkpoint.clone());
116            let inputs = source_authority
117                .retained_replay_inputs_on(source_records, &root)
118                .map_err(|error| DbError::context("checkpoint source retained inputs", error))?;
119            let source_floor = crate::connection_io::parse_seed(
120                crate::get_protocol_state_on(source, coven_protocol::hlc::HIGHWATER_STATE_KEY)?,
121                "received checkpoint clock floor",
122            )?;
123            let row_floor = crate::connection_io::parse_seed(
124                crate::connection_io::scan_max_updated_at(
125                    source,
126                    receiver.synced_tables,
127                    receiver_wall_ms.saturating_add(coven_protocol::hlc::MAX_FUTURE_SKEW_MS),
128                )?,
129                "received checkpoint row clock",
130            )?;
131            let schema_version = receiver.schema_version;
132            let routing_hash = receiver.sync_routing_hash;
133            receiver.verified_store_transaction(|transaction| {
134                if observed_store_publication::load_store_current_publication_on(
135                    transaction.store.transaction,
136                )? != expected
137                {
138                    return Err(DbError::StorePublicationChanged);
139                }
140                let frontier = coven_protocol::store_commit::CommitFrontier::from_refs(
141                    materialized_commit_index::materialized_frontier_on(
142                        transaction.store.transaction,
143                        None,
144                    )?,
145                )?;
146                let (local_image, folded) = transaction
147                    .capture_replay_baseline_at_cut(
148                        &root,
149                        &frontier,
150                        &frontier,
151                        snapshot.snapshot.snapshot_hash,
152                        routing_encryption,
153                    )
154                    .map_err(|error| {
155                        DbError::context("checkpoint receiver replay capture", error)
156                    })?;
157                let covered_suffix = StoreDatabase::covered_replay_suffix_on(
158                    StoreRecords::new(transaction.store.transaction, transaction.store.store_dir),
159                    &baseline,
160                    &folded,
161                )?;
162                let image = source_records
163                    .received_snapshot_image_with_local_rows(
164                        &local_image,
165                        transaction.gates,
166                        &covered_suffix,
167                    )
168                    .map_err(|error| DbError::context("checkpoint Local rows", error))?;
169                transaction
170                    .store
171                    .import_snapshot_device_states(source_records)
172                    .map_err(|error| DbError::context("checkpoint device states", error))?;
173                let installed_baseline = transaction
174                    .store
175                    .replace_received_snapshot_baseline(
176                        source_records,
177                        &baseline,
178                        image,
179                        &folded,
180                        &inputs,
181                        transaction.blob_decls,
182                    )
183                    .map_err(|error| DbError::context("checkpoint baseline replacement", error))?;
184                observed_store_publication::install_store_checkpoint_publication_on(
185                    transaction.store.transaction,
186                    &expected,
187                    &accepted,
188                    checkpoint,
189                )
190                .map_err(|error| DbError::context("checkpoint publication boundary", error))?;
191                transaction
192                    .store
193                    .import_received_snapshot_inputs(source_records, &inputs, &installed_baseline)
194                    .map_err(|error| {
195                        DbError::context("checkpoint retained inputs import", error)
196                    })?;
197                transaction
198                    .store
199                    .import_received_snapshot_blob_inventory(source_records)
200                    .map_err(|error| DbError::context("checkpoint blob inventory", error))?;
201                // The replacement was validated against the received image. Its
202                // registration rows reach the live projection below, so reopening
203                // it against the previous projection would mix those two states.
204                transaction
205                    .authority
206                    .replace_installed_replay_baseline(installed_baseline)?;
207                transaction.clock_floor = source_floor.clone().max(row_floor.clone());
208                if let Some(floor) = &transaction.clock_floor {
209                    transaction.clock.advance_past(floor);
210                }
211                let replay = accepted.interval().clone();
212                let applied = transaction
213                    .apply_received_store_publication_interval(
214                        materializations,
215                        accepted,
216                        replay,
217                        snapshots,
218                        membership,
219                        schema_version,
220                        routing_hash,
221                        routing_encryption,
222                        routing_key,
223                        receiver_wall_ms,
224                        Some(snapshot),
225                    )
226                    .map_err(|error| DbError::context("checkpoint received interval", error))?;
227                if matches!(applied.0.outcome, crate::MaterializationOutcome::Applied(_)) {
228                    transaction.store.retain_snapshot_device_states(
229                        &mut *transaction.authority,
230                        &root,
231                        checkpoint.metadata.coverage.clone().into_refs(),
232                    )?;
233                    let installed_records = StoreRecords::new(
234                        transaction.store.transaction,
235                        transaction.store.store_dir,
236                    );
237                    let installed = retained_replay::load_replay_baseline_on(installed_records)?
238                        .ok_or_else(|| {
239                            DbError::Message(
240                                "received checkpoint installation lost its baseline".into(),
241                            )
242                        })?;
243                    if installed.authority != baseline.authority {
244                        return Err(DbError::Message(
245                            "received checkpoint installation changed its accepted authority"
246                                .into(),
247                        ));
248                    }
249                    installed_records.declared_store_device_state(
250                        &checkpoint.metadata.history_summary.post_state,
251                    )?;
252                    installed_records.store_device_state_for_history_cut(
253                        &coven_protocol::store_commit::StoreHistoryCut(
254                            coven_protocol::store_commit::CommitFrontier::from_refs(
255                                installed_records.materialized_frontier()?,
256                            )?
257                            .0,
258                        ),
259                    )?;
260                    Ok(StoreTransactionOutcome::Commit(applied))
261                } else {
262                    Ok(StoreTransactionOutcome::Rollback(applied))
263                }
264            })
265        });
266        self.directory.finish(outcome)
267    }
268
269    /// Release a preparation whose verified interval cannot be installed.
270    #[cfg(test)]
271    pub(crate) fn discard(self) -> Result<(), DbError> {
272        self.directory.finish(Ok(()))
273    }
274}
275
276/// Created exclusively for one preparation, including its file-backed payloads.
277/// The connection is declared before this guard so it closes before cleanup.
278pub(crate) struct SnapshotPreparationDirectory {
279    path: std::path::PathBuf,
280    armed: bool,
281}
282
283impl SnapshotPreparationDirectory {
284    pub(crate) fn create(path: std::path::PathBuf) -> Result<Self, DbError> {
285        std::fs::create_dir(&path)
286            .map_err(|error| DbError::context("create snapshot preparation directory", error))?;
287        Ok(Self { path, armed: true })
288    }
289
290    pub(crate) fn finish<T>(mut self, outcome: Result<T, DbError>) -> Result<T, DbError> {
291        let cleanup = std::fs::remove_dir_all(&self.path);
292        self.armed = false;
293        match (outcome, cleanup) {
294            (Ok(value), Ok(())) => Ok(value),
295            (Err(error), Ok(())) => Err(error),
296            (Ok(_), Err(cleanup)) => Err(crate::SnapshotImageError::Cleanup {
297                path: self.path.clone(),
298                cleanup: cleanup.to_string(),
299            }
300            .into()),
301            (Err(operation), Err(cleanup)) => Err(crate::SnapshotImageError::CleanupAfterFailure {
302                path: self.path.clone(),
303                cleanup: cleanup.to_string(),
304                cause: Box::new(crate::SnapshotImageError::from(operation)),
305            }
306            .into()),
307        }
308    }
309}
310
311impl Drop for SnapshotPreparationDirectory {
312    fn drop(&mut self) {
313        if self.armed {
314            if let Err(error) = std::fs::remove_dir_all(&self.path) {
315                tracing::warn!(path = %self.path.display(), %error, "could not remove abandoned snapshot preparation");
316            }
317        }
318    }
319}
320
321#[cfg(test)]
322#[path = "received_snapshot_tests.rs"]
323mod tests;
coven_database/store/store_session/received_snapshot.rs

coven_database/store/store_session/
received_snapshot.rs
