coven_database/store/store_session/
circle_snapshot_publication.rs

1use std::collections::BTreeSet;
2
3use crate::*;
4use coven_protocol::circle::CircleId;
5use coven_protocol::store_commit::{
6    circle_snapshot_image_semantic_prefix, circle_snapshot_slot_prefix, CircleSnapshotMeta,
7    CircleSnapshotRef,
8};
9use rusqlite::OptionalExtension;
10
11use super::*;
12
13impl StoreSession<'_> {
14    fn pending_circle_snapshot_ids(&self) -> Result<Vec<CircleId>, DbError> {
15        let mut statement = self
16            .conn
17            .prepare("SELECT circle_id FROM outbound_circle_snapshot ORDER BY circle_id")
18            .map_err(DbError::from)?;
19        let rows = statement
20            .query_map([], |row| row.get::<_, String>(0))
21            .map_err(DbError::from)?;
22        rows.map(|row| {
23            row.map_err(DbError::from)?
24                .parse()
25                .map_err(|error| DbError::context("pending Circle snapshot identity", error))
26        })
27        .collect()
28    }
29
30    fn outbound_circle_snapshot_publication(
31        &mut self,
32        circle_id: CircleId,
33    ) -> Result<Option<DurableCircleSnapshotPublication>, DbError> {
34        let authority = self.local_store_authority()?;
35        load_outbound_circle_snapshot_on(self.conn, self.store_dir, &authority, circle_id)
36    }
37
38    fn latest_local_circle_snapshot(
39        &mut self,
40        circle_id: CircleId,
41    ) -> Result<Option<PublishedCircleSnapshot>, DbError> {
42        let authority = self.local_store_authority()?;
43        load_published_circle_snapshot_on(self.conn, &authority, circle_id)
44    }
45
46    fn stage_circle_snapshot_publication(
47        &mut self,
48        meta: CircleSnapshotMeta,
49        meta_prepared: PreparedExactObject,
50        image: SnapshotDatabaseImage,
51        image_prepared: PreparedExactObject,
52    ) -> Result<CircleSnapshotRef, DbError> {
53        let authority = self.local_store_authority()?;
54        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
55        let image_facts =
56            crate::payload_store::write_payload_file_blocking(&tx, self.store_dir, image.path())
57                .map_err(|source| SnapshotImageError::ProjectionPayloadStore {
58                    operation: "spool Circle snapshot image".to_string(),
59                    source,
60                });
61        let (image_hash, _) = image.finish(image_facts).map_err(snapshot_image_db_error)?;
62        let image_prepared_hash = crate::payload_store::write_payload_blocking(
63            &tx,
64            self.store_dir,
65            image_prepared.stored_bytes(),
66        )
67        .map_err(|error| DbError::context("spool prepared Circle snapshot image", error))?;
68        let image_prepared_size = image_prepared.stored_bytes().len() as u64;
69        let registration_ref = authority.reference();
70        let registration = authority.value();
71        validate_snapshot_author(&meta.author_registration, registration_ref, "Circle")?;
72        let device_id = registration.device_id.to_string();
73        validate_snapshot_image(
74            &meta.bootstrap.image,
75            &image_prepared,
76            image_hash,
77            image_prepared_hash,
78            image_prepared_size,
79            format!(
80                "{}.db",
81                circle_snapshot_image_semantic_prefix(
82                    meta.circle_id,
83                    &device_id,
84                    meta.bootstrap.image.image_hash,
85                )
86            ),
87            "Circle",
88        )?;
89        let reference = CircleSnapshotRef {
90            generation: meta.generation,
91            snapshot_hash: meta.snapshot_hash(),
92            object: meta_prepared.reference().clone(),
93        };
94        CircleSnapshotMeta::parse_at(
95            &meta.to_bytes(),
96            registration.store_root.store_root_hash,
97            &reference,
98            registration,
99        )
100        .map_err(|error| DbError::context("verify staged Circle snapshot metadata", error))?;
101        let previous = load_published_circle_snapshot_on(&tx, &authority, meta.circle_id)?;
102        let (expected_generation, expected_slot) = match &previous {
103            Some(previous) => (
104                previous
105                    .reference
106                    .generation
107                    .checked_add(1)
108                    .ok_or_else(|| {
109                        DbError::Message("Circle snapshot generation overflow".to_string())
110                    })?,
111                previous.successor_slot.clone(),
112            ),
113            None => (
114                0,
115                coven_protocol::objects::ObjectSlot::logical(format!(
116                    "{}.json",
117                    circle_snapshot_slot_prefix(meta.circle_id, &device_id, 0)
118                ))
119                .map_err(DbError::from)?,
120            ),
121        };
122        if meta.generation != expected_generation
123            || meta_prepared.reference().slot() != &expected_slot
124            || meta.successor.predecessor != previous.as_ref().map(|value| value.reference.clone())
125        {
126            return Err(DbError::Message(
127                "Circle snapshot does not extend the exact local stream".to_string(),
128            ));
129        }
130        let next_generation = meta
131            .generation
132            .checked_add(1)
133            .ok_or_else(|| DbError::Message("Circle snapshot generation overflow".to_string()))?;
134        if meta.successor.next_slot.logical_key()
135            != format!(
136                "{}.json",
137                circle_snapshot_slot_prefix(meta.circle_id, &device_id, next_generation)
138            )
139        {
140            return Err(DbError::Message(
141                "Circle snapshot successor is outside its activated exact stream".to_string(),
142            ));
143        }
144        tx.execute(
145            "INSERT INTO outbound_circle_snapshot \
146             (circle_id, snapshot_ref, meta_prepared, meta_bytes) \
147             VALUES (?1, ?2, ?3, ?4)",
148            rusqlite::params![
149                meta.circle_id.to_string(),
150                serde_json::to_string(&reference).map_err(|error| {
151                    DbError::context("serialize exact Circle snapshot ref", error)
152                })?,
153                serde_json::to_string(&meta_prepared).map_err(|error| {
154                    DbError::context("serialize prepared Circle snapshot metadata", error)
155                })?,
156                meta.to_bytes(),
157            ],
158        )
159        .map_err(DbError::from)?;
160        crate::payload_store::set_payload_owner_claims_on(
161            &tx,
162            &crate::payload_store::outbound_circle_snapshot_owner_key(meta.circle_id),
163            &BTreeSet::from([image_hash, image_prepared_hash]),
164        )?;
165        tx.commit().map_err(DbError::from)?;
166        Ok(reference)
167    }
168
169    fn complete_circle_snapshot_publication(
170        &mut self,
171        accepted: CircleSnapshotRef,
172    ) -> Result<(), DbError> {
173        let authority = self.local_store_authority()?;
174        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
175        let circle_id = {
176            let bytes: Vec<u8> = tx
177                .query_row(
178                    "SELECT meta_bytes FROM outbound_circle_snapshot \
179                     WHERE snapshot_ref = ?1",
180                    [serde_json::to_string(&accepted).map_err(|error| {
181                        DbError::context("serialize accepted Circle snapshot ref", error)
182                    })?],
183                    |row| row.get::<_, Vec<u8>>(0),
184                )
185                .optional()
186                .map_err(DbError::from)?
187                .ok_or_else(|| {
188                    DbError::Message("outbound Circle snapshot is absent".to_string())
189                })?;
190            let meta: CircleSnapshotMeta = serde_json::from_slice(&bytes)
191                .map_err(|error| DbError::context("accepted Circle snapshot metadata", error))?;
192            meta.circle_id
193        };
194        let outbound =
195            load_outbound_circle_snapshot_on(&tx, self.store_dir, &authority, circle_id)?
196                .ok_or_else(|| {
197                    DbError::Message("outbound Circle snapshot is absent".to_string())
198                })?;
199        if outbound.reference != accepted {
200            return Err(DbError::Message(
201                "accepted Circle snapshot differs from the prepared exact object".to_string(),
202            ));
203        }
204        let snapshot_owner = coven_protocol::remote_object::SnapshotObjectOwner::Circle {
205            activation: coven_protocol::store_commit::circle_snapshot_stream_activation(
206                outbound.meta.value.store_root_hash,
207                &outbound.meta.value.author_registration,
208                outbound.meta.value.circle_id,
209            )
210            .map_err(DbError::from)?,
211            generation: outbound.meta.value.generation,
212        };
213        persist_snapshot_image_on(
214            &tx,
215            self.store_dir,
216            &outbound.meta.value.bootstrap.image,
217            snapshot_owner,
218            "Circle snapshot image",
219        )?;
220        let deleted = tx
221            .execute(
222                "DELETE FROM outbound_circle_snapshot WHERE circle_id = ?1",
223                [circle_id.to_string()],
224            )
225            .map_err(DbError::from)?;
226        if deleted != 1 {
227            return Err(DbError::Message(
228                "outbound Circle snapshot ownership row is absent or changed".to_string(),
229            ));
230        }
231        crate::payload_store::release_payload_owner_on(
232            &tx,
233            &crate::payload_store::outbound_circle_snapshot_owner_key(circle_id),
234        )?;
235        let accepted_generation =
236            snapshot_generation_as_i64(accepted.generation, "Circle snapshot")?;
237        tx.execute(
238            "INSERT INTO published_circle_snapshot \
239             (circle_id, generation, snapshot_ref, successor_slot, cut, meta_bytes) \
240             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
241            rusqlite::params![
242                circle_id.to_string(),
243                accepted_generation,
244                serde_json::to_string(&accepted).map_err(|error| {
245                    DbError::context("serialize published Circle snapshot ref", error)
246                })?,
247                serde_json::to_string(&outbound.meta.value.successor.next_slot).map_err(
248                    |error| DbError::context("serialize Circle snapshot successor slot", error)
249                )?,
250                serde_json::to_string(&outbound.meta.value.bootstrap.coverage)
251                    .map_err(|error| DbError::context("serialize Circle snapshot cut", error))?,
252                outbound.meta.bytes,
253            ],
254        )
255        .map_err(DbError::from)?;
256        tx.commit().map_err(DbError::from)
257    }
258}
259
260impl StoreDatabase {
261    pub async fn pending_circle_snapshot_ids(&self) -> Result<Vec<CircleId>, DbError> {
262        self.call_store(move |session| session.pending_circle_snapshot_ids())
263            .await
264    }
265
266    pub async fn outbound_circle_snapshot_publication(
267        &self,
268        circle_id: CircleId,
269    ) -> Result<Option<DurableCircleSnapshotPublication>, DbError> {
270        self.call_store(move |session| session.outbound_circle_snapshot_publication(circle_id))
271            .await
272    }
273
274    pub async fn latest_local_circle_snapshot(
275        &self,
276        circle_id: CircleId,
277    ) -> Result<Option<PublishedCircleSnapshot>, DbError> {
278        self.call_store(move |session| session.latest_local_circle_snapshot(circle_id))
279            .await
280    }
281
282    pub async fn stage_circle_snapshot_publication(
283        &self,
284        meta: CircleSnapshotMeta,
285        meta_prepared: PreparedExactObject,
286        image: SnapshotDatabaseImage,
287        image_prepared: PreparedExactObject,
288    ) -> Result<CircleSnapshotRef, DbError> {
289        self.call_store(move |session| {
290            session.stage_circle_snapshot_publication(meta, meta_prepared, image, image_prepared)
291        })
292        .await
293    }
294
295    pub async fn complete_circle_snapshot_publication(
296        &self,
297        accepted: CircleSnapshotRef,
298    ) -> Result<(), DbError> {
299        self.call_store(move |session| session.complete_circle_snapshot_publication(accepted))
300            .await
301    }
302}
coven_database/store/store_session/circle_snapshot_publication.rs

coven_database/store/store_session/
circle_snapshot_publication.rs
