coven_core/
database.rs

1//! The owned SQLite connection.
2//!
3//! coven owns one `rusqlite::Connection` together with the sync bookkeeping
4//! beside it. Every database access — the host's app SQL, coven's bookkeeping,
5//! changeset capture and apply — runs against that one connection, so access is
6//! serialized.
7//!
8//! Hosts open coven with [`crate::Coven::builder`] and run app SQL through
9//! [`crate::CovenHandle::sql`] or [`crate::CovenHandle::write`].
10
11pub(crate) use crate::database::blob_records::{
12    load_activated_registration_on, remote_audience_to_db,
13};
14pub(crate) use crate::database::circle_snapshot_records::{
15    load_outbound_circle_snapshot_on, load_published_circle_snapshot_on,
16};
17pub(crate) use crate::database::cloud_outbox_records::consume_created_upload_handoff_on;
18use crate::database::connection_io::open_connection;
19use crate::database::connection_io::open_connection_read_only;
20use crate::database::connection_io::scan_max_updated_at;
21use crate::database::connection_io::seed_from;
22use crate::database::local_store_identity::pin_host_device_id_on;
23use crate::database::local_store_identity::validate_host_device_id_on;
24pub(crate) use crate::database::remote_object_records::begin_remote_candidate_nonactivation_on;
25pub(crate) use crate::database::remote_object_records::begin_remote_candidate_nonactivation_with_verified_head_on;
26pub(crate) use crate::database::remote_object_records::candidate_graph_exact_objects;
27pub(crate) use crate::database::remote_object_records::finish_remote_candidate_nonactivation_on;
28pub(crate) use crate::database::remote_object_records::index_retained_replay_owner_on;
29pub(crate) use crate::database::remote_object_records::load_protocol_inert_object_on;
30pub(crate) use crate::database::remote_object_records::load_remote_object_on;
31pub(crate) use crate::database::remote_object_records::mark_remote_object_uploaded_on;
32pub(crate) use crate::database::remote_object_records::mark_reusable_retained_authority_uploaded_on;
33pub(crate) use crate::database::remote_object_records::persist_exact_remote_object_on;
34pub(crate) use crate::database::remote_object_records::persist_prepared_remote_object_on;
35pub(crate) use crate::database::remote_object_records::record_reclaimed_store_package_on;
36pub(crate) use crate::database::remote_object_records::replace_prepared_merge_head_remote_on;
37pub(crate) use crate::database::remote_object_records::update_remote_object_on;
38pub(crate) use crate::database::remote_object_records::{
39    validate_prepared_blob_on, validate_prepared_package_on, validate_remote_object_on,
40    RemoteStoredRepresentationRef,
41};
42use crate::database::snapshot_objects::validate_snapshot_object_owners_on;
43pub(crate) use crate::database::snapshot_objects::{
44    install_snapshot_blob_plan_on, validate_snapshot_blob_plan_on,
45};
46pub(crate) use crate::database::snapshot_records::{
47    load_outbound_store_snapshot_on, load_published_store_snapshot_on,
48};
49pub(crate) use crate::database::store_ack_records::{
50    finish_outbound_circle_acks_on, finish_outbound_store_ack_on, load_outbound_store_ack_on,
51    load_published_store_ack_on, record_activated_circle_acks_on, record_activated_store_ack_on,
52    store_snapshot_first_slot, verify_next_local_store_ack_on,
53};
54pub(crate) use crate::database::store_authority_records::install_store_founder_state_on;
55pub(crate) use crate::database::store_reclaim_records::{
56    insert_store_reclaim_operation_on, load_store_reclaim_operation_on,
57    parse_store_reclaim_operation, record_store_reclaim_activation_on, store_reclaim_journal_error,
58    update_store_reclaim_operation_on,
59};
60pub(crate) use crate::database::stream_activation_records::load_registered_stream_activation_on;
61pub(crate) use crate::database::stream_activation_records::record_verified_stream_activations_on;
62
63use std::collections::{BTreeMap, BTreeSet, HashMap};
64use std::path::{Path, PathBuf};
65use std::sync::Arc;
66
67use rusqlite::{Connection, OptionalExtension};
68use tracing::error;
69
70use crate::blob::decl::BlobDecls;
71use crate::blob::locator::{BlobLocator, RemoteAudience, StoredBlobRef};
72use crate::blob::{BlobRef, RowBlobAuthority, RowBlobRef};
73use crate::db::{
74    apply_coven_schema, expected_coven_schema_manifest, is_reserved_table_name,
75    live_coven_schema_manifest, CovenSchemaManifest, ExternalBlob, OutboxEntry, OutboxOperation,
76    OutboxUploadState,
77};
78use crate::encryption::EncryptionService;
79use crate::migration::{run_migrations_in_transaction, Migration, MigrationError};
80use crate::sync::audience_package::{AudiencePackage, RowBlobLocatorBinding};
81use crate::sync::circle::Audience;
82use crate::sync::gate::{self, Gates};
83use crate::sync::hlc::{Hlc, Timestamp, UpdatedAtStamper, HIGHWATER_STATE_KEY, MAX_FUTURE_SKEW_MS};
84use crate::sync::membership::{AuthorHead, MembershipEntry, MembershipEntryRef, MembershipHeadRef};
85use crate::sync::remote_object::{
86    remote_object_id, CandidateExclusiveObjectDomain, RemoteObjectRecord, RetainedReplayOwner,
87    SharedLiveSetObjectDomain,
88};
89use crate::sync::routing_contract::SyncRoutingContract;
90use crate::sync::session::{quote_ident, SyncedTable};
91use crate::sync::storage::{ExactObjectRef, PreparedExactObject};
92use crate::sync::store::VerifiedStreamActivations;
93use crate::sync::store::{
94    DurableStoreReclaimOperation, ReclaimCommitActivation, ReclaimedStorePackage,
95    StoreReclaimJournalError,
96};
97use crate::sync::store::{
98    RetainedReplayAuthority, RetainedReplayBaseline, RetainedReplayGenesisAuthority,
99    RetainedReplaySnapshotAuthority,
100};
101use crate::sync::store_commit::{
102    ack_slot_prefix, CommitFrontier, ObjectHash, ResolvedStoreDeviceState, SnapshotImageRef,
103    SnapshotMeta, StoreAck, StoreAckRef, StoreBatchCommit, StoreBatchCommitRef, StoreCommitCoord,
104    StoreDeviceHead, StoreDeviceRegistration, StoreDeviceRegistrationRef, StoreProtocolRoot,
105    StoreSnapshotRef, StreamActivationId,
106};
107use crate::write::{PendingWrite, WriteId, WriteStatus};
108
109mod blob_bindings;
110mod blob_records;
111mod circle_operation_records;
112mod cloud_outbox;
113mod cloud_outbox_records;
114mod connection_io;
115pub(crate) use connection_io::{attach_session, capture_changeset, open_database_image};
116mod circle_snapshot_records;
117mod database_open;
118mod database_runtime;
119mod local_state;
120mod local_store_identity;
121mod make_remote;
122mod operation_models;
123mod prepared_audience_objects;
124mod provider_probes;
125mod remote_object_records;
126mod schema_contract;
127mod snapshot_objects;
128mod snapshot_records;
129mod store_ack_records;
130mod store_authority_records;
131mod store_coordinates;
132mod store_reclaim_records;
133mod stream_activation_records;
134mod write_lifecycle;
135mod write_models;
136
137pub(crate) use crate::sync::store::{
138    CandidateCleanupObject, OwnedVerifiedMergeMaterialization, RetainedMergeMaterializationKey,
139    RetainedPackageApplication, VerifiedMergeMaterialization, VerifiedMergeMembershipObjects,
140};
141pub(crate) use blob_records::{load_prepared_audience_objects_on, previous_row_blob_for_write_on};
142pub(crate) use circle_operation_records::{
143    load_circle_operation_on, parse_circle_operation_row, PreparedCircleOperationRow,
144};
145use database_open::{run_connection_thread, ConnectionThread, CovenMetadataOpen, DbJob};
146pub(crate) use local_store_identity::local_merge_stream_id_on;
147pub(crate) use local_store_identity::{
148    local_activated_registration_ref_on, local_store_authority_on,
149};
150pub(crate) use operation_models::{
151    DurableCircleSnapshotPublication, DurableDeviceRegistration, DurableMembershipMutation,
152    DurableSnapshotPublication, LocalDeviceRegistrationJournalRow, LocalDeviceRegistrationState,
153    MembershipMutationActivation, PreparedLocalDeviceRegistrationRow, PreparedSnapshotBlob,
154    PublishedCircleSnapshot, PublishedStoreSnapshot,
155};
156#[cfg(feature = "invariant-tests")]
157pub use prepared_audience_objects::exercise_exact_outbound_blob_graph;
158pub(crate) use prepared_audience_objects::{
159    validate_prepared_audience_blob_graph, BlobActivation, MakeRemoteIntentState,
160    PreparedAudienceBlob, PreparedAudienceObjects, PreparedAudiencePackage, PreparedRemoteObject,
161    StoredBlobReferenceState,
162};
163use schema_contract::validate_host_synced_tables;
164pub(crate) use schema_contract::DurablePreparedProtocolObject;
165pub(crate) use schema_contract::{StoreBatchCompletion, StoreBatchLocalCleanup};
166pub(crate) use store_authority_records::{
167    ensure_founder_replay_baseline_on, founder_graph_identity,
168    install_generation_zero_replay_baseline_on, install_snapshot_replay_baseline_on,
169    install_store_root_authority_on, load_local_store_founder_graph_on,
170    load_store_root_authority_on, validate_founder_graph, DurableFounderMembershipJournal,
171};
172pub(crate) use store_authority_records::{
173    load_generation_zero_replay_baseline_on, required_store_root_authority_on,
174};
175pub(crate) use store_authority_records::{
176    DurableFounderGraph, DurableFounderMembership, FounderMembershipRefs,
177};
178pub(crate) use write_models::{
179    AuthorExclusionActivationLocator, BlockedMergeCandidate, CompletePreparedStoreWriteOutcome,
180    ExactProtocolObject, InitialStoreMembershipAuthority, MergeAbandonmentState,
181    MergeReplayWriteOverlay, OutboundStoreAck, OutboundStoreAckActivation,
182    PreparedMergeAbandonmentCandidates, PreparedProtocolObject, PreparedStoreWrite,
183    PreparedStoreWriteCommit, PreparedStoreWritePartitions, PublishedStoreAck, StoreWriteBase,
184    StoreWriteBlobFact, StoreWriteBlobFacts, StoreWriteBlobMoveDestination, StoreWriteRemoteBlob,
185    StoreWriteRouting, TerminalCandidateAuthority, TerminalCandidateCleanupVerification,
186};
187
188pub const LOCAL_DEVICE_ID_STATE_KEY: &str = "local_device_id";
189const HOST_DEVICE_ID_STATE_KEY: &str = "host_device_id";
190pub(crate) const SYNC_ROUTING_CONTRACT_STATE_KEY: &str = "sync_routing_contract";
191pub const SYNC_ROUTING_HASH_STATE_KEY: &str = "sync_routing_hash";
192pub(crate) const COVEN_SCHEMA_MANIFEST_STATE_KEY: &str = "coven_schema_manifest";
193pub(crate) const COVEN_INITIALIZED_STATE_KEY: &str = "coven_initialized";
194pub(crate) const COVEN_INITIALIZED_STATE_VALUE: &str = "1";
195pub(crate) const STORE_DEVICE_GENESIS_STATE_KEY: &str = "store_device_genesis_state";
196const GATE_BASELINE_SCHEMA: &str = "coven_gate_empty";
197
198pub(crate) fn authorize_host_sql(
199    context: rusqlite::hooks::AuthContext<'_>,
200) -> rusqlite::hooks::Authorization {
201    use rusqlite::hooks::{AuthAction, Authorization};
202
203    if context
204        .database_name
205        .is_some_and(|name| name.eq_ignore_ascii_case(GATE_BASELINE_SCHEMA))
206        || matches!(
207            context.action,
208            AuthAction::Detach { database_name }
209                if database_name.eq_ignore_ascii_case(GATE_BASELINE_SCHEMA)
210        )
211        || matches!(
212            context.action,
213            AuthAction::Pragma { pragma_name, .. }
214                if pragma_name.eq_ignore_ascii_case("database_list")
215        )
216    {
217        Authorization::Deny
218    } else {
219        Authorization::Allow
220    }
221}
222
223/// An error from the owned database.
224#[derive(Debug, thiserror::Error)]
225pub enum DbError {
226    #[error("database error: {0}")]
227    Message(String),
228    #[error("database error: Store protocol root hash is absent")]
229    StoreRootHashMissing,
230    /// The local device was excluded from a Circle epoch close and has not yet
231    /// reset its projection from the successor bootstrap, so it cannot publish
232    /// into the Circle. Stays matchable at the publication boundary rather than
233    /// flattening into a message.
234    #[error(
235        "device excluded from circle {circle_id} close {close_id} must reset before publishing"
236    )]
237    ExcludedDeviceMustReset {
238        circle_id: crate::sync::circle::CircleId,
239        close_id: crate::sync::circle::CircleEpochCloseId,
240    },
241}
242
243impl DbError {
244    pub(crate) fn into_message(self) -> String {
245        match self {
246            Self::Message(message) => message,
247            Self::StoreRootHashMissing => "Store protocol root hash is absent".to_string(),
248            other @ Self::ExcludedDeviceMustReset { .. } => other.to_string(),
249        }
250    }
251}
252
253impl From<rusqlite::Error> for DbError {
254    fn from(e: rusqlite::Error) -> Self {
255        DbError::Message(e.to_string())
256    }
257}
258
259/// Why opening the database failed. Splits a migration-ladder failure from every
260/// other open-time database error so the [`MigrationError`] a host acts on —
261/// [`MigrationError::SchemaTooNew`], whose remedy is "update the app" — stays
262/// matchable at the open boundary instead of being flattened into a
263/// [`DbError`] string.
264#[derive(Debug, thiserror::Error)]
265pub enum OpenError {
266    #[error(transparent)]
267    Migration(#[from] MigrationError),
268    #[error(transparent)]
269    Db(#[from] DbError),
270}
271
272#[derive(Clone)]
273struct DatabaseState {
274    hlc: Arc<Hlc>,
275    synced_tables: Arc<Vec<SyncedTable>>,
276    schema_version: u32,
277    sync_routing_hash: ObjectHash,
278    gates: Arc<Gates>,
279    blob_decls: Arc<BlobDecls>,
280    /// The host's blob-tombstone convergence window, read by the tombstone GC to
281    /// age each tombstone's `deleted_at`. Host config carried here alongside
282    /// `synced_tables` so the sync layer reads it from the one owner rather than
283    /// threading a separately-passed copy that could diverge.
284    blob_tombstone_grace: chrono::Duration,
285    /// How many blob transfers each transfer loop runs at once, read by the upload
286    /// drain and the pin loop (both hold `&Database`). Open-time host config carried
287    /// here for the same single-owner reason as `blob_tombstone_grace`.
288    transfer_limits: crate::blob::TransferLimits,
289    store_runtime: crate::sync::store::StoreDatabaseRuntime,
290    /// Serializes the full durable-intent to filesystem-deletion to intent-removal
291    /// operation across every clone of this database.
292    local_blob_cleanup: Arc<tokio::sync::Mutex<()>>,
293    ids: crate::id_provider::IdRef,
294    write_statuses:
295        Arc<std::sync::Mutex<HashMap<WriteId, tokio::sync::watch::Sender<WriteStatus>>>>,
296    #[cfg(any(test, feature = "test-utils"))]
297    test_pause_points: Arc<TestPausePoints<DatabaseTestPoint>>,
298    #[cfg(any(test, feature = "test-utils"))]
299    merge_materialization_failure: Arc<std::sync::Mutex<Option<MergeMaterializationFailurePoint>>>,
300}
301
302/// Test-only checkpoints reached by database operations whose ordering matters.
303#[cfg(any(test, feature = "test-utils"))]
304#[doc(hidden)]
305#[derive(Clone, Debug, PartialEq, Eq)]
306pub enum DatabaseTestPoint {
307    LocalBlobCleanupRequested,
308    LocalBlobCleanupAcquired,
309    LocalBlobCleanupBeforeFilesystem { namespace: String, blob_id: String },
310    LocalBlobCleanupFinished,
311    PullAfterRemoteCommit { device_id: String, seq: u64 },
312    StoreWriteCommitUploaded { write_id: WriteId },
313    StoreWriteHeadReadBack { write_id: WriteId },
314    StoreDeviceExclusionCandidateStaged,
315}
316
317#[cfg(any(test, feature = "test-utils"))]
318#[doc(hidden)]
319#[derive(Clone, Copy, Debug, PartialEq, Eq)]
320pub enum MergeMaterializationFailurePoint {
321    SummaryMaterialization,
322    RetractionDeletion,
323    ProjectionReplacement,
324}
325
326#[cfg(any(test, feature = "test-utils"))]
327#[derive(Clone)]
328pub(crate) struct MergeMaterializationFailureInjection {
329    armed: Arc<std::sync::Mutex<Option<MergeMaterializationFailurePoint>>>,
330}
331
332#[cfg(any(test, feature = "test-utils"))]
333impl MergeMaterializationFailureInjection {
334    pub(crate) fn reach(&self, point: MergeMaterializationFailurePoint) -> Result<bool, DbError> {
335        let mut armed = self.armed.lock().map_err(|_| {
336            DbError::Message("Merge materialization failure lock poisoned".to_string())
337        })?;
338        if armed.as_ref() != Some(&point) {
339            return Ok(false);
340        }
341        armed.take();
342        Ok(true)
343    }
344}
345
346#[cfg(any(test, feature = "test-utils"))]
347struct ArmedTestPause<K> {
348    point: K,
349    reached: Arc<tokio::sync::Notify>,
350    resume: Arc<tokio::sync::Notify>,
351}
352
353#[cfg(any(test, feature = "test-utils"))]
354struct TestPauseState<K> {
355    armed: Option<ArmedTestPause<K>>,
356    observers: Vec<tokio::sync::mpsc::UnboundedSender<K>>,
357}
358
359#[cfg(any(test, feature = "test-utils"))]
360struct TestPausePoints<K> {
361    state: std::sync::Mutex<TestPauseState<K>>,
362}
363
364#[cfg(any(test, feature = "test-utils"))]
365impl<K> Default for TestPausePoints<K> {
366    fn default() -> Self {
367        Self {
368            state: std::sync::Mutex::new(TestPauseState {
369                armed: None,
370                observers: Vec::new(),
371            }),
372        }
373    }
374}
375
376#[cfg(any(test, feature = "test-utils"))]
377impl<K: Clone + PartialEq> TestPausePoints<K> {
378    fn arm(&self, point: K) -> (Arc<tokio::sync::Notify>, Arc<tokio::sync::Notify>) {
379        let reached = Arc::new(tokio::sync::Notify::new());
380        let resume = Arc::new(tokio::sync::Notify::new());
381        let prior = self
382            .state
383            .lock()
384            .expect("database test pause mutex poisoned")
385            .armed
386            .replace(ArmedTestPause {
387                point,
388                reached: reached.clone(),
389                resume: resume.clone(),
390            });
391        assert!(prior.is_none(), "database test pause already armed");
392        (reached, resume)
393    }
394
395    fn observe(&self) -> tokio::sync::mpsc::UnboundedReceiver<K> {
396        let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
397        self.state
398            .lock()
399            .expect("database test pause mutex poisoned")
400            .observers
401            .push(sender);
402        receiver
403    }
404
405    async fn reach(&self, point: K) {
406        let pause = {
407            let mut state = self
408                .state
409                .lock()
410                .expect("database test pause mutex poisoned");
411            state
412                .observers
413                .retain(|observer| observer.send(point.clone()).is_ok());
414            if state
415                .armed
416                .as_ref()
417                .is_some_and(|pause| pause.point == point)
418            {
419                state.armed.take()
420            } else {
421                None
422            }
423        };
424        if let Some(pause) = pause {
425            pause.reached.notify_one();
426            pause.resume.notified().await;
427        }
428    }
429}
430
431/// The owned SQLite connection and the sync bookkeeping resolved beside it at
432/// open. One connection thread owns this for the connection's whole life; every
433/// database access runs against it, so access is serialized. Changeset capture is
434/// per-transaction — [`Database::run_store_write_transaction_on`] attaches a
435/// session for the span of one host write and drains it into the existing write records —
436/// so no capture state lives on the core between calls.
437///
438/// `DatabaseCore` holds only `Send` fields (a `rusqlite::Connection`, which is
439/// `Send`, plus `Arc`s and a `u32`), so it is `Send` by construction — the
440/// connection thread receives it by value across a single `thread::spawn` with no
441/// manual `unsafe impl`.
442struct DatabaseCore {
443    conn: Connection,
444    hlc: Arc<Hlc>,
445    synced_tables: Arc<Vec<SyncedTable>>,
446    schema_version: u32,
447    sync_routing_hash: ObjectHash,
448    gates: Arc<Gates>,
449    blob_decls: Arc<BlobDecls>,
450    blob_tombstone_grace: chrono::Duration,
451    transfer_limits: crate::blob::TransferLimits,
452}
453
454/// One Circle's staged restore outcome, decided by selection against the
455/// restoring identity's re-resolved access. `Install` carries a verified image to
456/// project and record coverage for; `ClearCoverage` names a Circle the identity
457/// cannot decrypt, whose preserved coverage row must be deleted so replay never
458/// reconstructs an image the identity has no access to.
459pub(crate) enum StagedCircleDecision {
460    Install {
461        activation_commit: StoreBatchCommitRef,
462        image: crate::sync::store::VerifiedCircleImage,
463    },
464    ClearCoverage(crate::sync::circle::CircleId),
465}
466
467pub(crate) struct VerifiedSnapshotBootstrapInstall {
468    snapshot: PublishedStoreSnapshot,
469    store_root: crate::sync::store_objects::VerifiedObject<StoreProtocolRoot>,
470    founder: crate::sync::store_objects::VerifiedObject<StoreDeviceRegistration>,
471    stability: RetainedReplaySnapshotAuthority,
472    routing_key: Option<crate::sync::circle::RowRoutingKey>,
473    circle_decisions: Vec<StagedCircleDecision>,
474    /// Fail the Circle-decision step of the install transaction, after the Store
475    /// image has been installed within it — a test's stand-in for a crash between
476    /// the Store and Circle installs, exercising the single-transaction rollback.
477    #[cfg(any(test, feature = "test-utils"))]
478    fail_circle_install: bool,
479}
480
481impl VerifiedSnapshotBootstrapInstall {
482    pub(crate) fn new(
483        snapshot: PublishedStoreSnapshot,
484        store_root: crate::sync::store_objects::VerifiedObject<StoreProtocolRoot>,
485        founder: crate::sync::store_objects::VerifiedObject<StoreDeviceRegistration>,
486        stability: crate::sync::store::VerifiedStoreSnapshotStability,
487        routing_encryption: Option<&EncryptionService>,
488        circle_decisions: Vec<StagedCircleDecision>,
489    ) -> Result<Self, DbError> {
490        if store_root.value.to_bytes() != store_root.bytes
491            || store_root.value.object_hash() != store_root.semantic_hash
492        {
493            return Err(DbError::Message(
494                "bootstrap Store root differs from its verified object".to_string(),
495            ));
496        }
497        let root = crate::sync::store_commit::StoreRootRef {
498            store_root_id: store_root.value.descriptor.store_root_id(),
499            store_root_hash: store_root.semantic_hash,
500            object: store_root.object.clone(),
501        };
502        let founder_reference =
503            StoreDeviceRegistrationRef::from_registration(&founder.value, founder.object.clone());
504        if founder.semantic_hash != founder_reference.registration_hash {
505            return Err(DbError::Message(
506                "bootstrap founder semantic hash differs from its exact registration".to_string(),
507            ));
508        }
509        let stability = stability.into_authority();
510        stability.validate()?;
511        if stability.store_root != root
512            || stability.founder_registration != founder_reference
513            || stability.snapshot != snapshot.reference
514            || stability.metadata != snapshot.meta
515            || snapshot.meta.successor.next_slot != snapshot.successor_slot
516        {
517            return Err(DbError::Message(
518                "bootstrap snapshot differs from its verified stability authority".to_string(),
519            ));
520        }
521        let routing_key = routing_encryption
522            .map(|encryption| {
523                crate::sync::circle::derive_row_routing_key(encryption, root.store_root_hash)
524                    .map_err(|error| {
525                        DbError::Message(format!("derive bootstrap row-routing key: {error}"))
526                    })
527            })
528            .transpose()?;
529        Ok(Self {
530            snapshot,
531            store_root,
532            founder,
533            stability,
534            routing_key,
535            circle_decisions,
536            #[cfg(any(test, feature = "test-utils"))]
537            fail_circle_install: false,
538        })
539    }
540
541    /// Attach the Circle install/clear decisions selected against a throwaway
542    /// query copy opened through this same authority. Kept separate from `new` so
543    /// one verified install can first query (with no decisions) and then install
544    /// for real (with them), without re-verifying the Store authority.
545    pub(crate) fn with_circle_decisions(
546        mut self,
547        circle_decisions: Vec<StagedCircleDecision>,
548    ) -> Self {
549        self.circle_decisions = circle_decisions;
550        self
551    }
552
553    /// Arm the Circle-install failure injection: the install transaction rolls
554    /// back after the Store image is installed but before any Circle decision
555    /// commits, standing in for a crash between the two installs.
556    #[cfg(any(test, feature = "test-utils"))]
557    pub(crate) fn fail_circle_install_for_test(mut self) -> Self {
558        self.fail_circle_install = true;
559        self
560    }
561
562    fn install_on(
563        &self,
564        conn: &Connection,
565        schema_version: u32,
566        routing_hash: ObjectHash,
567        synced_tables: &[SyncedTable],
568    ) -> Result<(), DbError> {
569        let root = crate::sync::store_commit::StoreRootRef {
570            store_root_id: self.store_root.value.descriptor.store_root_id(),
571            store_root_hash: self.store_root.semantic_hash,
572            object: self.store_root.object.clone(),
573        };
574        let founder_reference = StoreDeviceRegistrationRef::from_registration(
575            &self.founder.value,
576            self.founder.object.clone(),
577        );
578        let genesis = ResolvedStoreDeviceState::founder(
579            &root,
580            founder_reference.clone(),
581            &self.store_root.value.descriptor.founder_pubkey,
582            self.store_root.value.descriptor.founder_grant.clone(),
583            &self.store_root.value.descriptor.founder_recovery,
584        )
585        .map_err(|error| DbError::Message(error.to_string()))?;
586        validate_snapshot_object_owners_on(conn, &root, &self.snapshot.meta)?;
587        install_store_root_authority_on(conn, &root, &self.store_root.bytes)?;
588        install_store_founder_state_on(
589            conn,
590            &root,
591            &founder_reference,
592            &self.founder.value,
593            &self.founder.bytes,
594            &genesis,
595        )?;
596        conn.execute("DELETE FROM snapshot_coverage", [])
597            .map_err(DbError::from)?;
598        for (stream_id, reference) in self.snapshot.meta.coverage.clone().into_refs() {
599            let encoded = serde_json::to_string(&reference).map_err(|error| {
600                DbError::Message(format!("serialize snapshot exact commit ref: {error}"))
601            })?;
602            conn.execute(
603                "INSERT INTO snapshot_coverage
604                 (device_id, seq, commit_ref, snapshot_hash) VALUES (?1, ?2, ?3, ?4)",
605                (
606                    &stream_id,
607                    Database::sequence_to_sqlite(&stream_id, reference.coord.sequence())?,
608                    encoded,
609                    self.snapshot.reference.snapshot_hash.to_string(),
610                ),
611            )
612            .map_err(DbError::from)?;
613        }
614        install_snapshot_replay_baseline_on(
615            conn,
616            schema_version,
617            routing_hash,
618            self.stability.clone(),
619        )?;
620        self.install_circle_decisions_on(conn, synced_tables)
621    }
622
623    /// Apply the staged Circle decisions inside the Store install's single
624    /// transaction: each `Install` projects the verified image and records its
625    /// coverage row (accepting a strictly newer cut, refusing a regression); each
626    /// `ClearCoverage` deletes the preserved row for a Circle the restoring
627    /// identity cannot decrypt. The whole set commits or rolls back with the Store
628    /// image — a partially installed union is never exposed.
629    fn install_circle_decisions_on(
630        &self,
631        conn: &Connection,
632        synced_tables: &[SyncedTable],
633    ) -> Result<(), DbError> {
634        use crate::sync::store::StoreDatabase;
635        #[cfg(any(test, feature = "test-utils"))]
636        if self.fail_circle_install {
637            return Err(DbError::Message(
638                "injected Circle install failure after Store install".to_string(),
639            ));
640        }
641        for decision in &self.circle_decisions {
642            match decision {
643                StagedCircleDecision::Install {
644                    activation_commit,
645                    image,
646                } => {
647                    let activation = StoreDatabase::verified_circle_activation_on(
648                        conn,
649                        image.circle_id(),
650                        image.control(),
651                    )?
652                    .ok_or_else(|| {
653                        DbError::Message(format!(
654                            "restored Circle {} image names a control absent from the installed \
655                             control indexes",
656                            image.circle_id()
657                        ))
658                    })?;
659                    crate::sync::store::install_circle_bootstrap_image_on(
660                        conn,
661                        synced_tables,
662                        activation_commit,
663                        image,
664                    )?;
665                    StoreDatabase::record_one_circle_bootstrap_coverage_on(
666                        conn,
667                        activation_commit,
668                        image,
669                        &activation.control,
670                    )?;
671                }
672                StagedCircleDecision::ClearCoverage(circle_id) => {
673                    StoreDatabase::clear_circle_bootstrap_coverage_on(conn, *circle_id)?;
674                }
675            }
676        }
677        Ok(())
678    }
679}
680
681/// A handle to the owned database. Cloneable; every clone sends work to the one
682/// connection thread over the same channel, so access serializes as the channel's
683/// FIFO.
684#[derive(Clone)]
685pub struct Database {
686    thread: Arc<ConnectionThread>,
687    state: DatabaseState,
688}
689
690#[cfg(test)]
691#[path = "database/tests.rs"]
692mod tests;
coven_core/database.rs

coven_core/
database.rs
