Skip to main content

coven_protocol/store_commit/
retained_history.rs

1use super::validation::require_version;
2use super::*;
3
4#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
5#[serde(deny_unknown_fields)]
6pub struct MembershipCausalFloor {
7    pub effective_coordinates: Vec<MembershipCoord>,
8    pub resolutions: Vec<StoreMembershipConflictResolutionRef>,
9}
10
11impl MembershipCausalFloor {
12    pub fn from_membership(membership: &crate::membership::MembershipChain) -> Self {
13        Self {
14            effective_coordinates: membership.effective_frontier(),
15            resolutions: membership.resolution_refs().to_vec(),
16        }
17    }
18
19    pub fn advance(
20        &mut self,
21        coordinate: crate::membership::MembershipCoord,
22        resolutions: &[StoreMembershipConflictResolutionRef],
23    ) -> Result<(), StoreProtocolError> {
24        let stream = coordinate.stream_key();
25        self.effective_coordinates
26            .retain(|current| current.stream_key() != stream);
27        self.effective_coordinates.push(coordinate);
28        self.effective_coordinates.sort();
29        self.resolutions.extend_from_slice(resolutions);
30        self.resolutions.sort();
31        self.resolutions.dedup();
32        self.validate()
33    }
34
35    pub fn is_included_in(&self, membership: &crate::membership::MembershipChain) -> bool {
36        self.effective_coordinates
37            .iter()
38            .all(|coordinate| membership.effectively_contains_coord(coordinate))
39            && self.resolutions.iter().all(|reference| {
40                membership
41                    .resolution_refs()
42                    .binary_search(reference)
43                    .is_ok()
44            })
45    }
46
47    fn validate(&self) -> Result<(), StoreProtocolError> {
48        if self
49            .effective_coordinates
50            .windows(2)
51            .any(|pair| pair[0] >= pair[1])
52            || self.resolutions.windows(2).any(|pair| pair[0] >= pair[1])
53        {
54            return Err(StoreProtocolError::Malformed(
55                "Merge history membership floor is not canonical".to_string(),
56            ));
57        }
58        Ok(())
59    }
60}
61
62/// The acknowledgement one commit activated, retained beside that commit.
63///
64/// One acknowledgement, not the chain behind it. A retained row describes its
65/// own commit, and an acknowledgement's predecessors are described by the rows
66/// that retained *them* — each acknowledgement names its predecessor's object,
67/// so contiguity follows from the rows in the same way a commit's ancestry
68/// follows from the commits, without every row carrying a copy of everything
69/// before it.
70///
71/// Storing the chain here instead made a retained row grow with the history in
72/// front of it: on a two-device store where nearly every commit acknowledges,
73/// the row at sequence N held N acknowledgements, so the table grew with the
74/// square of the history. A field store reached 223 MB over 385 rows, and both
75/// applying a commit and reading the retained rows back paid for it.
76#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
77#[serde(deny_unknown_fields)]
78pub struct RetainedVerifiedActivatedAck {
79    pub acknowledgement: (StoreAckRef, StoreAck),
80    pub activating_commit: StoreBatchCommitRef,
81}
82
83/// A device's acknowledgement chain, contiguous from sequence one, carried by a
84/// snapshot's portable summary.
85///
86/// This is the one place the whole chain belongs. A device restoring from a
87/// snapshot has no retained rows to walk, so the summary has to state the
88/// contiguity itself; it is folded once per snapshot generation from the rows
89/// the snapshot covers, rather than rebuilt into every row.
90#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
91#[serde(deny_unknown_fields)]
92pub struct RetainedAcknowledgementChain {
93    #[serde(with = "ordered_map_entries")]
94    pub chain: BTreeMap<u64, (StoreAckRef, StoreAck)>,
95    pub activating_commit: StoreBatchCommitRef,
96    pub activating_commit_value: StoreBatchCommit,
97}
98
99/// Everything a device needs to install one snapshot as its starting state and
100/// verify what arrives after it: the Store root and founder it belongs to, the
101/// signed metadata, the cut it covers, and the device state and registrations
102/// active at that cut.
103///
104/// Every field is re-derived from the signed `metadata` by
105/// [`validate`](Self::validate), so an installing device trusts the owner's
106/// signature over the snapshot and nothing local. What is deliberately absent
107/// is any claim about the *other* devices having caught up: that is
108/// [`AcknowledgedStoreSnapshot`], and only reclaim needs it. A device installing
109/// a baseline verifies each later commit against the registrations and device
110/// state carried here, exactly as a device that never installed a snapshot
111/// verifies them against its own history.
112#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
113#[serde(deny_unknown_fields)]
114pub struct RetainedReplaySnapshotAuthority {
115    pub store_root: StoreRootRef,
116    pub founder_registration: StoreDeviceRegistrationRef,
117    pub snapshot: StoreSnapshotRef,
118    pub metadata: SnapshotMeta,
119    pub snapshot_cut: StoreHistoryCut,
120    pub accepted_cut: StoreHistoryCut,
121    pub device_state: ResolvedStoreDeviceState,
122    #[serde(with = "ordered_map_entries")]
123    pub active_registrations: BTreeMap<StoreDeviceId, ReferencedStoreDeviceRegistration>,
124}
125
126/// One snapshot every device active at its cut has acknowledged.
127///
128/// This is the unanimity proof, and it answers only one question: may history
129/// behind this snapshot be deleted? It may, because every device that could
130/// still need that history has said in a signed acknowledgement — activated by
131/// a commit in the verified closure — that it holds this snapshot.
132///
133/// Installing a snapshot asks a different question and does not need this. A
134/// device joining or restoring wants a signed, owner-authored, history-
135/// consistent image; whether some other device has caught up has no bearing on
136/// that, and a device that is behind converges through an ordinary pull no
137/// matter which image the joiner installed. Requiring unanimity there made a
138/// store with one joined-and-idle device fall back to its generation-zero
139/// image forever.
140#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
141#[serde(deny_unknown_fields)]
142pub struct AcknowledgedStoreSnapshot {
143    pub authority: RetainedReplaySnapshotAuthority,
144    /// One chain per device that had to acknowledge, which is a subset of the
145    /// devices active at the coverage: those still active now. Which subset is
146    /// a question about the current device state, so it is decided by the
147    /// builder against verified history and recorded here — `validate` can
148    /// check that these devices were active at the coverage and that each chain
149    /// proves what it claims, but not that the set is the right one to have
150    /// asked. See the reclaim module for why the set is what it is.
151    #[serde(with = "ordered_map_entries")]
152    pub acknowledgements: BTreeMap<StoreDeviceId, RetainedAcknowledgementChain>,
153}
154
155/// The evidence required to retire local replay inputs behind one snapshot.
156///
157/// Cloud reclaim asks whether the devices active at the snapshot have made the
158/// exact snapshot promise. Local retirement asks a stronger and different
159/// question: whether every writer active now has crossed that cut, including a
160/// writer activated after the snapshot. Keeping the proofs separate prevents
161/// the local ordering rule from changing which cloud objects may be reclaimed.
162#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
163#[serde(deny_unknown_fields)]
164pub struct ReplayBaselineRetirementProof {
165    pub authority: RetainedReplaySnapshotAuthority,
166    pub current_cut: StoreHistoryCut,
167    pub current_state: StoreDeviceStateRef,
168    pub current_device_state: ResolvedStoreDeviceState,
169    pub current_membership: StoreMembershipStateRef,
170    pub membership_witness: ReplayRetirementMembershipWitness,
171    #[serde(with = "ordered_map_entries")]
172    pub current_registrations: BTreeMap<StoreDeviceId, ReferencedStoreDeviceRegistration>,
173    #[serde(with = "ordered_map_entries")]
174    pub acknowledgements: BTreeMap<StoreDeviceId, RetainedAcknowledgementChain>,
175}
176
177/// Accepted Store history that names the exact membership used to decide which
178/// writers must acknowledge a replay cut.
179#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
180#[serde(rename_all = "snake_case", deny_unknown_fields)]
181pub enum ReplayRetirementMembershipWitness {
182    Snapshot,
183    StoreCommit(StoreBatchCommitRef),
184}
185
186impl RetainedReplaySnapshotAuthority {
187    pub fn validate(&self) -> Result<(), StoreProtocolError> {
188        let metadata_bytes = self.metadata.to_bytes();
189        let author = self
190            .active_registrations
191            .get(&self.metadata.author_registration.device_id)
192            .filter(|registration| registration.reference() == &self.metadata.author_registration)
193            .ok_or_else(|| {
194                StoreProtocolError::Malformed(
195                    "retained snapshot author is absent from its active registrations".to_string(),
196                )
197            })?;
198        let parsed = SnapshotMeta::parse_at(
199            &metadata_bytes,
200            self.store_root.store_root_hash,
201            &self.snapshot,
202            author.value(),
203        )?;
204        if self.metadata.store_root_hash != self.store_root.store_root_hash
205            || self.metadata.generation != self.snapshot.generation
206            || self.metadata.snapshot_hash() != self.snapshot.snapshot_hash
207            || self.snapshot.object.verify(&metadata_bytes).is_err()
208            || self.snapshot_cut.frontier() != self.metadata.coverage
209            || !self
210                .accepted_cut
211                .frontier()
212                .covers(&self.snapshot_cut.frontier())
213            || parsed != self.metadata
214            || self.device_state.state_hash != self.metadata.state.devices.state_hash()
215            || self.device_state.recovery != self.metadata.state.devices.recovery()
216        {
217            return Err(StoreProtocolError::Malformed(
218                "retained snapshot replay authority differs from its signed snapshot state"
219                    .to_string(),
220            ));
221        }
222        let expected_active = self
223            .device_state
224            .devices
225            .iter()
226            .filter_map(|(device_id, record)| {
227                matches!(record.status, StoreDeviceStatus::Active)
228                    .then_some((*device_id, &record.registration))
229            })
230            .collect::<BTreeMap<_, _>>();
231        if expected_active.len() != self.active_registrations.len()
232            || expected_active.iter().any(|(device_id, reference)| {
233                self.active_registrations
234                    .get(device_id)
235                    .is_none_or(|registration| registration.reference() != *reference)
236            })
237        {
238            return Err(StoreProtocolError::Malformed(
239                "retained snapshot replay authority does not exactly cover active devices"
240                    .to_string(),
241            ));
242        }
243        for (device_id, registration) in &self.active_registrations {
244            let bytes = registration.value().to_bytes();
245            registration.reference().object.verify(&bytes)?;
246            let parsed = StoreDeviceRegistration::parse_at(&bytes, &self.store_root, *device_id)?;
247            if &parsed != registration.value() {
248                return Err(StoreProtocolError::Malformed(
249                    "retained snapshot registration is not canonical".to_string(),
250                ));
251            }
252            registration
253                .reference()
254                .verify_registration(registration.value())?;
255        }
256        Ok(())
257    }
258}
259
260impl AcknowledgedStoreSnapshot {
261    /// The latest acknowledgement each active device signed for this snapshot,
262    /// in a stable order. This is the evidence a reclaim claim carries: the
263    /// devices are named by what they signed, not by the chains behind it.
264    pub fn acknowledgement_refs(&self) -> Result<Vec<StoreAckRef>, StoreProtocolError> {
265        let mut references = self
266            .acknowledgements
267            .values()
268            .map(|acknowledgement| {
269                acknowledgement
270                    .latest()
271                    .map(|(reference, _)| reference.clone())
272                    .ok_or_else(|| {
273                        StoreProtocolError::Malformed(
274                            "acknowledged snapshot proof chain is empty".to_string(),
275                        )
276                    })
277            })
278            .collect::<Result<Vec<_>, _>>()?;
279        references.sort();
280        Ok(references)
281    }
282
283    /// The installable authority, plus proof that every device active at its cut
284    /// acknowledged this exact snapshot. Reclaim deletes history behind a
285    /// snapshot only against this.
286    pub fn validate(&self) -> Result<(), StoreProtocolError> {
287        self.authority.validate()?;
288        if self.acknowledgements.is_empty() {
289            return Err(StoreProtocolError::Malformed(
290                "acknowledged snapshot has no acknowledgements".to_string(),
291            ));
292        }
293        for (device_id, acknowledgement) in &self.acknowledgements {
294            let registration = self
295                .authority
296                .active_registrations
297                .get(device_id)
298                .ok_or_else(|| {
299                    StoreProtocolError::Malformed(
300                        "acknowledged snapshot names a device that was not active at its coverage"
301                            .to_string(),
302                    )
303                })?;
304            let acknowledgement_value = validate_acknowledgement_activation(
305                &self.authority.store_root,
306                &self.authority.accepted_cut,
307                registration,
308                acknowledgement,
309            )?;
310            if !acknowledgement_value
311                .snapshot
312                .as_ref()
313                .is_some_and(|acknowledged| {
314                    acknowledged.author_registration == self.authority.metadata.author_registration
315                        && acknowledged.snapshot == self.authority.snapshot
316                })
317                || acknowledgement_value.device_state != self.authority.metadata.state.devices
318                || !acknowledgement_value
319                    .store_cut
320                    .frontier()
321                    .covers(&self.authority.metadata.coverage)
322            {
323                return Err(StoreProtocolError::Malformed(
324                    "retained snapshot acknowledgement differs from its activated commit"
325                        .to_string(),
326                ));
327            }
328        }
329        Ok(())
330    }
331}
332
333impl ReplayBaselineRetirementProof {
334    pub fn validate(
335        &self,
336        membership: &crate::membership::MembershipChain,
337    ) -> Result<BTreeSet<StoreDeviceId>, StoreProtocolError> {
338        self.authority.validate()?;
339        let crate::membership::MembershipStatus::Resolved(resolved_membership) =
340            membership.status()
341        else {
342            return Err(StoreProtocolError::Malformed(
343                "replay baseline retirement membership is conflicted".to_string(),
344            ));
345        };
346        let expected_membership = StoreMembershipStateRef::from_parts(
347            membership.head_refs().to_vec(),
348            membership.resolution_refs().to_vec(),
349            self.current_device_state.recovery.clone(),
350            resolved_membership.state_hash,
351        )?;
352        let required_writer_ids = replay_retirement_writer_ids(
353            self.authority.store_root.store_root_hash,
354            &self.current_device_state,
355            &self.current_registrations,
356            membership,
357        )?;
358        let membership_is_witnessed = match &self.membership_witness {
359            ReplayRetirementMembershipWitness::Snapshot => {
360                self.current_membership == self.authority.metadata.state.membership
361            }
362            ReplayRetirementMembershipWitness::StoreCommit(reference) => {
363                self.current_cut.frontier().covers_commit(reference)
364            }
365        };
366        if required_writer_ids.is_empty()
367            || self.current_membership != expected_membership
368            || !membership_is_witnessed
369            || self.acknowledgements.len() != required_writer_ids.len()
370            || !self
371                .current_cut
372                .frontier()
373                .covers(&self.authority.accepted_cut.frontier())
374            || StoreDeviceStateRef::from_resolved(
375                self.current_cut.frontier(),
376                &self.current_device_state,
377            )? != self.current_state
378        {
379            return Err(StoreProtocolError::Malformed(
380                "replay baseline retirement has inconsistent current authority".to_string(),
381            ));
382        }
383        for device_id in &required_writer_ids {
384            let registration = self
385                .current_registrations
386                .get(device_id)
387                .expect("current writer derivation validates registration coverage");
388            let acknowledgement = self.acknowledgements.get(device_id).ok_or_else(|| {
389                StoreProtocolError::Malformed(
390                    "replay baseline retirement omits a required writer".to_string(),
391                )
392            })?;
393            let acknowledgement_value = validate_acknowledgement_activation(
394                &self.authority.store_root,
395                &self.current_cut,
396                registration,
397                acknowledgement,
398            )?;
399            if !acknowledgement_value
400                .store_cut
401                .frontier()
402                .covers(&self.authority.metadata.coverage)
403            {
404                return Err(StoreProtocolError::Malformed(
405                    "replay baseline retirement acknowledgement does not cross its cut".to_string(),
406                ));
407            }
408        }
409        Ok(required_writer_ids)
410    }
411}
412
413pub fn replay_retirement_writer_ids(
414    store_root_hash: ObjectHash,
415    current_device_state: &ResolvedStoreDeviceState,
416    current_registrations: &BTreeMap<StoreDeviceId, ReferencedStoreDeviceRegistration>,
417    membership: &crate::membership::MembershipChain,
418) -> Result<BTreeSet<StoreDeviceId>, StoreProtocolError> {
419    current_device_state.validate_canonical()?;
420    if current_registrations.len() != current_device_state.devices.len() {
421        return Err(StoreProtocolError::Malformed(
422            "replay baseline retirement registrations do not exactly cover current devices"
423                .to_string(),
424        ));
425    }
426    let mut writers = BTreeSet::new();
427    for (device_id, record) in &current_device_state.devices {
428        let registration = current_registrations
429            .get(device_id)
430            .filter(|registration| registration.reference() == &record.registration)
431            .ok_or_else(|| {
432                StoreProtocolError::Malformed(
433                    "replay baseline retirement registration differs from current device state"
434                        .to_string(),
435                )
436            })?;
437        let bytes = registration.value().to_bytes();
438        registration.reference().object.verify(&bytes)?;
439        let parsed = StoreDeviceRegistration::parse_at(
440            &bytes,
441            &registration.value().store_root,
442            *device_id,
443        )?;
444        if parsed != *registration.value()
445            || registration.value().store_root.store_root_hash != store_root_hash
446        {
447            return Err(StoreProtocolError::Malformed(
448                "replay baseline retirement registration is not canonical".to_string(),
449            ));
450        }
451        if matches!(record.status, StoreDeviceStatus::Active)
452            && membership.is_member_now(&registration.value().author_pubkey)
453        {
454            writers.insert(*device_id);
455        }
456    }
457    Ok(writers)
458}
459
460fn validate_acknowledgement_activation<'a>(
461    root: &StoreRootRef,
462    cut: &StoreHistoryCut,
463    registration: &ReferencedStoreDeviceRegistration,
464    acknowledgement: &'a RetainedAcknowledgementChain,
465) -> Result<&'a StoreAck, StoreProtocolError> {
466    acknowledgement.validate_chain(root, registration)?;
467    let (acknowledgement_ref, acknowledgement_value) =
468        acknowledgement.latest().ok_or_else(|| {
469            StoreProtocolError::Malformed(
470                "retained snapshot acknowledgement proof chain is empty".to_string(),
471            )
472        })?;
473    let commit_bytes = acknowledgement.activating_commit_value.to_bytes();
474    acknowledgement
475        .activating_commit
476        .object
477        .verify(&commit_bytes)?;
478    let parsed_commit = VerifiedStoreBatchCommit::parse(
479        &commit_bytes,
480        root.store_root_hash,
481        &acknowledgement.activating_commit,
482        registration.value(),
483    )?;
484    if parsed_commit.value() != &acknowledgement.activating_commit_value
485        || parsed_commit.commit_hash() != acknowledgement.activating_commit.commit_hash
486        || parsed_commit.acknowledgement() != Some(acknowledgement_ref)
487        || !history_cut_covers_commit(cut, &acknowledgement.activating_commit)
488    {
489        return Err(StoreProtocolError::Malformed(
490            "retained snapshot acknowledgement differs from its activated commit".to_string(),
491        ));
492    }
493    Ok(acknowledgement_value)
494}
495
496fn history_cut_covers_commit(cut: &StoreHistoryCut, reference: &StoreBatchCommitRef) -> bool {
497    let covered = CommitFrontier(BTreeMap::from([(
498        reference.coord.stream_id,
499        reference.clone(),
500    )]));
501    cut.frontier().covers(&covered)
502}
503
504impl RetainedVerifiedActivatedAck {
505    pub fn acknowledgement(&self) -> &(StoreAckRef, StoreAck) {
506        &self.acknowledgement
507    }
508}
509
510impl RetainedAcknowledgementChain {
511    /// Start a chain from the one acknowledgement a commit activated. Contiguity
512    /// is not claimed yet: [`extend`](Self::extend) adds the rest, and
513    /// [`validate_chain`](Self::validate_chain) is what asserts the result runs
514    /// from sequence one.
515    pub fn activated(
516        activated: &RetainedVerifiedActivatedAck,
517        activating_commit_value: &StoreBatchCommit,
518    ) -> Self {
519        let (reference, value) = activated.acknowledgement.clone();
520        Self {
521            chain: BTreeMap::from([(reference.sequence, (reference, value))]),
522            activating_commit: activated.activating_commit.clone(),
523            activating_commit_value: activating_commit_value.clone(),
524        }
525    }
526
527    /// Fold one more retained acknowledgement in. A sequence already present
528    /// must carry the same acknowledgement — two different ones at one sequence
529    /// is a forked chain, not a longer one. The activating commit tracks the
530    /// highest sequence, which is the one the summary reports.
531    pub fn extend(
532        &mut self,
533        activated: &RetainedVerifiedActivatedAck,
534        activating_commit_value: &StoreBatchCommit,
535    ) -> bool {
536        let (reference, value) = &activated.acknowledgement;
537        match self.chain.get(&reference.sequence) {
538            Some(existing) if existing == &activated.acknowledgement => {}
539            Some(_) => return false,
540            None => {
541                self.chain
542                    .insert(reference.sequence, (reference.clone(), value.clone()));
543            }
544        }
545        if self
546            .latest()
547            .is_some_and(|(latest, _)| latest.sequence == reference.sequence)
548        {
549            self.activating_commit = activated.activating_commit.clone();
550            self.activating_commit_value = activating_commit_value.clone();
551        }
552        true
553    }
554
555    pub fn latest(&self) -> Option<&(StoreAckRef, StoreAck)> {
556        self.chain
557            .last_key_value()
558            .map(|(_, acknowledgement)| acknowledgement)
559    }
560
561    pub fn exactly_extends(&self, predecessor: &Self) -> bool {
562        self.chain.len() > predecessor.chain.len()
563            && predecessor.chain.iter().all(|(sequence, acknowledgement)| {
564                self.chain.get(sequence) == Some(acknowledgement)
565            })
566    }
567
568    pub fn validate_chain(
569        &self,
570        root: &StoreRootRef,
571        registration: &ReferencedStoreDeviceRegistration,
572    ) -> Result<(), StoreProtocolError> {
573        if self.chain.is_empty() {
574            return Err(StoreProtocolError::DeviceStateMismatch);
575        }
576        let mut predecessor: Option<&StoreAckRef> = None;
577        for (expected_sequence, (sequence, (reference, value))) in (1_u64..).zip(self.chain.iter())
578        {
579            if *sequence != expected_sequence
580                || reference.sequence != expected_sequence
581                || value.sequence != expected_sequence
582                || reference.registration != *registration.reference()
583                || value.registration != *registration.reference()
584                || value.successor.predecessor.as_ref()
585                    != predecessor.map(|reference| &reference.object)
586            {
587                return Err(StoreProtocolError::DeviceStateMismatch);
588            }
589            reference.object.verify(&value.to_bytes())?;
590            StoreAck::parse_at(&value.to_bytes(), root, reference, registration.value())?;
591            predecessor = Some(reference);
592        }
593        Ok(())
594    }
595}
596
597#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
598#[serde(deny_unknown_fields)]
599pub struct RetainedAcceptedStoreAnnouncement {
600    pub reference: StoreDeviceHeadRef,
601    pub value: StoreDeviceHead,
602}
603
604#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
605#[serde(deny_unknown_fields)]
606pub struct RetainedMergeMembershipProof {
607    pub commit: StoreBatchCommitRef,
608    pub commit_value: StoreBatchCommit,
609    pub announcement: Option<RetainedAcceptedStoreAnnouncement>,
610    pub entry: MembershipEntryRef,
611    pub entry_value: MembershipEntry,
612    pub head: MembershipHeadRef,
613    pub head_value: AuthorHead,
614    pub resolution: Option<StoreMembershipConflictResolutionRef>,
615    pub resolution_value: Option<StoreMembershipConflictResolution>,
616}
617
618/// The proof values introduced by one verified Merge commit and retained with
619/// that commit after its remote authority objects can be reclaimed.
620#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
621#[serde(deny_unknown_fields)]
622pub struct RetainedMergeCommitEvidence {
623    pub acknowledgement: Option<Box<RetainedVerifiedActivatedAck>>,
624    pub membership_proof: Option<Box<RetainedMergeMembershipProof>>,
625}
626
627impl RetainedMergeCommitEvidence {
628    pub fn none() -> Self {
629        Self {
630            acknowledgement: None,
631            membership_proof: None,
632        }
633    }
634
635    pub fn validate_for(
636        &self,
637        commit_ref: &StoreBatchCommitRef,
638        commit: &StoreBatchCommit,
639    ) -> Result<(), StoreProtocolError> {
640        commit_ref.verify_commit(commit)?;
641        if commit.acknowledgement().is_some() != self.acknowledgement.is_some()
642            || commit.control().is_some() != self.membership_proof.is_some()
643        {
644            return Err(StoreProtocolError::DeviceStateMismatch);
645        }
646        if let Some(acknowledgement) = &self.acknowledgement {
647            let (reference, _) = acknowledgement.acknowledgement();
648            if acknowledgement.activating_commit != *commit_ref
649                || commit.acknowledgement() != Some(reference)
650            {
651                return Err(StoreProtocolError::DeviceStateMismatch);
652            }
653        }
654        if let Some(proof) = &self.membership_proof {
655            if proof.commit != *commit_ref || proof.commit_value != *commit {
656                return Err(StoreProtocolError::DeviceStateMismatch);
657            }
658            let control = commit
659                .control()
660                .ok_or(StoreProtocolError::DeviceStateMismatch)?;
661            if control.transition.body.entry != proof.entry
662                || proof.entry.coord != proof.entry_value.coord()
663                || !crate::membership::verify_membership_entry(&proof.entry_value)
664                || !control
665                    .transition
666                    .matches_head(&proof.head_value, &proof.head)
667                || !matches!(
668                    &proof.head_value.activation,
669                    crate::membership::MembershipHeadActivation::StoreCommit { commit }
670                        if commit == commit_ref
671                )
672            {
673                return Err(StoreProtocolError::DeviceStateMismatch);
674            }
675            proof
676                .entry
677                .object
678                .verify(&serde_json::to_vec(&proof.entry_value)?)?;
679            proof
680                .head
681                .object
682                .verify(&serde_json::to_vec(&proof.head_value)?)?;
683            match (
684                &proof.entry_value.change,
685                &proof.resolution,
686                &proof.resolution_value,
687            ) {
688                (
689                    crate::membership::MembershipChange::ResolutionActivation { resolution },
690                    Some(reference),
691                    Some(value),
692                ) if resolution == reference
693                    && value.store_root_hash == commit.store_root_hash
694                    && value.resolution_ref(reference.object.clone()) == *reference
695                    && value.verify_signature() =>
696                {
697                    reference.object.verify(&serde_json::to_vec(value)?)?;
698                }
699                (crate::membership::MembershipChange::ResolutionActivation { .. }, _, _)
700                | (_, Some(_), _)
701                | (_, _, Some(_)) => return Err(StoreProtocolError::DeviceStateMismatch),
702                _ => {}
703            }
704        }
705        Ok(())
706    }
707}
708
709#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
710#[serde(deny_unknown_fields)]
711pub struct RetainedVerifiedMergeHistorySummary {
712    pub version: u32,
713    pub store_root_hash: ObjectHash,
714    #[serde(with = "ordered_map_entries")]
715    pub causal_cut: BTreeMap<StoreCommitCoord, StoreBatchCommitRef>,
716    pub post_state: StoreDeviceStateRef,
717    pub membership_floor: MembershipCausalFloor,
718    #[serde(with = "ordered_map_entries")]
719    pub registrations: BTreeMap<StoreDeviceId, ReferencedStoreDeviceRegistration>,
720    #[serde(with = "ordered_map_entries")]
721    pub acknowledgements: BTreeMap<StoreDeviceId, RetainedAcknowledgementChain>,
722    #[serde(with = "ordered_map_entries")]
723    pub membership_proofs: BTreeMap<StoreBatchCommitRef, RetainedMergeMembershipProof>,
724    #[serde(with = "ordered_map_entries")]
725    pub announcement_frontier: BTreeMap<AuthorStreamId, RetainedAcceptedStoreAnnouncement>,
726}
727
728#[derive(Debug, Clone)]
729pub struct OpenedRetainedMergeHistorySummary {
730    pub summary: RetainedVerifiedMergeHistorySummary,
731    pub announcement_frontier: BTreeMap<AuthorStreamId, RetainedAcceptedStoreAnnouncement>,
732    pub post_state: ResolvedStoreDeviceState,
733}
734
735impl RetainedVerifiedMergeHistorySummary {
736    pub fn frontier(
737        &self,
738    ) -> Result<BTreeMap<AuthorStreamId, StoreBatchCommitRef>, StoreProtocolError> {
739        let mut frontier = BTreeMap::new();
740        for reference in self.causal_cut.values() {
741            let stream_id = reference.coord.stream_id;
742            let sequence = reference.coord.sequence;
743            match frontier.entry(stream_id) {
744                std::collections::btree_map::Entry::Vacant(entry) => {
745                    entry.insert(reference.clone());
746                }
747                std::collections::btree_map::Entry::Occupied(mut entry) => {
748                    if sequence > entry.get().coord.sequence() {
749                        entry.insert(reference.clone());
750                    }
751                }
752            }
753        }
754        Ok(frontier)
755    }
756
757    pub fn validate_shape(&self) -> Result<(), StoreProtocolError> {
758        require_version(self.version)?;
759        self.membership_floor.validate()?;
760        for (coord, reference) in &self.causal_cut {
761            if coord != &reference.coord {
762                return Err(StoreProtocolError::Malformed(
763                    "Merge history causal cut contains a mismatched coordinate".to_string(),
764                ));
765            }
766        }
767        let expected_frontier = CommitFrontier(self.frontier()?);
768        if self.post_state.frontier() != &expected_frontier {
769            return Err(StoreProtocolError::DeviceStateMismatch);
770        }
771        for (device_id, registration) in &self.registrations {
772            if device_id != &registration.reference().device_id
773                || registration.value().store_root.store_root_hash != self.store_root_hash
774            {
775                return Err(StoreProtocolError::DeviceStateMismatch);
776            }
777            registration
778                .reference()
779                .verify_registration(registration.value())?;
780            registration
781                .reference()
782                .object
783                .verify(&registration.value().to_bytes())?;
784            StoreDeviceRegistration::parse_at(
785                &registration.value().to_bytes(),
786                &registration.value().store_root,
787                *device_id,
788            )?;
789        }
790        for (device_id, acknowledgement) in &self.acknowledgements {
791            let registration = self
792                .registrations
793                .get(device_id)
794                .ok_or(StoreProtocolError::DeviceStateMismatch)?;
795            acknowledgement.validate_chain(&registration.value().store_root, registration)?;
796            let (acknowledgement_ref, acknowledgement_value) = acknowledgement
797                .latest()
798                .ok_or(StoreProtocolError::DeviceStateMismatch)?;
799            acknowledgement
800                .activating_commit
801                .verify_commit(&acknowledgement.activating_commit_value)?;
802            if device_id != &acknowledgement_ref.registration.device_id
803                || acknowledgement.activating_commit_value.acknowledgement()
804                    != Some(acknowledgement_ref)
805                || acknowledgement.activating_commit_value.author_registration
806                    != *registration.reference()
807                || self
808                    .causal_cut
809                    .get(&acknowledgement.activating_commit.coord)
810                    != Some(&acknowledgement.activating_commit)
811            {
812                return Err(StoreProtocolError::DeviceStateMismatch);
813            }
814            let predecessor_cut = acknowledgement
815                .activating_commit_value
816                .order
817                .predecessor_cut()?;
818            if acknowledgement_value.store_cut != predecessor_cut
819                || acknowledgement_value.device_state
820                    != acknowledgement.activating_commit_value.device_state
821            {
822                return Err(StoreProtocolError::DeviceStateMismatch);
823            }
824        }
825        for (reference, proof) in &self.membership_proofs {
826            if reference != &proof.commit
827                || self.causal_cut.get(&proof.commit.coord) != Some(&proof.commit)
828            {
829                return Err(StoreProtocolError::DeviceStateMismatch);
830            }
831            proof.commit.verify_commit(&proof.commit_value)?;
832            let Some(control) = proof.commit_value.control() else {
833                return Err(StoreProtocolError::DeviceStateMismatch);
834            };
835            let transition = &control.transition;
836            if transition.body.entry != proof.entry
837                || proof.entry.coord != proof.entry_value.coord()
838                || !crate::membership::verify_membership_entry(&proof.entry_value)
839            {
840                return Err(StoreProtocolError::DeviceStateMismatch);
841            }
842            proof
843                .entry
844                .object
845                .verify(&serde_json::to_vec(&proof.entry_value)?)?;
846            let head_author = self
847                .registrations
848                .get(&proof.head_value.body.author_registration.device_id)
849                .ok_or(StoreProtocolError::DeviceStateMismatch)?;
850            if !transition.matches_head(&proof.head_value, &proof.head)
851                || !proof.head_value.verify(head_author.value())
852                || !matches!(
853                    &proof.head_value.activation,
854                    crate::membership::MembershipHeadActivation::StoreCommit { commit }
855                        if commit == &proof.commit
856                )
857            {
858                return Err(StoreProtocolError::DeviceStateMismatch);
859            }
860            proof
861                .head
862                .object
863                .verify(&serde_json::to_vec(&proof.head_value)?)?;
864            match (
865                &proof.entry_value.change,
866                &proof.resolution,
867                &proof.resolution_value,
868            ) {
869                (
870                    crate::membership::MembershipChange::ResolutionActivation { resolution },
871                    Some(reference),
872                    Some(value),
873                ) if resolution == reference
874                    && value.store_root_hash == self.store_root_hash
875                    && value.resolution_ref(reference.object.clone()) == *reference
876                    && value.verify_signature() =>
877                {
878                    reference.object.verify(&serde_json::to_vec(value)?)?;
879                }
880                (crate::membership::MembershipChange::ResolutionActivation { .. }, _, _)
881                | (_, Some(_), _)
882                | (_, _, Some(_)) => return Err(StoreProtocolError::DeviceStateMismatch),
883                _ => {}
884            }
885            if let Some(announcement) = &proof.announcement {
886                self.validate_announcement(announcement)?;
887                if announcement.value.commit != proof.commit {
888                    return Err(StoreProtocolError::DeviceStateMismatch);
889                }
890            }
891        }
892        for (stream_id, announcement) in &self.announcement_frontier {
893            self.validate_announcement(announcement)?;
894            if announcement.value.commit.coord.stream_id != *stream_id
895                || self.causal_cut.get(&announcement.value.commit.coord)
896                    != Some(&announcement.value.commit)
897            {
898                return Err(StoreProtocolError::DeviceStateMismatch);
899            }
900        }
901        Ok(())
902    }
903
904    pub fn validate_snapshot_baseline(&self) -> Result<(), StoreProtocolError> {
905        self.validate_shape()?;
906        let frontier = self.frontier()?;
907        if self.announcement_frontier.len() != frontier.len()
908            || frontier.iter().any(|(stream_id, commit)| {
909                self.announcement_frontier
910                    .get(stream_id)
911                    .is_none_or(|announcement| announcement.value.commit != *commit)
912            })
913            || self
914                .membership_proofs
915                .values()
916                .any(|proof| proof.announcement.is_none())
917        {
918            return Err(StoreProtocolError::DeviceStateMismatch);
919        }
920        Ok(())
921    }
922
923    fn validate_announcement(
924        &self,
925        announcement: &RetainedAcceptedStoreAnnouncement,
926    ) -> Result<(), StoreProtocolError> {
927        let registration = self
928            .registrations
929            .get(&announcement.value.author_registration.device_id)
930            .ok_or(StoreProtocolError::DeviceStateMismatch)?;
931        if announcement.value.store_root_hash != self.store_root_hash
932            || announcement.value.author_registration != *registration.reference()
933            || announcement.reference.head_hash != announcement.value.head_hash()
934        {
935            return Err(StoreProtocolError::DeviceStateMismatch);
936        }
937        announcement
938            .reference
939            .object
940            .verify(&announcement.value.to_bytes())?;
941        StoreDeviceHead::parse_at(
942            &announcement.value.to_bytes(),
943            self.store_root_hash,
944            registration.value(),
945            &announcement.value.commit,
946        )?;
947        Ok(())
948    }
949}