coven_database/
write_models.rs

1use super::*;
2use coven_protocol::store_commit::VerifiedStoreBatchCommit;
3
4pub struct PreparedStoreWrite {
5    pub write_id: WriteId,
6    pub partitions: PreparedStoreWritePartitions,
7    pub base: StoreWriteBase,
8    pub blob_facts: StoreWriteBlobFacts,
9}
10
11#[derive(Debug, Clone, PartialEq, Eq)]
12pub struct PreparedStoreWritePartitions {
13    pub store: Option<gate::AudiencePartition>,
14    pub circles: Vec<gate::AudiencePartition>,
15    pub local: Option<gate::AudiencePartition>,
16}
17
18#[derive(Clone)]
19pub(crate) struct MergeReplayWriteEffect {
20    pub write_id: WriteId,
21    pub partitions: PreparedStoreWritePartitions,
22}
23
24#[derive(Clone)]
25pub(crate) enum MergeReplayWrite {
26    LocalOnly {
27        effect: MergeReplayWriteEffect,
28        observed: coven_protocol::store_commit::CommitFrontier,
29    },
30    Unaccepted {
31        effect: MergeReplayWriteEffect,
32        observed: coven_protocol::store_commit::CommitFrontier,
33    },
34    Accepted {
35        effect: MergeReplayWriteEffect,
36        observed: coven_protocol::store_commit::CommitFrontier,
37        commit: StoreBatchCommitRef,
38    },
39    Consumed {
40        write_id: WriteId,
41    },
42}
43
44impl MergeReplayWrite {
45    pub(crate) fn write_id(&self) -> &WriteId {
46        match self {
47            Self::LocalOnly { effect, .. }
48            | Self::Unaccepted { effect, .. }
49            | Self::Accepted { effect, .. } => &effect.write_id,
50            Self::Consumed { write_id } => write_id,
51        }
52    }
53}
54
55/// What a replay projection owes the write journal.
56///
57/// The journal is the only record of a local partition — no commit carries one
58/// and no image built for an audience may — so what a projection has to put
59/// back from it depends on what the projection is for.
60pub(crate) enum ReplayJournal<'a> {
61    /// Nothing. An image projected for an audience carries no local rows at
62    /// all, and a projection built only to count rows does not need them.
63    Omit,
64    /// Everything the journal still owes to a projection that will replace the
65    /// live database.
66    Owed,
67    /// The settled prefix a baseline at this cut absorbs.
68    Folded(&'a [SettledStoreWrite]),
69}
70
71/// One write of the journal prefix a baseline at some cut absorbs, and what the
72/// fold owes it.
73#[derive(Debug, Clone, PartialEq, Eq)]
74pub(crate) struct SettledStoreWrite {
75    pub ordinal: i64,
76    pub write_id: WriteId,
77    pub fold: SettledWriteFold,
78    pub status: coven_protocol::write::WriteStatus,
79    pub observed: StoreWriteBase,
80    pub changeset_hash: ObjectHash,
81    pub input_hash: ObjectHash,
82}
83
84pub(crate) struct RetainedStoreWriteManifest {
85    pub ordinal: i64,
86    pub write_id: String,
87    pub status: String,
88    pub base: String,
89    pub changeset_hash: String,
90    pub prepared: Option<String>,
91    pub input_hash: ObjectHash,
92}
93
94/// What a baseline at some cut has to do with one settled write.
95#[derive(Debug, Clone, Copy, PartialEq, Eq)]
96pub(crate) enum SettledWriteFold {
97    /// Local-only. The image takes its rows, and the journal keeps nothing:
98    /// local-only is the whole of what could ever be said about the write, and
99    /// its caller was told that when it committed. A row missing from the
100    /// journal therefore means exactly this, which is how the status of one is
101    /// still answerable afterwards.
102    LocalOnly,
103    /// Published, at a commit the cut covers. The image takes its local rows;
104    /// the row stays as this device's record of where the write landed, which
105    /// is the one answer that has to survive an advance now that the
106    /// per-position index does not.
107    Published,
108    /// Discarded or retracted. Its rows were reversed, so the image must not
109    /// put them back, and the row stays as the record of that.
110    Reversed,
111}
112
113impl SettledWriteFold {
114    /// Whether the baseline image has to state this write's local rows.
115    pub(crate) fn states_local_rows(self) -> bool {
116        matches!(self, Self::LocalOnly | Self::Published)
117    }
118
119    /// Whether the journal keeps the write's receipt — its id and status —
120    /// after the image has absorbed everything else about it.
121    pub(crate) fn keeps_receipt(self) -> bool {
122        !matches!(self, Self::LocalOnly)
123    }
124}
125
126#[derive(Clone, Copy)]
127pub enum StoreWriteRouting<'a> {
128    Unscoped,
129    MergeScoped(&'a EncryptionService),
130}
131
132impl PreparedStoreWritePartitions {
133    #[cfg(any(test, feature = "test-utils"))]
134    pub fn iter(&self) -> impl Iterator<Item = &gate::AudiencePartition> {
135        self.store
136            .iter()
137            .chain(self.circles.iter())
138            .chain(self.local.iter())
139    }
140}
141
142#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
143#[serde(deny_unknown_fields)]
144pub struct StoreWriteBase {
145    /// The complete accepted Store frontier visible to the host transaction.
146    /// Publication removes its own stream and represents that position through
147    /// the signed predecessor; replay uses every stream to place local effects.
148    pub dependencies: BTreeMap<String, StoreBatchCommitRef>,
149}
150
151#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
152#[serde(deny_unknown_fields)]
153pub struct StoreWriteBlobFacts {
154    pub blobs: Vec<StoreWriteBlobFact>,
155}
156
157#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
158#[serde(deny_unknown_fields)]
159pub struct StoreWriteBlobFact {
160    pub table: String,
161    pub row_id: String,
162    pub row_stamp: String,
163    pub column: String,
164    pub blob: BlobRef,
165    pub plaintext_size: u64,
166    pub plaintext_hash: ObjectHash,
167    pub external_path: Option<PathBuf>,
168    pub previous: Option<StoreWriteRemoteBlob>,
169    pub audience_move: Option<StoreWriteBlobMoveDestination>,
170}
171
172#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
173#[serde(deny_unknown_fields)]
174pub struct StoreWriteRemoteBlob {
175    pub authority: coven_protocol::audience_package::PackageAudience,
176    pub stored: StoredBlobRef,
177}
178
179#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
180#[serde(rename_all = "snake_case", deny_unknown_fields)]
181pub enum StoreWriteBlobMoveDestination {
182    Local,
183    Remote {
184        audience: coven_protocol::blob::locator::RemoteAudience,
185        locator: coven_protocol::blob::locator::BlobLocator,
186        spool_path: PathBuf,
187    },
188}
189
190impl StoreWriteBlobFact {
191    pub fn identity_key(&self) -> (String, String, String, String) {
192        (
193            self.table.clone(),
194            self.row_id.clone(),
195            self.column.clone(),
196            self.row_stamp.clone(),
197        )
198    }
199}
200
201#[derive(Debug, Clone)]
202pub struct PreparedStoreWriteCommit {
203    pub audiences: PreparedAudienceObjects,
204    pub commit: ExactProtocolObject<VerifiedStoreBatchCommit>,
205    pub head: ExactProtocolObject<StoreDeviceHead>,
206}
207
208/// A candidate whose activation is blocked: the commit and head it would have
209/// activated, each named by the reference that identifies it.
210///
211/// The upload bytes are deliberately absent. A blocked candidate is only ever
212/// examined and cleaned up — its objects are deleted from storage by reference,
213/// never written again — so carrying them would be carrying what no reader
214/// reads.
215#[derive(Debug, Clone)]
216pub struct BlockedMergeCandidate {
217    pub commit: VerifiedStoreBatchCommit,
218    pub commit_bytes: Vec<u8>,
219    pub commit_object: ExactObjectRef,
220    pub head: StoreDeviceHead,
221    pub head_object: ExactObjectRef,
222}
223
224#[derive(Debug, Clone)]
225pub struct PreparedMergeAbandonmentCandidates {
226    pub candidate: BlockedMergeCandidate,
227    pub authority: BlockedMergeCandidate,
228}
229
230#[derive(Debug, Clone, PartialEq, Eq)]
231pub enum CompletePreparedStoreWriteOutcome {
232    Published,
233    AuthorExcluded {
234        device_id: coven_protocol::store_commit::StoreDeviceId,
235    },
236}
237
238#[derive(Debug, Clone, PartialEq, Eq)]
239pub struct AuthorExclusionActivationLocator {
240    exclusion: coven_protocol::store_commit::StoreDeviceExclusionRef,
241    accepted_cut: BTreeMap<coven_protocol::causal_grants::AuthorStreamId, StoreBatchCommitRef>,
242    activation_commit: StoreBatchCommitRef,
243    activation_head: coven_protocol::store_commit::StoreDeviceHeadRef,
244}
245
246impl AuthorExclusionActivationLocator {
247    pub fn verified(
248        exclusion: coven_protocol::store_commit::StoreDeviceExclusionRef,
249        accepted_cut: BTreeMap<coven_protocol::causal_grants::AuthorStreamId, StoreBatchCommitRef>,
250        activation_commit: StoreBatchCommitRef,
251        activation_head: coven_protocol::store_commit::StoreDeviceHeadRef,
252    ) -> Self {
253        Self {
254            exclusion,
255            accepted_cut,
256            activation_commit,
257            activation_head,
258        }
259    }
260
261    pub fn exclusion(&self) -> &coven_protocol::store_commit::StoreDeviceExclusionRef {
262        &self.exclusion
263    }
264
265    pub fn accepted_cut(
266        &self,
267    ) -> &BTreeMap<coven_protocol::causal_grants::AuthorStreamId, StoreBatchCommitRef> {
268        &self.accepted_cut
269    }
270
271    pub fn activation_head(&self) -> &coven_protocol::store_commit::StoreDeviceHeadRef {
272        &self.activation_head
273    }
274
275    pub fn activation_commit(&self) -> &StoreBatchCommitRef {
276        &self.activation_commit
277    }
278}
279
280#[derive(Debug, Clone)]
281pub enum TerminalCandidateAuthority {
282    AuthorExclusion(AuthorExclusionActivationLocator),
283    MembershipGrantRevocation {
284        grant_id: coven_protocol::membership::MembershipGrantId,
285        membership: coven_protocol::circle_control::StoreMembershipStateRef,
286        activation_commit: StoreBatchCommitRef,
287        activation_head: coven_protocol::store_commit::StoreDeviceHeadRef,
288    },
289    DependencyRetraction(coven_protocol::remote_object::VerifiedDependencyRetractionAuthority),
290}
291
292#[derive(Debug, Clone)]
293pub struct TerminalCandidateCleanupVerification {
294    pub authority: TerminalCandidateAuthority,
295    pub candidate: BlockedMergeCandidate,
296}
297
298#[derive(Debug)]
299pub struct InitialStoreMembershipAuthority {
300    pub head_refs: Vec<coven_protocol::membership::MembershipHeadRef>,
301}
302
303impl InitialStoreMembershipAuthority {
304    const CURSOR_STATE_KEY_PREFIX: &'static str = "membership_head_cursor/";
305
306    pub fn cursor_state_key_for_stream(
307        owner_grant: &coven_protocol::membership::MembershipGrantId,
308        stream_id: coven_protocol::membership::AuthorStreamId,
309    ) -> String {
310        format!("{}{owner_grant}/{stream_id}", Self::CURSOR_STATE_KEY_PREFIX)
311    }
312
313    fn cursor_state_key(reference: &coven_protocol::membership::MembershipHeadRef) -> String {
314        Self::cursor_state_key_for_stream(
315            &reference.coord.author_owner_grant,
316            reference.coord.stream_id,
317        )
318    }
319
320    pub(crate) fn load_on(conn: &Connection) -> Result<Self, DbError> {
321        let mut statement = conn
322            .prepare(
323                "SELECT value FROM protocol_state \
324                 WHERE substr(key, 1, length(?1)) = ?1 ORDER BY key",
325            )
326            .map_err(DbError::from)?;
327        let rows = statement
328            .query_map([Self::CURSOR_STATE_KEY_PREFIX], |row| {
329                row.get::<_, String>(0)
330            })
331            .map_err(DbError::from)?;
332        let mut head_refs = Vec::new();
333        for row in rows {
334            let value = row.map_err(DbError::from)?;
335            let reference: coven_protocol::membership::MembershipHeadRef =
336                serde_json::from_str(&value).map_err(|error| {
337                    DbError::context("membership head cursor is malformed", error)
338                })?;
339            if reference.coord.seq == 0 {
340                return Err(DbError::Message(
341                    "membership head cursor has sequence zero".to_string(),
342                ));
343            }
344            head_refs.push(reference);
345        }
346        Ok(Self { head_refs })
347    }
348
349    pub(crate) fn install_on(&self, conn: &Connection) -> Result<(), DbError> {
350        for reference in &self.head_refs {
351            let key = Self::cursor_state_key(reference);
352            if let Some(existing) = get_protocol_state_on(conn, &key)? {
353                let existing: coven_protocol::membership::MembershipHeadRef =
354                    serde_json::from_str(&existing).map_err(|error| {
355                        DbError::context("membership head cursor is malformed", error)
356                    })?;
357                if existing.coord.stream_key() != reference.coord.stream_key() {
358                    return Err(DbError::Message(
359                        "membership head cursor key names a different stream".to_string(),
360                    ));
361                }
362                if existing.coord.seq > reference.coord.seq {
363                    continue;
364                }
365                if existing.coord.seq == reference.coord.seq {
366                    if existing == *reference {
367                        continue;
368                    }
369                    return Err(DbError::Message(
370                        "membership head cursor forks at the same sequence".to_string(),
371                    ));
372                }
373            }
374            let value = serde_json::to_string(reference)
375                .map_err(|error| DbError::context("serialize membership head cursor", error))?;
376            set_protocol_state_on(conn, &key, &value)?;
377        }
378        Ok(())
379    }
380
381    #[cfg(any(test, feature = "test-utils"))]
382    pub fn cursor_state_key_for_test(
383        reference: &coven_protocol::membership::MembershipHeadRef,
384    ) -> String {
385        Self::cursor_state_key(reference)
386    }
387}
388
389#[derive(Debug, Clone, Copy, PartialEq, Eq)]
390pub enum MergeAbandonmentState {
391    None,
392    Prepared,
393    Accepted,
394    CandidateWon,
395    OtherWon,
396    AuthorExcluded,
397}
398
399#[derive(Debug, Clone)]
400pub struct OutboundStoreAck {
401    pub reference: StoreAckRef,
402    pub ack: ExactProtocolObject<StoreAck>,
403    pub circle_acknowledgements: Vec<coven_protocol::prepared_commit::CircleAckActivation>,
404    pub activation: OutboundStoreAckActivation,
405}
406
407#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
408#[serde(rename_all = "snake_case", deny_unknown_fields)]
409pub enum OutboundStoreAckActivation {
410    AwaitingCandidate,
411    Prepared(coven_protocol::prepared_commit::PreparedStoreOperationCommit),
412    Nonactivating(coven_protocol::prepared_commit::PreparedStoreOperationCommit),
413}
414
415#[derive(Debug, Clone)]
416pub struct PublishedStoreAck {
417    pub reference: StoreAckRef,
418    pub successor_slot: coven_protocol::objects::ObjectSlot,
419    /// What that acknowledgement said, so the next cycle can tell whether it
420    /// still holds. `None` on an acknowledgement installed while bootstrapping
421    /// the device, which computed no assertion of its own: the first cycle after
422    /// one of those has no basis to skip, so it acknowledges and records what it
423    /// said.
424    pub standing: Option<coven_protocol::store_commit::StandingStoreAck>,
425}
426
427/// An acknowledgement a device has activated, and the commit that activated it.
428#[derive(Debug, Clone, PartialEq, Eq)]
429pub struct ActivatedStoreAck {
430    pub reference: StoreAckRef,
431    pub activating_commit: StoreBatchCommitRef,
432}
coven_database/write_models.rs

coven_database/
write_models.rs
