coven_replication/sync/store/device_join/
journal.rs

1use coven_protocol::store_commit::DeviceJoinAttemptId;
2
3use super::DeviceJoinError;
4use coven_database::StoreDatabase;
5use coven_protocol::store_commit::device_join_exchange::{
6    DeviceJoinActivation, DeviceJoinReadiness,
7};
8
9pub(crate) use coven_database::device_join_journal::validate_initial_progress;
10use coven_database::device_join_journal::validate_successor;
11pub(crate) use coven_protocol::store_commit::device_join_journal::attempt_key;
12pub(crate) use coven_protocol::store_commit::device_join_journal::{
13    device_join_action, DeviceJoinRoleProgress, DeviceJoinRoleProgressKind, JoinerJoinProgress,
14    OwnerJoinProgress, PreparedDeviceJoinObject,
15};
16pub use coven_protocol::store_commit::device_join_journal::{
17    DeviceJoinAction, DeviceJoinJournalRecord, DeviceJoinRole, DeviceJoinStatus,
18};
19
20/// One attempt's role journal in the Store database. An operation reads and
21/// advances a single attempt under a single role, so the handle carries both and
22/// each call names only the progress it moves to.
23pub(super) struct StoreJoinJournal<Progress> {
24    database: StoreDatabase,
25    attempt_id: DeviceJoinAttemptId,
26    progress: std::marker::PhantomData<Progress>,
27}
28
29impl<Progress: DeviceJoinRoleProgressKind> StoreJoinJournal<Progress> {
30    pub(super) fn new(database: &StoreDatabase, attempt_id: DeviceJoinAttemptId) -> Self {
31        Self {
32            database: database.clone(),
33            attempt_id,
34            progress: std::marker::PhantomData,
35        }
36    }
37
38    /// The durable record for this attempt and role, or `None` when this role has
39    /// never written one.
40    pub(super) async fn load(&self) -> Result<Option<DeviceJoinJournalRecord>, DeviceJoinError> {
41        Ok(self
42            .database
43            .load_device_join(self.attempt_id, Progress::ROLE)
44            .await?)
45    }
46
47    /// The durable record for this attempt and role. A role holding no record has
48    /// nothing to advance from, so its absence is a journal conflict.
49    pub(super) async fn current(&self) -> Result<DeviceJoinJournalRecord, DeviceJoinError> {
50        self.load().await?.ok_or(DeviceJoinError::JournalConflict)
51    }
52
53    pub(super) fn record(&self, progress: Progress) -> DeviceJoinJournalRecord {
54        DeviceJoinJournalRecord {
55            attempt_id: self.attempt_id,
56            progress: Box::new(progress.into()),
57        }
58    }
59
60    /// Advance from `previous`, returning the record now durable so the next step
61    /// advances from it.
62    pub(super) async fn advance(
63        &self,
64        previous: &DeviceJoinJournalRecord,
65        progress: Progress,
66    ) -> Result<DeviceJoinJournalRecord, DeviceJoinError> {
67        let next = self.record(progress);
68        self.advance_to(previous, &next).await?;
69        Ok(next)
70    }
71
72    /// Advance from `previous` to a record the caller already built — the shape a
73    /// caller needs when several durable predecessors advance to one successor.
74    pub(super) async fn advance_to(
75        &self,
76        previous: &DeviceJoinJournalRecord,
77        next: &DeviceJoinJournalRecord,
78    ) -> Result<(), DeviceJoinError> {
79        Ok(self
80            .database
81            .advance_device_join(previous, next.clone())
82            .await?)
83    }
84}
85
86/// Durable role journal. Each row stores a closed progress value; SQLite's
87/// compare-and-swap update rejects stale or skipped transitions.
88#[derive(Clone, Debug)]
89pub struct DeviceJoinJournalDatabase {
90    store: coven_database::DeviceJoinJournalStore,
91}
92
93impl DeviceJoinJournalDatabase {
94    pub fn open(path: impl AsRef<std::path::Path>) -> Result<Self, DeviceJoinError> {
95        Self::from_store(coven_database::DeviceJoinJournalStore::open(path))
96    }
97
98    #[cfg(any(test, feature = "test-utils"))]
99    pub fn open_for_test(path: impl AsRef<std::path::Path>) -> Result<Self, DeviceJoinError> {
100        Self::from_store(coven_database::DeviceJoinJournalStore::open_for_test(path))
101    }
102
103    fn from_store(
104        store: Result<coven_database::DeviceJoinJournalStore, coven_database::DbError>,
105    ) -> Result<Self, DeviceJoinError> {
106        Ok(Self {
107            store: store.map_err(database_error)?,
108        })
109    }
110
111    pub fn begin(
112        &self,
113        record: DeviceJoinJournalRecord,
114    ) -> Result<DeviceJoinJournalRecord, DeviceJoinError> {
115        validate_initial_progress(&record.progress)?;
116        let attempt_id = attempt_key(record.attempt_id);
117        let role = record.progress.role_name();
118        let payload = serde_json::to_string(&record)?;
119        let actual = self
120            .store
121            .insert_or_load(&attempt_id, role, &payload)
122            .map_err(database_error)?;
123        let actual = serde_json::from_str::<DeviceJoinJournalRecord>(&actual)?;
124        if actual != record {
125            return Err(DeviceJoinError::JournalConflict);
126        }
127        Ok(actual)
128    }
129
130    pub fn load(
131        &self,
132        attempt_id: DeviceJoinAttemptId,
133        role: DeviceJoinRole,
134    ) -> Result<Option<DeviceJoinJournalRecord>, DeviceJoinError> {
135        let raw = self
136            .store
137            .load(&attempt_key(attempt_id), role.as_str())
138            .map_err(database_error)?;
139        let record = raw
140            .map(|value| serde_json::from_str::<DeviceJoinJournalRecord>(&value))
141            .transpose()?;
142        if record
143            .as_ref()
144            .is_some_and(|record| record.attempt_id != attempt_id || record.progress.role() != role)
145        {
146            return Err(DeviceJoinError::JournalConflict);
147        }
148        Ok(record)
149    }
150
151    pub fn records(&self) -> Result<Vec<DeviceJoinJournalRecord>, DeviceJoinError> {
152        let mut records = Vec::new();
153        for (attempt_id, role, payload) in self.store.records().map_err(database_error)? {
154            let record: DeviceJoinJournalRecord = serde_json::from_str(&payload)?;
155            if attempt_key(record.attempt_id) != attempt_id || record.progress.role_name() != role {
156                return Err(DeviceJoinError::JournalConflict);
157            }
158            records.push(record);
159        }
160        records.sort_by_key(|record| (record.attempt_id, record.progress.role()));
161        Ok(records)
162    }
163
164    pub fn actions(&self) -> Result<Vec<DeviceJoinAction>, DeviceJoinError> {
165        Ok(self
166            .records()?
167            .iter()
168            .filter_map(device_join_action)
169            .collect())
170    }
171
172    pub fn status(
173        &self,
174        attempt_id: DeviceJoinAttemptId,
175    ) -> Result<Option<DeviceJoinStatus>, DeviceJoinError> {
176        self.load(attempt_id, DeviceJoinRole::Joiner)
177            .map(|record| record.as_ref().map(DeviceJoinJournalRecord::status))
178    }
179
180    pub fn completed_joiner_readiness(
181        &self,
182        attempt_id: DeviceJoinAttemptId,
183    ) -> Result<Option<DeviceJoinReadiness>, DeviceJoinError> {
184        let Some(record) = self.load(attempt_id, DeviceJoinRole::Joiner)? else {
185            return Ok(None);
186        };
187        match &*record.progress {
188            DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::Ready(readiness)) => {
189                if readiness.proof.attempt_id != attempt_id {
190                    return Err(DeviceJoinError::JournalConflict);
191                }
192                Ok(Some(readiness.clone()))
193            }
194            _ => Ok(None),
195        }
196    }
197
198    pub fn observe_joiner_activation_if_pending(
199        &self,
200        activation: &DeviceJoinActivation,
201    ) -> Result<Option<DeviceJoinReadiness>, DeviceJoinError> {
202        let attempt_id = activation.attempt_id;
203        let Some(current) = self.load(attempt_id, DeviceJoinRole::Joiner)? else {
204            return Ok(None);
205        };
206        match &*current.progress {
207            DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::Ready(readiness)) => {
208                let observed = DeviceJoinJournalRecord {
209                    attempt_id,
210                    progress: Box::new(DeviceJoinRoleProgress::Joiner(
211                        JoinerJoinProgress::ActivationObserved {
212                            readiness: readiness.clone(),
213                            activation: activation.clone(),
214                        },
215                    )),
216                };
217                self.advance(&current, observed)?;
218                Ok(Some(readiness.clone()))
219            }
220            DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::ActivationObserved {
221                readiness,
222                activation: existing,
223            }) if existing == activation => Ok(Some(readiness.clone())),
224            _ => Err(DeviceJoinError::JournalConflict),
225        }
226    }
227
228    pub fn advance(
229        &self,
230        previous: &DeviceJoinJournalRecord,
231        next: DeviceJoinJournalRecord,
232    ) -> Result<(), DeviceJoinError> {
233        validate_successor(previous, &next)?;
234        self.swap(previous, &next)
235    }
236
237    /// Replace `previous` with `next` only if the stored row still holds
238    /// `previous` exactly.
239    fn swap(
240        &self,
241        previous: &DeviceJoinJournalRecord,
242        next: &DeviceJoinJournalRecord,
243    ) -> Result<(), DeviceJoinError> {
244        let previous_payload = serde_json::to_string(previous)?;
245        let next_payload = serde_json::to_string(next)?;
246        if !self
247            .store
248            .compare_and_swap(
249                &attempt_key(previous.attempt_id),
250                previous.progress.role_name(),
251                &previous_payload,
252                &next_payload,
253            )
254            .map_err(database_error)?
255        {
256            return Err(DeviceJoinError::JournalConflict);
257        }
258        Ok(())
259    }
260
261    /// Drop this attempt's joiner row now the join has finished.
262    ///
263    /// The row held the join's last state so a repeated completion could tell
264    /// "already done" from "conflicting". The library's config file says that
265    /// now — it is written before this runs, and it is what the caller checks —
266    /// so the row has nothing left to answer and goes.
267    pub(super) fn retire(&self, current: &DeviceJoinJournalRecord) -> Result<(), DeviceJoinError> {
268        let payload = serde_json::to_string(current)?;
269        if !self
270            .store
271            .compare_and_forget(
272                &attempt_key(current.attempt_id),
273                current.progress.role_name(),
274                &payload,
275            )
276            .map_err(database_error)?
277        {
278            return Err(DeviceJoinError::JournalConflict);
279        }
280        Ok(())
281    }
282}
283
284pub(super) fn database_error(error: coven_database::DbError) -> DeviceJoinError {
285    DeviceJoinError::Database(error)
286}
coven_replication/sync/store/device_join/journal.rs

coven_replication/sync/store/device_join/
journal.rs
