coven_database/store/store_session/
snapshot_publication.rs

1use std::collections::BTreeSet;
2use std::path::{Path, PathBuf};
3
4use crate::*;
5use coven_protocol::store_commit::{
6    snapshot_image_semantic_prefix, snapshot_slot_prefix, SnapshotMeta, StoreSnapshotRef,
7};
8
9use super::*;
10
11impl StoreSession<'_> {
12    fn outbound_snapshot_publication(
13        &mut self,
14    ) -> Result<Option<DurableSnapshotPublication>, DbError> {
15        let authority = self.local_store_authority()?;
16        load_outbound_store_snapshot_on(self.conn, self.store_dir, &authority)
17    }
18
19    fn stage_snapshot_publication(
20        &mut self,
21        meta: SnapshotMeta,
22        meta_prepared: PreparedExactObject,
23        rollup_bytes: Vec<u8>,
24        rollup_prepared: PreparedExactObject,
25        image: SnapshotDatabaseImage,
26        image_prepared: PreparedExactObject,
27        blobs: Vec<PreparedSnapshotBlob>,
28    ) -> Result<StoreSnapshotRef, DbError> {
29        let authority = self.local_store_authority()?;
30        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
31        let image_facts =
32            crate::payload_store::write_payload_file_blocking(&tx, self.store_dir, image.path())
33                .map_err(|source| SnapshotImageError::ProjectionPayloadStore {
34                    operation: "spool Store snapshot image".to_string(),
35                    source,
36                });
37        let (image_hash, _) = image.finish(image_facts).map_err(snapshot_image_db_error)?;
38        let image_prepared_hash = crate::payload_store::write_payload_blocking(
39            &tx,
40            self.store_dir,
41            image_prepared.stored_bytes(),
42        )
43        .map_err(|error| DbError::context("spool prepared Store snapshot image", error))?;
44        let image_prepared_size = image_prepared.stored_bytes().len() as u64;
45        let registration_ref = authority.reference();
46        let registration = authority.value();
47        validate_snapshot_author(&meta.author_registration, registration_ref, "Store")?;
48        validate_snapshot_image(
49            &meta.image,
50            &image_prepared,
51            image_hash,
52            image_prepared_hash,
53            image_prepared_size,
54            format!(
55                "{}.db",
56                snapshot_image_semantic_prefix(
57                    &registration.device_id.to_string(),
58                    meta.image.image_hash,
59                )
60            ),
61            "Store",
62        )?;
63        let reference = StoreSnapshotRef {
64            generation: meta.generation,
65            snapshot_hash: meta.snapshot_hash(),
66            object: meta_prepared.reference().clone(),
67        };
68        let verified = SnapshotMeta::parse_at(
69            &meta.to_bytes(),
70            registration.store_root.store_root_hash,
71            &reference,
72            registration,
73        )
74        .map_err(|error| DbError::context("verify staged Store snapshot metadata", error))?;
75        if verified != meta {
76            return Err(DbError::Message(
77                "staged Store snapshot changed during exact verification".to_string(),
78            ));
79        }
80        coven_protocol::store_commit::MembershipRollup::parse_at(
81            &rollup_bytes,
82            registration.store_root.store_root_hash,
83            &meta.membership_rollup,
84            registration,
85        )
86        .map_err(|error| DbError::context("verify staged membership rollup", error))?;
87        if rollup_prepared.reference() != &meta.membership_rollup.object {
88            return Err(DbError::Message(
89                "staged membership rollup differs from the snapshot that names it".to_string(),
90            ));
91        }
92        // Spooled beside the image rather than carried in the row: a rollup
93        // holds every membership object the Store has, which is KB-class and
94        // belongs in the payload store.
95        let rollup_hash =
96            crate::payload_store::write_payload_blocking(&tx, self.store_dir, &rollup_bytes)
97                .map_err(|error| DbError::context("spool membership rollup", error))?;
98        let rollup_prepared_hash = crate::payload_store::write_payload_blocking(
99            &tx,
100            self.store_dir,
101            rollup_prepared.stored_bytes(),
102        )
103        .map_err(|error| DbError::context("spool prepared membership rollup", error))?;
104        if rollup_hash != meta.membership_rollup.rollup_hash {
105            return Err(DbError::Message(
106                "staged membership rollup bytes differ from the hash the snapshot names"
107                    .to_string(),
108            ));
109        }
110        let previous = load_published_store_snapshot_on(&tx, &authority)?;
111        let (expected_generation, expected_predecessor, expected_slot) = match &previous {
112            Some(previous) => (
113                previous
114                    .reference
115                    .generation
116                    .checked_add(1)
117                    .ok_or_else(|| {
118                        DbError::Message("Store snapshot generation overflow".to_string())
119                    })?,
120                Some(previous.reference.clone()),
121                previous.successor_slot.clone(),
122            ),
123            None => (0, None, store_snapshot_first_slot(registration)?.clone()),
124        };
125        if meta.generation != expected_generation
126            || meta.predecessor != expected_predecessor
127            || meta_prepared.reference().slot() != &expected_slot
128            || meta.successor.predecessor != previous.as_ref().map(|value| value.reference.clone())
129        {
130            return Err(DbError::Message(
131                "Store snapshot does not extend the exact local stream".to_string(),
132            ));
133        }
134        let next_generation = meta
135            .generation
136            .checked_add(1)
137            .ok_or_else(|| DbError::Message("Store snapshot generation overflow".to_string()))?;
138        if meta.successor.activation
139            != registration
140                .store_snapshot_activation(registration_ref)
141                .map_err(DbError::from)?
142                .activation_id()
143            || meta.successor.next_slot.logical_key()
144                != format!(
145                    "{}.json",
146                    snapshot_slot_prefix(&registration.device_id.to_string(), next_generation)
147                )
148        {
149            return Err(DbError::Message(
150                "Store snapshot successor is outside its activated exact stream".to_string(),
151            ));
152        }
153        let snapshot_owner = coven_protocol::remote_object::SnapshotObjectOwner {
154            activation: meta.successor.activation,
155            generation: meta.generation,
156        };
157        validate_snapshot_blob_plans_on(
158            self.conn,
159            self.gates,
160            self.synced_tables,
161            &snapshot_owner,
162            &blobs,
163        )?;
164        tx.execute(
165            "INSERT INTO outbound_store_snapshot \
166             (singleton, snapshot_ref, meta_prepared, image_ref, rollup_ref, \
167              meta_bytes, blobs) \
168             VALUES (1, ?1, ?2, ?3, ?4, ?5, ?6)",
169            rusqlite::params![
170                serde_json::to_string(&reference).map_err(|error| {
171                    DbError::context("serialize exact Store snapshot ref", error)
172                })?,
173                serde_json::to_string(&meta_prepared).map_err(|error| {
174                    DbError::context("serialize prepared Store snapshot metadata", error)
175                })?,
176                serde_json::to_string(&meta.image).map_err(|error| {
177                    DbError::context("serialize exact Store snapshot image ref", error)
178                })?,
179                serde_json::to_string(&meta.membership_rollup).map_err(|error| {
180                    DbError::context("serialize exact membership rollup ref", error)
181                })?,
182                meta.to_bytes(),
183                serde_json::to_string(&blobs).map_err(|error| {
184                    DbError::context("serialize prepared Store snapshot blobs", error)
185                })?,
186            ],
187        )
188        .map_err(DbError::from)?;
189        crate::payload_store::set_payload_owner_claims_on(
190            &tx,
191            crate::payload_store::OUTBOUND_STORE_SNAPSHOT_OWNER_KEY,
192            &BTreeSet::from([
193                image_hash,
194                image_prepared_hash,
195                rollup_hash,
196                rollup_prepared_hash,
197            ]),
198        )?;
199        tx.commit().map_err(DbError::from)?;
200        Ok(reference)
201    }
202
203    fn latest_local_store_snapshot(&mut self) -> Result<Option<PublishedStoreSnapshot>, DbError> {
204        let authority = self.local_store_authority()?;
205        load_published_store_snapshot_on(self.conn, &authority)
206    }
207
208    fn local_store_snapshots(&mut self) -> Result<Vec<PublishedStoreSnapshot>, DbError> {
209        let authority = self.local_store_authority()?;
210        load_published_store_snapshots_on(self.conn, &authority)
211    }
212
213    fn complete_snapshot_publication(&mut self, accepted: StoreSnapshotRef) -> Result<(), DbError> {
214        let authority = self.local_store_authority()?;
215        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
216        let outbound = load_outbound_store_snapshot_on(&tx, self.store_dir, &authority)?
217            .ok_or_else(|| DbError::Message("outbound Store snapshot is absent".to_string()))?;
218        if outbound.reference != accepted {
219            return Err(DbError::Message(
220                "accepted Store snapshot differs from the prepared exact object".to_string(),
221            ));
222        }
223        install_snapshot_blob_plans_on(&tx, &outbound.blobs)?;
224        let snapshot_owner = coven_protocol::remote_object::SnapshotObjectOwner {
225            activation: outbound.meta.value.successor.activation,
226            generation: outbound.meta.value.generation,
227        };
228        persist_snapshot_image_on(
229            &tx,
230            self.store_dir,
231            &outbound.meta.value.image,
232            snapshot_owner.clone(),
233            "Store snapshot image",
234        )?;
235        crate::snapshot_objects::persist_membership_rollup_on(
236            &tx,
237            self.store_dir,
238            &outbound.meta.value.membership_rollup,
239            snapshot_owner,
240            "Store membership rollup",
241        )?;
242        let deleted = tx
243            .execute(
244                "DELETE FROM outbound_store_snapshot \
245                 WHERE singleton = 1 AND snapshot_ref = ?1",
246                [serde_json::to_string(&accepted).map_err(|error| {
247                    DbError::context("serialize accepted Store snapshot ref", error)
248                })?],
249            )
250            .map_err(DbError::from)?;
251        if deleted != 1 {
252            return Err(DbError::Message(
253                "outbound snapshot ownership row is absent or changed".to_string(),
254            ));
255        }
256        crate::payload_store::release_payload_owner_on(
257            &tx,
258            crate::payload_store::OUTBOUND_STORE_SNAPSHOT_OWNER_KEY,
259        )?;
260        let accepted_generation =
261            snapshot_generation_as_i64(accepted.generation, "Store snapshot")?;
262        tx.execute(
263            "INSERT INTO published_store_snapshot \
264             (generation, snapshot_ref, successor_slot, meta_bytes) VALUES (?1, ?2, ?3, ?4)",
265            rusqlite::params![
266                accepted_generation,
267                serde_json::to_string(&accepted).map_err(|error| {
268                    DbError::context("serialize published Store snapshot ref", error)
269                })?,
270                serde_json::to_string(&outbound.meta.value.successor.next_slot).map_err(
271                    |error| DbError::context("serialize Store snapshot successor slot", error)
272                )?,
273                outbound.meta.bytes,
274            ],
275        )
276        .map_err(DbError::from)?;
277        tx.commit().map_err(DbError::from)
278    }
279
280    fn snapshot_blob_spool_cleanup_paths(&self) -> Result<Vec<PathBuf>, DbError> {
281        let mut statement = self
282            .conn
283            .prepare("SELECT path FROM snapshot_blob_spool_cleanup ORDER BY path")
284            .map_err(DbError::from)?;
285        let paths = statement
286            .query_map([], |row| row.get::<_, String>(0))
287            .map_err(DbError::from)?
288            .map(|row| row.map(PathBuf::from).map_err(DbError::from))
289            .collect();
290        paths
291    }
292
293    fn complete_snapshot_blob_spool_cleanup(&self, path: &str) -> Result<(), DbError> {
294        let deleted = self
295            .conn
296            .execute(
297                "DELETE FROM snapshot_blob_spool_cleanup WHERE path = ?1",
298                [path],
299            )
300            .map_err(DbError::from)?;
301        if deleted != 1 {
302            return Err(DbError::Message(
303                "snapshot blob spool cleanup ownership is absent".to_string(),
304            ));
305        }
306        Ok(())
307    }
308}
309
310impl StoreDatabase {
311    pub async fn outbound_snapshot_publication(
312        &self,
313    ) -> Result<Option<DurableSnapshotPublication>, DbError> {
314        let pending = self
315            .call_store(|session| session.outbound_snapshot_publication())
316            .await?;
317        if let Some(pending) = &pending {
318            verify_snapshot_blob_spools(&pending.blobs, "prepared").await?;
319        }
320        Ok(pending)
321    }
322
323    #[allow(clippy::too_many_arguments)]
324    pub async fn stage_snapshot_publication(
325        &self,
326        meta: SnapshotMeta,
327        meta_prepared: PreparedExactObject,
328        rollup_bytes: Vec<u8>,
329        rollup_prepared: PreparedExactObject,
330        image: SnapshotDatabaseImage,
331        image_prepared: PreparedExactObject,
332        blobs: Vec<PreparedSnapshotBlob>,
333    ) -> Result<StoreSnapshotRef, DbError> {
334        self.call_store(move |session| {
335            session.stage_snapshot_publication(
336                meta,
337                meta_prepared,
338                rollup_bytes,
339                rollup_prepared,
340                image,
341                image_prepared,
342                blobs,
343            )
344        })
345        .await
346    }
347
348    pub async fn latest_local_store_snapshot(
349        &self,
350    ) -> Result<Option<PublishedStoreSnapshot>, DbError> {
351        self.call_store(|session| session.latest_local_store_snapshot())
352            .await
353    }
354
355    pub async fn local_store_snapshots(&self) -> Result<Vec<PublishedStoreSnapshot>, DbError> {
356        self.call_store(|session| session.local_store_snapshots())
357            .await
358    }
359
360    pub async fn complete_snapshot_publication(
361        &self,
362        accepted: StoreSnapshotRef,
363    ) -> Result<(), DbError> {
364        self.call_store(move |session| session.complete_snapshot_publication(accepted))
365            .await
366    }
367
368    pub async fn snapshot_blob_spool_cleanup_paths(&self) -> Result<Vec<PathBuf>, DbError> {
369        self.call_store(|session| session.snapshot_blob_spool_cleanup_paths())
370            .await
371    }
372
373    pub async fn complete_snapshot_blob_spool_cleanup(&self, path: &Path) -> Result<(), DbError> {
374        let path = path.to_string_lossy().into_owned();
375        self.call_store(move |session| session.complete_snapshot_blob_spool_cleanup(&path))
376            .await
377    }
378}
coven_database/store/store_session/snapshot_publication.rs

coven_database/store/store_session/
snapshot_publication.rs
