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    /// Replay accepted work and the settled local prefix, leaving the unresolved
68    /// journal suffix to the recorded-edit capture owner.
69    Rebase,
70    /// The settled prefix a baseline at this cut absorbs.
71    Folded(&'a [SettledStoreWrite]),
72}
73
74/// One write of the journal prefix a baseline at some cut absorbs, and what the
75/// fold owes it.
76#[derive(Debug, Clone, PartialEq, Eq)]
77pub(crate) struct SettledStoreWrite {
78    pub ordinal: i64,
79    pub write_id: WriteId,
80    pub status: coven_protocol::write::WriteStatus,
81    pub observed: StoreWriteBase,
82    pub changeset_hash: ObjectHash,
83    pub input_hash: ObjectHash,
84}
85
86pub(crate) struct RetainedStoreWriteManifest {
87    pub ordinal: i64,
88    pub write_id: String,
89    pub status: String,
90    pub base: String,
91    pub changeset_hash: String,
92    pub prepared: Option<String>,
93    pub input_hash: ObjectHash,
94}
95
96#[derive(Clone, Copy)]
97pub enum StoreWriteRouting<'a> {
98    Unscoped,
99    MergeScoped(&'a EncryptionService),
100}
101
102impl PreparedStoreWritePartitions {
103    #[cfg(any(test, feature = "test-utils"))]
104    pub fn iter(&self) -> impl Iterator<Item = &gate::AudiencePartition> {
105        self.store
106            .iter()
107            .chain(self.circles.iter())
108            .chain(self.local.iter())
109    }
110}
111
112#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
113#[serde(deny_unknown_fields)]
114pub struct StoreWriteBase {
115    /// The complete accepted Store frontier visible to the host transaction.
116    /// Publication removes its own stream and represents that position through
117    /// the signed predecessor; replay uses every stream to place local effects.
118    pub dependencies: BTreeMap<String, StoreBatchCommitRef>,
119}
120
121/// A write's current replay base and actual effect for inverse discard.
122/// Publication keeps the captured partitions and timestamps. Blob facts retain
123/// exact sources, including verified uploads from retired candidates.
124#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
125#[serde(deny_unknown_fields)]
126pub(crate) struct RebasedStoreWrite {
127    pub base: StoreWriteBase,
128    pub publication_base: coven_protocol::store_commit::StorePublicationBase,
129    pub changeset_hash: ObjectHash,
130    pub blob_facts: StoreWriteBlobFacts,
131}
132
133#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
134#[serde(deny_unknown_fields)]
135pub struct StoreWriteBlobFacts {
136    pub blobs: Vec<StoreWriteBlobFact>,
137}
138
139#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
140#[serde(deny_unknown_fields)]
141pub struct StoreWriteBlobFact {
142    pub table: String,
143    pub row_id: String,
144    pub row_stamp: String,
145    pub column: String,
146    pub blob: BlobRef,
147    pub plaintext_size: u64,
148    pub plaintext_hash: ObjectHash,
149    pub external_path: Option<PathBuf>,
150    pub previous: Option<StoreWriteRemoteBlob>,
151    pub audience_move: Option<StoreWriteBlobMoveMaterialization>,
152}
153
154#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
155#[serde(deny_unknown_fields)]
156pub struct StoreWriteRemoteBlob {
157    pub authority: coven_protocol::audience_package::PackageAudience,
158    pub stored: StoredBlobRef,
159}
160
161#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
162#[serde(rename_all = "snake_case", deny_unknown_fields)]
163pub enum StoreWriteBlobMoveMaterialization {
164    Local,
165    /// Exact source bytes in PayloadStore, addressed by the containing fact's
166    /// plaintext hash. Destination signing and encryption belong to publication.
167    Payload,
168}
169
170impl StoreWriteBlobFacts {
171    pub(crate) fn captured_payloads(&self) -> impl Iterator<Item = ObjectHash> + '_ {
172        self.blobs.iter().filter_map(|fact| {
173            (fact.audience_move == Some(StoreWriteBlobMoveMaterialization::Payload))
174                .then_some(fact.plaintext_hash)
175        })
176    }
177}
178
179impl StoreWriteBlobFact {
180    pub fn identity_key(&self) -> (String, String, String, String) {
181        (
182            self.table.clone(),
183            self.row_id.clone(),
184            self.column.clone(),
185            self.row_stamp.clone(),
186        )
187    }
188}
189
190#[derive(Debug, Clone)]
191pub struct PreparedStoreWriteCommit {
192    pub audiences: PreparedAudienceObjects,
193    pub commit: ExactProtocolObject<VerifiedStoreBatchCommit>,
194    pub publication: coven_protocol::prepared_commit::PreparedStorePublication,
195}
196
197#[derive(Debug)]
198pub struct InitialStoreMembershipAuthority {
199    pub head_refs: Vec<coven_protocol::membership::MembershipHeadRef>,
200}
201
202impl InitialStoreMembershipAuthority {
203    const CURSOR_STATE_KEY_PREFIX: &'static str = "membership_head_cursor/";
204
205    pub fn cursor_state_key_for_stream(
206        owner_grant: &coven_protocol::membership::MembershipGrantId,
207        stream_id: coven_protocol::membership::AuthorStreamId,
208    ) -> String {
209        format!("{}{owner_grant}/{stream_id}", Self::CURSOR_STATE_KEY_PREFIX)
210    }
211
212    fn cursor_state_key(reference: &coven_protocol::membership::MembershipHeadRef) -> String {
213        Self::cursor_state_key_for_stream(
214            &reference.coord.author_owner_grant,
215            reference.coord.stream_id,
216        )
217    }
218
219    pub(crate) fn load_on(conn: &Connection) -> Result<Self, DbError> {
220        let mut statement = conn
221            .prepare(
222                "SELECT value FROM protocol_state \
223                 WHERE substr(key, 1, length(?1)) = ?1 ORDER BY key",
224            )
225            .map_err(DbError::from)?;
226        let rows = statement
227            .query_map([Self::CURSOR_STATE_KEY_PREFIX], |row| {
228                row.get::<_, String>(0)
229            })
230            .map_err(DbError::from)?;
231        let mut head_refs = Vec::new();
232        for row in rows {
233            let value = row.map_err(DbError::from)?;
234            let reference: coven_protocol::membership::MembershipHeadRef =
235                serde_json::from_str(&value).map_err(|error| {
236                    DbError::context("membership head cursor is malformed", error)
237                })?;
238            if reference.coord.seq == 0 {
239                return Err(DbError::Message(
240                    "membership head cursor has sequence zero".to_string(),
241                ));
242            }
243            head_refs.push(reference);
244        }
245        Ok(Self { head_refs })
246    }
247
248    pub(crate) fn install_on(&self, conn: &Connection) -> Result<(), DbError> {
249        for reference in &self.head_refs {
250            let key = Self::cursor_state_key(reference);
251            if let Some(existing) = get_protocol_state_on(conn, &key)? {
252                let existing: coven_protocol::membership::MembershipHeadRef =
253                    serde_json::from_str(&existing).map_err(|error| {
254                        DbError::context("membership head cursor is malformed", error)
255                    })?;
256                if existing.coord.stream_key() != reference.coord.stream_key() {
257                    return Err(DbError::Message(
258                        "membership head cursor key names a different stream".to_string(),
259                    ));
260                }
261                if existing.coord.seq > reference.coord.seq {
262                    continue;
263                }
264                if existing.coord.seq == reference.coord.seq {
265                    if existing == *reference {
266                        continue;
267                    }
268                    return Err(DbError::Message(
269                        "membership head cursor forks at the same sequence".to_string(),
270                    ));
271                }
272            }
273            let value = serde_json::to_string(reference)
274                .map_err(|error| DbError::context("serialize membership head cursor", error))?;
275            set_protocol_state_on(conn, &key, &value)?;
276        }
277        Ok(())
278    }
279
280    #[cfg(any(test, feature = "test-utils"))]
281    pub fn cursor_state_key_for_test(
282        reference: &coven_protocol::membership::MembershipHeadRef,
283    ) -> String {
284        Self::cursor_state_key(reference)
285    }
286}
287
288#[derive(Debug, Clone)]
289pub struct OutboundStoreAck {
290    pub reference: StoreAckRef,
291    pub ack: ExactProtocolObject<StoreAck>,
292    pub circle_acknowledgements: Vec<coven_protocol::prepared_commit::CircleAckActivation>,
293    pub activation: OutboundStoreAckActivation,
294}
295
296#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
297#[serde(rename_all = "snake_case", deny_unknown_fields)]
298pub enum OutboundStoreAckActivation {
299    AwaitingCandidate,
300    Created,
301    Prepared(coven_protocol::prepared_commit::PreparedStoreOperationCommit),
302}
303
304#[derive(Debug, Clone)]
305pub struct PublishedStoreAck {
306    pub reference: StoreAckRef,
307    pub successor_slot: coven_protocol::objects::ObjectSlot,
308    /// What that acknowledgement said, so the next cycle can tell whether it
309    /// still holds. `None` on an acknowledgement installed while bootstrapping
310    /// the device, which computed no assertion of its own: the first cycle after
311    /// one of those has no basis to skip, so it acknowledges and records what it
312    /// said.
313    pub standing: Option<coven_protocol::store_commit::StandingStoreAck>,
314}
315
316/// An acknowledgement a device has activated, and the commit that activated it.
317#[derive(Debug, Clone, PartialEq, Eq)]
318pub struct ActivatedStoreAck {
319    pub reference: StoreAckRef,
320    pub activating_commit: StoreBatchCommitRef,
321}
coven_database/write_models.rs

coven_database/
write_models.rs
