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

coven/
coven.rs
