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, PreparedMembershipTransition};
6use crate::objects::{ExactObjectRef, PreparedExactObject, StoreObjectError};
7use crate::store_commit::{
8    ActivatedStoreDeviceRegistration, StoreBatchCommit, StoreBatchCommitRef, StoreControl,
9    StoreDeviceHead, StoreDeviceHeadRef,
10};
11
12/// A prepared commit whose parts contradict each other or cannot form valid
13/// remote-object records. Workflow errors wrap it at the operation boundary.
14#[derive(Debug, thiserror::Error)]
15pub enum PreparedCommitError {
16    #[error("invalid prepared Store operation: {0}")]
17    Invariant(String),
18    #[error("prepared Store operation storage: {0}")]
19    Storage(#[from] crate::objects::StorageError),
20    #[error("prepared Store operation object: {0}")]
21    StoreObject(#[from] StoreObjectError),
22    #[error("prepared Store protocol: {0}")]
23    Protocol(#[from] crate::store_commit::StoreProtocolError),
24    #[error("prepared membership transition: {0}")]
25    Membership(#[from] crate::membership_mutation::MembershipPreparationError),
26    #[error("{operation}: {source}")]
27    Json {
28        operation: &'static str,
29        #[source]
30        source: serde_json::Error,
31    },
32    #[error("prepared Store remote object: {0}")]
33    RemoteObject(#[from] crate::remote_object::RemoteObjectRecordError),
34}
35
36/// A signed commit and the exact object it is published as.
37///
38/// `reference.object` names that object; the commit's bytes are what `commit`
39/// serializes to, so they are not carried beside it. Whoever uploads rebuilds
40/// them through [`PreparedExactObject::new`], which re-checks them against the
41/// reference on the way out.
42#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
43#[serde(deny_unknown_fields)]
44pub struct PreparedStoreOperationCommon {
45    pub commit: StoreBatchCommit,
46    pub reference: StoreBatchCommitRef,
47    pub registration_activation: Option<ActivatedStoreDeviceRegistration>,
48}
49
50impl PreparedStoreOperationCommon {
51    /// The commit prepared for upload: its canonical bytes, re-derived from the
52    /// value, under the exact reference the operation names.
53    pub fn prepared_commit(&self) -> Result<PreparedExactObject, PreparedCommitError> {
54        PreparedExactObject::new(self.reference.object.clone(), self.commit.to_bytes())
55            .map_err(PreparedCommitError::from)
56    }
57}
58
59#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
60#[serde(deny_unknown_fields)]
61pub struct PreparedStoreOperationCommit {
62    pub common: PreparedStoreOperationCommon,
63    pub head: StoreDeviceHead,
64    pub head_object: ExactObjectRef,
65    pub history_evidence: super::store_commit::RetainedMergeCommitEvidence,
66}
67
68impl std::ops::Deref for PreparedStoreOperationCommit {
69    type Target = PreparedStoreOperationCommon;
70
71    fn deref(&self) -> &Self::Target {
72        &self.common
73    }
74}
75
76impl std::ops::DerefMut for PreparedStoreOperationCommit {
77    fn deref_mut(&mut self) -> &mut Self::Target {
78        &mut self.common
79    }
80}
81
82impl PreparedStoreOperationCommit {
83    fn candidate_remote_objects(
84        &self,
85    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
86        let commit_bytes = self.commit.to_bytes();
87        let head_bytes = self.head.to_bytes();
88        // A Store commit and a Store head are signed plaintext: what goes to
89        // storage is the canonical value, so both arguments are the same bytes.
90        let mut objects = vec![crate::remote_object::RemoteObjectRecord::candidate_commit(
91            self.reference.clone(),
92            &commit_bytes,
93            &commit_bytes,
94        )
95        .map_err(PreparedCommitError::from)?];
96        objects.push(
97            crate::remote_object::RemoteObjectRecord::candidate_activated_store_head(
98                self.head_ref(),
99                &head_bytes,
100                &head_bytes,
101                self.reference.clone(),
102            )
103            .map_err(PreparedCommitError::from)?,
104        );
105        Ok(objects)
106    }
107
108    /// Validate the frame every Merge membership-activation candidate shares:
109    /// a closed commit, valid transition and publication, the commit's control
110    /// naming the transition, and the published head activating this candidate.
111    fn validate_merge_membership_activation(
112        &self,
113        transition: &PreparedMembershipTransition,
114        publication: &PreparedMembershipPublication,
115    ) -> Result<(), PreparedCommitError> {
116        self.validate_closed_shape()?;
117        transition.validate().map_err(PreparedCommitError::from)?;
118        publication.validate().map_err(PreparedCommitError::from)?;
119        if self.commit.control()
120            != Some(&StoreControl {
121                transition: transition.transition.clone(),
122            })
123            || !transition
124                .transition
125                .matches_head(&publication.head, &publication.head_ref)
126            || !matches!(
127                &publication.head.activation,
128                super::membership::MembershipHeadActivation::StoreCommit { commit }
129                    if commit == &self.reference
130            )
131        {
132            return Err(PreparedCommitError::Invariant(
133                "Merge membership authority graph differs from its activating Store candidate"
134                    .to_string(),
135            ));
136        }
137        Ok(())
138    }
139
140    pub fn merge_membership_activation_remote_objects(
141        &self,
142        transition: &PreparedMembershipTransition,
143        publication: &PreparedMembershipPublication,
144        wraps: &[super::wrapped_store_key::PreparedWrappedStoreKey],
145    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
146        self.validate_merge_membership_activation(transition, publication)?;
147        let expected_wraps = match &transition.entry.change {
148            super::membership::MembershipChange::RemoveMember { wrapped_keys, .. } => wrapped_keys,
149            _ => {
150                return Err(PreparedCommitError::Invariant(
151                    "Merge membership removal graph contains another change".to_string(),
152                ))
153            }
154        };
155        if expected_wraps.len() != wraps.len()
156            || expected_wraps
157                .iter()
158                .zip(wraps)
159                .any(|(reference, prepared)| reference != &prepared.reference)
160        {
161            return Err(PreparedCommitError::Invariant(
162                "Merge membership removal wraps differ from its exact entry".to_string(),
163            ));
164        }
165        self.close_merge_membership_remote_objects(transition, publication, wraps, Vec::new())
166    }
167
168    pub fn merge_membership_resolution_remote_objects(
169        &self,
170        transition: &PreparedMembershipTransition,
171        publication: &PreparedMembershipPublication,
172        resolution: &super::membership::StoreMembershipConflictResolution,
173        reference: &super::membership::StoreMembershipConflictResolutionRef,
174    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
175        self.validate_merge_membership_activation(transition, publication)?;
176        let resolution_bytes =
177            serde_json::to_vec(resolution).map_err(|source| PreparedCommitError::Json {
178                operation: "serialize Store membership resolution",
179                source,
180            })?;
181        if !matches!(
182            &transition.entry.change,
183            super::membership::MembershipChange::ResolutionActivation {
184                resolution: introduced,
185            } if introduced == reference
186        ) || reference.object.verify(&resolution_bytes).is_err()
187            || reference.resolution_hash != resolution.resolution_hash()
188            || reference.conflict_hash != resolution.conflict_hash
189            || reference.resolver_pubkey != resolution.resolver_pubkey
190        {
191            return Err(PreparedCommitError::Invariant(
192                "Merge membership resolution graph differs from its activating Store candidate"
193                    .to_string(),
194            ));
195        }
196        let authority =
197            crate::remote_object::RemoteObjectRecord::candidate_activated_store_membership_resolution(
198                reference.clone(),
199                &resolution_bytes,
200                &resolution_bytes,
201                self.reference.clone(),
202            )
203            .map_err(PreparedCommitError::from)?;
204        self.close_merge_membership_remote_objects(transition, publication, &[], vec![authority])
205    }
206
207    pub fn merge_owner_promotion_remote_objects(
208        &self,
209        transition: &PreparedMembershipTransition,
210        publication: &PreparedMembershipPublication,
211        wrapped_key: &super::wrapped_store_key::PreparedWrappedStoreKey,
212    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
213        self.validate_merge_membership_activation(transition, publication)?;
214        if !matches!(
215            &transition.entry.change,
216            super::membership::MembershipChange::SetMember { wrapped_key: expected, role: super::membership::StoreMembershipRoleGrant::Owner { .. }, .. }
217                if expected == &wrapped_key.reference
218        ) {
219            return Err(PreparedCommitError::Invariant(
220                "Merge Owner-promotion graph differs from its activating Store candidate"
221                    .to_string(),
222            ));
223        }
224        self.close_merge_membership_remote_objects(
225            transition,
226            publication,
227            std::slice::from_ref(wrapped_key),
228            Vec::new(),
229        )
230    }
231
232    fn close_merge_membership_remote_objects(
233        &self,
234        transition: &PreparedMembershipTransition,
235        publication: &PreparedMembershipPublication,
236        wraps: &[super::wrapped_store_key::PreparedWrappedStoreKey],
237        authorities: Vec<crate::remote_object::ClosedRemoteObject>,
238    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
239        let family = self.commit.candidate_family();
240        let mut objects = self.candidate_remote_objects()?;
241        let entry_bytes =
242            serde_json::to_vec(&transition.entry).map_err(|source| PreparedCommitError::Json {
243                operation: "serialize Merge membership candidate entry",
244                source,
245            })?;
246        let head_bytes =
247            serde_json::to_vec(&publication.head).map_err(|source| PreparedCommitError::Json {
248                operation: "serialize Merge membership candidate head",
249                source,
250            })?;
251        // Membership entries and heads are signed plaintext, so the canonical
252        // value is also what goes to storage.
253        objects.push(
254            crate::remote_object::RemoteObjectRecord::candidate_exclusive_merge_membership_entry(
255                family,
256                transition.entry_ref.clone(),
257                &entry_bytes,
258                &entry_bytes,
259                self.reference.clone(),
260            )
261            .map_err(PreparedCommitError::from)?,
262        );
263        objects.push(
264            crate::remote_object::RemoteObjectRecord::candidate_exclusive_merge_membership_head(
265                family,
266                publication.head_ref.clone(),
267                &head_bytes,
268                &head_bytes,
269                self.reference.clone(),
270            )
271            .map_err(PreparedCommitError::from)?,
272        );
273        for prepared in wraps {
274            let value = prepared.validate().map_err(PreparedCommitError::from)?;
275            let canonical =
276                serde_json::to_vec(&value).map_err(|source| PreparedCommitError::Json {
277                    operation: "serialize Merge membership candidate wrap",
278                    source,
279                })?;
280            objects.push(
281                crate::remote_object::RemoteObjectRecord::candidate_exclusive_merge_membership_wrapped_store_key(
282                    family,
283                    prepared.reference.clone(),
284                    &canonical,
285                    prepared.object.stored_bytes(),
286                    self.reference.clone(),
287                )
288                .map_err(PreparedCommitError::from)?,
289            );
290        }
291        objects.extend(authorities);
292        let mut unique = std::collections::BTreeSet::new();
293        if objects
294            .iter()
295            .any(|object| !unique.insert(object.record().object_id()))
296        {
297            return Err(PreparedCommitError::Invariant(
298                "Merge membership authority graph repeats an exact object".to_string(),
299            ));
300        }
301        Ok(objects)
302    }
303
304    pub fn validate_closed_shape(&self) -> Result<(), PreparedCommitError> {
305        self.reference.verify_commit(&self.commit)?;
306        self.reference.object.verify(&self.commit.to_bytes())?;
307        if self.head.commit != self.reference {
308            return Err(PreparedCommitError::Invariant(
309                "prepared Store operation head names another commit".to_string(),
310            ));
311        }
312        self.head_object.verify(&self.head.to_bytes())?;
313        self.history_evidence
314            .validate_for(&self.reference, &self.commit)?;
315        Ok(())
316    }
317
318    pub(crate) fn has_same_durable_activation_as(&self, other: &Self) -> bool {
319        self.reference == other.reference
320            && self.commit.to_bytes() == other.commit.to_bytes()
321            && self.registration_activation == other.registration_activation
322            && self.head.to_bytes() == other.head.to_bytes()
323            && self.head_object == other.head_object
324            && self.history_evidence == other.history_evidence
325    }
326
327    /// The activation head prepared for upload: its canonical bytes, re-derived
328    /// from the value, under the exact object the operation names.
329    pub fn prepared_head(&self) -> Result<PreparedExactObject, PreparedCommitError> {
330        PreparedExactObject::new(self.head_object.clone(), self.head.to_bytes())
331            .map_err(PreparedCommitError::from)
332    }
333
334    pub fn publication(&self) -> (&StoreDeviceHead, &ExactObjectRef) {
335        (&self.head, &self.head_object)
336    }
337
338    pub fn head_ref(&self) -> StoreDeviceHeadRef {
339        StoreDeviceHeadRef {
340            head_hash: self.head.head_hash(),
341            object: self.head_object.clone(),
342        }
343    }
344
345    pub fn acknowledgement_remote_objects(
346        &self,
347        acknowledgement: &crate::objects::ExactProtocolObject<super::store_commit::StoreAck>,
348    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
349        let reference = self.commit.acknowledgement().ok_or_else(|| {
350            PreparedCommitError::Invariant(
351                "prepared acknowledgement operation has no exact acknowledgement ref".to_string(),
352            )
353        })?;
354        if &reference.object != acknowledgement.prepared.reference()
355            || reference.ack_hash != acknowledgement.value.ack_hash()
356            || acknowledgement.value.to_bytes() != acknowledgement.bytes
357        {
358            return Err(PreparedCommitError::Invariant(
359                "prepared acknowledgement operation differs from its exact acknowledgement object"
360                    .to_string(),
361            ));
362        }
363        let authority =
364            crate::remote_object::RemoteObjectRecord::candidate_activated_store_acknowledgement(
365                reference.clone(),
366                &acknowledgement.bytes,
367                acknowledgement.prepared.stored_bytes(),
368                self.reference.clone(),
369            )
370            .map_err(PreparedCommitError::from)?;
371        self.retained_authority_remote_objects(vec![authority])
372    }
373
374    pub fn circle_acknowledgement_remote_objects(
375        &self,
376        acknowledgement: &crate::objects::ExactProtocolObject<super::store_commit::CircleAck>,
377    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
378        let reference = self
379            .commit
380            .circle_acknowledgements()
381            .iter()
382            .find(|reference| &reference.object == acknowledgement.prepared.reference())
383            .ok_or_else(|| {
384                PreparedCommitError::Invariant(
385                    "prepared activation does not name its Circle acknowledgement object"
386                        .to_string(),
387                )
388            })?;
389        if reference.circle_id != acknowledgement.value.circle_id
390            || reference.ack_hash != acknowledgement.value.ack_hash()
391            || acknowledgement.value.to_bytes() != acknowledgement.bytes
392        {
393            return Err(PreparedCommitError::Invariant(
394                "prepared Circle acknowledgement differs from its exact acknowledgement object"
395                    .to_string(),
396            ));
397        }
398        let authority =
399            crate::remote_object::RemoteObjectRecord::candidate_activated_circle_acknowledgement(
400                reference.clone(),
401                &acknowledgement.bytes,
402                acknowledgement.prepared.stored_bytes(),
403                self.reference.clone(),
404            )
405            .map_err(PreparedCommitError::from)?;
406        self.retained_authority_remote_objects(vec![authority])
407    }
408
409    pub fn retained_authority_remote_objects(
410        &self,
411        authorities: Vec<crate::remote_object::ClosedRemoteObject>,
412    ) -> Result<Vec<crate::remote_object::ClosedRemoteObject>, PreparedCommitError> {
413        if authorities.is_empty() {
414            return Err(PreparedCommitError::Invariant(
415                "Store operation has no retained authority objects".to_string(),
416            ));
417        }
418        let mut authority_ids = std::collections::BTreeSet::new();
419        for authority in &authorities {
420            if !matches!(authority.record(), crate::remote_object::RemoteObjectRecord::RetainedAuthority(record)
421                if matches!(&record.state, crate::remote_object::RetainedAuthorityObjectState::Prepared { ownership }
422                    if ownership.pending == std::collections::BTreeSet::from([self.reference.clone()])))
423            {
424                return Err(PreparedCommitError::Invariant(
425                    "Store operation retained authority has different candidate ownership"
426                        .to_string(),
427                ));
428            }
429            if !authority_ids.insert(authority.record().object_id()) {
430                return Err(PreparedCommitError::Invariant(
431                    "Store operation repeats a retained authority object".to_string(),
432                ));
433            }
434        }
435        let mut objects = self.candidate_remote_objects()?;
436        objects.extend(authorities);
437        Ok(objects)
438    }
439
440    pub fn adopt_merge_head(
441        &mut self,
442        winner: StoreDeviceHead,
443        object: ExactObjectRef,
444    ) -> Result<(), PreparedCommitError> {
445        let current = &mut self.head;
446        let current_object = &mut self.head_object;
447        if winner.commit != self.common.reference
448            || object.slot() != current_object.slot()
449            || object == *current_object
450            || winner.author_registration != current.author_registration
451            || winner.successor.activation != current.successor.activation
452            || winner.successor.predecessor != current.successor.predecessor
453        {
454            return Err(PreparedCommitError::Invariant(
455                "alternate Merge head differs from the prepared activation point".to_string(),
456            ));
457        }
458        *current = winner;
459        *current_object = object;
460        Ok(())
461    }
462
463    pub fn attach_merge_membership_proof_with(
464        &mut self,
465        publication: &PreparedMembershipPublication,
466        resolution_value: Option<&super::membership::StoreMembershipConflictResolution>,
467    ) -> Result<(), PreparedCommitError> {
468        publication.validate().map_err(PreparedCommitError::from)?;
469        let reference = self.common.reference.clone();
470        let commit = self.common.commit.clone();
471        let Some(StoreControl { transition }) = commit.control() else {
472            return Err(PreparedCommitError::Invariant(
473                "Merge membership proof accompanies another Store control".to_string(),
474            ));
475        };
476        if !transition.matches_head(&publication.head, &publication.head_ref)
477            || publication.entry_ref != transition.body.entry
478        {
479            return Err(PreparedCommitError::Invariant(
480                "Merge membership proof differs from its signed Store transition".to_string(),
481            ));
482        }
483        let resolution = match &publication.entry.change {
484            super::membership::MembershipChange::ResolutionActivation { resolution } => {
485                let value = resolution_value.ok_or_else(|| {
486                    PreparedCommitError::Invariant(
487                        "Merge resolution activation lacks its exact resolution proof".to_string(),
488                    )
489                })?;
490                if value.resolution_ref(resolution.object.clone()) != *resolution {
491                    return Err(PreparedCommitError::Invariant(
492                        "Merge resolution proof differs from its exact reference".to_string(),
493                    ));
494                }
495                (Some(resolution.clone()), Some(value.clone()))
496            }
497            _ if resolution_value.is_none() => (None, None),
498            _ => {
499                return Err(PreparedCommitError::Invariant(
500                    "non-resolution membership proof carries a resolution".to_string(),
501                ))
502            }
503        };
504        self.history_evidence.membership_proof = Some(Box::new(
505            super::store_commit::RetainedMergeMembershipProof {
506                commit: reference,
507                commit_value: commit,
508                announcement: None,
509                entry: publication.entry_ref.clone(),
510                entry_value: publication.entry.clone(),
511                head: publication.head_ref.clone(),
512                head_value: publication.head.clone(),
513                resolution: resolution.0,
514                resolution_value: resolution.1,
515            },
516        ));
517        self.validate_closed_shape()?;
518        Ok(())
519    }
520}
521
522/// One Circle acknowledgement object riding an activating Store commit: its
523/// exact reference (named in the signed commit body) and the exact object the
524/// commit uploads and takes ownership of.
525#[derive(Debug, Clone)]
526pub struct CircleAckActivation {
527    pub reference: crate::store_commit::CircleAckRef,
528    pub ack: crate::objects::ExactProtocolObject<crate::store_commit::CircleAck>,
529}
coven_protocol/prepared_commit.rs

coven_protocol/
prepared_commit.rs
