coven_protocol/store_commit/
identifiers.rs

1use super::device_state::merge_history_cuts;
2use super::*;
3
4pub use coven_foundation::object_hash::ObjectHash;
5
6/// Closed coordinate of one Store commit in its author stream.
7#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
8#[serde(deny_unknown_fields)]
9pub struct StoreCommitCoord {
10    pub stream_id: AuthorStreamId,
11    pub sequence: u64,
12}
13
14impl StoreCommitCoord {
15    pub fn sequence(&self) -> u64 {
16        self.sequence
17    }
18
19    pub fn validate(&self) -> Result<(), StoreProtocolError> {
20        if self.sequence() == 0 {
21            return Err(StoreProtocolError::Malformed(
22                "Store commit coordinate uses sequence zero".to_string(),
23            ));
24        }
25        Ok(())
26    }
27}
28
29/// Domain-separated family shared by replacements at one competition point.
30#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
31#[serde(transparent)]
32pub struct CandidateFamilyId(ObjectHash);
33
34impl CandidateFamilyId {
35    pub fn from_hash(hash: ObjectHash) -> Self {
36        Self(hash)
37    }
38
39    pub fn as_hash(self) -> ObjectHash {
40        self.0
41    }
42
43    pub fn derive(
44        store_root_hash: ObjectHash,
45        author_registration: &StoreDeviceRegistrationRef,
46        write_id: &WriteId,
47        order: &StoreCommitOrder,
48    ) -> Self {
49        #[derive(Serialize)]
50        struct Fields<'a> {
51            store_root_hash: ObjectHash,
52            author_registration: &'a StoreDeviceRegistrationRef,
53            write_id: &'a WriteId,
54            sequence: u64,
55            predecessor: Option<&'a StoreBatchCommitRef>,
56        }
57        let fields = Fields {
58            store_root_hash,
59            author_registration,
60            write_id,
61            sequence: order.seq(),
62            predecessor: order.predecessor.as_ref(),
63        };
64        Self(ObjectHash::digest(&domain_json(
65            CANDIDATE_FAMILY_DOMAIN,
66            &fields,
67        )))
68    }
69}
70
71/// Exact identity of one signed Store commit candidate.
72#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
73#[serde(deny_unknown_fields)]
74pub struct StoreBatchCommitRef {
75    pub coord: StoreCommitCoord,
76    pub commit_hash: ObjectHash,
77    pub object: ExactObjectRef,
78}
79
80/// Exact stored candidate commit retained as cleanup authority after abandonment.
81#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
82#[serde(deny_unknown_fields)]
83pub struct StoreBatchCommitDeletionTarget {
84    pub coord: StoreCommitCoord,
85    pub object: ExactObjectRef,
86    pub canonical_signed_bytes: Vec<u8>,
87}
88
89impl StoreBatchCommitDeletionTarget {
90    pub(crate) fn verify_candidate(
91        &self,
92        expected_store_root_hash: ObjectHash,
93        author: &StoreDeviceRegistration,
94    ) -> Result<VerifiedStoreBatchCommit, StoreProtocolError> {
95        let commit = self.verify_exact_candidate(expected_store_root_hash, author)?;
96        if matches!(&commit.body, StoreCommitBody::AbandonCandidates { .. }) {
97            return Err(StoreProtocolError::Malformed(
98                "retained authority cannot be a candidate cleanup target".to_string(),
99            ));
100        }
101        Ok(commit)
102    }
103
104    fn verify_exact_candidate(
105        &self,
106        expected_store_root_hash: ObjectHash,
107        author: &StoreDeviceRegistration,
108    ) -> Result<VerifiedStoreBatchCommit, StoreProtocolError> {
109        self.object.verify(&self.canonical_signed_bytes)?;
110        let commit = VerifiedStoreBatchCommit::parse_prepared(
111            &self.canonical_signed_bytes,
112            expected_store_root_hash,
113            self.coord.clone(),
114            self.object.clone(),
115            author,
116        )?;
117        if commit.to_bytes() != self.canonical_signed_bytes {
118            return Err(StoreProtocolError::Malformed(
119                "candidate commit bytes are not canonical".to_string(),
120            ));
121        }
122        Ok(commit)
123    }
124}
125
126#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
127#[serde(deny_unknown_fields)]
128pub struct CandidateCleanupManifest {
129    pub candidate: StoreBatchCommitDeletionTarget,
130}
131
132impl StoreBatchCommitRef {
133    pub fn from_commit(
134        commit: &StoreBatchCommit,
135        coord: StoreCommitCoord,
136        object: ExactObjectRef,
137    ) -> Result<Self, StoreProtocolError> {
138        if coord.sequence() != commit.seq() {
139            return Err(StoreProtocolError::Malformed(
140                "Store commit reference coordinate differs from the signed commit".to_string(),
141            ));
142        }
143        let reference = Self {
144            coord,
145            commit_hash: commit.commit_hash(),
146            object,
147        };
148        reference.verify_commit(commit)?;
149        Ok(reference)
150    }
151
152    pub fn verify_commit(&self, commit: &StoreBatchCommit) -> Result<(), StoreProtocolError> {
153        if self.coord.sequence() != commit.seq() || self.commit_hash != commit.commit_hash() {
154            return Err(StoreProtocolError::Malformed(
155                "exact Store commit reference differs from the signed commit".to_string(),
156            ));
157        }
158        let stream_id = commit_stream_id(&self.coord);
159        let expected = format!(
160            "{}.json",
161            commit_semantic_prefix(
162                commit.candidate_family(),
163                &stream_id,
164                self.coord.sequence(),
165                self.commit_hash,
166            )
167        );
168        if self.object.slot().logical_key() != expected {
169            return Err(StoreProtocolError::RelocatedSlot {
170                expected,
171                actual: self.object.slot().logical_key().to_string(),
172            });
173        }
174        Ok(())
175    }
176}
177
178#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
179#[serde(deny_unknown_fields)]
180pub struct StoreRootRef {
181    pub store_root_id: ObjectHash,
182    pub store_root_hash: ObjectHash,
183    pub object: ExactObjectRef,
184}
185
186#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
187#[serde(transparent)]
188pub struct StreamActivationId(ObjectHash);
189
190impl StreamActivationId {
191    pub fn from_digest(hash: ObjectHash) -> Self {
192        Self(hash)
193    }
194
195    pub fn as_hash(self) -> ObjectHash {
196        self.0
197    }
198}
199
200#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
201#[serde(rename_all = "snake_case", deny_unknown_fields)]
202pub enum StreamActivation {
203    GrantAuthorized {
204        store_root_hash: ObjectHash,
205        author_registration: StoreDeviceRegistrationRef,
206        grant_id: MembershipGrantId,
207        anchor: GrantStreamAnchor,
208    },
209    DeviceAuthorized {
210        store_root_hash: ObjectHash,
211        author_registration: StoreDeviceRegistrationRef,
212        anchor: DeviceStreamAnchor,
213    },
214}
215
216#[derive(Debug, Clone, PartialEq, Eq)]
217pub struct RegisteredStreamActivation {
218    activation: StreamActivation,
219    activating_commit: StoreBatchCommitRef,
220}
221
222impl RegisteredStreamActivation {
223    pub fn from_stored(
224        stored_activation_id: StreamActivationId,
225        stored_author_stream_id: AuthorStreamId,
226        activation: StreamActivation,
227        activating_commit: StoreBatchCommitRef,
228    ) -> Result<Self, StoreProtocolError> {
229        if activation.activation_id() != stored_activation_id {
230            return Err(StoreProtocolError::Malformed(
231                "stored stream activation id differs from its canonical descriptor".to_string(),
232            ));
233        }
234        if activation.author_stream_id() != stored_author_stream_id {
235            return Err(StoreProtocolError::Malformed(
236                "stored author stream id differs from its canonical descriptor".to_string(),
237            ));
238        }
239        Ok(Self {
240            activation,
241            activating_commit,
242        })
243    }
244
245    pub fn activation(&self) -> &StreamActivation {
246        &self.activation
247    }
248
249    pub fn activating_commit(&self) -> &StoreBatchCommitRef {
250        &self.activating_commit
251    }
252}
253
254#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)]
255#[serde(rename_all = "snake_case")]
256pub enum StreamAnchorDomain {
257    StoreMembership,
258    OwnerRecovery,
259    CircleControl {
260        circle_id: CircleId,
261    },
262    CircleRoster {
263        circle_id: CircleId,
264    },
265    CircleMetadata {
266        circle_id: CircleId,
267    },
268    CircleAcknowledgements {
269        circle_id: CircleId,
270    },
271    CircleSnapshots {
272        circle_id: CircleId,
273    },
274    /// Derives Store commit author coordinates from the exact device registration.
275    /// Accepted publications locate commits; this domain has no successor-slot anchor.
276    StoreAnnouncements,
277    StoreAcknowledgements,
278}
279
280impl GrantStreamAnchor {
281    fn domain(&self) -> StreamAnchorDomain {
282        match self {
283            Self::StoreMembership { .. } => StreamAnchorDomain::StoreMembership,
284            Self::OwnerRecovery { .. } => StreamAnchorDomain::OwnerRecovery,
285            Self::CircleControl { circle_id, .. } => StreamAnchorDomain::CircleControl {
286                circle_id: *circle_id,
287            },
288            Self::CircleRoster { circle_id, .. } => StreamAnchorDomain::CircleRoster {
289                circle_id: *circle_id,
290            },
291            Self::CircleMetadata { circle_id, .. } => StreamAnchorDomain::CircleMetadata {
292                circle_id: *circle_id,
293            },
294        }
295    }
296}
297
298impl DeviceStreamAnchor {
299    fn domain(&self) -> StreamAnchorDomain {
300        match self {
301            Self::StoreAcknowledgements { .. } => StreamAnchorDomain::StoreAcknowledgements,
302            Self::CircleAcknowledgements { circle_id, .. } => {
303                StreamAnchorDomain::CircleAcknowledgements {
304                    circle_id: *circle_id,
305                }
306            }
307            Self::CircleSnapshots { circle_id, .. } => StreamAnchorDomain::CircleSnapshots {
308                circle_id: *circle_id,
309            },
310        }
311    }
312}
313
314impl StreamActivation {
315    pub fn grant_authorized(
316        store_root_hash: ObjectHash,
317        author_registration: StoreDeviceRegistrationRef,
318        grant_id: MembershipGrantId,
319        anchor: GrantStreamAnchor,
320    ) -> Self {
321        Self::GrantAuthorized {
322            store_root_hash,
323            author_registration,
324            grant_id,
325            anchor,
326        }
327    }
328
329    pub fn device_authorized(
330        store_root_hash: ObjectHash,
331        author_registration: StoreDeviceRegistrationRef,
332        anchor: DeviceStreamAnchor,
333    ) -> Self {
334        Self::DeviceAuthorized {
335            store_root_hash,
336            author_registration,
337            anchor,
338        }
339    }
340
341    pub fn activation_id(&self) -> StreamActivationId {
342        StreamActivationId(ObjectHash::digest(&domain_json(
343            STREAM_ACTIVATION_ID_DOMAIN,
344            self,
345        )))
346    }
347
348    pub fn author_stream_id(&self) -> AuthorStreamId {
349        match self {
350            Self::GrantAuthorized {
351                store_root_hash,
352                author_registration,
353                grant_id,
354                anchor,
355            } => derive_grant_author_stream_id(
356                *store_root_hash,
357                author_registration,
358                grant_id,
359                anchor.domain(),
360            ),
361            Self::DeviceAuthorized {
362                store_root_hash,
363                author_registration,
364                anchor,
365            } => derive_device_author_stream_id(
366                *store_root_hash,
367                author_registration,
368                anchor.domain(),
369            ),
370        }
371    }
372
373    pub fn device_authorized_stream_id(
374        store_root_hash: ObjectHash,
375        author_registration: &StoreDeviceRegistrationRef,
376        domain: StreamAnchorDomain,
377    ) -> AuthorStreamId {
378        derive_device_author_stream_id(store_root_hash, author_registration, domain)
379    }
380
381    pub fn grant_authorized_stream_id(
382        store_root_hash: ObjectHash,
383        author_registration: &StoreDeviceRegistrationRef,
384        grant_id: &MembershipGrantId,
385        domain: StreamAnchorDomain,
386    ) -> AuthorStreamId {
387        derive_grant_author_stream_id(store_root_hash, author_registration, grant_id, domain)
388    }
389
390    pub fn first_slot(&self) -> &ObjectSlot {
391        match self {
392            Self::GrantAuthorized { anchor, .. } => anchor.first_slot(),
393            Self::DeviceAuthorized { anchor, .. } => anchor.first_slot(),
394        }
395    }
396
397    pub fn author_registration(&self) -> &StoreDeviceRegistrationRef {
398        match self {
399            Self::GrantAuthorized {
400                author_registration,
401                ..
402            }
403            | Self::DeviceAuthorized {
404                author_registration,
405                ..
406            } => author_registration,
407        }
408    }
409
410    pub fn store_root_hash(&self) -> ObjectHash {
411        match self {
412            Self::GrantAuthorized {
413                store_root_hash, ..
414            }
415            | Self::DeviceAuthorized {
416                store_root_hash, ..
417            } => *store_root_hash,
418        }
419    }
420}
421
422#[derive(Serialize)]
423struct GrantAuthorStreamFields<'a> {
424    store_root_hash: ObjectHash,
425    domain: StreamAnchorDomain,
426    author_registration: &'a StoreDeviceRegistrationRef,
427    grant_id: &'a MembershipGrantId,
428}
429
430#[derive(Serialize)]
431struct DeviceAuthorStreamFields<'a> {
432    store_root_hash: ObjectHash,
433    domain: StreamAnchorDomain,
434    author_registration: &'a StoreDeviceRegistrationRef,
435}
436
437fn derive_grant_author_stream_id(
438    store_root_hash: ObjectHash,
439    author_registration: &StoreDeviceRegistrationRef,
440    grant_id: &MembershipGrantId,
441    domain: StreamAnchorDomain,
442) -> AuthorStreamId {
443    derive_author_stream_id(&GrantAuthorStreamFields {
444        store_root_hash,
445        domain,
446        author_registration,
447        grant_id,
448    })
449}
450
451fn derive_device_author_stream_id(
452    store_root_hash: ObjectHash,
453    author_registration: &StoreDeviceRegistrationRef,
454    domain: StreamAnchorDomain,
455) -> AuthorStreamId {
456    derive_author_stream_id(&DeviceAuthorStreamFields {
457        store_root_hash,
458        domain,
459        author_registration,
460    })
461}
462
463fn derive_author_stream_id(fields: &impl Serialize) -> AuthorStreamId {
464    AuthorStreamId::from_digest(ObjectHash::digest(&domain_json(
465        AUTHOR_STREAM_ID_DOMAIN,
466        fields,
467    )))
468}
469
470#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
471#[serde(deny_unknown_fields)]
472pub struct SuccessorLink {
473    pub activation: StreamActivationId,
474    pub predecessor: Option<ExactObjectRef>,
475    pub next_slot: ObjectSlot,
476}
477
478/// Exact materialized cut across author streams.
479#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
480#[serde(transparent)]
481pub struct CommitFrontier(pub BTreeMap<AuthorStreamId, StoreBatchCommitRef>);
482
483/// Exact Store history cut across author streams.
484#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
485#[serde(transparent)]
486pub struct StoreHistoryCut(pub BTreeMap<AuthorStreamId, StoreBatchCommitRef>);
487
488impl StoreHistoryCut {
489    pub fn from_commits(commits: BTreeMap<AuthorStreamId, StoreBatchCommitRef>) -> Self {
490        Self(commits)
491    }
492
493    pub fn position_count(&self) -> usize {
494        self.0.len()
495    }
496
497    pub fn commits(&self) -> &BTreeMap<AuthorStreamId, StoreBatchCommitRef> {
498        &self.0
499    }
500
501    pub fn frontier(&self) -> CommitFrontier {
502        CommitFrontier(self.0.clone())
503    }
504
505    pub fn join(self, other: Self) -> Result<Self, StoreProtocolError> {
506        merge_history_cuts(self, other)
507    }
508}
509
510impl CommitFrontier {
511    pub fn from_refs(
512        commits: BTreeMap<String, StoreBatchCommitRef>,
513    ) -> Result<Self, StoreProtocolError> {
514        commits
515            .into_iter()
516            .map(|(stream_id, commit)| {
517                let stream_id = stream_id
518                    .parse()
519                    .map_err(StoreProtocolError::AuthorStreamId)?;
520                Ok((stream_id, commit))
521            })
522            .collect::<Result<BTreeMap<_, _>, _>>()
523            .map(Self)
524    }
525
526    pub fn into_refs(self) -> BTreeMap<String, StoreBatchCommitRef> {
527        self.0
528            .into_iter()
529            .map(|(stream_id, commit)| (stream_id.to_string(), commit))
530            .collect()
531    }
532
533    pub fn position_count(&self) -> usize {
534        self.0.len()
535    }
536
537    pub fn covers(&self, covered: &Self) -> bool {
538        covered
539            .0
540            .iter()
541            .all(|(stream, covered_ref)| self.covers_commit_on_stream(stream, covered_ref))
542    }
543
544    pub fn commits(&self) -> &BTreeMap<AuthorStreamId, StoreBatchCommitRef> {
545        &self.0
546    }
547
548    pub fn covers_commit(&self, commit: &StoreBatchCommitRef) -> bool {
549        self.covers_commit_on_stream(&commit.coord.stream_id, commit)
550    }
551
552    fn covers_commit_on_stream(
553        &self,
554        stream: &AuthorStreamId,
555        covered: &StoreBatchCommitRef,
556    ) -> bool {
557        self.0.get(stream).is_some_and(|current| {
558            current.coord.sequence() > covered.coord.sequence()
559                || current.coord.sequence() == covered.coord.sequence() && current == covered
560        })
561    }
562
563    pub fn join(self, other: Self) -> Result<Self, StoreProtocolError> {
564        StoreHistoryCut::from_commits(self.0)
565            .join(StoreHistoryCut::from_commits(other.0))
566            .map(|cut| Self(cut.0))
567    }
568}
569
570/// Predecessor and dependency order authenticated by one Store commit.
571#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
572#[serde(deny_unknown_fields)]
573pub struct StoreCommitOrder {
574    pub seq: u64,
575    pub predecessor: Option<StoreBatchCommitRef>,
576    pub dependencies: BTreeMap<AuthorStreamId, StoreBatchCommitRef>,
577}
578
579impl StoreCommitOrder {
580    pub fn seq(&self) -> u64 {
581        self.seq
582    }
583
584    pub fn predecessor(&self) -> Option<&StoreBatchCommitRef> {
585        self.predecessor.as_ref()
586    }
587
588    pub fn dependencies(&self) -> &BTreeMap<AuthorStreamId, StoreBatchCommitRef> {
589        &self.dependencies
590    }
591
592    pub fn predecessor_cut(&self) -> Result<StoreHistoryCut, StoreProtocolError> {
593        let mut cut = self.dependencies.clone();
594        if let Some(predecessor) = &self.predecessor {
595            if cut
596                .insert(predecessor.coord.stream_id, predecessor.clone())
597                .is_some_and(|existing| existing != *predecessor)
598            {
599                return Err(StoreProtocolError::Malformed(
600                    "Merge predecessor disagrees with the same-stream dependency".to_string(),
601                ));
602            }
603        }
604        Ok(StoreHistoryCut(cut))
605    }
606}
607
608pub(super) fn commit_stream_id(coord: &StoreCommitCoord) -> String {
609    coord.stream_id.to_string()
610}
coven_protocol/store_commit/identifiers.rs

coven_protocol/store_commit/
identifiers.rs
