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        use coven_protocol::store_commit::device_join_journal::{
366            DeviceJoinRoleProgress, OwnerJoinProgress,
367        };
368
369        // Publication preparation and completion also own the active author
370        // reservation and accepted authority. Their transaction is the only
371        // route into or out of the prepared publication state.
372        if [&*previous.progress, &*next.progress]
373            .into_iter()
374            .any(|progress| {
375                matches!(
376                    progress,
377                    DeviceJoinRoleProgress::Owner(OwnerJoinProgress::StorePublicationPrepared(_))
378                )
379            })
380        {
381            return Err(DeviceJoinJournalError::NonAdjacentJournalTransition);
382        }
383        validate_successor(previous, &next)?;
384        let key = previous.store_key();
385        let previous = serde_json::to_string(previous)?;
386        let next = serde_json::to_string(&next)?;
387        let changed = self
388            .call_database(move |session| session.advance_device_join(&key, &previous, &next))
389            .await
390            .map_err(DeviceJoinJournalError::Database)?;
391        if changed == 1 {
392            Ok(())
393        } else {
394            Err(DeviceJoinJournalError::JournalConflict)
395        }
396    }
397
398    /// Every attempt's journal record, for the sweeps that look across attempts.
399    ///
400    /// A row this binary cannot read is reported and skipped rather than failing
401    /// the sweep. The rows are one per attempt and role, and an attempt being
402    /// driven reads its own row by key through
403    /// [`load_device_join`](Self::load_device_join), which still refuses
404    /// anything it cannot parse. Aborting here instead meant one abandoned
405    /// attempt's record — left by an older binary, since the journal shape is
406    /// not carried across changes — stopped every later pairing on the device,
407    /// with nothing short of editing the database to recover.
408    async fn device_join_records(
409        &self,
410    ) -> Result<Vec<DeviceJoinJournalRecord>, DeviceJoinJournalError> {
411        let rows = self
412            .call_database(|session| session.device_join_records())
413            .await
414            .map_err(DeviceJoinJournalError::Database)?;
415        let mut records = Vec::with_capacity(rows.len());
416        for (key, value) in rows {
417            let record: DeviceJoinJournalRecord = match serde_json::from_str(&value) {
418                Ok(record) => record,
419                Err(error) => {
420                    tracing::warn!(
421                        journal_key = %key,
422                        %error,
423                        "Skipping a device join journal record this binary cannot read"
424                    );
425                    continue;
426                }
427            };
428            if record.store_key() != key {
429                tracing::warn!(
430                    journal_key = %key,
431                    record_key = %record.store_key(),
432                    "Skipping a device join journal record stored under another attempt's key"
433                );
434                continue;
435            }
436            records.push(record);
437        }
438        records.sort_by_key(DeviceJoinJournalRecord::sort_key);
439        Ok(records)
440    }
441
442    pub async fn device_join_status(
443        &self,
444        attempt_id: DeviceJoinAttemptId,
445        role: DeviceJoinRole,
446    ) -> Result<Option<DeviceJoinStatus>, DeviceJoinJournalError> {
447        self.load_device_join(attempt_id, role)
448            .await
449            .map(|record| record.as_ref().map(DeviceJoinJournalRecord::status))
450    }
451
452    pub async fn device_join_actions(
453        &self,
454    ) -> Result<Vec<DeviceJoinAction>, DeviceJoinJournalError> {
455        Ok(self
456            .device_join_records()
457            .await?
458            .iter()
459            .filter_map(DeviceJoinJournalRecord::action)
460            .collect())
461    }
462
463    /// Every owner journal row standing at a published activation, with the
464    /// registration of the device it activated.
465    ///
466    /// The owner's half of a join ends here and the row is never advanced past
467    /// it, so this is the whole set of attempts that could be finished.
468    pub async fn owner_device_joins_awaiting_arrival(
469        &self,
470    ) -> Result<
471        Vec<(
472            DeviceJoinAttemptId,
473            coven_protocol::store_commit::StoreDeviceRegistrationRef,
474        )>,
475        DeviceJoinJournalError,
476    > {
477        use coven_protocol::store_commit::device_join_journal::{
478            DeviceJoinRoleProgress, OwnerJoinProgress,
479        };
480
481        Ok(self
482            .device_join_records()
483            .await?
484            .into_iter()
485            .filter_map(|record| match &*record.progress {
486                // Both of the owner's ends: the cross-principal join hands the
487                // activation over, and the same-principal join hands the whole
488                // installation over. The second is the larger row by far — a
489                // snapshot's metadata and the bootstrap closure ride inside it.
490                DeviceJoinRoleProgress::Owner(
491                    OwnerJoinProgress::ActivationPrepared { registration, .. }
492                    | OwnerJoinProgress::SamePrincipalCompleted { registration, .. },
493                ) => Some((record.attempt_id, registration.clone())),
494                _ => None,
495            })
496            .collect())
497    }
498
499    /// Drop one attempt's journal row for one role.
500    ///
501    /// The row is this device's working notes on an exchange, not a record
502    /// anything later reads: what the join durably produced is the activation
503    /// commit and the outcome object it named, both of which live in history
504    /// and are what every other device verifies the join against.
505    pub async fn retire_device_join(
506        &self,
507        attempt_id: DeviceJoinAttemptId,
508        role: DeviceJoinRole,
509    ) -> Result<(), DeviceJoinJournalError> {
510        let key = DeviceJoinJournalRecord::store_key_for(attempt_id, role);
511        self.call_database(move |session| session.forget_device_join(&key))
512            .await
513            .map_err(DeviceJoinJournalError::Database)
514    }
515
516    #[cfg(any(test, feature = "test-utils"))]
517    pub async fn forget_for_test(
518        &self,
519        attempt_id: DeviceJoinAttemptId,
520        role: DeviceJoinRole,
521    ) -> Result<(), DeviceJoinJournalError> {
522        let key = DeviceJoinJournalRecord::store_key_for(attempt_id, role);
523        self.call_database(move |session| session.forget_device_join(&key))
524            .await
525            .map_err(DeviceJoinJournalError::Database)
526    }
527}
528
529#[cfg(test)]
530mod tests {
531    use super::*;
532
533    #[test]
534    fn pending_join_test_store_disables_commit_durability() {
535        let pending_dir = tempfile::tempdir().expect("create pending join directory");
536        let pending =
537            DeviceJoinJournalStore::open_for_test(pending_dir.path().join("pending.sqlite"))
538                .expect("open pending join journal");
539
540        let synchronous = pending
541            .synchronous_for_test()
542            .expect("read synchronous setting");
543
544        assert_eq!(synchronous, 0);
545    }
546
547    #[test]
548    fn retiring_the_last_attempt_deletes_the_journal_file() {
549        let pending_dir = tempfile::tempdir().expect("create pending join directory");
550        let path = pending_dir.path().join("pending.sqlite");
551        let pending = DeviceJoinJournalStore::open(&path).expect("open pending join journal");
552        pending
553            .insert_or_load("attempt-one", "joiner", "first")
554            .expect("begin the first attempt");
555        pending
556            .insert_or_load("attempt-two", "joiner", "second")
557            .expect("begin the second attempt");
558
559        assert!(pending
560            .compare_and_forget("attempt-one", "joiner", "first")
561            .expect("retire the first attempt"));
562        assert!(
563            path.exists(),
564            "a journal still holding an attempt keeps its file"
565        );
566
567        assert!(pending
568            .compare_and_forget("attempt-two", "joiner", "second")
569            .expect("retire the last attempt"));
570        assert!(!path.exists(), "an emptied journal deletes its file");
571        for sidecar in ["pending.sqlite-wal", "pending.sqlite-shm"] {
572            assert!(
573                !pending_dir.path().join(sidecar).exists(),
574                "an emptied journal deletes its {sidecar} sidecar"
575            );
576        }
577    }
578
579    #[test]
580    fn a_journal_used_after_its_file_was_deleted_writes_a_new_one() {
581        let pending_dir = tempfile::tempdir().expect("create pending join directory");
582        let path = pending_dir.path().join("pending.sqlite");
583        let pending = DeviceJoinJournalStore::open(&path).expect("open pending join journal");
584        pending
585            .insert_or_load("attempt-one", "joiner", "first")
586            .expect("begin the first attempt");
587        pending
588            .compare_and_forget("attempt-one", "joiner", "first")
589            .expect("retire the first attempt");
590
591        pending
592            .insert_or_load("attempt-two", "joiner", "second")
593            .expect("a later attempt reopens the deleted journal");
594
595        assert!(path.exists());
596        assert_eq!(
597            DeviceJoinJournalStore::open(&path)
598                .expect("reopen the journal")
599                .records()
600                .expect("read the reopened journal"),
601            vec![(
602                "attempt-two".to_string(),
603                "joiner".to_string(),
604                "second".to_string()
605            )],
606            "the later attempt is durable in the new file, not an unlinked one"
607        );
608    }
609
610    #[test]
611    fn a_failed_retire_leaves_the_journal_file_alone() {
612        let pending_dir = tempfile::tempdir().expect("create pending join directory");
613        let path = pending_dir.path().join("pending.sqlite");
614        let pending = DeviceJoinJournalStore::open(&path).expect("open pending join journal");
615        pending
616            .insert_or_load("attempt-one", "joiner", "first")
617            .expect("begin the attempt");
618
619        assert!(!pending
620            .compare_and_forget("attempt-one", "joiner", "stale")
621            .expect("refuse to retire a row whose payload moved on"));
622
623        assert!(path.exists());
624        assert_eq!(
625            pending
626                .load("attempt-one", "joiner")
627                .expect("read the attempt"),
628            Some("first".to_string())
629        );
630    }
631}
coven_database/store/device_join.rs

coven_database/store/
device_join.rs
