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

coven_database/store/store_session/
circle_acknowledgements.rs
