coven_database/
lib.rs

1//! Owned SQLite writes and concurrent application reads.
2//!
3//! [`Database`] serializes writes, sync bookkeeping, changeset capture and
4//! apply on its owned connection. [`store::StoreReads`] owns a bounded pool of
5//! read-only connections for application queries, with one transaction per
6//! operation and separate bounded workers for processing owned query results.
7//!
8//! Hosts open coven with `Coven::builder` and run app SQL through
9//! `CovenHandle::write` or `CovenHandle::read`.
10
11pub(crate) use crate::blob_records::load_activated_registration_on;
12pub use crate::blob_records::remote_audience_to_db;
13pub(crate) use crate::circle_snapshot_records::load_outbound_circle_snapshot_on;
14pub(crate) use crate::circle_snapshot_records::load_published_circle_snapshot_on;
15pub use crate::cloud_outbox_records::{
16    outbox_identity, row_to_outbox_entry, CloudOutboxRecords, OutboxIdentity,
17};
18use crate::connection_io::scan_max_updated_at;
19use crate::connection_io::seed_from;
20pub(crate) use crate::local_state::{
21    delete_protocol_state_on, get_protocol_state_on, required_protocol_state_on,
22    set_protocol_state_on,
23};
24use crate::local_store_identity::pin_host_device_id_on;
25use crate::local_store_identity::validate_host_device_id_on;
26pub(crate) use crate::remote_object_records::begin_remote_candidate_nonactivation_on;
27pub(crate) use crate::remote_object_records::begin_remote_candidate_nonactivation_with_verified_head_on;
28pub use crate::remote_object_records::candidate_graph_exact_objects;
29pub(crate) use crate::remote_object_records::finish_remote_candidate_nonactivation_on;
30pub(crate) use crate::remote_object_records::index_retained_replay_owner_on;
31pub(crate) use crate::remote_object_records::load_protocol_inert_object_on;
32pub(crate) use crate::remote_object_records::load_remote_object_on;
33pub(crate) use crate::remote_object_records::mark_remote_object_uploaded_on;
34pub(crate) use crate::remote_object_records::mark_reusable_retained_authority_uploaded_on;
35pub(crate) use crate::remote_object_records::persist_exact_remote_object_on;
36pub(crate) use crate::remote_object_records::persist_prepared_remote_object_on;
37pub(crate) use crate::remote_object_records::record_reclaimed_store_package_on;
38pub(crate) use crate::remote_object_records::reopen_remote_object_on;
39pub(crate) use crate::remote_object_records::replace_prepared_merge_head_remote_on;
40pub(crate) use crate::remote_object_records::update_remote_object_on;
41pub(crate) use crate::remote_object_records::{
42    validate_prepared_blob_on, validate_prepared_package_on, validate_remote_object_on,
43};
44use crate::snapshot_objects::validate_snapshot_object_owners_on;
45pub(crate) use crate::snapshot_objects::{
46    install_snapshot_blob_plan_on, install_snapshot_blob_plans_on, persist_snapshot_image_on,
47    validate_snapshot_blob_plans_on,
48};
49pub use crate::snapshot_objects::{
50    snapshot_generation_as_i64, validate_snapshot_author, validate_snapshot_image,
51    verify_snapshot_blob_spools,
52};
53pub(crate) use crate::snapshot_records::load_outbound_store_snapshot_on;
54pub(crate) use crate::snapshot_records::load_published_store_snapshot_on;
55pub(crate) use crate::snapshot_records::load_published_store_snapshots_on;
56pub use crate::store_ack_records::store_snapshot_first_slot;
57pub(crate) use crate::store_ack_records::{
58    finish_outbound_store_ack_on, load_published_store_ack_on, verify_next_local_store_ack_on,
59};
60pub(crate) use crate::store_authority_records::install_store_founder_state_on;
61pub(crate) use crate::store_reclaim_records::{
62    clear_store_reclaim_operation_stuck_on, insert_store_reclaim_operation_on,
63    load_store_reclaim_operation_on, mark_store_reclaim_operation_stuck_on,
64    update_store_reclaim_operation_on,
65};
66pub use crate::store_reclaim_records::{
67    parse_store_reclaim_operation, store_reclaim_journal_error,
68};
69use std::collections::{BTreeMap, BTreeSet};
70use std::path::{Path, PathBuf};
71use std::sync::Arc;
72
73use coven_keys::encryption::EncryptionService;
74use coven_protocol::audience_package::{AudiencePackage, RowBlobLocatorBinding};
75use coven_protocol::blob::locator::{BlobLocator, RemoteAudience, StoredBlobRef};
76use coven_protocol::blob::{BlobRef, RowBlobAuthority, RowBlobRef};
77use coven_protocol::circle::Audience;
78use coven_protocol::hlc::{Hlc, Timestamp, HIGHWATER_STATE_KEY, MAX_FUTURE_SKEW_MS};
79use coven_protocol::membership::{
80    AuthorHead, MembershipEntry, MembershipEntryRef, MembershipHeadRef,
81};
82use coven_protocol::objects::{ExactObjectRef, PreparedExactObject};
83use coven_protocol::remote_object::{
84    remote_object_id, CandidateExclusiveObjectDomain, RemoteObjectRecord, RetainedReplayOwner,
85    SharedLiveSetObjectDomain,
86};
87use coven_protocol::store_commit::{
88    ack_slot_prefix, ObjectHash, ResolvedStoreDeviceState, SnapshotImageRef, SnapshotMeta,
89    StoreAck, StoreAckRef, StoreBatchCommit, StoreBatchCommitRef, StoreCommitCoord,
90    StoreDeviceHead, StoreDeviceRegistration, StoreDeviceRegistrationRef, StoreProtocolRoot,
91    StoreSnapshotRef,
92};
93use coven_protocol::synced_schema::SyncedTable;
94use coven_protocol::write::{WriteId, WriteStatus};
95use rusqlite::{Connection, OptionalExtension};
96
97pub use rusqlite;
98
99mod blob_bindings;
100pub(crate) use blob_bindings::{
101    install_pulled_merge_membership_activations_on, install_pulled_package_activation_on,
102};
103mod blob_declarations;
104mod blob_records;
105mod changeset;
106mod changeset_identity;
107mod circle_operation_records;
108mod cloud_outbox_records;
109mod connection_io;
110mod coven_migration;
111mod coven_schema;
112mod coven_schema_definitions;
113mod database_connection;
114pub(crate) use connection_io::capture_changeset;
115#[cfg(any(test, feature = "test-utils"))]
116pub(crate) use coven_migration::COVEN_SCHEMA_VERSION_STATE_KEY;
117pub(crate) use coven_migration::{
118    initialize_coven_schema_version, run_coven_migrations_in_transaction,
119    run_uninitialized_snapshot_coven_migrations_in_transaction, validate_coven_schema_for_reader,
120};
121pub use coven_migration::{CovenMigrationError, CovenMigrationPolicy};
122#[cfg(test)]
123pub(crate) use coven_schema::all_table_names;
124pub(crate) use coven_schema::{
125    apply_coven_routing_schema, apply_coven_schema, live_coven_schema_manifest, user_table_names,
126};
127pub use coven_schema::{
128    expected_coven_schema_manifest, is_reserved_table_name, CovenSchemaManifest,
129};
130mod circle_snapshot_records;
131mod database_open;
132mod database_runtime;
133mod database_session;
134mod external_blob_records;
135mod gate;
136mod live_query;
137mod local_state;
138mod local_store_identity;
139mod make_remote;
140mod migration;
141mod operation_models;
142mod prepared_audience_objects;
143mod prepared_external_blob;
144mod remote_object_records;
145mod routing_contract;
146mod schema_contract;
147mod schema_introspection;
148mod snapshot_objects;
149mod snapshot_records;
150pub mod store;
151mod store_ack_records;
152mod store_authority_records;
153mod store_coordinates;
154mod store_reclaim_records;
155#[cfg(any(test, feature = "test-utils"))]
156mod test_sql;
157#[cfg(any(test, feature = "test-utils"))]
158pub mod test_support;
159#[cfg(any(test, feature = "test-utils"))]
160mod test_transaction;
161#[cfg(any(test, feature = "test-utils"))]
162pub use coven_schema::DatabaseTestTable;
163#[cfg(any(test, feature = "test-utils"))]
164pub(crate) use test_sql::DatabaseTestSql;
165#[cfg(any(test, feature = "test-utils"))]
166pub use test_support::synthetic_store;
167#[cfg(any(test, feature = "test-utils"))]
168pub use test_support::{
169    DatabaseImageTest, OutboxAttempt, RetainedRegistrationTamper, ScopedRoutingStateForTest,
170};
171#[cfg(any(test, feature = "test-utils"))]
172pub(crate) use test_transaction::DatabaseTestTransaction;
173mod write_lifecycle;
174mod write_models;
175
176#[cfg(any(test, feature = "test-utils"))]
177pub use blob_declarations::{from_tables_call_count, reset_from_tables_call_count};
178pub use blob_declarations::{BlobDeclError, BlobDecls, PublicationBlob};
179pub(crate) use blob_records::{load_prepared_audience_objects_on, previous_row_blob_for_write_on};
180pub use changeset::{
181    value_ref_to_string, walk as walk_changeset, walk_old as walk_old_changeset, ChangesetError,
182};
183pub use changeset_identity::ChangesetIdentityError;
184pub(crate) use circle_operation_records::{
185    circle_operation_ids_in_phase_on, circle_operation_phase_json,
186};
187pub(crate) use circle_operation_records::{
188    circle_operation_uploaded_steps_on, load_circle_operation_on,
189};
190pub use circle_operation_records::{parse_circle_operation_row, PreparedCircleOperationRow};
191pub use coven_protocol::objects::{ExactProtocolObject, PreparedProtocolObject};
192pub(crate) use database_connection::{DatabaseConnection, DatabaseCore};
193use database_open::CovenMetadataOpen;
194pub use database_runtime::Database;
195pub use external_blob_records::ExternalBlob;
196use external_blob_records::ExternalBlobRecords;
197pub(crate) use gate::query_truth;
198pub(crate) use gate::{
199    active_circle_control, align_inbound_scoped_root_audiences, audience_moves,
200    capture_routing_changes, filter_inbound_circle_changeset, filter_inbound_store_rows,
201    live_row_audience, normalize_inbound_store_changeset, partition_outbound,
202    prune_ineligible_scoped_rows, prune_private_routes_without_rows, retain_snapshot_audience_rows,
203    validate_scoped_foreign_key_audiences, validate_snapshot_routing_state,
204};
205#[cfg(any(test, feature = "test-utils"))]
206pub use gate::{
207    from_tables_call_count as gate_from_tables_call_count,
208    reset_from_tables_call_count as reset_gate_from_tables_call_count,
209};
210pub use gate::{
211    is_routing_table, store_audience_transitions, AudienceMove, AudiencePartition,
212    CircleControlFailure, CirclePartitionControl, CirclePartitionControlError, GateError, Gates,
213    RoutingChanges, StoreAudienceTransitions,
214};
215pub use live_query::{CommittedChanges, QueryDependencies};
216pub(crate) use local_store_identity::local_activated_registration_ref_on;
217pub use migration::supported_version;
218pub(crate) use migration::{ensure_schema_supported, run_migrations_in_transaction};
219pub use migration::{Migration, MigrationContext, MigrationError, MigrationStep};
220pub use operation_models::{
221    DurableCircleSnapshotPublication, DurableDeviceRegistration, DurableMembershipMutation,
222    DurableSnapshotPublication, LocalDeviceRegistrationJournalRow, LocalDeviceRegistrationState,
223    MembershipMutationActivation, OwnerRecoveryPublication, PreparedLocalDeviceRegistrationRow,
224    PreparedSnapshotBlob, PublishedCircleSnapshot, PublishedStoreSnapshot,
225};
226pub use prepared_audience_objects::{
227    validate_prepared_audience_blob_graph, BlobActivation, MakeRemoteIntentState,
228    PreparedAudienceBlob, PreparedAudienceObjects, PreparedAudiencePackage, PreparedRemoteObject,
229    StoredBlobReferenceState,
230};
231pub use prepared_external_blob::{prepare_external_blob, PreparedExternalBlob};
232pub use routing_contract::SyncRoutingContract;
233pub use routing_contract::SyncRoutingContractError;
234use schema_contract::validate_host_synced_tables;
235pub use schema_contract::DurablePreparedProtocolObject;
236pub use schema_contract::{StoreBatchCompletion, StoreBatchLocalCleanup};
237
238#[derive(Debug, Clone, PartialEq, Eq)]
239pub enum MaterializationHold {
240    ForeignKeyDependency,
241    ConstraintConflict(Vec<String>),
242    PrivateSharedConflict {
243        table: String,
244        row_id: String,
245        commit: coven_protocol::store_commit::StoreBatchCommitRef,
246    },
247    InvalidLocalCircleContext {
248        circle_id: coven_protocol::circle::CircleId,
249    },
250}
251
252pub type MaterializationOutcome = coven_protocol::membership::ApplyOutcome<MaterializationHold>;
253
254pub(crate) use schema_introspection::{create_table_sql, foreign_key_edges, table_columns};
255pub use schema_introspection::{
256    quote_ident, rewrite_create_into_schema, CreateTableSchemaError, ForeignKeyEdge,
257    ForeignKeySchemaError,
258};
259pub use store::device_join_journal;
260pub use store::device_join_journal::DeviceJoinJournalError;
261pub(crate) use store::payload_store;
262pub use store::PayloadStoreError;
263pub use store::{
264    activated_merge_membership_remote_objects, DeviceJoinBootstrapActivation,
265    DeviceJoinBootstrapCommit, DeviceJoinBootstrapPlan, DeviceJoinBootstrapRowData,
266    MembershipAuthorityBytes, PreparedMergeMaterialization, PreparedMergeMaterializationPackage,
267    ResolvedDeviceJoinBootstrap, VerifiedAcknowledgedStoreSnapshot,
268    VerifiedReplayBaselineRetirementProof, VerifiedStoreSnapshotAuthority,
269};
270pub use store::{
271    audience_moves_by_row, local_blob_cleanup_intents, AudienceBlobMoveStaging, PostUpload,
272    StagedAudienceBlobRollback,
273};
274pub(crate) use store::{
275    copy_table_with_conflicts, install_circle_bootstrap_image_on,
276    install_circle_bootstrap_remote_objects_on,
277};
278pub use store::{
279    projection_table_names, AdvancedReplayBaseline, BlobTransitionRoot, BlobUploadDrainPermit,
280    BlockedWriteDiscard, CandidateCleanupObject, CircleAckPublicationInput, CreatedSnapshot,
281    DeviceJoinJournalStore, DurableStoreReclaimObject, DurableStoreReclaimOperation,
282    HostWriteBlobTransaction, HostWriteError, HostWriteOperation, IncomingTimestampPolicy,
283    InstalledReplayBaseline, LocalBlobCleanup, MakeRemoteAdmission, MaterializedLocalBlob,
284    MergeCandidateAbandonmentPreparation, ObservedStorePublication, OutboxEntry, OutboxFailure,
285    OutboxFailureKind, OutboxOperation, OutboxUploadState, OwnStreamAuthorship,
286    OwnedVerifiedMergeMaterialization, PreparedCircleObjects, ReclaimCommitActivation,
287    ReclaimedStorePackage, RetainedAudiencePackage, RetainedMergeHistoryCheckpoint,
288    RetainedMergeMaterializationKey, RetainedPackageApplication, RetainedReplayAuthority,
289    RetainedReplayBaseline, RetainedReplayGenesisAuthority, SnapshotBlobAudience, SnapshotBlobFact,
290    SnapshotDatabaseImage, SnapshotImageError, SnapshotImageOperationError,
291    SnapshotPublicationPermit, StoreDatabase, StoreReclaimJournalError, StoreRowWrites,
292    StoreWritePreparation, StuckReclaimOperation, TableSchema, ValidatedChangeset,
293    VerifiedMergeMaterialization, VerifiedMergeMembershipObjects, WinningRow, GENERATION_ZERO,
294};
295#[cfg(any(test, feature = "test-utils"))]
296pub use store::{resolve_and_apply_changeset, ApplyResult};
297#[cfg(any(test, feature = "test-utils"))]
298pub use store::{select_author_exclusion_activation_locator, AuthorExclusionLocatorTamper};
299pub use store::{BlobFileFailure, BlobFileFailures, SqlContext, SqlReadContext, WriteBatch};
300pub use store::{
301    CloudOutboxSnapshot, MakeRemoteProgress, QueuedDelete, QueuedMakeRemote, QueuedUpload,
302    QueuedUploadPhase,
303};
304pub use store_authority_records::DurableFounderMembershipJournal;
305pub(crate) use store_authority_records::{
306    founder_graph_identity, install_store_root_authority_on, load_local_store_founder_graph_on,
307    load_store_root_authority_on,
308};
309pub use store_authority_records::{
310    DurableFounderGraph, DurableFounderMembership, FounderMembershipRefs, StoreOwnerAnchor,
311};
312pub use write_models::{
313    ActivatedStoreAck, AuthorExclusionActivationLocator, BlockedMergeCandidate,
314    CompletePreparedStoreWriteOutcome, InitialStoreMembershipAuthority, MergeAbandonmentState,
315    OutboundStoreAck, OutboundStoreAckActivation, PreparedMergeAbandonmentCandidates,
316    PreparedStoreWrite, PreparedStoreWriteCommit, PreparedStoreWritePartitions, PublishedStoreAck,
317    StoreWriteBase, StoreWriteBlobFact, StoreWriteBlobFacts, StoreWriteBlobMoveDestination,
318    StoreWriteRemoteBlob, StoreWriteRouting, TerminalCandidateAuthority,
319    TerminalCandidateCleanupVerification,
320};
321pub(crate) use write_models::{
322    MergeReplayWrite, MergeReplayWriteEffect, ReplayJournal, SettledStoreWrite, SettledWriteFold,
323};
324
325pub const LOCAL_DEVICE_ID_STATE_KEY: &str = "local_device_id";
326const HOST_DEVICE_ID_STATE_KEY: &str = "host_device_id";
327pub const SYNC_ROUTING_CONTRACT_STATE_KEY: &str = "sync_routing_contract";
328pub const SYNC_ROUTING_HASH_STATE_KEY: &str = "sync_routing_hash";
329pub const COVEN_SCHEMA_MANIFEST_STATE_KEY: &str = "coven_schema_manifest";
330pub const COVEN_INITIALIZED_STATE_KEY: &str = "coven_initialized";
331pub const COVEN_INITIALIZED_STATE_VALUE: &str = "1";
332pub const STORE_DEVICE_GENESIS_STATE_KEY: &str = "store_device_genesis_state";
333const GATE_BASELINE_SCHEMA: &str = "coven_gate_empty";
334const COVEN_CLEANUP_GUARD_PREFIX: &str = "coven_cleanup_guard_";
335
336fn is_coven_cleanup_guard_name(name: &str) -> bool {
337    name.get(..COVEN_CLEANUP_GUARD_PREFIX.len())
338        .is_some_and(|prefix| prefix.eq_ignore_ascii_case(COVEN_CLEANUP_GUARD_PREFIX))
339}
340
341thread_local! {
342    /// How many Coven-owned write operations are on the current call stack.
343    ///
344    /// The host-SQL authorizer denies statements that access Coven's reserved
345    /// tables, but Coven's own entry points are documented to run inside the
346    /// host's write closure (`register_external_blob`, `enqueue_blob_delete`,
347    /// `clear_external_blob` all bind to the row version the same write
348    /// produced). Those operations announce themselves through this depth so
349    /// the authorizer can tell "Coven writing its own bookkeeping" apart from
350    /// "host SQL reaching into it" — the statement text is identical; the
351    /// caller is not. Thread-local is sound because a write closure and every
352    /// statement it executes run synchronously on one thread.
353    static COVEN_SQL_AUTHORITY_DEPTH: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
354    static HOST_SQL_WRITE_SEEN: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
355}
356
357pub(crate) fn reset_host_sql_write_observation() {
358    HOST_SQL_WRITE_SEEN.with(|seen| seen.set(false));
359}
360
361pub(crate) fn host_sql_write_was_observed() -> bool {
362    HOST_SQL_WRITE_SEEN.with(std::cell::Cell::get)
363}
364
365pub(crate) fn observe_host_sql_write() {
366    HOST_SQL_WRITE_SEEN.with(|seen| seen.set(true));
367}
368
369/// Run `f` with Coven's own SQL authority, so the host-SQL authorizer permits
370/// the reserved-table writes it performs. Panic-safe: the depth restores when
371/// the guard drops.
372pub(crate) fn with_coven_sql_authority<R>(f: impl FnOnce() -> R) -> R {
373    struct DepthGuard;
374    impl Drop for DepthGuard {
375        fn drop(&mut self) {
376            COVEN_SQL_AUTHORITY_DEPTH.with(|depth| depth.set(depth.get() - 1));
377        }
378    }
379    COVEN_SQL_AUTHORITY_DEPTH.with(|depth| depth.set(depth.get() + 1));
380    let _guard = DepthGuard;
381    f()
382}
383
384pub(crate) fn authorize_host_sql(
385    context: rusqlite::hooks::AuthContext<'_>,
386) -> rusqlite::hooks::Authorization {
387    use rusqlite::hooks::{AuthAction, Authorization};
388
389    let coven_owned = COVEN_SQL_AUTHORITY_DEPTH.with(|depth| depth.get()) > 0;
390    if !coven_owned
391        && matches!(
392            context.action,
393            AuthAction::Delete { .. } | AuthAction::Insert { .. } | AuthAction::Update { .. }
394        )
395    {
396        HOST_SQL_WRITE_SEEN.with(|seen| seen.set(true));
397    }
398    if coven_owned {
399        return Authorization::Allow;
400    }
401
402    let runs_from_coven_cleanup_guard = context.accessor.is_some_and(is_coven_cleanup_guard_name);
403    let mut accesses_coven_table = match context.action {
404        AuthAction::Delete { table_name }
405        | AuthAction::Insert { table_name }
406        | AuthAction::CreateTable { table_name }
407        | AuthAction::DropTable { table_name }
408        | AuthAction::CreateVtable { table_name, .. }
409        | AuthAction::DropVtable { table_name, .. } => is_reserved_table_name(table_name),
410        AuthAction::Update { table_name, .. }
411        | AuthAction::Read { table_name, .. }
412        | AuthAction::CreateIndex { table_name, .. }
413        | AuthAction::DropIndex { table_name, .. }
414        | AuthAction::CreateTrigger { table_name, .. }
415        | AuthAction::DropTrigger { table_name, .. }
416        | AuthAction::AlterTable { table_name, .. } => is_reserved_table_name(table_name),
417        _ => false,
418    };
419    if runs_from_coven_cleanup_guard {
420        accesses_coven_table = false;
421    }
422    let changes_coven_cleanup_guard = match context.action {
423        AuthAction::CreateTempTrigger { trigger_name, .. }
424        | AuthAction::CreateTrigger { trigger_name, .. }
425        | AuthAction::DropTempTrigger { trigger_name, .. }
426        | AuthAction::DropTrigger { trigger_name, .. } => is_coven_cleanup_guard_name(trigger_name),
427        _ => false,
428    };
429    if accesses_coven_table
430        || changes_coven_cleanup_guard
431        || matches!(
432            context.action,
433            AuthAction::Transaction { .. } | AuthAction::Savepoint { .. }
434        )
435        || context
436            .database_name
437            .is_some_and(|name| name.eq_ignore_ascii_case(GATE_BASELINE_SCHEMA))
438        || matches!(
439            context.action,
440            AuthAction::Detach { database_name }
441                if database_name.eq_ignore_ascii_case(GATE_BASELINE_SCHEMA)
442        )
443        || matches!(
444            context.action,
445            AuthAction::Pragma { pragma_name, .. }
446                if pragma_name.eq_ignore_ascii_case("database_list")
447        )
448    {
449        Authorization::Deny
450    } else {
451        Authorization::Allow
452    }
453}
454
455/// A staged audience-move blob file that could not be rolled back, and why.
456/// Names the file so a host learns which staged bytes are left on disk.
457#[derive(Debug)]
458pub struct StagedBlobRollbackFailure {
459    pub path: PathBuf,
460    pub reason: StagedBlobRollbackReason,
461}
462
463#[derive(Debug, thiserror::Error)]
464pub enum StagedBlobRollbackReason {
465    #[error("staged audience blob disappeared before rollback")]
466    Missing,
467    #[error("{0}")]
468    File(#[from] coven_foundation::atomic_file::FileError),
469}
470
471impl std::fmt::Display for StagedBlobRollbackFailure {
472    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
473        write!(formatter, "{}: {}", self.path.display(), self.reason)
474    }
475}
476
477/// Every staged file that could not be rolled back, in the order attempted.
478#[derive(Debug)]
479pub struct StagedBlobRollbackFailures(pub Vec<StagedBlobRollbackFailure>);
480
481impl std::fmt::Display for StagedBlobRollbackFailures {
482    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
483        for (index, failure) in self.0.iter().enumerate() {
484            if index > 0 {
485                formatter.write_str("; ")?;
486            }
487            write!(formatter, "{failure}")?;
488        }
489        Ok(())
490    }
491}
492
493/// An error from the owned database.
494#[derive(Debug, thiserror::Error)]
495pub enum DbError {
496    #[error("database error: {0}")]
497    Message(String),
498    #[error("Store writes depend on state being removed: {writes:?}")]
499    WriteDependencyConflict { writes: Vec<WriteId> },
500    #[error("replay retirement cut is not a canonical application prefix")]
501    ReplayRetirementCutNotPrefix,
502    #[error(
503        "write callback prepared no INSERT, UPDATE, or DELETE statement; pure reads belong on read"
504    )]
505    ReadOnlyWriteTransaction,
506    #[error("{0}")]
507    Sqlite(#[from] rusqlite::Error),
508    #[error("failed to construct the expected Coven schema manifest: {0}")]
509    ExpectedSchema(#[source] &'static rusqlite::Error),
510    /// A JSON column's bytes did not read back as the value they encode, or a
511    /// value would not encode. Every synced-protocol column is stored as JSON,
512    /// so this is the shape of every column-level decode failure.
513    #[error("{0}")]
514    Serde(#[from] serde_json::Error),
515    /// Durable bytes failed the Store protocol's own validation — a hash that
516    /// does not match, a signature that does not verify, an object in the wrong
517    /// slot. The database read them back intact; the protocol refused them.
518    #[error("{0}")]
519    Protocol(#[source] Box<coven_protocol::store_commit::StoreProtocolError>),
520    #[error("{0}")]
521    RemoteObject(#[source] Box<coven_protocol::remote_object::RemoteObjectRecordError>),
522    #[error("{0}")]
523    AudiencePackage(#[source] Box<coven_protocol::audience_package::AudiencePackageError>),
524    #[error("{0}")]
525    ObjectHash(#[from] coven_foundation::object_hash::InvalidObjectHash),
526    #[error("{0}")]
527    BlobLocator(#[from] coven_protocol::blob::locator::BlobLocatorError),
528    #[error("{0}")]
529    CircleId(#[from] coven_protocol::circle::CircleIdError),
530    #[error("{0}")]
531    RowRoutingKey(#[from] coven_protocol::circle::RowRoutingKeyError),
532    #[error("{0}")]
533    CirclePartitionControl(#[from] crate::CirclePartitionControlError),
534    #[error("{0}")]
535    Gate(#[from] crate::gate::GateError),
536    #[error("{0}")]
537    Storage(#[from] coven_protocol::objects::StorageError),
538    #[error("unsafe blob path: {0}")]
539    BlobPath(#[from] coven_foundation::store_dir::PathTokenError),
540    #[error("{0}")]
541    Io(#[from] std::io::Error),
542    #[error("{0}")]
543    File(#[from] coven_foundation::atomic_file::FileError),
544    #[error("{0}")]
545    LocalBlobRemoval(#[from] coven_foundation::store_dir::LocalBlobRemovalError),
546    #[error("{0}")]
547    CachedLocatorRemoval(#[from] coven_foundation::store_dir::CachedLocatorRemovalError),
548    /// A stored integer column did not fit the type the schema says it holds.
549    #[error("stored value is out of range: {0}")]
550    IntRange(#[from] std::num::TryFromIntError),
551    #[error("stored value is not an integer: {0}")]
552    ParseInt(#[from] std::num::ParseIntError),
553    #[error("stored value is not UTF-8: {0}")]
554    Utf8(#[from] std::string::FromUtf8Error),
555    #[error("{0}")]
556    Utf8Slice(#[from] std::str::Utf8Error),
557    #[error("{0}")]
558    BlobDecl(#[from] crate::BlobDeclError),
559    #[error("{0}")]
560    ChangesetIdentity(#[from] crate::ChangesetIdentityError),
561    #[error("{0}")]
562    Changeset(#[from] crate::ChangesetError),
563    #[error("{0}")]
564    AuthorStreamId(#[from] coven_protocol::causal_grants::AuthorStreamIdParseError),
565    #[error("{0}")]
566    RowBlobRef(#[from] coven_protocol::blob::RowBlobRefError),
567    #[error("{0}")]
568    WriteRetraction(#[source] Box<coven_protocol::write::WriteRetractionError>),
569    #[error("{0}")]
570    RotationGate(#[from] coven_protocol::objects::RotationGateError),
571    #[error("{0}")]
572    Encryption(#[from] coven_keys::encryption::EncryptionError),
573    #[error("{0}")]
574    SnapshotImage(#[source] Box<crate::store::SnapshotImageError>),
575    #[error("{0}")]
576    PayloadStore(#[source] Box<PayloadStoreError>),
577    #[error("{operation}; payload cleanup failed: {cleanup}")]
578    PayloadCleanupFailed {
579        operation: Box<DbError>,
580        cleanup: Box<DbError>,
581    },
582    #[error("{operation}; committed-change capture failed: {capture}")]
583    ChangeCaptureFailed {
584        operation: Box<DbError>,
585        capture: Box<DbError>,
586    },
587    #[error("{0}")]
588    FromSql(#[from] rusqlite::types::FromSqlError),
589    #[error("{0}")]
590    BlobOpeningAuthority(#[from] coven_protocol::blob::BlobOpeningAuthorityError),
591    #[error("{0}")]
592    OwnerPromotionJournal(
593        #[source] Box<coven_protocol::owner_promotion_journal::OwnerPromotionJournalError>,
594    ),
595    #[error("{0}")]
596    CircleJournal(#[source] Box<coven_protocol::circle_journal::CircleJournalError>),
597    #[error("{0}")]
598    SyncRoutingContract(#[from] SyncRoutingContractError),
599    #[error("{0}")]
600    RowIdentity(#[from] coven_protocol::synced_schema::RowIdentityError),
601    #[error("{0}")]
602    PreparedCommit(#[source] Box<coven_protocol::prepared_commit::PreparedCommitError>),
603    #[error("{0}")]
604    CircleState(#[source] Box<coven_protocol::circle_activation::CircleStateError>),
605    #[error("{0}")]
606    DeviceExclusionJournal(
607        #[source] Box<coven_protocol::device_exclusion_journal::StoreDeviceExclusionJournalError>,
608    ),
609    #[error("{0}")]
610    StoreReclaimJournal(#[source] Box<StoreReclaimJournalError>),
611    #[error("{0}")]
612    CommitNewFile(#[from] coven_foundation::local_file::CommitNewFileError),
613    /// Staging a write's audience-move blobs failed. The implementation of
614    /// `AudienceBlobMoveStaging` is injected from above, so its failure is
615    /// carried as an opaque source rather than named here.
616    #[error("audience blob staging: {0}")]
617    AudienceBlobStaging(#[source] Box<dyn std::error::Error + Send + Sync + 'static>),
618    /// Staging a write's audience-move blobs failed AND rolling the staged
619    /// files back failed, so those files are left on disk. Carries both
620    /// failures rather than reporting one and describing the other.
621    #[error("{operation}; audience blob rollback failed: {rollback}")]
622    AudienceBlobRollbackFailed {
623        operation: Box<DbError>,
624        rollback: StagedBlobRollbackFailures,
625    },
626    #[error("staged audience blob rollback failed: {0}")]
627    StagedBlobRollback(StagedBlobRollbackFailures),
628    /// An audience move needs its blob materialized locally and it is not —
629    /// the row's bytes are absent, stale, or refuse their declared identity.
630    /// Names the row so the caller can act on it.
631    #[error(
632        "blob move requires materialization for {table}/{row_id}/{column} at {row_stamp}: {reason}"
633    )]
634    BlobMoveRequiresMaterialization {
635        table: String,
636        row_id: String,
637        column: String,
638        row_stamp: String,
639        reason: Box<DbError>,
640    },
641    /// A queued outbox entry's `last_attempt_at` is not an RFC 3339 timestamp,
642    /// so whether the entry is still inside its retry backoff cannot be
643    /// decided. Names the entry so the caller can act on that row.
644    #[error("outbox entry {entry_id} has unparseable last_attempt_at {value:?}: {source}")]
645    UnparseableOutboxAttemptTime {
646        entry_id: i64,
647        value: String,
648        source: chrono::ParseError,
649    },
650    /// A [`DbError`] with the operation that produced it named in front of it.
651    /// Carries the cause as a [`DbError`] so callers keep matching on it after
652    /// it crosses the layer that added the description.
653    #[error("{context}: {source}")]
654    Context {
655        context: String,
656        source: Box<DbError>,
657    },
658    #[error("database error: Store protocol root hash is absent")]
659    StoreRootHashMissing,
660    /// The local device was excluded from a Circle epoch close and has not yet
661    /// reset its projection from the successor bootstrap, so it cannot publish
662    /// into the Circle. Stays matchable at the publication boundary rather than
663    /// flattening into a message.
664    #[error(
665        "device excluded from circle {circle_id} close {close_id} must reset before publishing"
666    )]
667    ExcludedDeviceMustReset {
668        circle_id: coven_protocol::circle::CircleId,
669        close_id: coven_protocol::circle::CircleEpochCloseId,
670    },
671}
672
673impl DbError {
674    /// Name the operation `source` failed in without flattening it: the cause
675    /// stays a [`DbError`] the caller can still match on.
676    pub fn context(context: impl Into<String>, source: impl Into<DbError>) -> DbError {
677        DbError::Context {
678            context: context.into(),
679            source: Box::new(source.into()),
680        }
681    }
682}
683
684macro_rules! boxed_db_error_from {
685    ($source:path, $variant:ident) => {
686        impl From<$source> for DbError {
687            fn from(source: $source) -> Self {
688                Self::$variant(Box::new(source))
689            }
690        }
691    };
692}
693
694boxed_db_error_from!(
695    coven_protocol::owner_promotion_journal::OwnerPromotionJournalError,
696    OwnerPromotionJournal
697);
698boxed_db_error_from!(
699    coven_protocol::device_exclusion_journal::StoreDeviceExclusionJournalError,
700    DeviceExclusionJournal
701);
702boxed_db_error_from!(StoreReclaimJournalError, StoreReclaimJournal);
703boxed_db_error_from!(
704    coven_protocol::remote_object::RemoteObjectRecordError,
705    RemoteObject
706);
707boxed_db_error_from!(coven_protocol::write::WriteRetractionError, WriteRetraction);
708boxed_db_error_from!(crate::store::SnapshotImageError, SnapshotImage);
709boxed_db_error_from!(coven_protocol::store_commit::StoreProtocolError, Protocol);
710boxed_db_error_from!(
711    coven_protocol::audience_package::AudiencePackageError,
712    AudiencePackage
713);
714boxed_db_error_from!(PayloadStoreError, PayloadStore);
715boxed_db_error_from!(
716    coven_protocol::circle_activation::CircleStateError,
717    CircleState
718);
719boxed_db_error_from!(
720    coven_protocol::circle_journal::CircleJournalError,
721    CircleJournal
722);
723boxed_db_error_from!(
724    coven_protocol::prepared_commit::PreparedCommitError,
725    PreparedCommit
726);
727
728#[cfg(test)]
729mod db_error_tests;
730
731/// Run `sql`, map every row through `mapper`, and collect the results.
732///
733/// It returns the SQLite failure as it happened, so every caller's `?` converts
734/// it into whatever error that caller already returns — one helper, no error
735/// vocabulary of its own.
736pub(crate) fn query_mapped_rows<T, P, F>(
737    conn: &Connection,
738    sql: &str,
739    params: P,
740    mut mapper: F,
741) -> Result<Vec<T>, rusqlite::Error>
742where
743    P: rusqlite::Params,
744    F: FnMut(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
745{
746    let mut statement = conn.prepare_cached(sql)?;
747    let rows = statement.query_map(params, |row| mapper(row))?;
748    let mut mapped = Vec::new();
749    for row in rows {
750        mapped.push(row?);
751    }
752    Ok(mapped)
753}
754
755/// Why opening the database failed. Splits a migration-ladder failure from every
756/// other open-time database error so the [`MigrationError`] a host acts on —
757/// [`MigrationError::SchemaTooNew`], whose remedy is "update the app" — stays
758/// matchable at the open boundary instead of being flattened into a
759/// [`DbError`] string.
760#[derive(Debug, thiserror::Error)]
761pub enum OpenError {
762    #[error(transparent)]
763    CovenMigration(#[from] CovenMigrationError),
764    #[error(transparent)]
765    Migration(#[from] MigrationError),
766    #[error(transparent)]
767    Db(#[from] DbError),
768}
769
770/// Test-only checkpoints reached by database operations whose ordering matters.
771#[cfg(any(test, feature = "test-utils"))]
772#[doc(hidden)]
773#[derive(Clone, Debug, PartialEq, Eq)]
774pub enum DatabaseTestPoint {
775    LocalBlobCleanupRequested,
776    LocalBlobCleanupAcquired,
777    LocalBlobCleanupBeforeFilesystem {
778        namespace: String,
779        blob_id: String,
780    },
781    LocalBlobCleanupFinished,
782    PullAfterRemoteCommit {
783        device_id: String,
784        seq: u64,
785    },
786    StoreWriteCommitUploaded {
787        write_id: WriteId,
788    },
789    StoreWriteHeadReadBack {
790        write_id: WriteId,
791    },
792    StoreDeviceExclusionCandidateStaged,
793    /// The owner's device-join acceptance has read the position its attempt
794    /// will be bound to and holds the turn to author it, but has not yet
795    /// published the head that takes it.
796    DeviceJoinAttemptPositionHeld,
797}
798
799#[cfg(any(test, feature = "test-utils"))]
800#[doc(hidden)]
801#[derive(Clone, Copy, Debug, PartialEq, Eq)]
802pub enum MergeMaterializationFailurePoint {
803    SummaryMaterialization,
804    RetractionDeletion,
805    ProjectionReplacement,
806}
807
808#[cfg(any(test, feature = "test-utils"))]
809struct ArmedTestPause<K> {
810    point: K,
811    reached: Arc<tokio::sync::Notify>,
812    resume: Arc<tokio::sync::Notify>,
813}
814
815#[cfg(any(test, feature = "test-utils"))]
816struct TestPauseState<K> {
817    armed: Option<ArmedTestPause<K>>,
818    observers: Vec<tokio::sync::mpsc::UnboundedSender<K>>,
819}
820
821#[cfg(any(test, feature = "test-utils"))]
822struct TestPausePoints<K> {
823    state: std::sync::Mutex<TestPauseState<K>>,
824}
825
826#[cfg(any(test, feature = "test-utils"))]
827impl<K> Default for TestPausePoints<K> {
828    fn default() -> Self {
829        Self {
830            state: std::sync::Mutex::new(TestPauseState {
831                armed: None,
832                observers: Vec::new(),
833            }),
834        }
835    }
836}
837
838#[cfg(any(test, feature = "test-utils"))]
839impl<K: Clone + PartialEq> TestPausePoints<K> {
840    fn arm(&self, point: K) -> (Arc<tokio::sync::Notify>, Arc<tokio::sync::Notify>) {
841        let reached = Arc::new(tokio::sync::Notify::new());
842        let resume = Arc::new(tokio::sync::Notify::new());
843        let prior = self
844            .state
845            .lock()
846            .expect("database test pause mutex poisoned")
847            .armed
848            .replace(ArmedTestPause {
849                point,
850                reached: reached.clone(),
851                resume: resume.clone(),
852            });
853        assert!(prior.is_none(), "database test pause already armed");
854        (reached, resume)
855    }
856
857    fn observe(&self) -> tokio::sync::mpsc::UnboundedReceiver<K> {
858        let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
859        self.state
860            .lock()
861            .expect("database test pause mutex poisoned")
862            .observers
863            .push(sender);
864        receiver
865    }
866
867    async fn reach(&self, point: K) {
868        let pause = {
869            let mut state = self
870                .state
871                .lock()
872                .expect("database test pause mutex poisoned");
873            state
874                .observers
875                .retain(|observer| observer.send(point.clone()).is_ok());
876            if state
877                .armed
878                .as_ref()
879                .is_some_and(|pause| pause.point == point)
880            {
881                state.armed.take()
882            } else {
883                None
884            }
885        };
886        if let Some(pause) = pause {
887            pause.reached.notify_one();
888            pause.resume.notified().await;
889        }
890    }
891}
892
893/// One Circle image selected against the restoring identity's re-resolved
894/// access. Coverage references imported from the Store snapshot are removed as
895/// a set before these locally verified images are installed.
896pub struct StagedCircleInstall {
897    pub activation_commit: StoreBatchCommitRef,
898    pub image: coven_protocol::circle_activation::VerifiedCircleImage,
899}
900
901enum CircleRestoreSelection {
902    Pending,
903    Selected(Vec<StagedCircleInstall>),
904}
905
906pub struct VerifiedSnapshotBootstrapInstall {
907    snapshot: PublishedStoreSnapshot,
908    store_root: coven_protocol::objects::VerifiedObject<StoreProtocolRoot>,
909    founder: coven_protocol::objects::VerifiedObject<StoreDeviceRegistration>,
910    authority: coven_protocol::store_commit::RetainedReplaySnapshotAuthority,
911    membership: InitialStoreMembershipAuthority,
912    routing_key: Option<coven_protocol::circle::RowRoutingKey>,
913    circle_selection: CircleRestoreSelection,
914    /// Fail the Circle-install step of the install transaction, after the Store
915    /// image has been installed within it — a test's stand-in for a crash between
916    /// the Store and Circle installs, exercising the single-transaction rollback.
917    #[cfg(any(test, feature = "test-utils"))]
918    fail_circle_install: bool,
919}
920
921impl VerifiedSnapshotBootstrapInstall {
922    pub fn new(
923        snapshot: PublishedStoreSnapshot,
924        store_root: coven_protocol::objects::VerifiedObject<StoreProtocolRoot>,
925        founder: coven_protocol::objects::VerifiedObject<StoreDeviceRegistration>,
926        authority: crate::VerifiedStoreSnapshotAuthority,
927        membership: InitialStoreMembershipAuthority,
928        routing_encryption: Option<&EncryptionService>,
929    ) -> Result<Self, DbError> {
930        if store_root.value.to_bytes() != store_root.bytes
931            || store_root.value.object_hash() != store_root.semantic_hash
932        {
933            return Err(DbError::Message(
934                "bootstrap Store root differs from its verified object".to_string(),
935            ));
936        }
937        let root = coven_protocol::store_commit::StoreRootRef {
938            store_root_id: store_root.value.descriptor.store_root_id(),
939            store_root_hash: store_root.semantic_hash,
940            object: store_root.object.clone(),
941        };
942        let founder_reference =
943            StoreDeviceRegistrationRef::from_registration(&founder.value, founder.object.clone());
944        if founder.semantic_hash != founder_reference.registration_hash {
945            return Err(DbError::Message(
946                "bootstrap founder semantic hash differs from its exact registration".to_string(),
947            ));
948        }
949        let authority = authority.into_authority();
950        authority.validate()?;
951        if authority.store_root != root
952            || authority.founder_registration != founder_reference
953            || authority.snapshot != snapshot.reference
954            || authority.metadata != snapshot.meta
955            || snapshot.meta.successor.next_slot != snapshot.successor_slot
956        {
957            return Err(DbError::Message(
958                "bootstrap snapshot differs from its verified authority authority".to_string(),
959            ));
960        }
961        let routing_key = routing_encryption
962            .map(|encryption| {
963                coven_protocol::circle::derive_row_routing_key(encryption, root.store_root_hash)
964                    .map_err(|error| DbError::context("derive bootstrap row-routing key", error))
965            })
966            .transpose()?;
967        Ok(Self {
968            snapshot,
969            store_root,
970            founder,
971            authority,
972            membership,
973            routing_key,
974            circle_selection: CircleRestoreSelection::Pending,
975            #[cfg(any(test, feature = "test-utils"))]
976            fail_circle_install: false,
977        })
978    }
979
980    /// Attach the Circle images selected against a throwaway query copy opened
981    /// through this same authority. Kept separate from `new` so one verified
982    /// install can first query and then install for real without re-verifying the
983    /// Store authority.
984    pub fn with_circle_installs(mut self, circle_installs: Vec<StagedCircleInstall>) -> Self {
985        self.circle_selection = CircleRestoreSelection::Selected(circle_installs);
986        self
987    }
988
989    /// Arm the Circle-install failure injection: the install transaction rolls
990    /// back after the Store image is installed but before any Circle image
991    /// commits, standing in for a crash between the two installs.
992    #[cfg(any(test, feature = "test-utils"))]
993    pub fn fail_circle_install_for_test(mut self) -> Self {
994        self.fail_circle_install = true;
995        self
996    }
997}
998
999#[cfg(test)]
1000mod tests;
coven_database/lib.rs

coven_database/
lib.rs
