coven_database/store/
device_join.rs

1use crate::query_mapped_rows;
2use crate::store::device_join_journal::{
3    require_initial, validate_successor, DeviceJoinJournalError,
4};
5use crate::StoreDatabase;
6use coven_protocol::store_commit::device_join_journal::{
7    DeviceJoinAction, DeviceJoinJournalRecord, DeviceJoinRole, DeviceJoinStatus,
8};
9use coven_protocol::store_commit::{DeviceJoinAttemptId, ObjectHash};
10
11/// The joining device's own journal of in-flight join attempts, kept in its own
12/// SQLite file under the stores root because the Store database it is joining
13/// does not exist yet while the attempt runs.
14///
15/// The file lives exactly as long as the attempts in it. Retiring the last row
16/// closes the connection and deletes the file with its WAL sidecars, so a
17/// device that has joined a store does not leave a dead journal behind forever;
18/// the next attempt reopens the path and gets a new file.
19#[derive(Clone, Debug)]
20pub struct DeviceJoinJournalStore {
21    path: std::path::PathBuf,
22    durability: crate::connection_io::ConnectionDurability,
23    /// `None` between the delete of an emptied journal and the next operation
24    /// that reopens it. The connection is closed before the file is unlinked so
25    /// no later write can land in a file nothing can be read back from.
26    connection: std::sync::Arc<std::sync::Mutex<Option<rusqlite::Connection>>>,
27}
28
29impl DeviceJoinJournalStore {
30    pub fn open(path: impl AsRef<std::path::Path>) -> Result<Self, crate::DbError> {
31        Self::open_with_durability(path, crate::connection_io::ConnectionDurability::Full)
32    }
33
34    #[cfg(any(test, feature = "test-utils"))]
35    pub fn open_for_test(path: impl AsRef<std::path::Path>) -> Result<Self, crate::DbError> {
36        Self::open_with_durability(path, crate::connection_io::ConnectionDurability::Disabled)
37    }
38
39    fn open_with_durability(
40        path: impl AsRef<std::path::Path>,
41        durability: crate::connection_io::ConnectionDurability,
42    ) -> Result<Self, crate::DbError> {
43        let store = Self {
44            path: path.as_ref().to_path_buf(),
45            durability,
46            connection: std::sync::Arc::new(std::sync::Mutex::new(None)),
47        };
48        // Open here rather than on the first operation: a journal file that
49        // cannot be opened is a failure of whoever asked for the journal, and
50        // they find out at the call that asked.
51        store.with_connection(|_| Ok(()))?;
52        Ok(store)
53    }
54
55    /// Put a live connection in `held`, creating the journal file and its one
56    /// table. Called at construction and again by the first operation after a
57    /// completed attempt deleted the file.
58    fn open_into(&self, held: &mut Option<rusqlite::Connection>) -> Result<(), crate::DbError> {
59        if let Some(directory) = self.path.parent() {
60            std::fs::create_dir_all(directory).map_err(crate::DbError::from)?;
61        }
62        let opened = rusqlite::Connection::open(&self.path).map_err(crate::DbError::from)?;
63        crate::connection_io::configure_connection_durability(&opened, self.durability)?;
64        opened
65            .execute_batch(
66                "PRAGMA foreign_keys = ON;
67                 CREATE TABLE IF NOT EXISTS device_join_journals (
68                     attempt_id TEXT NOT NULL,
69                     role TEXT NOT NULL,
70                     payload TEXT NOT NULL,
71                     PRIMARY KEY (attempt_id, role)
72                 ) STRICT, WITHOUT ROWID;",
73            )
74            .map_err(crate::DbError::from)?;
75        *held = Some(opened);
76        Ok(())
77    }
78
79    /// Run `operation` on the journal's connection, opening the file first when
80    /// a completed attempt deleted it.
81    fn with_connection<R>(
82        &self,
83        operation: impl FnOnce(&rusqlite::Connection) -> Result<R, crate::DbError>,
84    ) -> Result<R, crate::DbError> {
85        let mut held = self
86            .connection
87            .lock()
88            .map_err(|_| pending_join_connection_poisoned())?;
89        if held.is_none() {
90            self.open_into(&mut held)?;
91        }
92        let connection = held.as_ref().ok_or_else(|| {
93            crate::DbError::Message(
94                "pending device-join journal connection is absent after opening it".to_string(),
95            )
96        })?;
97        operation(connection)
98    }
99
100    pub fn insert_or_load(
101        &self,
102        attempt_id: &str,
103        role: &str,
104        payload: &str,
105    ) -> Result<String, crate::DbError> {
106        self.with_connection(|connection| {
107            let transaction = connection
108                .unchecked_transaction()
109                .map_err(crate::DbError::from)?;
110            transaction
111                .execute(
112                    "INSERT OR IGNORE INTO device_join_journals (attempt_id, role, payload)
113                     VALUES (?1, ?2, ?3)",
114                    (attempt_id, role, payload),
115                )
116                .map_err(crate::DbError::from)?;
117            let actual = transaction
118                .query_row(
119                    "SELECT payload FROM device_join_journals WHERE attempt_id = ?1 AND role = ?2",
120                    (attempt_id, role),
121                    |row| row.get(0),
122                )
123                .map_err(crate::DbError::from)?;
124            transaction.commit().map_err(crate::DbError::from)?;
125            Ok(actual)
126        })
127    }
128
129    pub fn load(&self, attempt_id: &str, role: &str) -> Result<Option<String>, crate::DbError> {
130        use rusqlite::OptionalExtension;
131
132        self.with_connection(|connection| {
133            connection
134                .query_row(
135                    "SELECT payload FROM device_join_journals WHERE attempt_id = ?1 AND role = ?2",
136                    (attempt_id, role),
137                    |row| row.get(0),
138                )
139                .optional()
140                .map_err(crate::DbError::from)
141        })
142    }
143
144    pub fn records(&self) -> Result<Vec<(String, String, String)>, crate::DbError> {
145        self.with_connection(|connection| {
146            query_mapped_rows(
147                connection,
148                "SELECT attempt_id, role, payload FROM device_join_journals
149                     ORDER BY attempt_id, role",
150                [],
151                |row| {
152                    Ok((
153                        row.get::<_, String>(0)?,
154                        row.get::<_, String>(1)?,
155                        row.get::<_, String>(2)?,
156                    ))
157                },
158            )
159            .map_err(crate::DbError::from)
160        })
161    }
162
163    pub fn compare_and_swap(
164        &self,
165        attempt_id: &str,
166        role: &str,
167        previous_payload: &str,
168        next_payload: &str,
169    ) -> Result<bool, crate::DbError> {
170        self.with_connection(|connection| {
171            let changed = connection
172                .execute(
173                    "UPDATE device_join_journals SET payload = ?1
174                     WHERE attempt_id = ?2 AND role = ?3 AND payload = ?4",
175                    (next_payload, attempt_id, role, previous_payload),
176                )
177                .map_err(crate::DbError::from)?;
178            Ok(changed == 1)
179        })
180    }
181
182    /// Drop one attempt's joiner row, but only while it still holds exactly the
183    /// payload the caller last read, and delete the journal file when that row
184    /// was the last one in it.
185    ///
186    /// This is how a joining device finishes: the row is its working notes on
187    /// an exchange that is over, and what says the join happened is the
188    /// library's own config file, written before this runs. An emptied journal
189    /// answers nothing either, so the file goes with the row rather than
190    /// accumulating one dead SQLite database per store the device ever joined.
191    ///
192    /// The row delete, the emptiness check, and the unlink all happen under the
193    /// one connection lock, so a row begun by another attempt cannot land in a
194    /// file that is about to be deleted.
195    pub fn compare_and_forget(
196        &self,
197        attempt_id: &str,
198        role: &str,
199        expected_payload: &str,
200    ) -> Result<bool, crate::DbError> {
201        let mut held = self
202            .connection
203            .lock()
204            .map_err(|_| pending_join_connection_poisoned())?;
205        if held.is_none() {
206            self.open_into(&mut held)?;
207        }
208        let connection = held.as_ref().ok_or_else(|| {
209            crate::DbError::Message(
210                "pending device-join journal connection is absent after opening it".to_string(),
211            )
212        })?;
213        let removed = connection
214            .execute(
215                "DELETE FROM device_join_journals
216                 WHERE attempt_id = ?1 AND role = ?2 AND payload = ?3",
217                (attempt_id, role, expected_payload),
218            )
219            .map_err(crate::DbError::from)?;
220        if removed != 1 {
221            return Ok(false);
222        }
223        let remaining: i64 = connection
224            .query_row("SELECT COUNT(*) FROM device_join_journals", [], |row| {
225                row.get(0)
226            })
227            .map_err(crate::DbError::from)?;
228        if remaining == 0 {
229            if let Some(connection) = held.take() {
230                connection.close().map_err(|(_, error)| error)?;
231            }
232            remove_pending_join_files(&self.path)?;
233        }
234        Ok(true)
235    }
236
237    #[cfg(test)]
238    fn synchronous_for_test(&self) -> Result<i64, crate::DbError> {
239        self.with_connection(|connection| {
240            connection
241                .query_row("PRAGMA synchronous", [], |row| row.get(0))
242                .map_err(crate::DbError::from)
243        })
244    }
245}
246
247/// Delete an emptied journal and the WAL sidecars its durable mode writes beside
248/// it. Closing the connection already checkpoints and removes those in the
249/// ordinary case; naming them here is what covers a file left by an earlier
250/// process that did not close.
251fn remove_pending_join_files(path: &std::path::Path) -> Result<(), crate::DbError> {
252    for candidate in [
253        path.to_path_buf(),
254        std::path::PathBuf::from(format!("{}-wal", path.display())),
255        std::path::PathBuf::from(format!("{}-shm", path.display())),
256    ] {
257        match std::fs::remove_file(&candidate) {
258            Ok(()) => {}
259            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
260            Err(error) => return Err(crate::DbError::from(error)),
261        }
262    }
263    Ok(())
264}
265
266fn pending_join_connection_poisoned() -> crate::DbError {
267    crate::DbError::Message("pending device-join journal connection lock was poisoned".to_string())
268}
269
270pub(crate) fn begin_device_join_on(
271    conn: &rusqlite::Connection,
272    key: &str,
273    value: &str,
274) -> Result<DeviceJoinJournalRecord, crate::DbError> {
275    conn.execute(
276        "INSERT OR IGNORE INTO protocol_state (key, value) VALUES (?1, ?2)",
277        (key, value),
278    )
279    .map_err(crate::DbError::from)?;
280    let actual = crate::required_protocol_state_on(conn, key)?;
281    serde_json::from_str(&actual).map_err(crate::DbError::from)
282}
283
284pub(crate) fn advance_device_join_on(
285    conn: &rusqlite::Connection,
286    key: &str,
287    previous: &str,
288    next: &str,
289) -> Result<usize, crate::DbError> {
290    conn.execute(
291        "UPDATE protocol_state SET value = ?1 WHERE key = ?2 AND value = ?3",
292        (next, key, previous),
293    )
294    .map_err(crate::DbError::from)
295}
296
297pub(crate) fn device_join_records_on(
298    conn: &rusqlite::Connection,
299) -> Result<Vec<(String, String)>, crate::DbError> {
300    let mut statement = conn
301        .prepare(
302            "SELECT key, value FROM protocol_state
303                 WHERE key GLOB 'device_join/*' ORDER BY key",
304        )
305        .map_err(crate::DbError::from)?;
306    let rows = statement
307        .query_map([], |row| {
308            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
309        })
310        .map_err(crate::DbError::from)?;
311    rows.collect::<Result<Vec<_>, _>>()
312        .map_err(crate::DbError::from)
313}
314
315pub(crate) fn forget_device_join_on(
316    conn: &rusqlite::Connection,
317    key: &str,
318) -> Result<(), crate::DbError> {
319    conn.execute("DELETE FROM protocol_state WHERE key = ?1", [key])
320        .map(|_| ())
321        .map_err(crate::DbError::from)
322}
323
324impl StoreDatabase {
325    pub fn new_device_join_attempt_id(&self) -> DeviceJoinAttemptId {
326        DeviceJoinAttemptId::from_hash(ObjectHash::digest(
327            self.new_store_write_id().as_str().as_bytes(),
328        ))
329    }
330
331    pub async fn begin_device_join(
332        &self,
333        record: DeviceJoinJournalRecord,
334    ) -> Result<DeviceJoinJournalRecord, DeviceJoinJournalError> {
335        require_initial(&record)?;
336        let key = record.store_key();
337        let value = serde_json::to_string(&record)?;
338        self.call_database(move |session| session.begin_device_join(&key, &value))
339            .await
340            .map_err(DeviceJoinJournalError::Database)
341    }
342
343    pub async fn load_device_join(
344        &self,
345        attempt_id: DeviceJoinAttemptId,
346        role: DeviceJoinRole,
347    ) -> Result<Option<DeviceJoinJournalRecord>, DeviceJoinJournalError> {
348        let key = DeviceJoinJournalRecord::store_key_for(attempt_id, role);
349        let value = self
350            .get_protocol_state(&key)
351            .await
352            .map_err(DeviceJoinJournalError::Database)?;
353        value
354            .map(|value| {
355                serde_json::from_str(&value).map_err(DeviceJoinJournalError::Serialization)
356            })
357            .transpose()
358    }
359
360    pub async fn advance_device_join(
361        &self,
362        previous: &DeviceJoinJournalRecord,
363        next: DeviceJoinJournalRecord,
364    ) -> Result<(), DeviceJoinJournalError> {
365        validate_successor(previous, &next)?;
366        let key = previous.store_key();
367        let previous = serde_json::to_string(previous)?;
368        let next = serde_json::to_string(&next)?;
369        let changed = self
370            .call_database(move |session| session.advance_device_join(&key, &previous, &next))
371            .await
372            .map_err(DeviceJoinJournalError::Database)?;
373        if changed == 1 {
374            Ok(())
375        } else {
376            Err(DeviceJoinJournalError::JournalConflict)
377        }
378    }
379
380    /// Every attempt's journal record, for the sweeps that look across attempts.
381    ///
382    /// A row this binary cannot read is reported and skipped rather than failing
383    /// the sweep. The rows are one per attempt and role, and an attempt being
384    /// driven reads its own row by key through
385    /// [`load_device_join`](Self::load_device_join), which still refuses
386    /// anything it cannot parse. Aborting here instead meant one abandoned
387    /// attempt's record — left by an older binary, since the journal shape is
388    /// not carried across changes — stopped every later pairing on the device,
389    /// with nothing short of editing the database to recover.
390    async fn device_join_records(
391        &self,
392    ) -> Result<Vec<DeviceJoinJournalRecord>, DeviceJoinJournalError> {
393        let rows = self
394            .call_database(|session| session.device_join_records())
395            .await
396            .map_err(DeviceJoinJournalError::Database)?;
397        let mut records = Vec::with_capacity(rows.len());
398        for (key, value) in rows {
399            let record: DeviceJoinJournalRecord = match serde_json::from_str(&value) {
400                Ok(record) => record,
401                Err(error) => {
402                    tracing::warn!(
403                        journal_key = %key,
404                        %error,
405                        "Skipping a device join journal record this binary cannot read"
406                    );
407                    continue;
408                }
409            };
410            if record.store_key() != key {
411                tracing::warn!(
412                    journal_key = %key,
413                    record_key = %record.store_key(),
414                    "Skipping a device join journal record stored under another attempt's key"
415                );
416                continue;
417            }
418            records.push(record);
419        }
420        records.sort_by_key(DeviceJoinJournalRecord::sort_key);
421        Ok(records)
422    }
423
424    pub async fn device_join_status(
425        &self,
426        attempt_id: DeviceJoinAttemptId,
427        role: DeviceJoinRole,
428    ) -> Result<Option<DeviceJoinStatus>, DeviceJoinJournalError> {
429        self.load_device_join(attempt_id, role)
430            .await
431            .map(|record| record.as_ref().map(DeviceJoinJournalRecord::status))
432    }
433
434    pub async fn device_join_actions(
435        &self,
436    ) -> Result<Vec<DeviceJoinAction>, DeviceJoinJournalError> {
437        Ok(self
438            .device_join_records()
439            .await?
440            .iter()
441            .filter_map(DeviceJoinJournalRecord::action)
442            .collect())
443    }
444
445    /// Every owner journal row standing at a published activation, with the
446    /// registration of the device it activated.
447    ///
448    /// The owner's half of a join ends here and the row is never advanced past
449    /// it, so this is the whole set of attempts that could be finished.
450    pub async fn owner_device_joins_awaiting_arrival(
451        &self,
452    ) -> Result<
453        Vec<(
454            DeviceJoinAttemptId,
455            coven_protocol::store_commit::StoreDeviceRegistrationRef,
456        )>,
457        DeviceJoinJournalError,
458    > {
459        use coven_protocol::store_commit::device_join_journal::{
460            DeviceJoinRoleProgress, OwnerJoinProgress,
461        };
462
463        Ok(self
464            .device_join_records()
465            .await?
466            .into_iter()
467            .filter_map(|record| match &*record.progress {
468                // Both of the owner's ends: the cross-principal join hands the
469                // activation over, and the same-principal join hands the whole
470                // installation over. The second is the larger row by far — a
471                // snapshot's metadata and the bootstrap closure ride inside it.
472                DeviceJoinRoleProgress::Owner(
473                    OwnerJoinProgress::ActivationPrepared { registration, .. }
474                    | OwnerJoinProgress::SamePrincipalCompleted { registration, .. },
475                ) => Some((record.attempt_id, registration.clone())),
476                _ => None,
477            })
478            .collect())
479    }
480
481    /// Drop one attempt's journal row for one role.
482    ///
483    /// The row is this device's working notes on an exchange, not a record
484    /// anything later reads: what the join durably produced is the activation
485    /// commit and the outcome object it named, both of which live in history
486    /// and are what every other device verifies the join against.
487    pub async fn retire_device_join(
488        &self,
489        attempt_id: DeviceJoinAttemptId,
490        role: DeviceJoinRole,
491    ) -> Result<(), DeviceJoinJournalError> {
492        let key = DeviceJoinJournalRecord::store_key_for(attempt_id, role);
493        self.call_database(move |session| session.forget_device_join(&key))
494            .await
495            .map_err(DeviceJoinJournalError::Database)
496    }
497
498    #[cfg(any(test, feature = "test-utils"))]
499    pub async fn forget_for_test(
500        &self,
501        attempt_id: DeviceJoinAttemptId,
502        role: DeviceJoinRole,
503    ) -> Result<(), DeviceJoinJournalError> {
504        let key = DeviceJoinJournalRecord::store_key_for(attempt_id, role);
505        self.call_database(move |session| session.forget_device_join(&key))
506            .await
507            .map_err(DeviceJoinJournalError::Database)
508    }
509}
510
511#[cfg(test)]
512mod tests {
513    use super::*;
514
515    #[test]
516    fn pending_join_test_store_disables_commit_durability() {
517        let pending_dir = tempfile::tempdir().expect("create pending join directory");
518        let pending =
519            DeviceJoinJournalStore::open_for_test(pending_dir.path().join("pending.sqlite"))
520                .expect("open pending join journal");
521
522        let synchronous = pending
523            .synchronous_for_test()
524            .expect("read synchronous setting");
525
526        assert_eq!(synchronous, 0);
527    }
528
529    #[test]
530    fn retiring_the_last_attempt_deletes_the_journal_file() {
531        let pending_dir = tempfile::tempdir().expect("create pending join directory");
532        let path = pending_dir.path().join("pending.sqlite");
533        let pending = DeviceJoinJournalStore::open(&path).expect("open pending join journal");
534        pending
535            .insert_or_load("attempt-one", "joiner", "first")
536            .expect("begin the first attempt");
537        pending
538            .insert_or_load("attempt-two", "joiner", "second")
539            .expect("begin the second attempt");
540
541        assert!(pending
542            .compare_and_forget("attempt-one", "joiner", "first")
543            .expect("retire the first attempt"));
544        assert!(
545            path.exists(),
546            "a journal still holding an attempt keeps its file"
547        );
548
549        assert!(pending
550            .compare_and_forget("attempt-two", "joiner", "second")
551            .expect("retire the last attempt"));
552        assert!(!path.exists(), "an emptied journal deletes its file");
553        for sidecar in ["pending.sqlite-wal", "pending.sqlite-shm"] {
554            assert!(
555                !pending_dir.path().join(sidecar).exists(),
556                "an emptied journal deletes its {sidecar} sidecar"
557            );
558        }
559    }
560
561    #[test]
562    fn a_journal_used_after_its_file_was_deleted_writes_a_new_one() {
563        let pending_dir = tempfile::tempdir().expect("create pending join directory");
564        let path = pending_dir.path().join("pending.sqlite");
565        let pending = DeviceJoinJournalStore::open(&path).expect("open pending join journal");
566        pending
567            .insert_or_load("attempt-one", "joiner", "first")
568            .expect("begin the first attempt");
569        pending
570            .compare_and_forget("attempt-one", "joiner", "first")
571            .expect("retire the first attempt");
572
573        pending
574            .insert_or_load("attempt-two", "joiner", "second")
575            .expect("a later attempt reopens the deleted journal");
576
577        assert!(path.exists());
578        assert_eq!(
579            DeviceJoinJournalStore::open(&path)
580                .expect("reopen the journal")
581                .records()
582                .expect("read the reopened journal"),
583            vec![(
584                "attempt-two".to_string(),
585                "joiner".to_string(),
586                "second".to_string()
587            )],
588            "the later attempt is durable in the new file, not an unlinked one"
589        );
590    }
591
592    #[test]
593    fn a_failed_retire_leaves_the_journal_file_alone() {
594        let pending_dir = tempfile::tempdir().expect("create pending join directory");
595        let path = pending_dir.path().join("pending.sqlite");
596        let pending = DeviceJoinJournalStore::open(&path).expect("open pending join journal");
597        pending
598            .insert_or_load("attempt-one", "joiner", "first")
599            .expect("begin the attempt");
600
601        assert!(!pending
602            .compare_and_forget("attempt-one", "joiner", "stale")
603            .expect("refuse to retire a row whose payload moved on"));
604
605        assert!(path.exists());
606        assert_eq!(
607            pending
608                .load("attempt-one", "joiner")
609                .expect("read the attempt"),
610            Some("first".to_string())
611        );
612    }
613}
coven_database/store/device_join.rs

coven_database/store/
device_join.rs
