coven_database/
database_runtime.rs

1use super::*;
2
3/// A cloneable handle to one owned database. The connection capability retains
4/// both the worker and its matching database context; this handle has no second
5/// path to either.
6#[derive(Clone)]
7pub struct Database {
8    connection: DatabaseConnection,
9}
10
11/// Bind a database to the store directory that owns its payload files.
12///
13/// Production databases live under their store directory. In-memory databases
14/// have no parent path, so the database opening boundary creates their
15/// process-local directory once and passes that dependency into the core.
16fn store_dir_of(path: &Path) -> coven_foundation::store_dir::StoreDir {
17    if path == Path::new(":memory:") {
18        return coven_foundation::store_dir::StoreDir::new_ephemeral(
19            std::env::temp_dir().join(format!("coven-in-memory-store-{}", uuid::Uuid::new_v4())),
20        );
21    }
22    let parent = path
23        .parent()
24        .filter(|parent| !parent.as_os_str().is_empty())
25        .unwrap_or_else(|| Path::new("."));
26    coven_foundation::store_dir::StoreDir::new(parent)
27}
28
29impl Database {
30    pub(crate) async fn prepare_received_snapshot_circles(
31        &self,
32        selection: crate::StagedCircleRestore,
33        receiver_wall_ms: u64,
34    ) -> Result<(), DbError> {
35        self.connection
36            .prepare_received_snapshot_circles(selection, receiver_wall_ms)
37            .await
38    }
39
40    pub(crate) async fn prepare_snapshot_database(
41        &self,
42        plaintext: Vec<u8>,
43        install: VerifiedSnapshotBootstrapInstall,
44    ) -> Result<Database, OpenError> {
45        let connection = self
46            .connection
47            .prepare_snapshot_database(plaintext, install)
48            .await?;
49        Ok(Self { connection })
50    }
51
52    pub(crate) async fn into_prepared_snapshot(
53        self,
54    ) -> Result<crate::PreparedStoreSnapshot, DbError> {
55        self.connection.into_prepared_snapshot().await
56    }
57
58    pub(crate) async fn discard_snapshot_preparation(self) -> Result<(), DbError> {
59        self.connection.discard_snapshot_preparation().await
60    }
61
62    pub(crate) fn from_core(core: DatabaseCore, thread_name: &str) -> Result<Self, DbError> {
63        Ok(Self {
64            connection: DatabaseConnection::start(core, thread_name)?,
65        })
66    }
67
68    pub(crate) fn call_database<F, R>(
69        &self,
70        operation: F,
71    ) -> impl std::future::Future<Output = Result<R, DbError>> + Send + '_
72    where
73        F: for<'session> FnOnce(
74                &mut crate::database_session::DatabaseSession<'session>,
75            ) -> Result<R, DbError>
76            + Send
77            + 'static,
78        R: Send + 'static,
79    {
80        self.connection.call_database(operation)
81    }
82
83    pub(crate) fn call_store<F, R>(
84        &self,
85        operation: F,
86    ) -> impl std::future::Future<Output = Result<R, DbError>> + Send + '_
87    where
88        F: for<'session> FnOnce(&mut crate::store::StoreSession<'session>) -> Result<R, DbError>
89            + Send
90            + 'static,
91        R: Send + 'static,
92    {
93        self.connection.call_store(operation)
94    }
95
96    pub(crate) fn read_store<F, R, E>(
97        &self,
98        read: F,
99    ) -> impl std::future::Future<Output = Result<Result<R, E>, DbError>> + Send + '_
100    where
101        F: for<'connection> FnOnce(crate::store::SqlReadContext<'connection>) -> Result<R, E>
102            + Send
103            + 'static,
104        R: Send + 'static,
105        E: Send + 'static,
106    {
107        self.connection.read_store(read)
108    }
109
110    pub(crate) fn store_schema_version(&self) -> u32 {
111        self.connection.store_schema_version()
112    }
113
114    pub(crate) fn store_sync_routing_hash(&self) -> ObjectHash {
115        self.connection.store_sync_routing_hash()
116    }
117
118    pub(crate) fn store_has_synced_tables(&self) -> bool {
119        self.connection.store_has_synced_tables()
120    }
121
122    pub(crate) fn store_blob_transition_root(&self, table_name: &str) -> BlobTransitionRoot {
123        self.connection.store_blob_transition_root(table_name)
124    }
125
126    pub(crate) fn store_transfer_limits(&self) -> coven_protocol::blob::TransferLimits {
127        self.connection.store_transfer_limits()
128    }
129
130    pub(crate) fn set_store_transfer_limits(&self, limits: coven_protocol::blob::TransferLimits) {
131        self.connection.set_store_transfer_limits(limits)
132    }
133
134    pub(crate) fn store_blob_tombstone_grace(&self) -> chrono::Duration {
135        self.connection.store_blob_tombstone_grace()
136    }
137
138    pub(crate) fn store_has_scoped_graph(&self) -> bool {
139        self.connection.store_has_scoped_graph()
140    }
141
142    pub(crate) fn store_stamp(&self) -> String {
143        self.connection.store_stamp()
144    }
145
146    pub(crate) fn store_hlc_high_water(&self) -> String {
147        self.connection.store_hlc_high_water()
148    }
149
150    pub(crate) fn store_blob_ref_from_change(
151        &self,
152        change: &coven_foundation::changeset::RowChange,
153    ) -> Result<Option<coven_protocol::blob::BlobRef>, BlobDeclError> {
154        self.connection.store_blob_ref_from_change(change)
155    }
156
157    pub(crate) fn validate_store_local_blob_cleanup_changes(
158        &self,
159        old_changes: &[coven_foundation::changeset::RowChange],
160        new_changes: &[coven_foundation::changeset::RowChange],
161    ) -> Result<(), BlobDeclError> {
162        self.connection
163            .validate_store_local_blob_cleanup_changes(old_changes, new_changes)
164    }
165
166    pub(crate) fn store_receive_wall_ms(&self) -> u64 {
167        self.connection.store_receive_wall_ms()
168    }
169
170    #[cfg(any(test, feature = "test-utils"))]
171    pub(crate) fn assert_owns_payload_directory_for_test(
172        &self,
173        store_dir: &coven_foundation::store_dir::StoreDir,
174    ) {
175        self.connection
176            .assert_owns_payload_directory_for_test(store_dir);
177    }
178
179    pub(crate) fn new_store_id(&self) -> String {
180        self.connection.new_store_id()
181    }
182
183    pub(crate) fn notify_store_write_status(&self, write_id: WriteId, status: WriteStatus) {
184        self.connection.notify_store_write_status(write_id, status);
185    }
186
187    pub(crate) fn subscribe_store_write_status(
188        &self,
189        write_id: WriteId,
190        current: WriteStatus,
191    ) -> tokio::sync::watch::Receiver<WriteStatus> {
192        self.connection
193            .subscribe_store_write_status(write_id, current)
194    }
195
196    pub(crate) fn subscribe_committed_changes(
197        &self,
198    ) -> tokio::sync::broadcast::Receiver<Arc<crate::CommittedChanges>> {
199        self.connection.subscribe_committed_changes()
200    }
201
202    pub(crate) async fn membership_load_permit(&self) -> crate::store::MembershipLoadPermit {
203        self.connection.membership_load_permit().await
204    }
205
206    pub(crate) async fn membership_mutation_permit(
207        &self,
208    ) -> crate::store::MembershipMutationPermit {
209        self.connection.membership_mutation_permit().await
210    }
211
212    pub(crate) async fn store_creation_permit(&self) -> crate::store::StoreCreationPermit {
213        self.connection.store_creation_permit().await
214    }
215
216    pub(crate) async fn device_exclusion_permit(&self) -> crate::store::DeviceExclusionPermit {
217        self.connection.device_exclusion_permit().await
218    }
219
220    pub(crate) async fn author_own_store_stream(&self) -> tokio::sync::OwnedMutexGuard<()> {
221        self.connection.author_own_store_stream().await
222    }
223
224    pub(crate) async fn blob_upload_drain_permit(&self) -> crate::store::BlobUploadDrainPermit {
225        self.connection.blob_upload_drain_permit().await
226    }
227
228    pub(crate) async fn snapshot_publication_permit(
229        &self,
230    ) -> crate::store::SnapshotPublicationPermit {
231        self.connection.snapshot_publication_permit().await
232    }
233
234    pub(crate) async fn local_blob_cleanup_permit(&self) -> crate::store::LocalBlobCleanupPermit {
235        self.connection.local_blob_cleanup_permit().await
236    }
237
238    pub(crate) async fn apply_local_blob_cleanup_intent(
239        &self,
240        intent: &crate::local_blob_cleanup_intents::LocalBlobCleanupIntent,
241    ) -> Result<(), DbError> {
242        self.connection
243            .apply_local_blob_cleanup_intent(intent)
244            .await
245    }
246
247    pub(crate) async fn stage_host_write_blobs<E>(
248        &self,
249        blobs: Vec<crate::store::NewBlob>,
250    ) -> Result<crate::store::StagedBlobBatch, crate::HostWriteError<E>> {
251        self.connection.stage_host_write_blobs(blobs).await
252    }
253
254    pub(crate) async fn sync_store_parent_dir(
255        &self,
256        path: &Path,
257    ) -> Result<(), coven_foundation::atomic_file::FileError> {
258        self.connection.sync_store_parent_dir(path).await
259    }
260
261    #[cfg(any(test, feature = "test-utils"))]
262    pub(crate) async fn reach_store_test_point(&self, point: DatabaseTestPoint) {
263        self.connection.reach_store_test_point(point).await;
264    }
265
266    /// Open and own the connection at `path`.
267    ///
268    /// Runs the host migration ladder and validates its final sync-routing
269    /// contract in one transaction. A fresh database creates Coven metadata in
270    /// that transaction; an initialized database commits only when the final
271    /// contract exactly matches its pinned bytes. Then seeds the register clock
272    /// from on-disk rows. The `_updated_at` stamper remains inside the database
273    /// boundary and is used by every synced-row write.
274    pub fn open(
275        path: &Path,
276        synced_tables: Vec<SyncedTable>,
277        blob_tombstone_grace: chrono::Duration,
278        transfer_limits: coven_protocol::blob::TransferLimits,
279        device_id: String,
280        clock: coven_foundation::clock::ClockRef,
281        coven_migration_policy: CovenMigrationPolicy,
282        migrations: &[Migration],
283    ) -> Result<Database, OpenError> {
284        let hlc = Hlc::try_new(device_id, clock).map_err(|e| DbError::context("device_id", e))?;
285        Self::open_with_hlc_and_coven_metadata(
286            path,
287            synced_tables,
288            blob_tombstone_grace,
289            transfer_limits,
290            Arc::new(hlc),
291            coven_migration_policy,
292            migrations,
293            CovenMetadataOpen::Detect,
294        )
295    }
296
297    pub fn open_initialized_store(
298        path: &Path,
299        install: &VerifiedSnapshotBootstrapInstall,
300        synced_tables: Vec<SyncedTable>,
301        blob_tombstone_grace: chrono::Duration,
302        transfer_limits: coven_protocol::blob::TransferLimits,
303        device_id: String,
304        clock: coven_foundation::clock::ClockRef,
305        coven_migration_policy: CovenMigrationPolicy,
306        migrations: &[Migration],
307    ) -> Result<Database, OpenError> {
308        if ObjectHash::digest(&std::fs::read(path).map_err(DbError::from)?)
309            != install.snapshot.meta.image.image_hash
310        {
311            return Err(DbError::Message(
312                "snapshot database image differs from its authenticated plaintext hash".into(),
313            )
314            .into());
315        }
316        let hlc = Hlc::try_new(device_id, clock).map_err(|e| DbError::context("device_id", e))?;
317        Self::open_with_hlc_and_coven_metadata(
318            path,
319            synced_tables,
320            blob_tombstone_grace,
321            transfer_limits,
322            Arc::new(hlc),
323            coven_migration_policy,
324            migrations,
325            CovenMetadataOpen::VerifiedSnapshot(install),
326        )
327    }
328
329    fn open_with_hlc_and_coven_metadata(
330        path: &Path,
331        synced_tables: Vec<SyncedTable>,
332        blob_tombstone_grace: chrono::Duration,
333        transfer_limits: coven_protocol::blob::TransferLimits,
334        hlc: Arc<Hlc>,
335        coven_migration_policy: CovenMigrationPolicy,
336        migrations: &[Migration],
337        metadata_open: CovenMetadataOpen<'_>,
338    ) -> Result<Database, OpenError> {
339        let store_dir = store_dir_of(path);
340        Self::open_with_hlc_and_coven_metadata_in_store_dir(
341            path,
342            store_dir,
343            crate::connection_io::ConnectionDurability::Full,
344            synced_tables,
345            blob_tombstone_grace,
346            transfer_limits,
347            hlc,
348            coven_migration_policy,
349            migrations,
350            metadata_open,
351        )
352    }
353
354    fn open_with_hlc_and_coven_metadata_in_store_dir(
355        path: &Path,
356        store_dir: coven_foundation::store_dir::StoreDir,
357        connection_durability: crate::connection_io::ConnectionDurability,
358        synced_tables: Vec<SyncedTable>,
359        blob_tombstone_grace: chrono::Duration,
360        transfer_limits: coven_protocol::blob::TransferLimits,
361        hlc: Arc<Hlc>,
362        coven_migration_policy: CovenMigrationPolicy,
363        migrations: &[Migration],
364        metadata_open: CovenMetadataOpen<'_>,
365    ) -> Result<Database, OpenError> {
366        let core = DatabaseCore::open(
367            path,
368            store_dir,
369            connection_durability,
370            synced_tables,
371            blob_tombstone_grace,
372            transfer_limits,
373            hlc,
374            coven_migration_policy,
375            migrations,
376            metadata_open,
377        )?;
378
379        Self::from_core(core, "coven-db").map_err(OpenError::from)
380    }
381
382    #[cfg(any(test, feature = "test-utils"))]
383    pub fn open_in_store_dir_for_test(
384        path: &Path,
385        store_dir: coven_foundation::store_dir::StoreDir,
386        synced_tables: Vec<SyncedTable>,
387        blob_tombstone_grace: chrono::Duration,
388        transfer_limits: coven_protocol::blob::TransferLimits,
389        device_id: String,
390        clock: coven_foundation::clock::ClockRef,
391        coven_migration_policy: CovenMigrationPolicy,
392        migrations: &[Migration],
393    ) -> Result<Database, OpenError> {
394        let hlc = Hlc::try_new(device_id, clock).map_err(|e| DbError::context("device_id", e))?;
395        Self::open_with_hlc_in_store_dir_for_test(
396            path,
397            store_dir,
398            synced_tables,
399            blob_tombstone_grace,
400            transfer_limits,
401            Arc::new(hlc),
402            coven_migration_policy,
403            migrations,
404        )
405    }
406
407    #[cfg(any(test, feature = "test-utils"))]
408    pub fn open_with_hlc_in_store_dir_for_test(
409        path: &Path,
410        store_dir: coven_foundation::store_dir::StoreDir,
411        synced_tables: Vec<SyncedTable>,
412        blob_tombstone_grace: chrono::Duration,
413        transfer_limits: coven_protocol::blob::TransferLimits,
414        hlc: Arc<Hlc>,
415        coven_migration_policy: CovenMigrationPolicy,
416        migrations: &[Migration],
417    ) -> Result<Database, OpenError> {
418        Self::open_with_hlc_and_coven_metadata_in_store_dir(
419            path,
420            store_dir,
421            crate::connection_io::ConnectionDurability::Disabled,
422            synced_tables,
423            blob_tombstone_grace,
424            transfer_limits,
425            hlc,
426            coven_migration_policy,
427            migrations,
428            CovenMetadataOpen::Detect,
429        )
430    }
431
432    /// Open the store at `path` read-only for a same-store secondary reader
433    /// (e.g. a separate process reading while another holds the writer open).
434    ///
435    /// Distinct from [`Database::open`] in three ways, all so the reader never
436    /// mutates shared state a concurrent writer owns: the connection is
437    /// `SQLITE_OPEN_READONLY`; no migration ladder or bookkeeping DDL runs (it
438    /// opens against the schema the writer left, and refuses one newer than this
439    /// binary knows — the writer's `SchemaTooNew` policy); and it returns no
440    /// stamper, because a reader mints no `_updated_at`. Reads are safe across
441    /// processes because a read-only connection can coexist with the writer and
442    /// observes commits after each read transaction ends.
443    ///
444    /// The caller takes no store open-lock for a read-only open: the exclusive
445    /// advisory lock guards against a second *writer*, and a read-only connection
446    /// cannot write, so multiple readers can coexist with one writer.
447    pub fn open_read_only(
448        path: &Path,
449        synced_tables: Vec<SyncedTable>,
450        blob_tombstone_grace: chrono::Duration,
451        transfer_limits: coven_protocol::blob::TransferLimits,
452        device_id: String,
453        clock: coven_foundation::clock::ClockRef,
454        migrations: &[Migration],
455    ) -> Result<Database, OpenError> {
456        let hlc = Hlc::try_new(device_id, clock).map_err(|e| DbError::context("device_id", e))?;
457        let store_dir = store_dir_of(path);
458        let core = DatabaseCore::open_read_only(
459            path,
460            store_dir,
461            synced_tables,
462            blob_tombstone_grace,
463            transfer_limits,
464            Arc::new(hlc),
465            migrations,
466        )?;
467        Self::from_core(core, "coven-db-ro").map_err(OpenError::from)
468    }
469
470    #[cfg(any(test, feature = "test-utils"))]
471    #[doc(hidden)]
472    pub fn arm_test_pause(
473        &self,
474        point: DatabaseTestPoint,
475    ) -> (Arc<tokio::sync::Notify>, Arc<tokio::sync::Notify>) {
476        self.connection.arm_test_pause(point)
477    }
478
479    #[cfg(any(test, feature = "test-utils"))]
480    #[doc(hidden)]
481    pub fn observe_test_points(&self) -> tokio::sync::mpsc::UnboundedReceiver<DatabaseTestPoint> {
482        self.connection.observe_test_points()
483    }
484
485    #[cfg(any(test, feature = "test-utils"))]
486    #[doc(hidden)]
487    pub fn fail_next_merge_materialization_at(&self, point: MergeMaterializationFailurePoint) {
488        self.connection.fail_next_merge_materialization_at(point);
489    }
490
491    /// Open with a caller-supplied register clock instead of a fresh
492    /// system-wall-clock one. Lets a test inject an [`Hlc`] over a controlled
493    /// wall clock to exercise the skew/restart-seeding guarantees, sharing the
494    /// production open path (migration, seed, session) so the test drives the
495    /// real unit.
496    ///
497    #[cfg(any(test, feature = "test-utils"))]
498    pub fn open_with_hlc(
499        path: &Path,
500        synced_tables: Vec<SyncedTable>,
501        blob_tombstone_grace: chrono::Duration,
502        transfer_limits: coven_protocol::blob::TransferLimits,
503        hlc: Arc<Hlc>,
504        coven_migration_policy: CovenMigrationPolicy,
505        migrations: &[Migration],
506    ) -> Result<Database, OpenError> {
507        Self::open_with_hlc_and_coven_metadata(
508            path,
509            synced_tables,
510            blob_tombstone_grace,
511            transfer_limits,
512            hlc,
513            coven_migration_policy,
514            migrations,
515            CovenMetadataOpen::Detect,
516        )
517    }
518
519    #[cfg(any(test, feature = "test-utils"))]
520    pub fn schema_version(&self) -> u32 {
521        self.connection.store_schema_version()
522    }
523
524    #[cfg(any(test, feature = "test-utils"))]
525    pub fn synced_tables_for_test(&self) -> Vec<SyncedTable> {
526        self.connection.store_synced_tables()
527    }
528
529    #[cfg(any(test, feature = "test-utils"))]
530    pub async fn replace_with_database_image_for_test(
531        &self,
532        image: Vec<u8>,
533    ) -> Result<(), DbError> {
534        self.connection
535            .replace_with_database_image_for_test(image)
536            .await
537    }
538
539    #[cfg(any(test, feature = "test-utils"))]
540    pub async fn database_image_for_test(&self) -> Result<Vec<u8>, DbError> {
541        self.connection.database_image_for_test().await
542    }
543
544    #[cfg(any(test, feature = "test-utils"))]
545    pub fn sync_routing_hash(&self) -> ObjectHash {
546        self.connection.store_sync_routing_hash()
547    }
548
549    /// The receiver's current wall-clock millis, read from this database's
550    /// register clock. The pull reads it once and passes it down to bound an
551    /// incoming `_updated_at`'s physical component (a grossly-future stamp must not
552    /// win last-writer-wins or ratchet the clock).
553    #[cfg(any(test, feature = "test-utils"))]
554    pub fn receive_wall_ms(&self) -> u64 {
555        self.connection.store_receive_wall_ms()
556    }
557}
558
559#[cfg(test)]
560mod tests {
561    use super::*;
562
563    #[test]
564    fn a_relative_database_uses_the_working_directory_as_its_store_directory() {
565        assert_eq!(
566            store_dir_of(Path::new("store.sqlite")).as_ref(),
567            Path::new(".")
568        );
569    }
570}
coven_database/database_runtime.rs

coven_database/
database_runtime.rs
