coven_protocol/
prepared_commit.rs

1//! A signed Store operation commit prepared for publication: the exact commit
2//! bytes, their reference, and the remote-object records a candidate or
3//! activation derives from them.
4
5use crate::membership_mutation::PreparedMembershipPublication;
6use crate::objects::{ExactObjectRef, ExactObjectVersion, PreparedExactObject, StoreObjectError};
7use crate::store_commit::{
8    ActivatedStoreDeviceRegistration, SnapshotMeta, StoreBatchCommit, StoreBatchCommitRef,
9    StoreControl, StoreCurrentPublicationRecord, StorePublicationEntry, StorePublicationPayload,
10    StorePublicationRef, StoreSnapshotRef,
11};
12
13/// A prepared commit whose parts contradict each other or cannot form valid
14/// remote-object records. Workflow errors wrap it at the operation boundary.
15#[derive(Debug, thiserror::Error)]
16pub enum PreparedCommitError {
17    #[error("invalid prepared Store operation: {0}")]
18    Invariant(String),
19    #[error("prepared Store operation storage: {0}")]
20    Storage(#[from] crate::objects::StorageError),
21    #[error("prepared Store operation object: {0}")]
22    StoreObject(#[from] StoreObjectError),
23    #[error("prepared Store protocol: {0}")]
24    Protocol(#[from] crate::store_commit::StoreProtocolError),
25    #[error("prepared membership transition: {0}")]
26    Membership(#[from] crate::membership_mutation::MembershipPreparationError),
27    #[error("{operation}: {source}")]
28    Json {
29        operation: &'static str,
30        #[source]
31        source: serde_json::Error,
32    },
33    #[error("prepared Store remote object: {0}")]
34    RemoteObject(#[from] crate::remote_object::RemoteObjectRecordError),
35}
36
37/// A signed commit and the exact object it is published as.
38///
39/// `reference.object` names that object; the commit's bytes are what `commit`
40/// serializes to, so they are not carried beside it. Whoever uploads rebuilds
41/// them through [`PreparedExactObject::new`], which re-checks them against the
42/// reference on the way out.
43#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
44#[serde(deny_unknown_fields)]
45pub struct PreparedStoreOperationCommon {
46    pub commit: StoreBatchCommit,
47    pub reference: StoreBatchCommitRef,
48    pub registration_activation: Option<ActivatedStoreDeviceRegistration>,
49}
50
51/// One immutable Store publication entry and the conditional replacement that
52/// can accept it. The observed record and provider version remain together so
53/// a retry cannot apply the replacement against another boundary.
54#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
55#[serde(deny_unknown_fields)]
56pub struct PreparedStorePublication {
57    pub previous: StoreCurrentPublicationRecord,
58    pub previous_version: ExactObjectVersion,
59    pub entry: StorePublicationEntry,
60    pub entry_object: ExactObjectRef,
61    pub replacement: StoreCurrentPublicationRecord,
62}
63
64impl PreparedStorePublication {
65    pub fn reference(&self) -> Result<StorePublicationRef, PreparedCommitError> {
66        StorePublicationRef::from_entry(&self.entry, self.entry_object.clone())
67            .map_err(PreparedCommitError::from)
68    }
69
70    pub fn prepared_entry(&self) -> Result<PreparedExactObject, PreparedCommitError> {
71        PreparedExactObject::new(self.entry_object.clone(), self.entry.to_bytes())
72            .map_err(PreparedCommitError::from)
73    }
74
75    pub fn verify_commit(
76        &self,
77        commit: &crate::store_commit::VerifiedStoreBatchCommit,
78    ) -> Result<(), PreparedCommitError> {
79        let reference = self.reference()?;
80        let signing_pubkey = &commit.author().device_signing_pubkey;
81        StorePublicationEntry::parse_at(
82            &self.entry.to_bytes(),
83            commit.store_root_hash(),
84            &reference,
85            signing_pubkey,
86        )?;
87        self.replacement.verify_commit_transition(
88            &self.previous,
89            &self.entry,
90            &reference,
91            commit,
92            signing_pubkey,
93        )?;
94        Ok(())
95    }
96
97    pub fn validate_commit_shape(
98        &self,
99        commit: &StoreBatchCommit,
100        reference: &StoreBatchCommitRef,
101    ) -> Result<(), PreparedCommitError> {
102        let publication = self.reference()?;
103        if self.entry.predecessor.as_ref() != self.previous.accepted()
104            || self.entry.previous_state_hash != self.previous.state_hash()
105            || self.replacement.accepted() != Some(&publication)
106            || self.replacement.store_root_hash != self.previous.store_root_hash
107            || self.entry.store_root_hash != commit.store_root_hash
108            || self.entry.author_registration != commit.author_registration
109            || !matches!(&self.entry.payload, StorePublicationPayload::Commit(published) if published == reference)
110        {
111            return Err(PreparedCommitError::Invariant(
112                "prepared Store publication differs from its commit or predecessor".to_string(),
113            ));
114        }
115        Ok(())
116    }
117
118    pub fn validate_snapshot_shape(
119        &self,
120        snapshot: &SnapshotMeta,
121        reference: &StoreSnapshotRef,
122    ) -> Result<(), PreparedCommitError> {
123        let publication = self.reference()?;
124        if self.entry.predecessor.as_ref() != self.previous.accepted()
125            || self.entry.previous_state_hash != self.previous.state_hash()
126            || self.replacement.accepted() != Some(&publication)
127            || self.replacement.store_root_hash != self.previous.store_root_hash
128            || snapshot.publication_predecessor != self.previous
129            || self.entry.store_root_hash != snapshot.store_root_hash
130            || self.entry.author_registration != snapshot.author_registration
131            || !matches!(&self.entry.payload, StorePublicationPayload::Snapshot(published) if published == reference)
132        {
133            return Err(PreparedCommitError::Invariant(
134                "prepared Store publication differs from its snapshot or predecessor".to_string(),
135            ));
136        }
137        Ok(())
138    }
139}
140
141impl PreparedStoreOperationCommon {
142    /// The commit prepared for upload: its canonical bytes, re-derived from the
143    /// value, under the exact reference the operation names.
144    pub fn prepared_commit(&self) -> Result<PreparedExactObject, PreparedCommitError> {
145        PreparedExactObject::new(self.reference.object.clone(), self.commit.to_bytes())
146            .map_err(PreparedCommitError::from)
147    }
148}
149
150#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
151#[serde(deny_unknown_fields)]
152pub struct PreparedStoreOperationCommit {
153    pub common: PreparedStoreOperationCommon,
154    pub publication: PreparedStorePublication,
155    pub history_evidence: super::store_commit::RetainedMergeCommitEvidence,
156}
157
158impl std::ops::Deref for PreparedStoreOperationCommit {
159    type Target = PreparedStoreOperationCommon;
160
161    fn deref(&self) -> &Self::Target {
162        &self.common
163    }
164}
165
166impl std::ops::DerefMut for PreparedStoreOperationCommit {
167    fn deref_mut(&mut self) -> &mut Self::Target {
168        &mut self.common
169    }
170}
171
172impl PreparedStoreOperationCommit {
173    pub(crate) fn candidate_remote_object(
174        &self,
175    ) -> Result<crate::remote_object::ClosedRemoteObject, PreparedCommitError> {
176        let commit_bytes = self.commit.to_bytes();
177        // Store commits are signed plaintext: their canonical bytes are also
178        // their stored bytes. The publication attempt owns its own entry.
179        crate::remote_object::RemoteObjectRecord::candidate_commit(
180            self.reference.clone(),
181            &commit_bytes,
182            &commit_bytes,
183        )
184        .map_err(PreparedCommitError::from)
185    }
186
187    pub fn prepared_membership_publication(
188        &self,
189    ) -> Result<PreparedMembershipPublication, PreparedCommitError> {
190        let proof = self
191            .history_evidence
192            .membership_proof
193            .as_ref()
194            .ok_or_else(|| {
195                PreparedCommitError::Invariant(
196                    "Store control lacks its prepared authority proof".into(),
197                )
198            })?;
199        let publication = PreparedMembershipPublication {
200            entry: proof.entry_value.clone(),
201            entry_ref: proof.entry.clone(),
202            head: proof.head_value.clone(),
203            head_ref: proof.head.clone(),
204        };
205        self.validate_closed_shape()?;
206        publication.validate()?;
207        Ok(publication)
208    }
209
210    pub(crate) fn retained_control_remote_objects(
211        &self,
212        authorities: Vec<crate::remote_object::ClosedRemoteObject>,
213    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
214        let publication = self.prepared_membership_publication()?;
215        self.close_merge_membership_remote_objects(&publication, &[], authorities)
216    }
217
218    pub fn merge_membership_activation_remote_objects(
219        &self,
220        wraps: &[super::wrapped_store_key::PreparedWrappedStoreKey],
221    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
222        let publication = self.prepared_membership_publication()?;
223        let expected_wraps: &[super::wrapped_store_key::WrappedStoreKeyRef] =
224            match &publication.entry.change {
225                super::membership::StoreAuthorityChange::RemoveMember { wrapped_keys, .. } => {
226                    wrapped_keys
227                }
228                super::membership::StoreAuthorityChange::SetMember { wrapped_key, .. } => {
229                    std::slice::from_ref(wrapped_key)
230                }
231                _ => {
232                    return Err(PreparedCommitError::Invariant(
233                        "Merge membership mutation graph contains another change".to_string(),
234                    ));
235                }
236            };
237        if expected_wraps.len() != wraps.len()
238            || expected_wraps
239                .iter()
240                .zip(wraps)
241                .any(|(reference, prepared)| reference != &prepared.reference)
242        {
243            return Err(PreparedCommitError::Invariant(
244                "Merge membership mutation wraps differ from its exact entry".to_string(),
245            ));
246        }
247        self.close_merge_membership_remote_objects(&publication, wraps, Vec::new())
248    }
249
250    fn close_merge_membership_remote_objects(
251        &self,
252        publication: &PreparedMembershipPublication,
253        wraps: &[super::wrapped_store_key::PreparedWrappedStoreKey],
254        authorities: Vec<crate::remote_object::ClosedRemoteObject>,
255    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
256        let family = self.commit.candidate_family();
257        let mut objects = vec![self.candidate_remote_object()?];
258        objects.extend(publication.candidate_remote_objects(&self.commit, &self.reference)?);
259        for prepared in wraps {
260            let value = prepared.validate().map_err(PreparedCommitError::from)?;
261            let canonical =
262                serde_json::to_vec(&value).map_err(|source| PreparedCommitError::Json {
263                    operation: "serialize Merge membership candidate wrap",
264                    source,
265                })?;
266            objects.push(
267                crate::remote_object::RemoteObjectRecord::candidate_exclusive_merge_membership_wrapped_store_key(
268                    family,
269                    prepared.reference.clone(),
270                    &canonical,
271                    prepared.object.stored_bytes(),
272                    self.reference.clone(),
273                )
274                .map_err(PreparedCommitError::from)?,
275            );
276        }
277        objects.extend(authorities);
278        let mut unique = std::collections::BTreeSet::new();
279        if objects
280            .iter()
281            .any(|object| !unique.insert(object.record().object_id()))
282        {
283            return Err(PreparedCommitError::Invariant(
284                "Merge membership authority graph repeats an exact object".to_string(),
285            ));
286        }
287        Ok(objects)
288    }
289
290    pub fn validate_closed_shape(&self) -> Result<(), PreparedCommitError> {
291        self.reference.verify_commit(&self.commit)?;
292        self.reference.object.verify(&self.commit.to_bytes())?;
293        self.publication
294            .validate_commit_shape(&self.commit, &self.reference)?;
295        self.history_evidence
296            .validate_for(&self.reference, &self.commit)?;
297        Ok(())
298    }
299
300    pub(crate) fn has_same_durable_activation_as(&self, other: &Self) -> bool {
301        self.reference == other.reference
302            && self.commit.to_bytes() == other.commit.to_bytes()
303            && self.registration_activation == other.registration_activation
304            && self.publication == other.publication
305            && self.history_evidence == other.history_evidence
306    }
307
308    pub fn acknowledgement_remote_objects(
309        &self,
310        acknowledgement: &crate::objects::ExactProtocolObject<super::store_commit::StoreAck>,
311    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
312        let reference = self.commit.acknowledgement().ok_or_else(|| {
313            PreparedCommitError::Invariant(
314                "prepared acknowledgement operation has no exact acknowledgement ref".to_string(),
315            )
316        })?;
317        if &reference.object != acknowledgement.prepared.reference()
318            || reference.ack_hash != acknowledgement.value.ack_hash()
319            || acknowledgement.value.to_bytes() != acknowledgement.bytes
320        {
321            return Err(PreparedCommitError::Invariant(
322                "prepared acknowledgement operation differs from its exact acknowledgement object"
323                    .to_string(),
324            ));
325        }
326        let authority =
327            crate::remote_object::RemoteObjectRecord::candidate_activated_store_acknowledgement(
328                reference.clone(),
329                &acknowledgement.bytes,
330                acknowledgement.prepared.stored_bytes(),
331                self.reference.clone(),
332            )
333            .map_err(PreparedCommitError::from)?;
334        let mut authorities = vec![authority];
335        let retained = self
336            .history_evidence
337            .acknowledgement
338            .as_ref()
339            .ok_or_else(|| {
340                PreparedCommitError::Invariant(
341                    "prepared acknowledgement omits its retained proof".into(),
342                )
343            })?;
344        retained.validate_predecessors()?;
345        for (reference, value) in &retained.predecessors {
346            let bytes = value.to_bytes();
347            authorities.push(
348                crate::remote_object::RemoteObjectRecord::candidate_activated_store_acknowledgement(
349                    reference.clone(), &bytes, &bytes, self.reference.clone(),
350                )?,
351            );
352        }
353        self.retained_authority_remote_objects(authorities)
354    }
355
356    pub fn circle_acknowledgement_remote_objects(
357        &self,
358        acknowledgement: &crate::objects::ExactProtocolObject<super::store_commit::CircleAck>,
359    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
360        let reference = self
361            .commit
362            .circle_acknowledgements()
363            .iter()
364            .find(|reference| &reference.object == acknowledgement.prepared.reference())
365            .ok_or_else(|| {
366                PreparedCommitError::Invariant(
367                    "prepared activation does not name its Circle acknowledgement object"
368                        .to_string(),
369                )
370            })?;
371        if reference.circle_id != acknowledgement.value.circle_id
372            || reference.ack_hash != acknowledgement.value.ack_hash()
373            || acknowledgement.value.to_bytes() != acknowledgement.bytes
374        {
375            return Err(PreparedCommitError::Invariant(
376                "prepared Circle acknowledgement differs from its exact acknowledgement object"
377                    .to_string(),
378            ));
379        }
380        let authority =
381            crate::remote_object::RemoteObjectRecord::candidate_activated_circle_acknowledgement(
382                reference.clone(),
383                &acknowledgement.bytes,
384                acknowledgement.prepared.stored_bytes(),
385                self.reference.clone(),
386            )
387            .map_err(PreparedCommitError::from)?;
388        self.retained_authority_remote_objects(vec![authority])
389    }
390
391    pub fn retained_authority_remote_objects(
392        &self,
393        authorities: Vec<crate::remote_object::ClosedRemoteObject>,
394    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
395        if authorities.is_empty() {
396            return Err(PreparedCommitError::Invariant(
397                "Store operation has no retained authority objects".to_string(),
398            ));
399        }
400        let mut authority_ids = std::collections::BTreeSet::new();
401        for authority in &authorities {
402            if !matches!(authority.record(), crate::remote_object::RemoteObjectRecord::RetainedAuthority(record)
403                if matches!(&record.state, crate::remote_object::RetainedAuthorityObjectState::Prepared { ownership }
404                    if ownership.pending == std::collections::BTreeSet::from([self.reference.clone()])))
405            {
406                return Err(PreparedCommitError::Invariant(
407                    "Store operation retained authority has different candidate ownership"
408                        .to_string(),
409                ));
410            }
411            if !authority_ids.insert(authority.record().object_id()) {
412                return Err(PreparedCommitError::Invariant(
413                    "Store operation repeats a retained authority object".to_string(),
414                ));
415            }
416        }
417        let mut objects = vec![self.candidate_remote_object()?];
418        objects.extend(authorities);
419        Ok(objects)
420    }
421
422    pub fn attach_merge_membership_proof(
423        &mut self,
424        publication: &PreparedMembershipPublication,
425    ) -> Result<(), PreparedCommitError> {
426        publication.validate().map_err(PreparedCommitError::from)?;
427        let reference = self.common.reference.clone();
428        let commit = self.common.commit.clone();
429        let Some(StoreControl { transition }) = commit.control() else {
430            return Err(PreparedCommitError::Invariant(
431                "Merge membership proof accompanies another Store control".to_string(),
432            ));
433        };
434        if !transition.matches_head(&publication.head, &publication.head_ref)
435            || publication.entry_ref != transition.body.entry
436        {
437            return Err(PreparedCommitError::Invariant(
438                "Merge membership proof differs from its signed Store transition".to_string(),
439            ));
440        }
441        self.history_evidence.membership_proof = Some(Box::new(
442            super::store_commit::RetainedMergeMembershipProof {
443                commit: reference,
444                commit_value: commit,
445                entry: publication.entry_ref.clone(),
446                entry_value: publication.entry.clone(),
447                head: publication.head_ref.clone(),
448                head_value: publication.head.clone(),
449            },
450        ));
451        self.validate_closed_shape()?;
452        Ok(())
453    }
454}
455
456/// One Circle acknowledgement object riding an activating Store commit: its
457/// exact reference (named in the signed commit body) and the exact object the
458/// commit uploads and takes ownership of.
459#[derive(Debug, Clone)]
460pub struct CircleAckActivation {
461    pub reference: crate::store_commit::CircleAckRef,
462    pub ack: crate::objects::ExactProtocolObject<crate::store_commit::CircleAck>,
463}
coven_protocol/prepared_commit.rs

coven_protocol/
prepared_commit.rs
