coven_database/store/store_session/
device_streams.rs

1use crate::*;
2use coven_protocol::store_commit::{SnapshotMeta, StoreAck, StoreAckRef, StoreSnapshotRef};
3
4use super::*;
5
6impl StoreDatabase {
7    /// Record where the local device's published streams stand on the
8    /// provider: the acknowledgement head the pulled history activated for it
9    /// and the snapshot its stream ends on. A restore that adopts a
10    /// registration the device registered in an earlier life finds the
11    /// registration's own first slots already written, so its streams resume
12    /// from these heads rather than restarting there.
13    ///
14    /// The acknowledgement only ever advances the recorded head; the snapshot
15    /// is recorded when the local stream is empty and must match when it is
16    /// not.
17    pub async fn resume_local_device_streams(
18        &self,
19        latest_ack: (StoreAckRef, StoreAck),
20        latest_snapshot: Option<(StoreSnapshotRef, SnapshotMeta)>,
21    ) -> Result<(), DbError> {
22        self.call_store(move |session| {
23            session.resume_local_device_streams(latest_ack, latest_snapshot)
24        })
25        .await
26    }
27}
28
29impl StoreSession<'_> {
30    fn resume_local_device_streams(
31        &mut self,
32        (latest_ack_ref, latest_ack): (StoreAckRef, StoreAck),
33        latest_snapshot: Option<(StoreSnapshotRef, SnapshotMeta)>,
34    ) -> Result<(), DbError> {
35        let root = self.required_root_authority()?;
36        let Some(registration_ref) = local_activated_registration_ref_on(self.conn)? else {
37            return Err(DbError::Message(
38                "resuming device streams requires a local activated registration".into(),
39            ));
40        };
41        let activated = self.activated_registration(&registration_ref)?;
42        if latest_ack_ref.registration != registration_ref
43            || latest_ack.registration != registration_ref
44            || latest_ack.sequence != latest_ack_ref.sequence
45        {
46            return Err(DbError::Message(
47                "resumed acknowledgement head belongs to another registration".into(),
48            ));
49        }
50        let verified = StoreAck::parse_at(
51            &latest_ack.to_bytes(),
52            &root,
53            &latest_ack_ref,
54            activated.value(),
55        )
56        .map_err(DbError::from)?;
57        if verified != latest_ack {
58            return Err(DbError::Message(
59                "resumed acknowledgement head changed during exact verification".into(),
60            ));
61        }
62        let conn = self.conn;
63        let tx = conn.unchecked_transaction().map_err(DbError::from)?;
64        let recorded = load_published_store_ack_on(&tx)?;
65        if recorded
66            .as_ref()
67            .is_none_or(|recorded| recorded.reference.sequence < latest_ack_ref.sequence)
68        {
69            let ack_ref = serde_json::to_string(&latest_ack_ref)
70                .map_err(|error| DbError::context("resumed acknowledgement head", error))?;
71            let successor = serde_json::to_string(&latest_ack.successor.next_slot)
72                .map_err(|error| DbError::context("resumed acknowledgement successor", error))?;
73            tx.execute(
74                "INSERT INTO published_store_acks (singleton, ack_ref, successor_slot) \
75                     VALUES (1, ?1, ?2) \
76                 ON CONFLICT (singleton) DO UPDATE SET \
77                     ack_ref = excluded.ack_ref, successor_slot = excluded.successor_slot, \
78                     standing = NULL",
79                (&ack_ref, &successor),
80            )
81            .map_err(DbError::from)?;
82        }
83        let existing_snapshot = load_published_store_snapshot_on(&tx, &activated)?;
84        match (existing_snapshot, latest_snapshot) {
85            (None, None) => {}
86            (None, Some((reference, meta))) => {
87                let verified = SnapshotMeta::parse_stream_entry_at(
88                    &meta.to_bytes(),
89                    &root,
90                    &registration_ref,
91                    activated.value(),
92                    &reference,
93                )
94                .map_err(DbError::from)?;
95                if verified != meta {
96                    return Err(DbError::Message(
97                        "resumed snapshot head changed during exact verification".into(),
98                    ));
99                }
100                let generation = i64::try_from(reference.generation).map_err(|_| {
101                    DbError::Message("resumed snapshot generation exceeds SQLite INTEGER".into())
102                })?;
103                tx.execute(
104                    "INSERT INTO published_store_snapshot \
105                         (generation, snapshot_ref, successor_slot, meta_bytes) \
106                         VALUES (?1, ?2, ?3, ?4)",
107                    rusqlite::params![
108                        generation,
109                        serde_json::to_string(&reference)
110                            .map_err(|error| DbError::context("resumed snapshot ref", error))?,
111                        serde_json::to_string(&meta.successor.next_slot).map_err(|error| {
112                            DbError::context("resumed snapshot successor", error)
113                        })?,
114                        meta.to_bytes(),
115                    ],
116                )
117                .map_err(DbError::from)?;
118            }
119            (Some(existing), Some((reference, meta))) => {
120                if existing.reference != reference || existing.meta != meta {
121                    return Err(DbError::Message(
122                        "local snapshot stream differs from the provider's head".into(),
123                    ));
124                }
125            }
126            (Some(_), None) => {
127                return Err(DbError::Message(
128                    "local snapshot stream names a snapshot the provider does not hold".into(),
129                ));
130            }
131        }
132        tx.commit().map_err(DbError::from)
133    }
134}
coven_database/store/store_session/device_streams.rs

coven_database/store/store_session/
device_streams.rs
