coven_database/store/store_session/retained_merge_replay/
circle_coverage.rs

1use super::*;
2use crate::query_mapped_rows;
3#[cfg(any(test, feature = "test-utils"))]
4use crate::store::store_session::StoreRecords;
5
6impl StoreSession<'_> {
7    fn prepare_circle_restore_selection(
8        &mut self,
9        root: &coven_protocol::store_commit::StoreRootRef,
10    ) -> Result<CircleRestoreSelectionIndex, DbError> {
11        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
12        crate::store::store_session::StoreTransaction::new(&tx, self.store_dir)
13            .seed_stream_activation_index_from_retained(self.verified_store_authority, root)?;
14        let rows = query_mapped_rows(
15            &tx,
16            "SELECT circle_id, control_coord FROM circle_control_activations
17             ORDER BY circle_id, control_coord",
18            [],
19            |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
20        )?;
21        let mut circles: Vec<(
22            coven_protocol::circle::CircleId,
23            Vec<coven_protocol::circle::CircleControlCoord>,
24        )> = Vec::new();
25        for (circle_id, control_coord) in rows {
26            let circle_id: coven_protocol::circle::CircleId = circle_id
27                .parse()
28                .map_err(|error| DbError::context("parse retained Circle id", error))?;
29            let control: coven_protocol::circle::CircleControlCoord =
30                serde_json::from_str(&control_coord).map_err(|error| {
31                    DbError::context("parse retained Circle control coordinate", error)
32                })?;
33            match circles.last_mut() {
34                Some((last_circle, controls)) if *last_circle == circle_id => {
35                    controls.push(control)
36                }
37                _ => circles.push((circle_id, vec![control])),
38            }
39        }
40        let preserved_bootstraps = circle_bootstrap_coverage_refs_on(&tx)?;
41        tx.commit().map_err(DbError::from)?;
42        Ok(CircleRestoreSelectionIndex {
43            circles,
44            preserved_bootstraps,
45        })
46    }
47
48    fn retained_merge_materialization_by_ref(
49        &mut self,
50        root: &coven_protocol::store_commit::StoreRootRef,
51        reference: &StoreBatchCommitRef,
52    ) -> Result<OwnedVerifiedMergeMaterialization, DbError> {
53        let retained = self
54            .verified_store_authority
55            .retained_materialization_by_ref_on(
56                crate::store::store_session::StoreRecords::new(self.conn, self.store_dir),
57                reference,
58            )?;
59        if retained.root() != root {
60            return Err(DbError::Message(
61                "retained Merge materialization belongs to another Store root".to_string(),
62            ));
63        }
64        Ok(retained)
65    }
66
67    fn circle_replay_epoch_index(
68        &mut self,
69        root: &coven_protocol::store_commit::StoreRootRef,
70    ) -> Result<CircleReplayEpochIndex, DbError> {
71        self.verified_store_authority.retained_replay_inputs_on(
72            crate::store::store_session::StoreRecords::new(self.conn, self.store_dir),
73            root,
74        )?;
75        self.verified_store_authority.circle_replay_epoch_index_on(
76            crate::store::store_session::StoreRecords::new(self.conn, self.store_dir),
77        )
78    }
79}
80
81pub(crate) fn circle_bootstrap_coverage_ref_on(
82    conn: &Connection,
83    circle_id: coven_protocol::circle::CircleId,
84) -> Result<Option<coven_protocol::circle::CircleBootstrapCoverageRef>, DbError> {
85    let row: Option<(String, String, String, String, Vec<u8>)> = conn
86        .query_row(
87            "SELECT control_coord, activation_commit, exact_cut, image_hash, bootstrap_ref
88             FROM circle_bootstrap_coverage WHERE circle_id = ?1",
89            [circle_id.to_string()],
90            |row| {
91                Ok((
92                    row.get(0)?,
93                    row.get(1)?,
94                    row.get(2)?,
95                    row.get(3)?,
96                    row.get(4)?,
97                ))
98            },
99        )
100        .optional()
101        .map_err(DbError::from)?;
102    let Some((control, activation_commit, exact_cut, image_hash, bootstrap_ref)) = row else {
103        return Ok(None);
104    };
105    decode_circle_bootstrap_coverage_ref(
106        circle_id,
107        control,
108        activation_commit,
109        exact_cut,
110        image_hash,
111        bootstrap_ref,
112    )
113    .map(Some)
114}
115
116pub(crate) fn circle_bootstrap_coverage_refs_on(
117    conn: &Connection,
118) -> Result<Vec<coven_protocol::circle::CircleBootstrapCoverageRef>, DbError> {
119    let rows = query_mapped_rows(
120        conn,
121        "SELECT circle_id, control_coord, activation_commit, exact_cut,
122                    image_hash, bootstrap_ref
123             FROM circle_bootstrap_coverage ORDER BY circle_id",
124        [],
125        |row| {
126            Ok((
127                row.get::<_, String>(0)?,
128                row.get::<_, String>(1)?,
129                row.get::<_, String>(2)?,
130                row.get::<_, String>(3)?,
131                row.get::<_, String>(4)?,
132                row.get::<_, Vec<u8>>(5)?,
133            ))
134        },
135    )?;
136    let mut bootstraps = Vec::with_capacity(rows.len());
137    for (circle_id, control, activation_commit, exact_cut, image_hash, encoded_reference) in rows {
138        let circle_id: coven_protocol::circle::CircleId = circle_id
139            .parse()
140            .map_err(|error| DbError::context("parse retained Circle bootstrap id", error))?;
141        bootstraps.push(decode_circle_bootstrap_coverage_ref(
142            circle_id,
143            control,
144            activation_commit,
145            exact_cut,
146            image_hash,
147            encoded_reference,
148        )?);
149    }
150    Ok(bootstraps)
151}
152
153fn decode_circle_bootstrap_coverage_ref(
154    circle_id: coven_protocol::circle::CircleId,
155    control: String,
156    activation_commit: String,
157    exact_cut: String,
158    image_hash: String,
159    encoded_reference: Vec<u8>,
160) -> Result<coven_protocol::circle::CircleBootstrapCoverageRef, DbError> {
161    let control = serde_json::from_str(&control)
162        .map_err(|error| DbError::context("parse retained Circle bootstrap control", error))?;
163    let activation_commit = serde_json::from_str(&activation_commit)
164        .map_err(|error| DbError::context("parse retained Circle bootstrap activation", error))?;
165    let exact_cut: CommitFrontier = serde_json::from_str(&exact_cut)
166        .map_err(|error| DbError::context("parse retained Circle bootstrap coverage", error))?;
167    let bootstrap: coven_protocol::circle::CircleBootstrapRef =
168        serde_json::from_slice(&encoded_reference).map_err(|error| {
169            DbError::context("parse retained Circle bootstrap reference", error)
170        })?;
171    if serde_json::to_vec(&bootstrap)
172        .map_err(|error| DbError::context("serialize retained Circle bootstrap reference", error))?
173        != encoded_reference
174        || bootstrap.coverage != exact_cut
175        || bootstrap.image.image_hash.to_string() != image_hash
176    {
177        return Err(DbError::Message(
178            "retained Circle bootstrap row differs from its exact reference".to_string(),
179        ));
180    }
181    Ok(coven_protocol::circle::CircleBootstrapCoverageRef {
182        circle_id,
183        control,
184        activation_commit,
185        bootstrap,
186    })
187}
188
189impl StoreDatabase {
190    pub async fn prepare_circle_restore_selection(
191        &self,
192        root: coven_protocol::store_commit::StoreRootRef,
193    ) -> Result<CircleRestoreSelectionIndex, DbError> {
194        self.call_store(move |session| session.prepare_circle_restore_selection(&root))
195            .await
196    }
197
198    pub async fn retained_merge_materialization_by_ref(
199        &self,
200        root: coven_protocol::store_commit::StoreRootRef,
201        reference: StoreBatchCommitRef,
202    ) -> Result<OwnedVerifiedMergeMaterialization, DbError> {
203        self.call_store(move |session| {
204            session.retained_merge_materialization_by_ref(&root, &reference)
205        })
206        .await
207    }
208
209    pub async fn circle_replay_epoch_index(
210        &self,
211        root: coven_protocol::store_commit::StoreRootRef,
212    ) -> Result<CircleReplayEpochIndex, DbError> {
213        self.call_store(move |session| session.circle_replay_epoch_index(&root))
214            .await
215    }
216
217    #[cfg(any(test, feature = "test-utils"))]
218    pub(crate) fn circle_bootstrap_replay_inputs_on(
219        records: StoreRecords<'_>,
220    ) -> Result<
221        Vec<(
222            StoreBatchCommitRef,
223            coven_protocol::circle_activation::VerifiedCircleImage,
224        )>,
225        DbError,
226    > {
227        records
228            .claimed_circle_bootstrap_coverage_refs()?
229            .into_iter()
230            .map(|coverage| {
231                let image_bytes = records.payload(coverage.bootstrap.image.image_hash)?;
232                let image =
233                    coven_protocol::circle_activation::VerifiedCircleImage::from_stored_image(
234                        coverage.circle_id,
235                        coverage.control,
236                        coverage.bootstrap,
237                        image_bytes,
238                    )
239                    .map_err(DbError::from)?;
240                Ok((coverage.activation_commit, image))
241            })
242            .collect()
243    }
244}
245
246#[cfg(any(test, feature = "test-utils"))]
247impl crate::store::store_session::StoreTransaction<'_, '_> {
248    pub(crate) fn circle_bootstrap_replay_inputs(
249        self,
250    ) -> Result<
251        Vec<(
252            StoreBatchCommitRef,
253            coven_protocol::circle_activation::VerifiedCircleImage,
254        )>,
255        DbError,
256    > {
257        StoreDatabase::circle_bootstrap_replay_inputs_on(
258            crate::store::store_session::StoreRecords::new(self.transaction, self.store_dir),
259        )
260    }
261}
coven_database/store/store_session/retained_merge_replay/circle_coverage.rs

coven_database/store/store_session/retained_merge_replay/
circle_coverage.rs
