coven_database/store/store_session/
circle_acknowledgements.rs

1use crate::query_mapped_rows;
2use crate::*;
3use coven_protocol::circle::{
4    CircleBootstrapCoverageRef, CircleControlCoord, CircleEpochId, CircleId,
5};
6use coven_protocol::objects::PreparedExactObject;
7use coven_protocol::store_commit::{
8    CircleAck, CircleAckRef, CommitFrontier, StoreDeviceId, StoreDeviceStatus, StoreHistoryCut,
9};
10use rusqlite::OptionalExtension;
11use std::collections::{BTreeMap, BTreeSet};
12
13use super::{StoreDatabase, StoreSession};
14
15/// Everything one active Circle contributes to staging its device's next Circle
16/// acknowledgement: the exact control and epoch the live projection derives from,
17/// the access authority that seals the acknowledgement, and the retained bootstrap
18/// coverage the projection was seeded from (`None` for a founder/source device).
19pub struct CircleAckPublicationInput {
20    control: CircleControlCoord,
21    epoch_id: CircleEpochId,
22    access: coven_protocol::circle_activation::CircleEpochAccess,
23    seeded_from: Option<CircleBootstrapCoverageRef>,
24}
25
26impl CircleAckPublicationInput {
27    pub fn circle_id(&self) -> CircleId {
28        self.access.circle_id()
29    }
30
31    pub fn control(&self) -> &CircleControlCoord {
32        &self.control
33    }
34
35    pub fn epoch_id(&self) -> CircleEpochId {
36        self.epoch_id
37    }
38
39    pub fn seeded_from(&self) -> Option<&CircleBootstrapCoverageRef> {
40        self.seeded_from.as_ref()
41    }
42
43    pub fn protocol_context(
44        &self,
45        store_root_hash: coven_protocol::store_commit::ObjectHash,
46        domain: coven_protocol::objects::CircleProtocolObjectDomain,
47    ) -> coven_protocol::objects::ProtocolObjectContext {
48        self.access.protocol_context(store_root_hash, domain)
49    }
50
51    pub fn key_fingerprint(&self) -> coven_keys::encryption::KeyFingerprint {
52        self.access.key_fingerprint()
53    }
54}
55
56/// The last Circle acknowledgement this device published for one Circle: its
57/// exact reference, the successor slot its next acknowledgement occupies, and the
58/// coverage it named (used to skip re-staging an unchanged acknowledgement).
59pub struct PublishedCircleAck {
60    pub reference: CircleAckRef,
61    pub successor_slot: coven_protocol::objects::ObjectSlot,
62    pub store_cut: CommitFrontier,
63    pub control: CircleControlCoord,
64}
65
66impl StoreSession<'_> {
67    fn circle_acknowledgement_publication_inputs(
68        &self,
69    ) -> Result<Vec<CircleAckPublicationInput>, DbError> {
70        let conn = self.conn;
71        let mut inputs = Vec::new();
72        for state in super::circle_operations::circle_current_states_on(conn)? {
73            let circle_id = state.circle_id();
74            let Some(authoring) = state.authoring_state() else {
75                tracing::debug!(
76                    circle_id = %circle_id,
77                    "skip Circle acknowledgement: recipient holds no active access"
78                );
79                continue;
80            };
81            let control = authoring.control.coord.clone();
82            let epoch_id = authoring.control.value.epoch_id();
83            let access = super::circle_publication_context_on(conn, circle_id, &control)?;
84            let seeded_from =
85                super::retained_merge_replay::circle_bootstrap_coverage_ref_on(conn, circle_id)?;
86            inputs.push(CircleAckPublicationInput {
87                control,
88                epoch_id,
89                access,
90                seeded_from,
91            });
92        }
93        Ok(inputs)
94    }
95
96    fn activated_circle_ack(
97        &self,
98        circle_id: CircleId,
99        device_id: StoreDeviceId,
100    ) -> Result<Option<CircleAckRef>, DbError> {
101        self.conn
102            .query_row(
103                "SELECT ack_ref FROM activated_circle_acks
104                 WHERE circle_id = ?1 AND device_id = ?2",
105                rusqlite::params![circle_id.to_string(), device_id.to_string()],
106                |row| row.get::<_, String>(0),
107            )
108            .optional()
109            .map_err(DbError::from)?
110            .map(|raw| parse_circle_ack_ref(&raw, circle_id, "activated"))
111            .transpose()
112    }
113
114    fn circle_current_roster_members(
115        &self,
116        circle_id: CircleId,
117    ) -> Result<BTreeSet<String>, DbError> {
118        let Some(state) = super::circle_operations::circle_current_state_on(self.conn, circle_id)?
119        else {
120            return Err(DbError::Message(format!(
121                "Circle {circle_id} has no current state"
122            )));
123        };
124        let Some((_current, _access, roster, _metadata)) = state.active() else {
125            return Ok(BTreeSet::new());
126        };
127        Ok(roster.members().into_keys().collect())
128    }
129
130    fn activated_circle_acks(&self, circle_id: CircleId) -> Result<Vec<CircleAckRef>, DbError> {
131        query_mapped_rows(
132            self.conn,
133            "SELECT ack_ref FROM activated_circle_acks
134             WHERE circle_id = ?1 ORDER BY device_id",
135            [circle_id.to_string()],
136            |row| row.get::<_, String>(0),
137        )?
138        .into_iter()
139        .map(|raw| parse_circle_ack_ref(&raw, circle_id, "activated"))
140        .collect()
141    }
142
143    fn latest_published_circle_ack(
144        &self,
145        circle_id: CircleId,
146    ) -> Result<Option<PublishedCircleAck>, DbError> {
147        let row: Option<(String, String, String, String)> = self
148            .conn
149            .query_row(
150                "SELECT ack_ref, successor_slot, store_cut, control_coord
151                 FROM published_circle_acks WHERE circle_id = ?1",
152                [circle_id.to_string()],
153                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
154            )
155            .optional()
156            .map_err(DbError::from)?;
157        let Some((reference, successor_slot, store_cut, control)) = row else {
158            return Ok(None);
159        };
160        let reference = parse_circle_ack_ref(&reference, circle_id, "published")?;
161        if reference.sequence == 0 {
162            return Err(DbError::Message(
163                "published Circle acknowledgement names sequence zero".to_string(),
164            ));
165        }
166        Ok(Some(PublishedCircleAck {
167            reference,
168            successor_slot: serde_json::from_str(&successor_slot).map_err(|error| {
169                DbError::context("published Circle acknowledgement successor slot", error)
170            })?,
171            store_cut: serde_json::from_str(&store_cut)
172                .map_err(|error| DbError::context("published Circle acknowledgement cut", error))?,
173            control: serde_json::from_str(&control).map_err(|error| {
174                DbError::context("published Circle acknowledgement control", error)
175            })?,
176        }))
177    }
178
179    /// Whether any Circle acknowledgement is waiting to be published.
180    ///
181    /// A Store acknowledgement is what carries them to the cloud, so it stages
182    /// itself when any are queued even if it has nothing of its own to say.
183    fn outbound_circle_acks_pending(&self) -> Result<bool, DbError> {
184        self.conn
185            .query_row(
186                "SELECT EXISTS(SELECT 1 FROM outbound_circle_acks)",
187                [],
188                |row| row.get::<_, bool>(0),
189            )
190            .map_err(DbError::from)
191    }
192
193    fn stage_circle_ack(
194        &mut self,
195        ack: CircleAck,
196        prepared: PreparedExactObject,
197    ) -> Result<CircleAckRef, DbError> {
198        let authority = self.local_store_authority()?;
199        let registration = authority.value();
200        let bytes = ack.to_bytes();
201        let reference = CircleAckRef {
202            registration: ack.registration.clone(),
203            circle_id: ack.circle_id,
204            control: ack.control.clone(),
205            sequence: ack.sequence,
206            ack_hash: ack.ack_hash(),
207            object: prepared.reference().clone(),
208        };
209        CircleAck::parse_at(&bytes, &registration.store_root, &reference, registration)
210            .map_err(|error| DbError::context("stage Circle acknowledgement", error))?;
211        let ack_ref = serde_json::to_string(&reference).map_err(|error| {
212            DbError::context("serialize exact Circle acknowledgement ref", error)
213        })?;
214        let prepared = serde_json::to_string(&prepared).map_err(|error| {
215            DbError::context("serialize prepared Circle acknowledgement", error)
216        })?;
217        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
218        tx.execute(
219            "INSERT INTO outbound_circle_acks (circle_id, ack_ref, ack_bytes, prepared_object)
220             VALUES (?1, ?2, ?3, ?4)",
221            rusqlite::params![reference.circle_id.to_string(), ack_ref, bytes, prepared],
222        )
223        .map_err(DbError::from)?;
224        tx.commit().map_err(DbError::from)?;
225        Ok(reference)
226    }
227}
228
229fn parse_circle_ack_ref(
230    raw: &str,
231    circle_id: CircleId,
232    state: &str,
233) -> Result<CircleAckRef, DbError> {
234    let reference: CircleAckRef = serde_json::from_str(raw)
235        .map_err(|error| DbError::context(format!("{state} Circle acknowledgement ref"), error))?;
236    if reference.circle_id != circle_id {
237        return Err(DbError::Message(format!(
238            "{state} Circle acknowledgement names another Circle"
239        )));
240    }
241    Ok(reference)
242}
243
244impl StoreDatabase {
245    pub async fn circle_acknowledgement_publication_inputs(
246        &self,
247    ) -> Result<Vec<CircleAckPublicationInput>, DbError> {
248        self.call_store(|session| session.circle_acknowledgement_publication_inputs())
249            .await
250    }
251
252    /// The latest activated Circle acknowledgement `device_id` published for
253    /// `circle_id`, or `None` if that device has never had an acknowledgement
254    /// activated. Snapshot stability reads this per access-holding device.
255    pub async fn activated_circle_ack(
256        &self,
257        circle_id: CircleId,
258        device_id: StoreDeviceId,
259    ) -> Result<Option<CircleAckRef>, DbError> {
260        self.call_store(move |session| session.activated_circle_ack(circle_id, device_id))
261            .await
262    }
263
264    /// The device ids that currently hold active Circle access to `circle_id`:
265    /// every active Store device whose owner is a current member of the Circle's
266    /// resolved roster. Snapshot stability requires each of these devices to have
267    /// published a dominating acknowledgement, so a device that holds access but
268    /// has never acknowledged keeps the snapshot unstable (fail closed). A
269    /// `Closing`/`Inactive`/conflicted Circle authors no snapshot and returns an
270    /// empty set.
271    pub async fn active_circle_access_devices(
272        &self,
273        circle_id: CircleId,
274    ) -> Result<BTreeSet<StoreDeviceId>, DbError> {
275        let members = self.circle_current_roster_members(circle_id).await?;
276        if members.is_empty() {
277            return Ok(BTreeSet::new());
278        }
279        let frontier = CommitFrontier::from_refs(self.materialized_frontier().await?)
280            .map_err(|error| DbError::context("shape current Store frontier", error))?;
281        let (_, device_state) = self
282            .store_device_state_for_history_cut(&StoreHistoryCut(frontier.0))
283            .await?;
284        let owners: BTreeMap<StoreDeviceId, String> = self
285            .activated_store_device_registration_records()
286            .await?
287            .into_iter()
288            .map(|registration| {
289                (
290                    registration.value().device_id,
291                    registration.value().author_pubkey.clone(),
292                )
293            })
294            .collect();
295        let mut devices = BTreeSet::new();
296        for (device_id, record) in device_state.devices {
297            if !matches!(record.status, StoreDeviceStatus::Active) {
298                continue;
299            }
300            let owner = owners.get(&device_id).ok_or_else(|| {
301                DbError::Message(format!(
302                    "active Store device {device_id} has no activated registration"
303                ))
304            })?;
305            if members.contains(owner) {
306                devices.insert(device_id);
307            }
308        }
309        Ok(devices)
310    }
311
312    /// The pubkeys in `circle_id`'s current resolved roster, or an empty set when
313    /// the Circle is not in an active local state (so it has no snapshot quorum).
314    pub async fn circle_current_roster_members(
315        &self,
316        circle_id: CircleId,
317    ) -> Result<BTreeSet<String>, DbError> {
318        self.call_store(move |session| session.circle_current_roster_members(circle_id))
319            .await
320    }
321
322    /// The latest activated Circle acknowledgement every device that has ever
323    /// acknowledged `circle_id` published — one per device, including devices whose
324    /// owner has since been removed from the roster (rows are never deleted, so a
325    /// removed recipient's last acknowledgement persists as the evidence bootstrap
326    /// reclamation reads to prove that recipient lost authority).
327    pub async fn activated_circle_acks(
328        &self,
329        circle_id: CircleId,
330    ) -> Result<Vec<CircleAckRef>, DbError> {
331        self.call_store(move |session| session.activated_circle_acks(circle_id))
332            .await
333    }
334
335    pub async fn latest_published_circle_ack(
336        &self,
337        circle_id: CircleId,
338    ) -> Result<Option<PublishedCircleAck>, DbError> {
339        self.call_store(move |session| session.latest_published_circle_ack(circle_id))
340            .await
341    }
342
343    pub async fn outbound_circle_acks_pending(&self) -> Result<bool, DbError> {
344        self.call_store(|session| session.outbound_circle_acks_pending())
345            .await
346    }
347
348    pub async fn stage_circle_ack(
349        &self,
350        ack: CircleAck,
351        prepared: PreparedExactObject,
352    ) -> Result<CircleAckRef, DbError> {
353        self.call_store(move |session| session.stage_circle_ack(ack, prepared))
354            .await
355    }
356}
coven_database/store/store_session/circle_acknowledgements.rs

coven_database/store/store_session/
circle_acknowledgements.rs
