coven_database/
store.rs

1mod activated_registration_records;
2#[cfg(any(test, feature = "test-utils"))]
3pub(crate) use circle_operations::circle_current_state_on;
4use circle_operations::circle_publication_context_on;
5pub(crate) use store_session::active_store_publication::{
6    clear_active_store_commit_for_owner_on, load_active_store_publication_on,
7};
8pub(crate) use store_session::payload_store;
9use store_session::{
10    blob_outbox, blob_transitions, circle_authority, circle_controls, circle_operations,
11    host_write_capture, host_write_operation, local_blob_cleanup, materialized_commit_index,
12    merge_materialization_transaction, pull_replay, replay_projection, retained_merge_replay,
13    retained_replay, snapshot_image, stream_activation_records, verified_store_authority,
14    write_lifecycle,
15};
16pub use store_session::{candidate_records, payload_store::PayloadStoreError, reclaim};
17mod device_join;
18pub(crate) use device_join::{
19    advance_device_join_on, begin_device_join_on, device_join_records_on, forget_device_join_on,
20};
21pub mod device_join_journal;
22mod host_sql;
23mod host_sql_reads;
24mod store_reads;
25pub use store_reads::StoreReads;
26mod host_sql_transaction;
27pub(crate) use host_write_operation::{NewBlob, StagedBlobBatch};
28pub(crate) use local_blob_cleanup::{
29    complete_local_blob_cleanup_on, local_blob_cleanup_intents_on,
30};
31pub mod local_blob_cleanup_intents;
32pub mod materialization_models;
33use activated_registration_records::record_activated_store_device_registrations_on;
34#[cfg(any(test, feature = "test-utils"))]
35pub(crate) use store_session::prepared_remote_objects::persist_prepared_audience_objects_on;
36pub mod publication_state;
37use replay_projection::ReplayProjection;
38#[cfg(any(test, feature = "test-utils"))]
39pub(crate) use retained_merge_replay::remove_retained_replay_ownership_from_snapshot_on;
40mod store_database;
41mod store_device_state;
42pub use store_database::StoreDatabase;
43#[cfg(any(test, feature = "test-utils"))]
44pub(crate) use store_device_state::{
45    prune_unreferenced_store_device_states_on, record_store_device_snapshot_on,
46};
47mod store_session;
48pub use store_session::PreparedStoreSnapshot;
49pub(crate) use store_session::{SnapshotPreparationDirectory, StoreSession};
50pub(crate) use verified_store_authority::VerifiedStoreAuthority;
51
52use crate::{
53    load_remote_object_on, persist_exact_remote_object_on, Database, DbError,
54    OutboundStoreAckActivation,
55};
56use coven_protocol::prepared_commit::PreparedStoreOperationCommit;
57use coven_protocol::store_commit::{StoreAckRef, StoreBatchCommitRef};
58
59const CACHE_BUDGET_STATE_KEY_PREFIX: &str = "cache_budget:";
60
61fn cache_budget_state_key(namespace: &str) -> String {
62    format!("{CACHE_BUDGET_STATE_KEY_PREFIX}{namespace}")
63}
64
65pub use blob_outbox::{
66    CloudOutboxSnapshot, MakeRemoteProgress, QueuedDelete, QueuedMakeRemote, QueuedUpload,
67    QueuedUploadPhase,
68};
69pub use blob_outbox::{
70    OutboxEntry, OutboxFailure, OutboxFailureKind, OutboxOperation, OutboxUploadState,
71};
72pub use blob_transitions::{
73    BlobTransitionRoot, MakeRemoteAdmission, MaterializedLocalBlob, PostUpload,
74};
75pub use candidate_records::CandidateCleanupObject;
76pub use circle_authority::CirclePackageAccess;
77pub use circle_controls::PreparedCircleObjects;
78pub use device_join::DeviceJoinJournalStore;
79pub use host_sql::{SqlContext, SqlReadContext};
80pub use host_write_capture::{
81    audience_moves_by_row, AudienceBlobMoveStaging, HostWriteBlobTransaction,
82    StagedAudienceBlobRollback,
83};
84pub use host_write_operation::StoreRowWrites;
85pub use host_write_operation::{BlobFileFailure, BlobFileFailures, WriteBatch};
86pub use host_write_operation::{HostWriteError, HostWriteOperation};
87pub use local_blob_cleanup::LocalBlobCleanup;
88pub use materialization_models::{
89    activated_merge_membership_remote_objects, DeviceJoinBootstrapCommit, DeviceJoinBootstrapPlan,
90    DeviceJoinBootstrapRowData, InstalledReplayBaseline, MembershipAuthorityBytes,
91    OwnedVerifiedMergeMaterialization, PreparedMergeMaterialization,
92    PreparedMergeMaterializationPackage, ResolvedDeviceJoinBootstrap, RetainedAudiencePackage,
93    RetainedMergeHistoryCheckpoint, RetainedMergeMaterializationKey, RetainedPackageApplication,
94    VerifiedMergeMaterialization, VerifiedMergeMembershipObjects, VerifiedStoreSnapshotAuthority,
95};
96#[cfg(test)]
97pub(crate) use merge_materialization_transaction::test_install_winning_blob_bindings;
98#[cfg(any(test, feature = "test-utils"))]
99pub use merge_materialization_transaction::{resolve_and_apply_changeset, ApplyResult};
100#[cfg(any(test, feature = "test-utils"))]
101pub(crate) use merge_materialization_transaction::{
102    test_apply_changeset, test_record_verified_circle_activations,
103};
104pub(crate) use merge_materialization_transaction::{
105    AppliedMergeMaterialization, MergeMaterializationTransaction,
106};
107pub use merge_materialization_transaction::{
108    IncomingTimestampPolicy, TableSchema, ValidatedChangeset, WinningRow,
109};
110pub use publication_state::{StorePublicationPreparation, StoreWritePreparation};
111pub(crate) use pull_replay::{
112    install_circle_bootstrap_connection_on, install_circle_bootstrap_image_on,
113    install_circle_bootstrap_remote_objects_on,
114};
115pub use reclaim::journal::{
116    DurableStoreReclaimObject, DurableStoreReclaimOperation, ReclaimedStorePackage,
117    StoreReclaimJournalError, StuckReclaimOperation,
118};
119pub(crate) use retained_replay::copy_table_with_conflicts;
120pub(crate) use retained_replay::migrate_retained_replay_schema_on;
121pub use retained_replay::{
122    projection_table_names, RetainedReplayAuthority, RetainedReplayBaseline,
123    RetainedReplayGenesisAuthority,
124};
125pub(crate) use snapshot_image::verify_circle_bootstrap_connection;
126pub use snapshot_image::{
127    CreatedSnapshot, SnapshotBlobFact, SnapshotDatabaseImage, SnapshotImageError,
128    SnapshotImageOperationError,
129};
130pub use store_session::circle_acknowledgements::CircleAckPublicationInput;
131pub use store_session::observed_store_publication::{
132    AcceptedStoreCommitEvidence, AcceptedStoreCommitPublication, AcceptedStorePublicationInterval,
133    ObservedStorePublication, StoreCommitPublicationOutcome, StorePublicationBoundary,
134};
135pub use store_session::store_records::{CoveredStoreWrite, CoveredStoreWriteCompletion};
136pub use store_session::AdvancedReplayBaseline;
137pub use write_lifecycle::BlockedWriteDiscard;
138
139pub(crate) use store_session::install_verified_snapshot_bootstrap_on;
140#[cfg(any(test, feature = "test-utils"))]
141pub(crate) use store_session::{
142    circle_bootstrap_replay_inputs_for_test, retained_merge_replay_inputs_for_test,
143};
144
145#[derive(Clone)]
146pub(crate) struct StoreDatabaseRuntime {
147    /// Serializes complete membership-chain loads that share this database, so a
148    /// load cannot return an older chain after another load commits a newer floor.
149    membership_load: std::sync::Arc<tokio::sync::Mutex<()>>,
150    /// Serializes construction and execution of the one local membership mutation
151    /// whose exact signed bytes are held in `outbound_membership_mutation`.
152    membership_mutation: std::sync::Arc<tokio::sync::Mutex<()>>,
153    /// Serializes publication of the one durable founder graph.
154    store_creation: std::sync::Arc<tokio::sync::Mutex<()>>,
155    /// Serializes the exact local device-exclusion object and its Store-stream
156    /// activation candidate across every database-handle clone.
157    device_exclusion: std::sync::Arc<tokio::sync::Mutex<()>>,
158    /// Serializes staging and publication of the one exact snapshot generation
159    /// held in `outbound_store_snapshot`.
160    snapshot_publication: std::sync::Arc<tokio::sync::Mutex<()>>,
161    /// Serializes this device's authorship of its own Store stream: reading the
162    /// position a commit extends, and publishing the head that takes it.
163    ///
164    /// The device owns that stream, so two of its own writers contending for one
165    /// position is an implementation accident with no meaning in the protocol —
166    /// not a conflict any peer could observe. Held across the pair, it cannot
167    /// happen.
168    own_stream_authorship: std::sync::Arc<tokio::sync::Mutex<()>>,
169    /// Serializes the full durable-intent to filesystem-deletion to
170    /// intent-removal operation across every clone of this database.
171    local_blob_cleanup: std::sync::Arc<tokio::sync::Mutex<()>>,
172    /// Serializes draining the blob upload queue: reading the pending entries
173    /// and running every attempt admitted from that read.
174    ///
175    /// The queue has two drainers — the sync cycle's and the host's explicit
176    /// one — and an entry is only claimed by the compare-and-set that hands off
177    /// its prepared object. Everything before that handoff is unguarded, so two
178    /// drainers reading the same pending row both sealed the whole blob, both
179    /// wrote a spool, and both reported preparation progress for it, with only
180    /// the loser's work thrown away at the end. Held across the read and the
181    /// attempts it admits, one entry cannot be in two attempts at once.
182    blob_upload_drain: std::sync::Arc<tokio::sync::Mutex<()>>,
183}
184
185impl StoreDatabaseRuntime {
186    pub(crate) fn new() -> Self {
187        Self {
188            membership_load: Default::default(),
189            membership_mutation: Default::default(),
190            store_creation: Default::default(),
191            device_exclusion: Default::default(),
192            snapshot_publication: Default::default(),
193            own_stream_authorship: Default::default(),
194            local_blob_cleanup: Default::default(),
195            blob_upload_drain: Default::default(),
196        }
197    }
198
199    pub(crate) async fn membership_load_permit(&self) -> MembershipLoadPermit {
200        MembershipLoadPermit {
201            _guard: self.membership_load.clone().lock_owned().await,
202        }
203    }
204
205    pub(crate) async fn membership_mutation_permit(&self) -> MembershipMutationPermit {
206        MembershipMutationPermit {
207            _guard: self.membership_mutation.clone().lock_owned().await,
208        }
209    }
210
211    pub(crate) async fn store_creation_permit(&self) -> StoreCreationPermit {
212        StoreCreationPermit {
213            _guard: self.store_creation.clone().lock_owned().await,
214        }
215    }
216
217    pub(crate) async fn device_exclusion_permit(&self) -> DeviceExclusionPermit {
218        DeviceExclusionPermit {
219            _guard: self.device_exclusion.clone().lock_owned().await,
220        }
221    }
222
223    pub(crate) async fn author_own_stream(&self) -> tokio::sync::OwnedMutexGuard<()> {
224        self.own_stream_authorship.clone().lock_owned().await
225    }
226
227    pub(crate) async fn snapshot_publication_permit(&self) -> SnapshotPublicationPermit {
228        SnapshotPublicationPermit {
229            _guard: self.snapshot_publication.clone().lock_owned().await,
230        }
231    }
232
233    pub(crate) async fn local_blob_cleanup_permit(&self) -> LocalBlobCleanupPermit {
234        LocalBlobCleanupPermit {
235            _guard: self.local_blob_cleanup.clone().lock_owned().await,
236        }
237    }
238
239    pub(crate) async fn blob_upload_drain_permit(&self) -> BlobUploadDrainPermit {
240        BlobUploadDrainPermit {
241            _guard: self.blob_upload_drain.clone().lock_owned().await,
242        }
243    }
244}
245
246pub struct MembershipLoadPermit {
247    _guard: tokio::sync::OwnedMutexGuard<()>,
248}
249
250pub struct MembershipMutationPermit {
251    _guard: tokio::sync::OwnedMutexGuard<()>,
252}
253
254pub struct StoreCreationPermit {
255    _guard: tokio::sync::OwnedMutexGuard<()>,
256}
257
258pub struct DeviceExclusionPermit {
259    _guard: tokio::sync::OwnedMutexGuard<()>,
260}
261
262/// This device's exclusive turn to author its own next Store commit, held from
263/// reading the position through publishing the head that takes it.
264pub struct OwnStreamAuthorship {
265    _guard: tokio::sync::OwnedMutexGuard<()>,
266    database: StoreDatabase,
267}
268
269/// This drain's exclusive turn over the blob upload queue, held from reading
270/// the pending entries through the last attempt admitted from that read.
271///
272/// A caller that waits for it reads the queue afterwards and so sees what the
273/// drain ahead of it left, rather than a second view of entries already being
274/// uploaded.
275pub struct BlobUploadDrainPermit {
276    _guard: tokio::sync::OwnedMutexGuard<()>,
277}
278
279pub struct SnapshotPublicationPermit {
280    _guard: tokio::sync::OwnedMutexGuard<()>,
281}
282
283pub struct LocalBlobCleanupPermit {
284    _guard: tokio::sync::OwnedMutexGuard<()>,
285}
coven_database/store.rs

coven_database/
store.rs
