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

coven_database/store/store_session/
received_snapshot.rs
