coven_core/sync/store_commit/
identifiers.rs

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

coven_core/sync/store_commit/
identifiers.rs
