coven_database/store/store_session/
snapshot_capture.rs

1use std::path::{Path, PathBuf};
2
3use super::snapshot_image::{snapshot_image_db_error, verify_circle_bootstrap_image};
4use super::*;
5use crate::*;
6
7impl VerifiedStoreTransaction<'_, '_, '_, '_> {
8    /// Reconstruct the Store as of `cut` and serialize it as a replay baseline,
9    /// alongside the write-journal prefix the image now states.
10    ///
11    /// The live database is left exactly as it was: the projection is built
12    /// from this transaction into a separate replay database. Its frontier is
13    /// `cut` — checked, not assumed — which is the one property a baseline
14    /// image must have, because replay applies the retained commits the cut
15    /// does not cover on top of it.
16    ///
17    /// It also folds in the local partitions of the writes settled at `cut`.
18    /// A local partition is stated nowhere else — no commit carries one, and an
19    /// image projected for an audience may not — so without this the journal is
20    /// the durable home of every local row a device has ever written, replayed
21    /// in full on every rebuild and never shorter. Folding them in is what lets
22    /// the advance adopting this image delete them, and the returned write ids
23    /// are exactly what it may delete.
24    pub(super) fn capture_replay_baseline_at_cut(
25        &mut self,
26        root: &coven_protocol::store_commit::StoreRootRef,
27        cut: &coven_protocol::store_commit::CommitFrontier,
28        current_cut: &coven_protocol::store_commit::CommitFrontier,
29        snapshot_hash: crate::ObjectHash,
30        routing_encryption: Option<&coven_keys::encryption::EncryptionService>,
31    ) -> Result<(Vec<u8>, Vec<crate::SettledStoreWrite>), DbError> {
32        if self.authority.root() != root {
33            return Err(DbError::Message(
34                "replay baseline belongs to another Store root".to_string(),
35            ));
36        }
37        let routing_key = if self.gates.has_scoped_graph() {
38            let encryption = routing_encryption.ok_or_else(|| {
39                DbError::Message(
40                    "scoped replay baseline capture requires Store routing encryption".to_string(),
41                )
42            })?;
43            Some(
44                coven_protocol::circle::derive_row_routing_key(encryption, root.store_root_hash)
45                    .map_err(DbError::from)?,
46            )
47        } else {
48            None
49        };
50        let folded = crate::StoreDatabase::settled_store_write_prefix_on(
51            crate::store::store_session::StoreRecords::new(
52                self.store.transaction,
53                self.store.store_dir,
54            ),
55            cut,
56        )?;
57        let current_replay = self.authority.replay_projection_result_on(
58            self.store,
59            self.blob_decls,
60            self.gates,
61            self.synced_tables,
62            routing_key.as_ref(),
63            Some(current_cut),
64            crate::ReplayJournal::Omit,
65            coven_protocol::membership::LocalStoreMembership::Current,
66        )?;
67        if current_replay.materialized_frontier()? != *current_cut {
68            return Err(DbError::Message(
69                "replay retirement proof does not cover the current Store frontier".to_string(),
70            ));
71        }
72        let mut crossed_cut = false;
73        for reference in current_replay.applied_order() {
74            if cut.covers_commit(reference) {
75                if crossed_cut {
76                    return Err(DbError::ReplayRetirementCutNotPrefix);
77                }
78            } else {
79                crossed_cut = true;
80            }
81        }
82        let replay = self.authority.replay_projection_result_on(
83            self.store,
84            self.blob_decls,
85            self.gates,
86            self.synced_tables,
87            routing_key.as_ref(),
88            Some(cut),
89            crate::ReplayJournal::Folded(&folded),
90            coven_protocol::membership::LocalStoreMembership::Current,
91        )?;
92        let replay_frontier = replay.materialized_frontier()?;
93        if replay_frontier != *cut {
94            return Err(DbError::Message(
95                "retained replay baseline cut is not an exact Store frontier".to_string(),
96            ));
97        }
98        Ok((
99            replay.capture_replay_baseline(root, cut, snapshot_hash)?,
100            folded,
101        ))
102    }
103}
104
105impl StoreSession<'_> {
106    fn capture_shared_snapshot_cut(
107        &mut self,
108        root: &coven_protocol::store_commit::StoreRootRef,
109        temp_dir: &Path,
110        routing_encryption: Option<&coven_keys::encryption::EncryptionService>,
111        audience: coven_protocol::circle::Audience,
112    ) -> Result<
113        (
114            CreatedSnapshot,
115            coven_protocol::store_commit::CommitFrontier,
116        ),
117        DbError,
118    > {
119        let routing_key = if self.gates.has_scoped_graph() {
120            let encryption = routing_encryption.ok_or_else(|| {
121                DbError::Message(
122                    "scoped snapshot capture requires Store routing encryption".to_string(),
123                )
124            })?;
125            Some(
126                coven_protocol::circle::derive_row_routing_key(encryption, root.store_root_hash)
127                    .map_err(DbError::from)?,
128            )
129        } else {
130            None
131        };
132        self.verified_store_transaction(|transaction| {
133            if transaction.authority.root() != root {
134                return Err(DbError::Message(
135                    "snapshot capture belongs to another Store root".to_string(),
136                ));
137            }
138            let coverage = coven_protocol::store_commit::CommitFrontier::from_refs(
139                crate::store::materialized_commit_index::materialized_frontier_on(
140                    transaction.store.transaction,
141                    None,
142                )?,
143            )
144            .map_err(|error| DbError::context("snapshot coverage", error))?;
145            for entry in super::observed_store_publication::load_store_publication_entries_on(
146                transaction.store.transaction,
147            )? {
148                if let coven_protocol::store_commit::StorePublicationPayload::Commit(reference) =
149                    &entry.value.payload
150                {
151                    if !coverage.covers_commit(reference) {
152                        return Err(DbError::Message(format!(
153                            "snapshot capture is missing accepted commit {reference:?}"
154                        )));
155                    }
156                }
157            }
158            let replay = transaction.authority.replay_projection_result_on(
159                transaction.store,
160                transaction.blob_decls,
161                transaction.gates,
162                transaction.synced_tables,
163                routing_key.as_ref(),
164                Some(&coverage),
165                crate::ReplayJournal::Omit,
166                coven_protocol::membership::LocalStoreMembership::Current,
167            )?;
168            if replay.materialized_frontier()? != coverage {
169                return Err(DbError::Message(
170                    "snapshot capture could not reconstruct the accepted Store frontier"
171                        .to_string(),
172                ));
173            }
174            let snapshot = SnapshotDatabaseImage::prepare_snapshot(temp_dir)
175                .and_then(|image| {
176                    replay.capture_snapshot(
177                        image,
178                        root,
179                        transaction.synced_tables,
180                        routing_encryption,
181                        &audience,
182                    )
183                })
184                .map_err(snapshot_image_db_error)?;
185            Ok(StoreTransactionOutcome::Rollback((snapshot, coverage)))
186        })
187    }
188
189    #[allow(clippy::too_many_arguments)]
190    fn capture_circle_snapshot_at_cutoff(
191        &mut self,
192        root: &coven_protocol::store_commit::StoreRootRef,
193        temp_dir: &Path,
194        routing_encryption: &coven_keys::encryption::EncryptionService,
195        routing_key: &coven_protocol::circle::RowRoutingKey,
196        circle_id: coven_protocol::circle::CircleId,
197        cutoff: &coven_protocol::store_commit::CommitFrontier,
198    ) -> Result<CreatedSnapshot, DbError> {
199        let transaction = self.conn.unchecked_transaction().map_err(DbError::from)?;
200        let replay =
201            crate::store::store_session::StoreTransaction::new(&transaction, self.store_dir)
202                .replay_projection_with_authority(
203                    self.verified_store_authority,
204                    root,
205                    self.blob_decls,
206                    self.gates,
207                    self.synced_tables,
208                    Some(routing_key),
209                    &std::collections::BTreeSet::new(),
210                    Some(cutoff),
211                    crate::ReplayJournal::Omit,
212                    coven_protocol::membership::LocalStoreMembership::Current,
213                )?;
214        transaction.rollback().map_err(DbError::from)?;
215        let replay_frontier = replay.materialized_frontier()?;
216        if replay_frontier != *cutoff {
217            return Err(DbError::Message(
218                "Circle close cutoff is not an exact retained Store frontier".to_string(),
219            ));
220        }
221        SnapshotDatabaseImage::prepare_snapshot(temp_dir)
222            .and_then(|image| {
223                replay.capture_snapshot(
224                    image,
225                    root,
226                    self.synced_tables,
227                    Some(routing_encryption),
228                    &coven_protocol::circle::Audience::Circle(circle_id),
229                )
230            })
231            .map_err(snapshot_image_db_error)
232    }
233
234    #[cfg(any(test, feature = "test-utils"))]
235    fn capture_snapshot_image_for_test(
236        &self,
237        root: &coven_protocol::store_commit::StoreRootRef,
238        temp_dir: &Path,
239        routing_encryption: Option<&coven_keys::encryption::EncryptionService>,
240        audience: coven_protocol::circle::Audience,
241    ) -> Result<Vec<u8>, DbError> {
242        SnapshotDatabaseImage::prepare_snapshot(temp_dir)
243            .and_then(|image| {
244                StoreRecords::new(self.conn, self.store_dir).capture_snapshot(
245                    image,
246                    root,
247                    self.synced_tables,
248                    routing_encryption,
249                    &audience,
250                )
251            })
252            .and_then(|snapshot| snapshot.into_parts().0.read_and_discard())
253            .map_err(snapshot_image_db_error)
254    }
255}
256
257impl StoreDatabase {
258    pub async fn capture_store_snapshot_cut(
259        &self,
260        root: coven_protocol::store_commit::StoreRootRef,
261        temp_dir: PathBuf,
262        routing_encryption: Option<coven_keys::encryption::EncryptionService>,
263    ) -> Result<
264        (
265            CreatedSnapshot,
266            coven_protocol::store_commit::CommitFrontier,
267        ),
268        DbError,
269    > {
270        self.call_store(move |session| {
271            session.capture_shared_snapshot_cut(
272                &root,
273                &temp_dir,
274                routing_encryption.as_ref(),
275                coven_protocol::circle::Audience::Store,
276            )
277        })
278        .await
279    }
280
281    pub async fn capture_circle_snapshot_cut(
282        &self,
283        root: coven_protocol::store_commit::StoreRootRef,
284        temp_dir: PathBuf,
285        routing_encryption: coven_keys::encryption::EncryptionService,
286        circle_id: coven_protocol::circle::CircleId,
287    ) -> Result<
288        (
289            CreatedSnapshot,
290            coven_protocol::store_commit::CommitFrontier,
291        ),
292        DbError,
293    > {
294        self.call_store(move |session| {
295            session.capture_shared_snapshot_cut(
296                &root,
297                &temp_dir,
298                Some(&routing_encryption),
299                coven_protocol::circle::Audience::Circle(circle_id),
300            )
301        })
302        .await
303    }
304
305    #[allow(clippy::too_many_arguments)]
306    pub async fn capture_circle_snapshot_at_cutoff(
307        &self,
308        root: coven_protocol::store_commit::StoreRootRef,
309        temp_dir: PathBuf,
310        routing_encryption: coven_keys::encryption::EncryptionService,
311        routing_key: coven_protocol::circle::RowRoutingKey,
312        circle_id: coven_protocol::circle::CircleId,
313        cutoff: coven_protocol::store_commit::CommitFrontier,
314    ) -> Result<CreatedSnapshot, DbError> {
315        self.call_store(move |session| {
316            session.capture_circle_snapshot_at_cutoff(
317                &root,
318                &temp_dir,
319                &routing_encryption,
320                &routing_key,
321                circle_id,
322                &cutoff,
323            )
324        })
325        .await
326    }
327
328    pub async fn verify_circle_bootstrap_image(
329        &self,
330        image: Vec<u8>,
331        reference: coven_protocol::circle::CircleBootstrapRef,
332        circle_id: coven_protocol::circle::CircleId,
333        routing_key: Option<coven_protocol::circle::RowRoutingKey>,
334    ) -> Result<Vec<u8>, SnapshotImageError> {
335        self.call_store(move |session| {
336            let verification = verify_circle_bootstrap_image(
337                &image,
338                &reference,
339                circle_id,
340                session.synced_tables,
341                routing_key.as_ref(),
342            );
343            Ok(verification.map(|()| image))
344        })
345        .await
346        .map_err(SnapshotImageError::from)?
347    }
348
349    #[cfg(any(test, feature = "test-utils"))]
350    pub async fn capture_snapshot_image_for_test(
351        &self,
352        root: coven_protocol::store_commit::StoreRootRef,
353        temp_dir: PathBuf,
354        routing_encryption: Option<coven_keys::encryption::EncryptionService>,
355    ) -> Result<Vec<u8>, DbError> {
356        self.call_store(move |session| {
357            session.capture_snapshot_image_for_test(
358                &root,
359                &temp_dir,
360                routing_encryption.as_ref(),
361                coven_protocol::circle::Audience::Store,
362            )
363        })
364        .await
365    }
366
367    #[cfg(any(test, feature = "test-utils"))]
368    pub async fn capture_circle_snapshot_image_for_test(
369        &self,
370        root: coven_protocol::store_commit::StoreRootRef,
371        temp_dir: PathBuf,
372        routing_encryption: coven_keys::encryption::EncryptionService,
373        circle_id: coven_protocol::circle::CircleId,
374    ) -> Result<Vec<u8>, DbError> {
375        self.call_store(move |session| {
376            session.capture_snapshot_image_for_test(
377                &root,
378                &temp_dir,
379                Some(&routing_encryption),
380                coven_protocol::circle::Audience::Circle(circle_id),
381            )
382        })
383        .await
384    }
385}
coven_database/store/store_session/snapshot_capture.rs

coven_database/store/store_session/
snapshot_capture.rs
