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

coven_database/
store.rs
