coven_database/store/
store_database.rs

1use super::*;
2
3/// Store operations backed by one owned database.
4///
5/// This capability retains the database as a whole. It never copies out the
6/// connection, Store directory, schema configuration, clock, gates, blob
7/// declarations, or coordination state that database owns.
8#[derive(Clone)]
9pub struct StoreDatabase {
10    database: Database,
11}
12
13impl StoreDatabase {
14    #[doc(hidden)]
15    pub fn from_database(database: Database) -> Self {
16        Self { database }
17    }
18
19    #[doc(hidden)]
20    pub fn subscribe_committed_changes(
21        &self,
22    ) -> tokio::sync::broadcast::Receiver<std::sync::Arc<crate::CommittedChanges>> {
23        self.database.subscribe_committed_changes()
24    }
25
26    pub(super) async fn call_store<F, R>(&self, operation: F) -> Result<R, DbError>
27    where
28        F: for<'session> FnOnce(&mut StoreSession<'session>) -> Result<R, DbError> + Send + 'static,
29        R: Send + 'static,
30    {
31        self.database.call_store(operation).await
32    }
33
34    pub(super) async fn call_database<F, R>(&self, operation: F) -> Result<R, DbError>
35    where
36        F: for<'session> FnOnce(
37                &mut crate::database_session::DatabaseSession<'session>,
38            ) -> Result<R, DbError>
39            + Send
40            + 'static,
41        R: Send + 'static,
42    {
43        self.database.call_database(operation).await
44    }
45
46    pub async fn read<F, R, E>(&self, read: F) -> Result<Result<R, E>, DbError>
47    where
48        F: for<'connection> FnOnce(SqlReadContext<'connection>) -> Result<R, E> + Send + 'static,
49        R: Send + 'static,
50        E: Send + 'static,
51    {
52        self.database.read_store(read).await
53    }
54
55    pub fn schema_version(&self) -> u32 {
56        self.database.store_schema_version()
57    }
58
59    #[cfg(any(test, feature = "test-utils"))]
60    pub fn assert_owns_payload_directory_for_test(
61        &self,
62        store_dir: &coven_foundation::store_dir::StoreDir,
63    ) {
64        self.database
65            .assert_owns_payload_directory_for_test(store_dir);
66    }
67
68    pub fn sync_routing_hash(&self) -> coven_protocol::store_commit::ObjectHash {
69        self.database.store_sync_routing_hash()
70    }
71
72    pub fn has_synced_tables(&self) -> bool {
73        self.database.store_has_synced_tables()
74    }
75
76    pub fn blob_transition_root(&self, table_name: &str) -> crate::BlobTransitionRoot {
77        self.database.store_blob_transition_root(table_name)
78    }
79
80    pub fn transfer_limits(&self) -> coven_protocol::blob::TransferLimits {
81        self.database.store_transfer_limits()
82    }
83
84    /// Replace the transfer limits for every later upload-drain pass and pin
85    /// call. A pass already running keeps the limit it admitted under.
86    pub fn set_transfer_limits(&self, limits: coven_protocol::blob::TransferLimits) {
87        self.database.set_store_transfer_limits(limits)
88    }
89
90    pub fn blob_tombstone_grace(&self) -> chrono::Duration {
91        self.database.store_blob_tombstone_grace()
92    }
93
94    pub fn has_scoped_graph(&self) -> bool {
95        self.database.store_has_scoped_graph()
96    }
97
98    pub fn stamp(&self) -> String {
99        self.database.store_stamp()
100    }
101
102    pub async fn persist_hlc_high_water(&self) -> Result<(), DbError> {
103        self.set_protocol_state(
104            coven_protocol::hlc::HIGHWATER_STATE_KEY,
105            &self.database.store_hlc_high_water(),
106        )
107        .await
108    }
109
110    pub fn blob_ref_from_change(
111        &self,
112        change: &coven_foundation::changeset::RowChange,
113    ) -> Result<Option<coven_protocol::blob::BlobRef>, crate::BlobDeclError> {
114        self.database.store_blob_ref_from_change(change)
115    }
116
117    pub fn validate_local_blob_cleanup_changes(
118        &self,
119        old_changes: &[coven_foundation::changeset::RowChange],
120        new_changes: &[coven_foundation::changeset::RowChange],
121    ) -> Result<(), crate::BlobDeclError> {
122        self.database
123            .validate_store_local_blob_cleanup_changes(old_changes, new_changes)
124    }
125
126    pub fn receive_wall_ms(&self) -> u64 {
127        self.database.store_receive_wall_ms()
128    }
129
130    pub fn new_store_write_id(&self) -> coven_protocol::write::WriteId {
131        coven_protocol::write::WriteId::from_generated(self.database.new_store_id())
132    }
133
134    pub async fn get_protocol_state(&self, key: &str) -> Result<Option<String>, DbError> {
135        let key = key.to_string();
136        self.call_store(move |session| session.protocol_state(&key))
137            .await
138    }
139
140    pub async fn set_protocol_state(&self, key: &str, value: &str) -> Result<(), DbError> {
141        let key = key.to_string();
142        let value = value.to_string();
143        self.call_store(move |session| session.set_protocol_state(&key, &value))
144            .await
145    }
146
147    pub async fn get_cache_budget(&self, namespace: &str) -> Result<Option<u64>, DbError> {
148        let key = cache_budget_state_key(namespace);
149        match self.get_protocol_state(&key).await? {
150            Some(raw) => raw.parse::<u64>().map(Some).map_err(|error| {
151                DbError::context(
152                    format!("cache budget for {namespace:?} in protocol_state is not a byte count"),
153                    error,
154                )
155            }),
156            None => Ok(None),
157        }
158    }
159
160    #[doc(hidden)]
161    pub async fn set_cache_budget(&self, namespace: &str, max_bytes: u64) -> Result<(), DbError> {
162        let key = cache_budget_state_key(namespace);
163        self.set_protocol_state(&key, &max_bytes.to_string()).await
164    }
165
166    pub async fn write_status(
167        &self,
168        write_id: &coven_protocol::write::WriteId,
169    ) -> Result<coven_protocol::write::WriteStatus, DbError> {
170        let write_id = write_id.clone();
171        self.call_store(move |session| session.write_status(&write_id))
172            .await
173    }
174
175    pub async fn store_current_publication(
176        &self,
177    ) -> Result<crate::ObservedStorePublication, DbError> {
178        self.call_store(|session| session.store_current_publication())
179            .await
180    }
181
182    pub fn notify_write_status(
183        &self,
184        write_id: coven_protocol::write::WriteId,
185        status: coven_protocol::write::WriteStatus,
186    ) {
187        self.database.notify_store_write_status(write_id, status);
188    }
189
190    pub(super) fn subscribe_store_write_status(
191        &self,
192        write_id: coven_protocol::write::WriteId,
193        current: coven_protocol::write::WriteStatus,
194    ) -> tokio::sync::watch::Receiver<coven_protocol::write::WriteStatus> {
195        self.database
196            .subscribe_store_write_status(write_id, current)
197    }
198
199    pub async fn membership_load_permit(&self) -> MembershipLoadPermit {
200        self.database.membership_load_permit().await
201    }
202
203    pub async fn membership_mutation_permit(&self) -> MembershipMutationPermit {
204        self.database.membership_mutation_permit().await
205    }
206
207    pub async fn store_creation_permit(&self) -> StoreCreationPermit {
208        self.database.store_creation_permit().await
209    }
210
211    pub async fn device_exclusion_permit(&self) -> DeviceExclusionPermit {
212        self.database.device_exclusion_permit().await
213    }
214
215    /// Wait for this device's turn to author its own next Store commit.
216    ///
217    /// Every path that reads the local position to compose a commit, and every
218    /// path that publishes a device head, takes this and holds it across the
219    /// pair. Never taken twice in one call chain: a composer holds it until its
220    /// candidate is either activated or durably persisted, and a publisher of an
221    /// already-persisted candidate takes it for that publication alone.
222    pub async fn author_own_stream(&self) -> OwnStreamAuthorship {
223        self.database.author_own_store_stream().await
224    }
225
226    /// Wait for this drain's exclusive turn over the blob upload queue.
227    ///
228    /// Taken before the queue is read and held for the whole pass, so the
229    /// entries a drain admits are entries no other drain is already running.
230    /// Never taken twice in one call chain: an upload attempt performs no
231    /// second drain.
232    pub async fn blob_upload_drain_permit(&self) -> BlobUploadDrainPermit {
233        self.database.blob_upload_drain_permit().await
234    }
235
236    pub async fn snapshot_publication_permit(&self) -> SnapshotPublicationPermit {
237        self.database.snapshot_publication_permit().await
238    }
239
240    pub(super) async fn local_blob_cleanup_permit(&self) -> LocalBlobCleanupPermit {
241        self.database.local_blob_cleanup_permit().await
242    }
243
244    pub(super) async fn apply_local_blob_cleanup_intent(
245        &self,
246        intent: &crate::local_blob_cleanup_intents::LocalBlobCleanupIntent,
247    ) -> Result<(), DbError> {
248        self.database.apply_local_blob_cleanup_intent(intent).await
249    }
250
251    pub(super) async fn stage_host_write_blobs<E>(
252        &self,
253        blobs: Vec<super::NewBlob>,
254    ) -> Result<super::StagedBlobBatch, crate::HostWriteError<E>> {
255        self.database.stage_host_write_blobs(blobs).await
256    }
257
258    pub async fn begin_store_creation_attempt(
259        &self,
260        initialized: coven_protocol::store_creation::StoreCreationAttempt,
261    ) -> Result<coven_protocol::store_creation::StoreCreationAttempt, DbError> {
262        let value = serde_json::to_string(&initialized)
263            .map_err(|error| DbError::context("serialize Store creation attempt", error))?;
264        self.call_store(move |session| session.begin_store_creation_attempt(&value))
265            .await
266    }
267
268    pub async fn load_store_creation_attempt(
269        &self,
270    ) -> Result<Option<coven_protocol::store_creation::StoreCreationAttempt>, DbError> {
271        self.call_store(|session| session.load_store_creation_attempt())
272            .await
273    }
274
275    pub async fn advance_store_creation_attempt(
276        &self,
277        previous: coven_protocol::store_creation::StoreCreationAttempt,
278        next: coven_protocol::store_creation::StoreCreationAttempt,
279    ) -> Result<(), DbError> {
280        let previous = serde_json::to_string(&previous)
281            .map_err(|error| DbError::context("serialize Store creation predecessor", error))?;
282        let next = serde_json::to_string(&next)
283            .map_err(|error| DbError::context("serialize Store creation successor", error))?;
284        self.call_store(move |session| session.advance_store_creation_attempt(&previous, &next))
285            .await
286    }
287
288    pub(super) async fn sync_store_parent_dir(
289        &self,
290        path: &std::path::Path,
291    ) -> Result<(), coven_foundation::atomic_file::FileError> {
292        self.database.sync_store_parent_dir(path).await
293    }
294
295    #[cfg(any(test, feature = "test-utils"))]
296    pub fn new(database: &Database) -> Self {
297        Self::from_database(database.clone())
298    }
299
300    #[cfg(any(test, feature = "test-utils"))]
301    pub fn arm_test_pause(
302        &self,
303        point: crate::DatabaseTestPoint,
304    ) -> (
305        std::sync::Arc<tokio::sync::Notify>,
306        std::sync::Arc<tokio::sync::Notify>,
307    ) {
308        self.database.arm_test_pause(point)
309    }
310
311    #[cfg(any(test, feature = "test-utils"))]
312    pub async fn set_invalid_cache_budget_for_test(
313        &self,
314        namespace: &str,
315        value: &str,
316    ) -> Result<(), DbError> {
317        let key = cache_budget_state_key(namespace);
318        self.set_protocol_state(&key, value).await
319    }
320
321    #[cfg(any(test, feature = "test-utils"))]
322    pub async fn reach_test_point(&self, point: crate::DatabaseTestPoint) {
323        self.database.reach_store_test_point(point).await;
324    }
325
326    #[cfg(any(test, feature = "test-utils"))]
327    pub async fn required_store_root_hash(
328        &self,
329    ) -> Result<coven_protocol::store_commit::ObjectHash, DbError> {
330        self.call_store(|session| Ok(session.required_root_authority()?.store_root_hash))
331            .await
332    }
333
334    #[cfg(any(test, feature = "test-utils"))]
335    pub async fn scoped_snapshot_counts_for_test(&self) -> Result<(i64, i64, i64), DbError> {
336        self.call_store(|session| session.scoped_snapshot_counts())
337            .await
338    }
339
340    #[cfg(any(test, feature = "test-utils"))]
341    pub async fn migrated_scoped_snapshot_facts_for_test(
342        &self,
343    ) -> Result<(i64, i64, String), DbError> {
344        self.call_store(|session| session.migrated_scoped_snapshot_facts())
345            .await
346    }
347
348    #[cfg(any(test, feature = "test-utils"))]
349    pub async fn generation_zero_replay_baseline_for_test(
350        &self,
351    ) -> Result<crate::RetainedReplayBaseline, DbError> {
352        self.call_store(|session| session.generation_zero_replay_baseline())
353            .await
354    }
355
356    #[cfg(any(test, feature = "test-utils"))]
357    pub async fn replace_generation_zero_replay_authority_for_test(
358        &self,
359        authority_bytes: Vec<u8>,
360    ) -> Result<(), DbError> {
361        self.call_store(move |session| {
362            session.replace_generation_zero_replay_authority(&authority_bytes)
363        })
364        .await
365    }
366
367    #[cfg(any(test, feature = "test-utils"))]
368    pub async fn circle_bootstrap_coverage_ref(
369        &self,
370        circle_id: coven_protocol::circle::CircleId,
371    ) -> Result<Option<coven_protocol::circle::CircleBootstrapCoverageRef>, DbError> {
372        self.call_store(move |session| session.circle_bootstrap_coverage_ref(circle_id))
373            .await
374    }
375
376    #[cfg(any(test, feature = "test-utils"))]
377    pub async fn circle_bootstrap_replay_inputs(
378        &self,
379    ) -> Result<
380        Vec<(
381            StoreBatchCommitRef,
382            coven_protocol::circle_activation::VerifiedCircleImage,
383        )>,
384        DbError,
385    > {
386        self.call_store(|session| session.circle_bootstrap_replay_inputs())
387            .await
388    }
389
390    #[cfg(any(test, feature = "test-utils"))]
391    pub async fn circle_control_activation_count_for_test(
392        &self,
393        circle_id: coven_protocol::circle::CircleId,
394    ) -> Result<i64, DbError> {
395        self.call_store(move |session| session.circle_control_activation_count(circle_id))
396            .await
397    }
398}
399
400impl coven_foundation::id_provider::IdProvider for StoreDatabase {
401    fn new_id(&self) -> String {
402        self.database.new_store_id()
403    }
404}
405
406#[cfg(test)]
407mod tests {
408    use super::*;
409    use coven_protocol::blob::{TransferLimits, BLOB_TOMBSTONE_GRACE};
410    use std::{collections::BTreeSet, sync::Arc};
411
412    #[tokio::test]
413    async fn read_only_store_reads_leave_writer_payload_cleanup_owed() {
414        let directory = tempfile::tempdir().expect("temp dir");
415        let path = directory.path().join("read-only-store.sqlite");
416        let writer = StoreDatabase::from_database(
417            Database::open(
418                &path,
419                Vec::new(),
420                BLOB_TOMBSTONE_GRACE,
421                TransferLimits::one_at_a_time(),
422                "writer".to_string(),
423                Arc::new(coven_foundation::clock::SystemClock),
424                crate::CovenMigrationPolicy::ApplyPending,
425                &[],
426            )
427            .expect("open writer"),
428        );
429
430        writer
431            .call_database(|session| {
432                session.run_test_sql(|database| {
433                    let hash = database.install_payload(b"pending cleanup")?;
434                    database.set_payload_owner_claims("owner", &BTreeSet::from([hash]))?;
435                    database.set_payload_owner_claims("owner", &BTreeSet::new())
436                })
437            })
438            .await
439            .expect("create pending payload cleanup");
440
441        let reader = StoreDatabase::from_database(
442            Database::open_read_only(
443                &path,
444                Vec::new(),
445                BLOB_TOMBSTONE_GRACE,
446                TransferLimits::one_at_a_time(),
447                "writer".to_string(),
448                Arc::new(coven_foundation::clock::SystemClock),
449                &[],
450            )
451            .expect("open reader"),
452        );
453
454        let value = reader
455            .read(|database| database.query_row("SELECT 1", [], |row| row.get::<_, i64>(0)))
456            .await
457            .expect("run read-only Store operation")
458            .expect("read value");
459        assert_eq!(value, 1);
460
461        let (tracked_value, _) = StoreReads::open(&path)
462            .expect("open application readers")
463            .read_tracked(|database| database.query_row("SELECT 2", [], |row| row.get::<_, i64>(0)))
464            .await
465            .expect("run tracked read-only Store operation");
466        assert_eq!(tracked_value.expect("read tracked value"), 2);
467
468        let cleanup_count: i64 = writer
469            .call_database(|session| {
470                session.run_test_sql(|database| {
471                    database
472                        .query_row("SELECT COUNT(*) FROM payload_cleanup", [], |row| row.get(0))
473                        .map_err(DbError::from)
474                })
475            })
476            .await
477            .expect("count pending payload cleanup");
478        assert_eq!(cleanup_count, 1);
479    }
480}
coven_database/store/store_database.rs

coven_database/store/
store_database.rs
