coven/
coven.rs

1//! Top-level API: open one handle and drive rows, blobs, sync, and
2//! membership through it.
3
4use std::num::NonZeroUsize;
5use std::path::PathBuf;
6use std::sync::Arc;
7
8use crate::handle::CovenHandle;
9use crate::store_sync::ConfigProvider;
10use crate::{Migration, MigrationError};
11use coven_database::store::StoreReads;
12use coven_database::{CovenMigrationPolicy, Database, DbError, OpenError};
13use coven_foundation::clock::{ClockRef, SystemClock};
14use coven_foundation::config::{Config, HomeStorage};
15use coven_foundation::store_dir::{LocalBlobStoreError, PathTokenError};
16use coven_foundation::store_dir::{StoreDir, StoreOpenGuard};
17use coven_keys::custody::KeyCustody;
18use coven_keys::identity_custody::IdentityCustody;
19use coven_keys::keys::StoreKeys;
20use coven_protocol::blob::BlobTransitionObserver;
21use coven_protocol::synced_schema::SyncedTable;
22
23pub type CovenResult<T> = Result<T, CovenError>;
24
25#[derive(Debug, thiserror::Error)]
26pub enum CovenError {
27    /// A host callback's error, retained with its concrete type and source chain.
28    #[error("host callback failed: {0}")]
29    Host(#[source] Box<dyn std::error::Error + Send + Sync + 'static>),
30    #[error("database error: {0}")]
31    Database(#[source] Box<DbError>),
32    #[error("migration error: {0}")]
33    Migration(MigrationError),
34    #[error("Coven schema migration error: {0}")]
35    CovenMigration(coven_database::CovenMigrationError),
36    #[error("snapshot preparation failed: {0}")]
37    SnapshotPreparation(#[source] Box<OpenError>),
38    #[error("sqlite error: {0}")]
39    Sqlite(#[from] rusqlite::Error),
40    #[error("file error: {0}")]
41    File(#[from] coven_foundation::atomic_file::FileError),
42    #[error("local blob {} has {actual_size} bytes, expected {expected_size}", path.display())]
43    LocalBlobSizeMismatch {
44        path: PathBuf,
45        expected_size: u64,
46        actual_size: u64,
47    },
48    #[error("unsafe blob path: {0}")]
49    UnsafeBlobPath(#[from] PathTokenError),
50    #[error("row routing key: {0}")]
51    RoutingEncryption(#[from] coven_keys::keys::RoutingEncryptionError),
52    #[error("store database path has no parent: {}", path.display())]
53    StorePathHasNoParent { path: PathBuf },
54    #[error("the write SQL closure panicked")]
55    WriteClosurePanicked,
56    #[error(
57        "write failed: {write}; failed to remove installed local blobs during rollback: {rollback}"
58    )]
59    WriteRollbackFailed {
60        #[source]
61        write: Box<CovenError>,
62        rollback: coven_database::BlobFileFailures,
63    },
64    #[error("write failed: {operation}; failed to remove unpublished local blobs: {cleanup}")]
65    BlobCleanupFailed {
66        #[source]
67        operation: Box<CovenError>,
68        cleanup: coven_database::BlobFileFailures,
69    },
70    #[error("synced_tables must be set before opening a coven store")]
71    MissingSyncedTables,
72    #[error("migrations must be set before opening a coven store")]
73    MissingMigrations,
74    #[error("coven_migration_policy must be set before opening a coven store for writing")]
75    MissingCovenMigrationPolicy,
76    #[error("candidate resolution failed: {0}")]
77    CandidateResolution(Box<coven_replication::sync::SyncError>),
78    #[error("blob declaration failed: {0}")]
79    BlobDeclaration(#[from] coven_database::BlobDeclError),
80    #[error("blob_tombstone_grace must be a positive duration")]
81    InvalidBlobTombstoneGrace,
82    #[error("browsable cloud storage cannot be used with scoped table {table:?}")]
83    BrowsableStorageWithScopedTable { table: String },
84    #[error("blob {namespace}/{id} is still referenced by a row after the write")]
85    BlobStillReferenced { namespace: String, id: String },
86    #[error("blob {namespace}/{id} is already referenced by a row")]
87    BlobAlreadyReferenced { namespace: String, id: String },
88    #[error("blob {namespace}/{id} is owned by an unpublished write")]
89    BlobOwnedByPendingWrite { namespace: String, id: String },
90    #[error("store is already open: {}", store_dir.display())]
91    AlreadyOpen { store_dir: PathBuf },
92    #[error("I/O error: {0}")]
93    Io(#[from] std::io::Error),
94    #[cfg(test)]
95    #[error("test failure: {0}")]
96    TestFailure(&'static str),
97}
98
99impl From<DbError> for CovenError {
100    fn from(error: DbError) -> Self {
101        Self::Database(Box::new(error))
102    }
103}
104
105impl From<OpenError> for CovenError {
106    fn from(value: OpenError) -> Self {
107        match value {
108            OpenError::CovenMigration(e) => CovenError::CovenMigration(e),
109            OpenError::Migration(e) => CovenError::Migration(e),
110            OpenError::Db(e) => CovenError::from(e),
111            error @ OpenError::PreparationCleanup { .. } => {
112                CovenError::SnapshotPreparation(Box::new(error))
113            }
114        }
115    }
116}
117
118impl From<LocalBlobStoreError> for CovenError {
119    fn from(value: LocalBlobStoreError) -> Self {
120        match value {
121            LocalBlobStoreError::Path(error) => CovenError::UnsafeBlobPath(error),
122            LocalBlobStoreError::File(error) => CovenError::File(error),
123            LocalBlobStoreError::SizeMismatch {
124                path,
125                expected_size,
126                actual_size,
127            } => CovenError::LocalBlobSizeMismatch {
128                path,
129                expected_size,
130                actual_size,
131            },
132        }
133    }
134}
135
136#[derive(Clone)]
137pub struct CovenConfig(ConfigProvider);
138
139impl CovenConfig {
140    fn current(&self) -> Config {
141        (self.0)()
142    }
143
144    fn provider(&self) -> ConfigProvider {
145        self.0.clone()
146    }
147}
148
149impl From<Config> for CovenConfig {
150    fn from(value: Config) -> Self {
151        let config = value;
152        Self(Arc::new(move || config.clone()))
153    }
154}
155
156impl<F> From<F> for CovenConfig
157where
158    F: Fn() -> Config + Send + Sync + 'static,
159{
160    fn from(value: F) -> Self {
161        Self(Arc::new(value))
162    }
163}
164
165pub struct Coven;
166
167impl Coven {
168    pub fn builder(store_dir: StoreDir, config: impl Into<CovenConfig>) -> CovenBuilder {
169        let config = config.into();
170        CovenBuilder {
171            store_dir,
172            config,
173            synced_tables: None,
174            migrations: None,
175            coven_migration_policy: None,
176            blob_tombstone_grace: coven_protocol::blob::BLOB_TOMBSTONE_GRACE,
177            max_concurrent_uploads: NonZeroUsize::MIN,
178            max_concurrent_downloads: NonZeroUsize::MIN,
179            clock: Arc::new(SystemClock),
180            key_custody: KeyCustody::Keyring,
181            identity_custody: IdentityCustody::Keyring,
182            oauth_clients: coven_storage::oauth::OAuthClients::empty(),
183            cloudkit_ops: None,
184            observer: None,
185        }
186    }
187
188    /// Remove the master key for a closed store that cannot be opened.
189    ///
190    /// An open store performs this through [`CovenHandle::forget_master_key`],
191    /// which also disconnects operations retaining the unlocked value. This
192    /// entry point exists for host deletion flows whose damaged local database
193    /// prevents constructing a handle at all; Coven still owns the keyring
194    /// account and slot selection.
195    pub fn forget_keyring_master_key(store_id: &str) -> Result<(), coven_keys::keys::KeyError> {
196        StoreKeys::bind(store_id.to_string()).delete_encryption_key()
197    }
198}
199
200pub struct CovenBuilder {
201    store_dir: StoreDir,
202    config: CovenConfig,
203    synced_tables: Option<Vec<SyncedTable>>,
204    migrations: Option<Vec<Migration>>,
205    coven_migration_policy: Option<CovenMigrationPolicy>,
206    blob_tombstone_grace: chrono::Duration,
207    max_concurrent_uploads: NonZeroUsize,
208    max_concurrent_downloads: NonZeroUsize,
209    clock: ClockRef,
210    key_custody: KeyCustody,
211    identity_custody: IdentityCustody,
212    oauth_clients: coven_storage::oauth::OAuthClients,
213    cloudkit_ops: Option<Arc<dyn coven_storage::cloud::cloudkit::CloudKitOps>>,
214    observer: Option<Arc<dyn BlobTransitionObserver>>,
215}
216
217impl From<coven_foundation::store_dir::StoreOpenGuardError> for CovenError {
218    fn from(error: coven_foundation::store_dir::StoreOpenGuardError) -> Self {
219        match error {
220            coven_foundation::store_dir::StoreOpenGuardError::AlreadyOpen { store_dir } => {
221                CovenError::AlreadyOpen { store_dir }
222            }
223            coven_foundation::store_dir::StoreOpenGuardError::NoParent { path } => {
224                CovenError::StorePathHasNoParent { path }
225            }
226            coven_foundation::store_dir::StoreOpenGuardError::File(error) => {
227                CovenError::File(error)
228            }
229        }
230    }
231}
232
233impl CovenBuilder {
234    pub fn synced_tables(mut self, tables: Vec<SyncedTable>) -> Self {
235        self.synced_tables = Some(tables);
236        self
237    }
238
239    /// How long a deleted blob is kept after its tombstone is written before the
240    /// tombstone GC erases it: the cross-device convergence window. Defaults to
241    /// [`coven_protocol::blob::BLOB_TOMBSTONE_GRACE`]. Must be positive — a
242    /// zero-or-negative grace is refused at [`open`](Self::open), since it would
243    /// let the GC erase a blob a lagging peer still references.
244    pub fn blob_tombstone_grace(mut self, grace: chrono::Duration) -> Self {
245        self.blob_tombstone_grace = grace;
246        self
247    }
248
249    /// How many blob uploads the sync cycle's upload drain runs at once. Defaults
250    /// to one (one at a time). A [`NonZeroUsize`] so a zero — which would leave the
251    /// drain admitting nothing and never completing — cannot be set.
252    pub fn max_concurrent_uploads(mut self, n: NonZeroUsize) -> Self {
253        self.max_concurrent_uploads = n;
254        self
255    }
256
257    /// How many blob downloads a [`pin`](CovenHandle::pin) call fetches at once.
258    /// Defaults to one (one at a time). A [`NonZeroUsize`] so a zero — which would
259    /// leave the pin loop admitting nothing and never completing — cannot be set.
260    pub fn max_concurrent_downloads(mut self, n: NonZeroUsize) -> Self {
261        self.max_concurrent_downloads = n;
262        self
263    }
264
265    /// The host's synced-schema migration ladder, applied over `PRAGMA
266    /// user_version` at open. The top version is the wire `schema_version` every
267    /// changeset is stamped with.
268    pub fn migrations(mut self, migrations: Vec<Migration>) -> Self {
269        self.migrations = Some(migrations);
270        self
271    }
272
273    /// Whether this writer may apply pending changes to Coven's own
274    /// bookkeeping schema while opening the store.
275    pub fn coven_migration_policy(mut self, policy: CovenMigrationPolicy) -> Self {
276        self.coven_migration_policy = Some(policy);
277        self
278    }
279
280    pub fn clock(mut self, clock: ClockRef) -> Self {
281        self.clock = clock;
282        self
283    }
284
285    /// How the store's master key is protected: the OS keyring (the
286    /// default), a passphrase-wrapped file, an in-memory session value, or a
287    /// host's own [`MasterKeyCustody`](crate::MasterKeyCustody) implementation.
288    /// coven builds every cipher internally from what this custody supplies —
289    /// the host never touches a crypto type.
290    pub fn key_custody(mut self, custody: KeyCustody) -> Self {
291        self.key_custody = custody;
292        self
293    }
294
295    /// How this store's device-signing identity is protected: the OS keyring
296    /// (the default), a passphrase-wrapped file, an in-memory session value,
297    /// or a host's own
298    /// [`DeviceIdentityCustody`](crate::DeviceIdentityCustody) implementation.
299    /// Selected next to [`key_custody`](Self::key_custody) — the identity is
300    /// scoped to this store, established as part of creating, joining, or
301    /// restoring it (see [`CovenHandle::initialize_identity`]).
302    pub fn identity_custody(mut self, custody: IdentityCustody) -> Self {
303        self.identity_custody = custody;
304        self
305    }
306
307    /// The OAuth applications this app uses for consumer cloud providers.
308    pub fn oauth_clients(mut self, clients: coven_storage::oauth::OAuthClients) -> Self {
309        self.oauth_clients = clients;
310        self
311    }
312
313    pub fn apply_cloudkit_ops(
314        mut self,
315        ops: Option<Arc<dyn coven_storage::cloud::cloudkit::CloudKitOps>>,
316    ) -> Self {
317        self.cloudkit_ops = ops;
318        self
319    }
320
321    pub fn observer(mut self, observer: Arc<dyn BlobTransitionObserver>) -> Self {
322        self.observer = Some(observer);
323        self
324    }
325
326    /// Open the store, returning the [`CovenHandle`].
327    ///
328    /// Opening performs no keyring interaction: it opens the database, runs
329    /// migrations, and resolves the master-key custody selection to a value
330    /// (constructing the trait object, never calling its `unlock`) — a locked
331    /// agent (no OS keyring session, no established master key or device
332    /// identity) can `open()` a store and use it fully for rows and Local
333    /// blobs. The first read of any key happens lazily, at the specific call
334    /// that needs it ([`CovenHandle::connect_sync`],
335    /// [`CovenHandle::cloud_home_key_state`], and similar).
336    pub fn open(self) -> CovenResult<CovenHandle> {
337        let config = self.config.current();
338        let tables = validated_synced_tables(&config, self.synced_tables)?;
339        let migrations = self.migrations.ok_or(CovenError::MissingMigrations)?;
340        let coven_migration_policy = self
341            .coven_migration_policy
342            .ok_or(CovenError::MissingCovenMigrationPolicy)?;
343        if self.blob_tombstone_grace <= chrono::Duration::zero() {
344            return Err(CovenError::InvalidBlobTombstoneGrace);
345        }
346        let store_dir = self.store_dir;
347        let db_path = store_dir.db_path();
348        let provider = self.config.provider();
349        let transfer_limits = coven_protocol::blob::TransferLimits {
350            uploads: self.max_concurrent_uploads,
351            downloads: self.max_concurrent_downloads,
352        };
353        let open_guard = Arc::new(StoreOpenGuard::acquire(&store_dir)?);
354        store_dir.remove_orphaned_write_temps(self.clock.now().into())?;
355        let db = Database::open(
356            &db_path,
357            tables.clone(),
358            self.blob_tombstone_grace,
359            transfer_limits,
360            config.device_id.clone(),
361            self.clock.clone(),
362            coven_migration_policy,
363            &migrations,
364        )?;
365        // Application reads get independent snapshots after the writer has
366        // completed schema validation. Opening every reader is part of open.
367        let read_db = StoreReads::open(&db_path)?;
368        let (key_service, key_custody, identity_custody) =
369            resolve_custody(&config, &store_dir, self.key_custody, self.identity_custody);
370        Ok(CovenHandle::new(
371            db,
372            read_db,
373            store_dir,
374            provider,
375            key_service,
376            key_custody,
377            identity_custody,
378            self.oauth_clients,
379            self.clock,
380            self.cloudkit_ops,
381            self.observer,
382            open_guard,
383            coven_storage::BlobChunking::DEFAULT,
384        ))
385    }
386
387    /// Open the store read-only for a same-store secondary reader: a separate
388    /// process (or a second handle) that must read rows and blobs while another
389    /// handle holds the full [`open`](Self::open). Returns a [`crate::CovenReadHandle`],
390    /// whose surface is reads only — SQL queries and blob reads — with no write,
391    /// sync, migration, or stamp API by construction.
392    ///
393    /// Unlike [`open`](Self::open) this takes no store lock (see
394    /// `StoreOpenGuard`): it succeeds while a writer holds the exclusive lock,
395    /// and any number of read-only opens coexist. It opens a `SQLITE_OPEN_READONLY`
396    /// connection against the schema on disk and runs no migration ladder. It
397    /// refuses pending changes to Coven's bookkeeping schema and host schemas
398    /// newer than this binary supports. It runs no orphan-temp cleanup either
399    /// (that is a write the lock-holding writer owns).
400    ///
401    /// SQLite's locking coordinates the read-only connection with the writer; a
402    /// blob read that misses locally fetches from the cloud into the per-device
403    /// cache (files written atomically), which is device scratch and touches no
404    /// synced state.
405    pub fn open_read_only(self) -> CovenResult<crate::read_handle::CovenReadHandle> {
406        let config = self.config.current();
407        let tables = validated_synced_tables(&config, self.synced_tables)?;
408        let migrations = self.migrations.ok_or(CovenError::MissingMigrations)?;
409        let store_dir = self.store_dir;
410        let db_path = store_dir.db_path();
411        let provider = self.config.provider();
412        // No StoreOpenGuard and no orphan-temp cleanup: both are writer concerns
413        // (see StoreOpenGuard). A reader must not take the exclusive lock the
414        // writer holds, nor write the filesystem the writer owns.
415        let db = Database::open_read_only(
416            &db_path,
417            tables,
418            self.blob_tombstone_grace,
419            coven_protocol::blob::TransferLimits {
420                uploads: self.max_concurrent_uploads,
421                downloads: self.max_concurrent_downloads,
422            },
423            config.device_id.clone(),
424            self.clock.clone(),
425            &migrations,
426        )?;
427        let (key_service, key_custody, identity_custody) =
428            resolve_custody(&config, &store_dir, self.key_custody, self.identity_custody);
429        let reads = StoreReads::open(&db_path)?;
430        Ok(crate::read_handle::CovenReadHandle::new(
431            db,
432            reads,
433            store_dir,
434            provider,
435            key_service,
436            key_custody,
437            identity_custody,
438            self.oauth_clients,
439            self.clock,
440            self.cloudkit_ops,
441            coven_storage::BlobChunking::DEFAULT,
442        ))
443    }
444}
445
446/// The host's synced tables, refused when the store's storage mode cannot carry
447/// them. Both kinds of open check this the same way, so a read-only open never
448/// accepts a schema the writer would refuse.
449fn validated_synced_tables(
450    config: &Config,
451    tables: Option<Vec<SyncedTable>>,
452) -> CovenResult<Vec<SyncedTable>> {
453    let tables = tables.ok_or(CovenError::MissingSyncedTables)?;
454    validate_storage_scope(config, &tables)?;
455    Ok(tables)
456}
457
458/// Bind this store's key service and resolve the host's custody selections
459/// against it. A store's keys are the same keys whether or not the handle over
460/// them can write, so both kinds of open resolve them identically.
461fn resolve_custody(
462    config: &Config,
463    store_dir: &StoreDir,
464    key_custody: KeyCustody,
465    identity_custody: IdentityCustody,
466) -> (
467    StoreKeys,
468    Arc<dyn coven_keys::keys::MasterKeyCustody>,
469    Arc<dyn coven_keys::keys::DeviceIdentityCustody>,
470) {
471    let key_service = StoreKeys::bind(config.store_id.clone());
472    let master = key_custody.resolve(&key_service, store_dir);
473    let identity = identity_custody.resolve(&key_service, store_dir);
474    (key_service, master, identity)
475}
476
477fn validate_storage_scope(config: &Config, tables: &[SyncedTable]) -> CovenResult<()> {
478    if config.cloud_home.storage == HomeStorage::Browsable {
479        if let Some(table) = tables
480            .iter()
481            .find(|table| table.audience_column().is_some())
482        {
483            return Err(CovenError::BrowsableStorageWithScopedTable {
484                table: table.name().to_string(),
485            });
486        }
487    }
488    Ok(())
489}
490
491#[cfg(test)]
492#[path = "coven_tests.rs"]
493mod tests;
494
495#[cfg(test)]
496mod error_size_tests {
497    #[test]
498    fn coven_error_fits_below_clippys_large_result_threshold() {
499        let size = std::mem::size_of::<super::CovenError>();
500        assert!(size <= 128, "CovenError occupies {size} bytes");
501    }
502}
coven/coven.rs

coven/
coven.rs
