coven_database/store/store_session/
provider_probe.rs

1use super::*;
2
3use async_trait::async_trait;
4
5use coven_protocol::objects::{StorageBackendFailure, StorageError};
6use coven_protocol::provider::{ProviderProbeId, ProviderProbeJournal, ProviderProbeJournalRecord};
7
8impl StoreSession<'_> {
9    fn load_provider_probe_journal(
10        &self,
11        key: &str,
12    ) -> Result<Option<ProviderProbeJournalRecord>, DbError> {
13        crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
14            .protocol_state(key)?
15            .map(|value| {
16                serde_json::from_str(&value)
17                    .map_err(|error| DbError::context("parse provider probe journal", error))
18            })
19            .transpose()
20    }
21
22    fn begin_provider_probe_journal(
23        &self,
24        key: &str,
25        value: &str,
26    ) -> Result<ProviderProbeJournalRecord, DbError> {
27        let actual = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
28            .begin_protocol_state(key, value)?;
29        serde_json::from_str(&actual)
30            .map_err(|error| DbError::context("parse provider probe journal", error))
31    }
32
33    fn advance_provider_probe_journal(
34        &self,
35        key: &str,
36        previous: &str,
37        next: &str,
38    ) -> Result<(), DbError> {
39        if !crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
40            .compare_exchange_protocol_state(key, previous, next)?
41        {
42            return Err(DbError::Message(
43                "provider probe journal advance lost its exact predecessor".to_string(),
44            ));
45        }
46        Ok(())
47    }
48}
49
50#[async_trait]
51impl ProviderProbeJournal for StoreDatabase {
52    async fn load(
53        &self,
54        probe_id: ProviderProbeId,
55    ) -> Result<Option<ProviderProbeJournalRecord>, StorageError> {
56        let key = format!("provider_probe/{}", hex::encode(probe_id.as_bytes()));
57        self.call_store(move |session| session.load_provider_probe_journal(&key))
58            .await
59            .map_err(|error| {
60                StorageError::backend(
61                    StorageBackendFailure::Internal,
62                    "load provider probe journal",
63                    error,
64                )
65            })
66    }
67
68    async fn begin(
69        &self,
70        prepared: ProviderProbeJournalRecord,
71    ) -> Result<ProviderProbeJournalRecord, StorageError> {
72        prepared.validate_begin()?;
73        let key = format!(
74            "provider_probe/{}",
75            hex::encode(prepared.probe_id().as_bytes())
76        );
77        let value = serde_json::to_string(&prepared)?;
78        self.call_store(move |session| session.begin_provider_probe_journal(&key, &value))
79            .await
80            .map_err(|error| {
81                StorageError::backend(
82                    StorageBackendFailure::Internal,
83                    "begin provider probe journal",
84                    error,
85                )
86            })
87    }
88
89    async fn advance(
90        &self,
91        previous: &ProviderProbeJournalRecord,
92        next: ProviderProbeJournalRecord,
93    ) -> Result<(), StorageError> {
94        previous.validate_transition(&next)?;
95        let key = format!(
96            "provider_probe/{}",
97            hex::encode(previous.probe_id().as_bytes())
98        );
99        let previous = serde_json::to_string(previous)?;
100        let next = serde_json::to_string(&next)?;
101        self.call_store(move |session| {
102            session.advance_provider_probe_journal(&key, &previous, &next)
103        })
104        .await
105        .map_err(|error| {
106            StorageError::backend(
107                StorageBackendFailure::Internal,
108                "advance provider probe journal",
109                error,
110            )
111        })
112    }
113}
114
115#[cfg(test)]
116mod tests {
117    use super::*;
118    use coven_protocol::objects::ObjectSlot;
119    use coven_protocol::provider::test_fixtures::{
120        test_device_binding, test_exact_receipt, test_store_binding,
121    };
122    use coven_protocol::provider::{ExactProbeJournal, ExactProbeProgress};
123
124    #[tokio::test]
125    async fn database_probe_journal_rejects_a_skipped_progress_state() {
126        let db_store_dir = crate::synthetic_store::test_store_dir();
127        let db = crate::synthetic_store::open_test_db(db_store_dir.clone());
128        let journal = crate::StoreDatabase::new(&db);
129        let probe_id = ProviderProbeId::from_bytes([44; 32]);
130        let binding = coven_protocol::objects::ResolvedProviderBinding {
131            store: test_store_binding(),
132            device: test_device_binding(),
133        };
134        let prepared = ProviderProbeJournalRecord::Exact(ExactProbeJournal {
135            probe_id,
136            binding,
137            slot: ObjectSlot::logical("__coven_probe__/exact/journal".to_string()).unwrap(),
138            conditional_slot: ObjectSlot::logical(
139                "__coven_probe__/conditional/journal".to_string(),
140            )
141            .unwrap(),
142            lost_response_slot: ObjectSlot::logical(
143                "__coven_probe__/lost-response/journal".to_string(),
144            )
145            .unwrap(),
146            progress: ExactProbeProgress::Prepared,
147        });
148        assert_eq!(journal.begin(prepared.clone()).await.unwrap(), prepared);
149        let ProviderProbeJournalRecord::Exact(mut final_record) = prepared.clone() else {
150            unreachable!()
151        };
152        final_record.progress = ExactProbeProgress::ReceiptReady {
153            receipt: test_exact_receipt(),
154        };
155        let final_record = ProviderProbeJournalRecord::Exact(final_record);
156        assert!(journal.advance(&prepared, final_record).await.is_err());
157        assert_eq!(journal.load(probe_id).await.unwrap(), Some(prepared));
158    }
159}
coven_database/store/store_session/provider_probe.rs

coven_database/store/store_session/
provider_probe.rs
