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

coven_protocol/store_commit/
identifiers.rs
