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    /// Prepare an independently owned snapshot database with this receiver's
15    /// registered schema migrations. Preparation does not advance the live
16    /// register clock or replace the receiving connection.
17    pub async fn prepare_snapshot_database(
18        &self,
19        plaintext: Vec<u8>,
20        install: crate::VerifiedSnapshotBootstrapInstall,
21    ) -> Result<Self, crate::OpenError> {
22        self.database
23            .prepare_snapshot_database(plaintext, install)
24            .await
25            .map(Self::from_database)
26    }
27
28    /// Stop the preparation worker and transfer its verified image to the
29    /// checkpoint installer. Every borrowed query handle must have been released.
30    pub async fn into_prepared_snapshot(self) -> Result<crate::PreparedStoreSnapshot, DbError> {
31        self.database.into_prepared_snapshot().await
32    }
33
34    /// Close and release an abandoned checkpoint preparation and its payloads.
35    pub async fn discard_snapshot_preparation(self) -> Result<(), DbError> {
36        self.database.discard_snapshot_preparation().await
37    }
38
39    /// Resolve this preparation's recipient Circle images before its accepted tail.
40    pub async fn prepare_received_snapshot_circles(
41        &self,
42        selection: crate::StagedCircleRestore,
43        receiver_wall_ms: u64,
44    ) -> Result<(), DbError> {
45        self.database
46            .prepare_received_snapshot_circles(selection, receiver_wall_ms)
47            .await
48    }
49
50    /// Install an independently prepared checkpoint and its accepted tail in
51    /// one receiving transaction, preserving the local write journal.
52    pub async fn install_received_snapshot(
53        &self,
54        prepared: crate::PreparedStoreSnapshot,
55        expected: crate::StorePublicationBoundary,
56        materializations: Vec<crate::PreparedMergeMaterialization>,
57        accepted: crate::AcceptedStorePublicationInterval,
58        snapshots: Vec<crate::VerifiedStoreSnapshotAuthority>,
59        membership: coven_protocol::membership::LocalStoreMembership,
60        routing_encryption: Option<coven_keys::encryption::EncryptionService>,
61        routing_key: Option<coven_protocol::circle::RowRoutingKey>,
62        receiver_wall_ms: u64,
63    ) -> Result<
64        (
65            crate::MaterializationOutcome,
66            Vec<coven_protocol::store_commit::StoreBatchCommitRef>,
67        ),
68        DbError,
69    > {
70        #[cfg(any(test, feature = "test-utils"))]
71        self.reach_test_point(crate::DatabaseTestPoint::ReceivedSnapshotInstallRequested)
72            .await;
73        let snapshots = snapshots
74            .into_iter()
75            .map(crate::VerifiedStoreSnapshotAuthority::into_authority)
76            .collect();
77        let (applied, installed) = self
78            .call_store(move |receiver| {
79                prepared.install_on(
80                    receiver,
81                    expected,
82                    materializations,
83                    accepted,
84                    snapshots,
85                    membership,
86                    routing_encryption.as_ref(),
87                    routing_key,
88                    receiver_wall_ms,
89                )
90            })
91            .await?;
92        for (write_id, status) in applied.write_status_notifications {
93            self.notify_write_status(write_id, status);
94        }
95        Ok((applied.outcome, installed))
96    }
97
98    #[doc(hidden)]
99    pub fn from_database(database: Database) -> Self {
100        Self { database }
101    }
102
103    #[doc(hidden)]
104    pub fn subscribe_committed_changes(
105        &self,
106    ) -> tokio::sync::broadcast::Receiver<std::sync::Arc<crate::CommittedChanges>> {
107        self.database.subscribe_committed_changes()
108    }
109
110    pub(super) fn call_store<F, R>(
111        &self,
112        operation: F,
113    ) -> impl std::future::Future<Output = Result<R, DbError>> + Send + '_
114    where
115        F: for<'session> FnOnce(&mut StoreSession<'session>) -> Result<R, DbError> + Send + 'static,
116        R: Send + 'static,
117    {
118        self.database.call_store(operation)
119    }
120
121    pub(super) fn call_database<F, R>(
122        &self,
123        operation: F,
124    ) -> impl std::future::Future<Output = Result<R, DbError>> + Send + '_
125    where
126        F: for<'session> FnOnce(
127                &mut crate::database_session::DatabaseSession<'session>,
128            ) -> Result<R, DbError>
129            + Send
130            + 'static,
131        R: Send + 'static,
132    {
133        self.database.call_database(operation)
134    }
135
136    pub fn read<F, R, E>(
137        &self,
138        read: F,
139    ) -> impl std::future::Future<Output = Result<Result<R, E>, DbError>> + Send + '_
140    where
141        F: for<'connection> FnOnce(SqlReadContext<'connection>) -> Result<R, E> + Send + 'static,
142        R: Send + 'static,
143        E: Send + 'static,
144    {
145        self.database.read_store(read)
146    }
147
148    pub fn schema_version(&self) -> u32 {
149        self.database.store_schema_version()
150    }
151
152    #[cfg(any(test, feature = "test-utils"))]
153    pub fn synced_tables_for_test(&self) -> Vec<coven_protocol::synced_schema::SyncedTable> {
154        self.database.synced_tables_for_test()
155    }
156
157    #[cfg(any(test, feature = "test-utils"))]
158    pub async fn replace_with_database_image_for_test(
159        &self,
160        image: Vec<u8>,
161    ) -> Result<(), DbError> {
162        self.database
163            .replace_with_database_image_for_test(image)
164            .await
165    }
166
167    #[cfg(any(test, feature = "test-utils"))]
168    pub async fn database_image_for_test(&self) -> Result<Vec<u8>, DbError> {
169        self.database.database_image_for_test().await
170    }
171
172    #[cfg(any(test, feature = "test-utils"))]
173    pub fn assert_owns_payload_directory_for_test(
174        &self,
175        store_dir: &coven_foundation::store_dir::StoreDir,
176    ) {
177        self.database
178            .assert_owns_payload_directory_for_test(store_dir);
179    }
180
181    pub fn sync_routing_hash(&self) -> coven_protocol::store_commit::ObjectHash {
182        self.database.store_sync_routing_hash()
183    }
184
185    pub fn has_synced_tables(&self) -> bool {
186        self.database.store_has_synced_tables()
187    }
188
189    pub fn blob_transition_root(&self, table_name: &str) -> crate::BlobTransitionRoot {
190        self.database.store_blob_transition_root(table_name)
191    }
192
193    pub fn transfer_limits(&self) -> coven_protocol::blob::TransferLimits {
194        self.database.store_transfer_limits()
195    }
196
197    /// Replace the transfer limits for every later upload-drain pass and pin
198    /// call. A pass already running keeps the limit it admitted under.
199    pub fn set_transfer_limits(&self, limits: coven_protocol::blob::TransferLimits) {
200        self.database.set_store_transfer_limits(limits)
201    }
202
203    pub fn blob_tombstone_grace(&self) -> chrono::Duration {
204        self.database.store_blob_tombstone_grace()
205    }
206
207    pub fn has_scoped_graph(&self) -> bool {
208        self.database.store_has_scoped_graph()
209    }
210
211    pub fn stamp(&self) -> String {
212        self.database.store_stamp()
213    }
214
215    pub async fn persist_hlc_high_water(&self) -> Result<(), DbError> {
216        let floor = self.database.store_hlc_high_water();
217        self.call_store(move |session| session.persist_clock_floor(&floor))
218            .await
219    }
220
221    pub fn blob_ref_from_change(
222        &self,
223        change: &coven_foundation::changeset::RowChange,
224    ) -> Result<Option<coven_protocol::blob::BlobRef>, crate::BlobDeclError> {
225        self.database.store_blob_ref_from_change(change)
226    }
227
228    pub fn validate_local_blob_cleanup_changes(
229        &self,
230        old_changes: &[coven_foundation::changeset::RowChange],
231        new_changes: &[coven_foundation::changeset::RowChange],
232    ) -> Result<(), crate::BlobDeclError> {
233        self.database
234            .validate_store_local_blob_cleanup_changes(old_changes, new_changes)
235    }
236
237    pub fn receive_wall_ms(&self) -> u64 {
238        self.database.store_receive_wall_ms()
239    }
240
241    pub fn new_store_write_id(&self) -> coven_protocol::write::WriteId {
242        coven_protocol::write::WriteId::from_generated(self.database.new_store_id())
243    }
244
245    pub async fn get_protocol_state(&self, key: &str) -> Result<Option<String>, DbError> {
246        let key = key.to_string();
247        self.call_store(move |session| session.protocol_state(&key))
248            .await
249    }
250
251    pub async fn set_protocol_state(&self, key: &str, value: &str) -> Result<(), DbError> {
252        let key = key.to_string();
253        let value = value.to_string();
254        self.call_store(move |session| session.set_protocol_state(&key, &value))
255            .await
256    }
257
258    pub async fn get_cache_budget(&self, namespace: &str) -> Result<Option<u64>, DbError> {
259        let key = cache_budget_state_key(namespace);
260        match self.get_protocol_state(&key).await? {
261            Some(raw) => raw.parse::<u64>().map(Some).map_err(|error| {
262                DbError::context(
263                    format!("cache budget for {namespace:?} in protocol_state is not a byte count"),
264                    error,
265                )
266            }),
267            None => Ok(None),
268        }
269    }
270
271    #[doc(hidden)]
272    pub async fn set_cache_budget(&self, namespace: &str, max_bytes: u64) -> Result<(), DbError> {
273        let key = cache_budget_state_key(namespace);
274        self.set_protocol_state(&key, &max_bytes.to_string()).await
275    }
276
277    pub async fn write_status(
278        &self,
279        write_id: &coven_protocol::write::WriteId,
280    ) -> Result<coven_protocol::write::WriteStatus, DbError> {
281        let write_id = write_id.clone();
282        self.call_store(move |session| session.write_status(&write_id))
283            .await
284    }
285
286    pub async fn store_current_publication(
287        &self,
288    ) -> Result<crate::StorePublicationBoundary, DbError> {
289        self.call_store(|session| session.store_current_publication())
290            .await
291    }
292
293    pub async fn store_publication_entries(
294        &self,
295    ) -> Result<
296        Vec<
297            coven_protocol::objects::ExactProtocolObject<
298                coven_protocol::store_commit::StorePublicationEntry,
299            >,
300        >,
301        DbError,
302    > {
303        self.call_store(|session| session.store_publication_entries())
304            .await
305    }
306
307    pub fn notify_write_status(
308        &self,
309        write_id: coven_protocol::write::WriteId,
310        status: coven_protocol::write::WriteStatus,
311    ) {
312        self.database.notify_store_write_status(write_id, status);
313    }
314
315    pub(super) fn subscribe_store_write_status(
316        &self,
317        write_id: coven_protocol::write::WriteId,
318        current: coven_protocol::write::WriteStatus,
319    ) -> tokio::sync::watch::Receiver<coven_protocol::write::WriteStatus> {
320        self.database
321            .subscribe_store_write_status(write_id, current)
322    }
323
324    pub async fn membership_load_permit(&self) -> MembershipLoadPermit {
325        self.database.membership_load_permit().await
326    }
327
328    pub async fn membership_mutation_permit(&self) -> MembershipMutationPermit {
329        self.database.membership_mutation_permit().await
330    }
331
332    pub async fn store_creation_permit(&self) -> StoreCreationPermit {
333        self.database.store_creation_permit().await
334    }
335
336    pub async fn device_exclusion_permit(&self) -> DeviceExclusionPermit {
337        self.database.device_exclusion_permit().await
338    }
339
340    /// Wait for this device's turn to author its own next Store commit.
341    ///
342    /// Every path that reads the local position to compose a commit, and every
343    /// path that publishes a device head, takes this and holds it across the
344    /// pair. Never taken twice in one call chain: a composer holds it until its
345    /// candidate is either activated or durably persisted, and a publisher of an
346    /// already-persisted candidate takes it for that publication alone.
347    pub async fn author_own_stream(&self) -> OwnStreamAuthorship {
348        OwnStreamAuthorship {
349            _guard: self.database.author_own_store_stream().await,
350            database: self.clone(),
351        }
352    }
353
354    /// Wait for this drain's exclusive turn over the blob upload queue.
355    ///
356    /// Taken before the queue is read and held for the whole pass, so the
357    /// entries a drain admits are entries no other drain is already running.
358    /// Never taken twice in one call chain: an upload attempt performs no
359    /// second drain.
360    pub async fn blob_upload_drain_permit(&self) -> BlobUploadDrainPermit {
361        self.database.blob_upload_drain_permit().await
362    }
363
364    pub async fn snapshot_publication_permit(&self) -> SnapshotPublicationPermit {
365        self.database.snapshot_publication_permit().await
366    }
367
368    pub(super) async fn local_blob_cleanup_permit(&self) -> LocalBlobCleanupPermit {
369        self.database.local_blob_cleanup_permit().await
370    }
371
372    pub(super) async fn apply_local_blob_cleanup_intent(
373        &self,
374        intent: &crate::local_blob_cleanup_intents::LocalBlobCleanupIntent,
375    ) -> Result<(), DbError> {
376        self.database.apply_local_blob_cleanup_intent(intent).await
377    }
378
379    pub(super) async fn stage_host_write_blobs<E>(
380        &self,
381        blobs: Vec<super::NewBlob>,
382    ) -> Result<super::StagedBlobBatch, crate::HostWriteError<E>> {
383        self.database.stage_host_write_blobs(blobs).await
384    }
385
386    pub async fn begin_store_creation_attempt(
387        &self,
388        initialized: coven_protocol::store_creation::StoreCreationAttempt,
389    ) -> Result<coven_protocol::store_creation::StoreCreationAttempt, DbError> {
390        let value = serde_json::to_string(&initialized)
391            .map_err(|error| DbError::context("serialize Store creation attempt", error))?;
392        self.call_store(move |session| session.begin_store_creation_attempt(&value))
393            .await
394    }
395
396    pub async fn load_store_creation_attempt(
397        &self,
398    ) -> Result<Option<coven_protocol::store_creation::StoreCreationAttempt>, DbError> {
399        self.call_store(|session| session.load_store_creation_attempt())
400            .await
401    }
402
403    pub async fn advance_store_creation_attempt(
404        &self,
405        previous: coven_protocol::store_creation::StoreCreationAttempt,
406        next: coven_protocol::store_creation::StoreCreationAttempt,
407    ) -> Result<(), DbError> {
408        let previous = serde_json::to_string(&previous)
409            .map_err(|error| DbError::context("serialize Store creation predecessor", error))?;
410        let next = serde_json::to_string(&next)
411            .map_err(|error| DbError::context("serialize Store creation successor", error))?;
412        self.call_store(move |session| session.advance_store_creation_attempt(&previous, &next))
413            .await
414    }
415
416    pub(super) async fn sync_store_parent_dir(
417        &self,
418        path: &std::path::Path,
419    ) -> Result<(), coven_foundation::atomic_file::FileError> {
420        self.database.sync_store_parent_dir(path).await
421    }
422
423    #[cfg(any(test, feature = "test-utils"))]
424    pub fn new(database: &Database) -> Self {
425        Self::from_database(database.clone())
426    }
427
428    #[cfg(any(test, feature = "test-utils"))]
429    pub fn arm_test_pause(
430        &self,
431        point: crate::DatabaseTestPoint,
432    ) -> (
433        std::sync::Arc<tokio::sync::Notify>,
434        std::sync::Arc<tokio::sync::Notify>,
435    ) {
436        self.database.arm_test_pause(point)
437    }
438
439    #[cfg(any(test, feature = "test-utils"))]
440    pub async fn set_invalid_cache_budget_for_test(
441        &self,
442        namespace: &str,
443        value: &str,
444    ) -> Result<(), DbError> {
445        let key = cache_budget_state_key(namespace);
446        self.set_protocol_state(&key, value).await
447    }
448
449    #[cfg(any(test, feature = "test-utils"))]
450    pub async fn reach_test_point(&self, point: crate::DatabaseTestPoint) {
451        self.database.reach_store_test_point(point).await;
452    }
453
454    #[cfg(any(test, feature = "test-utils"))]
455    pub async fn required_store_root_hash(
456        &self,
457    ) -> Result<coven_protocol::store_commit::ObjectHash, DbError> {
458        self.call_store(|session| Ok(session.required_root_authority()?.store_root_hash))
459            .await
460    }
461
462    #[cfg(any(test, feature = "test-utils"))]
463    pub async fn scoped_snapshot_counts_for_test(&self) -> Result<(i64, i64, i64), DbError> {
464        self.call_store(|session| session.scoped_snapshot_counts())
465            .await
466    }
467
468    #[cfg(any(test, feature = "test-utils"))]
469    pub async fn migrated_scoped_snapshot_facts_for_test(
470        &self,
471    ) -> Result<(i64, i64, String), DbError> {
472        self.call_store(|session| session.migrated_scoped_snapshot_facts())
473            .await
474    }
475
476    #[cfg(any(test, feature = "test-utils"))]
477    pub async fn replay_baseline_for_test(&self) -> Result<crate::RetainedReplayBaseline, DbError> {
478        self.call_store(|session| session.load_replay_baseline())
479            .await
480    }
481
482    #[cfg(any(test, feature = "test-utils"))]
483    pub async fn replace_replay_authority_for_test(
484        &self,
485        authority_bytes: Vec<u8>,
486    ) -> Result<(), DbError> {
487        self.call_store(move |session| session.replace_replay_authority(&authority_bytes))
488            .await
489    }
490
491    #[cfg(any(test, feature = "test-utils"))]
492    pub async fn circle_bootstrap_coverage_ref(
493        &self,
494        circle_id: coven_protocol::circle::CircleId,
495    ) -> Result<Option<coven_protocol::circle::CircleBootstrapCoverageRef>, DbError> {
496        self.call_store(move |session| session.circle_bootstrap_coverage_ref(circle_id))
497            .await
498    }
499
500    #[cfg(any(test, feature = "test-utils"))]
501    pub async fn circle_bootstrap_replay_inputs(
502        &self,
503    ) -> Result<
504        Vec<(
505            StoreBatchCommitRef,
506            coven_protocol::circle_activation::VerifiedCircleImage,
507        )>,
508        DbError,
509    > {
510        self.call_store(|session| session.circle_bootstrap_replay_inputs())
511            .await
512    }
513
514    #[cfg(any(test, feature = "test-utils"))]
515    pub async fn circle_control_activation_count_for_test(
516        &self,
517        circle_id: coven_protocol::circle::CircleId,
518    ) -> Result<i64, DbError> {
519        self.call_store(move |session| session.circle_control_activation_count(circle_id))
520            .await
521    }
522}
523
524impl coven_foundation::id_provider::IdProvider for StoreDatabase {
525    fn new_id(&self) -> String {
526        self.database.new_store_id()
527    }
528}
529
530#[cfg(test)]
531mod tests {
532    use super::*;
533    use coven_protocol::blob::{TransferLimits, BLOB_TOMBSTONE_GRACE};
534    use std::{collections::BTreeSet, sync::Arc};
535
536    #[tokio::test]
537    async fn store_calls_dispatch_in_poll_order_and_ignore_unpolled_calls() {
538        let store = StoreDatabase::from_database(crate::tests::fixtures::open_outbox_database(
539            "poll-order",
540        ));
541        let unpolled = store.call_store(|session| session.set_protocol_state("dropped", "written"));
542        drop(unpolled);
543        assert_eq!(store.get_protocol_state("dropped").await.unwrap(), None);
544
545        let first = store.call_store(|session| session.set_protocol_state("order", "first"));
546        let second = store.call_store(|session| session.set_protocol_state("order", "second"));
547        second.await.unwrap();
548        first.await.unwrap();
549        assert_eq!(
550            store.get_protocol_state("order").await.unwrap().as_deref(),
551            Some("first")
552        );
553    }
554
555    #[tokio::test]
556    async fn cancelling_a_dispatched_store_call_preserves_its_commit() {
557        let store = StoreDatabase::from_database(crate::tests::fixtures::open_outbox_database(
558            "cancel-dispatched",
559        ));
560        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
561        let (release_tx, release_rx) = std::sync::mpsc::channel();
562        let mut call = Box::pin(store.call_store(move |session| {
563            started_tx.send(()).expect("signal dispatched operation");
564            release_rx.recv().expect("release dispatched operation");
565            session.set_protocol_state("cancelled", "committed")
566        }));
567        tokio::select! {
568            result = started_rx => result.expect("operation started"),
569            result = &mut call => panic!("operation completed before release: {result:?}"),
570        }
571        drop(call);
572        release_tx.send(()).expect("release database worker");
573        assert_eq!(
574            store
575                .get_protocol_state("cancelled")
576                .await
577                .unwrap()
578                .as_deref(),
579            Some("committed")
580        );
581    }
582
583    #[tokio::test]
584    async fn read_only_store_reads_leave_writer_payload_cleanup_owed() {
585        let directory = tempfile::tempdir().expect("temp dir");
586        let path = directory.path().join("read-only-store.sqlite");
587        let writer = StoreDatabase::from_database(
588            Database::open(
589                &path,
590                Vec::new(),
591                BLOB_TOMBSTONE_GRACE,
592                TransferLimits::one_at_a_time(),
593                "writer".to_string(),
594                Arc::new(coven_foundation::clock::SystemClock),
595                crate::CovenMigrationPolicy::ApplyPending,
596                &[],
597            )
598            .expect("open writer"),
599        );
600
601        writer
602            .call_database(|session| {
603                session.run_test_sql(|database| {
604                    let hash = database.install_payload(b"pending cleanup")?;
605                    database.set_payload_owner_claims("owner", &BTreeSet::from([hash]))?;
606                    database.set_payload_owner_claims("owner", &BTreeSet::new())
607                })
608            })
609            .await
610            .expect("create pending payload cleanup");
611
612        let reader = StoreDatabase::from_database(
613            Database::open_read_only(
614                &path,
615                Vec::new(),
616                BLOB_TOMBSTONE_GRACE,
617                TransferLimits::one_at_a_time(),
618                "writer".to_string(),
619                Arc::new(coven_foundation::clock::SystemClock),
620                &[],
621            )
622            .expect("open reader"),
623        );
624
625        let value = reader
626            .read(|database| database.query_row("SELECT 1", [], |row| row.get::<_, i64>(0)))
627            .await
628            .expect("run read-only Store operation")
629            .expect("read value");
630        assert_eq!(value, 1);
631
632        let (tracked_value, _) = StoreReads::open(&path)
633            .expect("open application readers")
634            .read_tracked(|database| database.query_row("SELECT 2", [], |row| row.get::<_, i64>(0)))
635            .await
636            .expect("run tracked read-only Store operation");
637        assert_eq!(tracked_value.expect("read tracked value"), 2);
638
639        let cleanup_count: i64 = writer
640            .call_database(|session| {
641                session.run_test_sql(|database| {
642                    database
643                        .query_row("SELECT COUNT(*) FROM payload_cleanup", [], |row| row.get(0))
644                        .map_err(DbError::from)
645                })
646            })
647            .await
648            .expect("count pending payload cleanup");
649        assert_eq!(cleanup_count, 1);
650    }
651}
coven_database/store/store_database.rs

coven_database/store/
store_database.rs
