coven_protocol/store_commit/
ack_snapshot.rs

1use super::validation::{
2    validate_ack_state, validate_commit_frontier, validate_store_device_state_ref,
3    validate_successor_sequence,
4};
5use super::*;
6
7#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
8#[serde(deny_unknown_fields)]
9pub struct StoreAckBody {
10    pub store_root_hash: ObjectHash,
11    pub registration: StoreDeviceRegistrationRef,
12    pub sequence: u64,
13    pub store_cut: StoreHistoryCut,
14    pub device_state: StoreDeviceStateRef,
15    pub snapshot: Option<StoreSnapshotLocator>,
16    pub exclusions: StoreAckExclusionState,
17    pub last_sync: String,
18    pub successor: SuccessorLink,
19}
20
21impl SignedBody for StoreAckBody {
22    const DOMAIN: &'static [u8] = ACK_DOMAIN;
23}
24
25/// What a Store acknowledgement asserts, apart from the bookkeeping every
26/// acknowledgement carries fresh: its sequence, the wall clock it was written
27/// at, and its links to the neighbours in the device's acknowledgement chain.
28///
29/// Two acknowledgements with equal assertions tell every reader the same thing.
30/// That matters because publishing an acknowledgement appends a commit, and a
31/// commit that says nothing new still lands in every device's history, every
32/// retained materialization, and every snapshot — a store that is doing nothing
33/// grows one commit per device per cycle, forever.
34#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
35#[serde(deny_unknown_fields)]
36pub struct StoreAckAssertion {
37    pub registration: StoreDeviceRegistrationRef,
38    pub store_cut: StoreHistoryCut,
39    pub device_state: StoreDeviceStateRef,
40    pub snapshot: Option<StoreSnapshotLocator>,
41    pub exclusions: StoreAckExclusionState,
42}
43
44/// The acknowledgement a device currently stands behind: what it asserted, and
45/// the commit that carried it.
46///
47/// Kept so the next cycle can ask whether that acknowledgement still says
48/// everything true, instead of publishing another one to find out.
49#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
50#[serde(deny_unknown_fields)]
51pub struct StandingStoreAck {
52    pub assertion: StoreAckAssertion,
53    /// The commit that published it, when it activated one.
54    ///
55    /// An acknowledgement cannot cover the commit that carries it — that commit
56    /// does not exist until the acknowledgement is signed. So this is the one
57    /// position in the Store's history the assertion leaves behind on purpose,
58    /// and the one advance that is not a reason to acknowledge again. Without it
59    /// a device acknowledges its own acknowledgement, cycle after cycle, forever.
60    pub activating_commit: Option<StoreBatchCommitRef>,
61}
62
63impl StandingStoreAck {
64    /// Whether `assertion` says exactly what this acknowledgement already said.
65    ///
66    /// Destructured exhaustively for the same reason [`StoreAckBody::assertion`]
67    /// is: a field added to the assertion has to be compared here, or the
68    /// acknowledgement could change without anything noticing.
69    pub fn still_holds(&self, assertion: &StoreAckAssertion) -> bool {
70        let StoreAckAssertion {
71            registration,
72            store_cut,
73            device_state,
74            snapshot,
75            exclusions,
76        } = assertion;
77        if registration != &self.assertion.registration
78            || snapshot != &self.assertion.snapshot
79            || exclusions != &self.assertion.exclusions
80        {
81            return false;
82        }
83        // A device-state reference names both the device set and the cut it was
84        // read at. The cut half moves with `store_cut` and is compared there; the
85        // device set is what this reference actually asserts.
86        if device_state.state_hash() != self.assertion.device_state.state_hash()
87            || device_state.recovery() != self.assertion.device_state.recovery()
88        {
89            return false;
90        }
91        *store_cut == self.covered_cut()
92    }
93
94    /// The Store history this acknowledgement leaves behind it: what it asserted,
95    /// plus the commit that carried it. A frontier equal to this one holds
96    /// nothing the standing acknowledgement has not already accounted for.
97    fn covered_cut(&self) -> StoreHistoryCut {
98        let mut cut = self.assertion.store_cut.0.clone();
99        if let Some(commit) = &self.activating_commit {
100            cut.insert(commit.coord.stream_id, commit.clone());
101        }
102        StoreHistoryCut(cut)
103    }
104}
105
106impl StoreAckBody {
107    /// The assertion this body makes.
108    ///
109    /// Destructured exhaustively on purpose: a field added to the body has to be
110    /// classified here as asserted or as bookkeeping, or this stops compiling. A
111    /// silently unclassified field would be one an acknowledgement could change
112    /// without anything noticing it had changed.
113    pub fn assertion(&self) -> StoreAckAssertion {
114        let Self {
115            store_root_hash: _,
116            registration,
117            sequence: _,
118            store_cut,
119            device_state,
120            snapshot,
121            exclusions,
122            last_sync: _,
123            successor: _,
124        } = self;
125        StoreAckAssertion {
126            registration: registration.clone(),
127            store_cut: store_cut.clone(),
128            device_state: device_state.clone(),
129            snapshot: snapshot.clone(),
130            exclusions: exclusions.clone(),
131        }
132    }
133}
134
135pub type StoreAck = Signed<StoreAckBody>;
136
137#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
138#[serde(deny_unknown_fields)]
139pub struct StoreAckRef {
140    pub registration: StoreDeviceRegistrationRef,
141    pub sequence: u64,
142    pub ack_hash: ObjectHash,
143    pub object: ExactObjectRef,
144}
145
146#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
147#[serde(deny_unknown_fields)]
148pub struct StoreSnapshotLocator {
149    pub author_registration: StoreDeviceRegistrationRef,
150    pub snapshot: StoreSnapshotRef,
151}
152
153/// The exact membership and device state represented by one Store snapshot.
154#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
155#[serde(deny_unknown_fields)]
156pub struct StoreSnapshotState {
157    pub membership: StoreMembershipStateRef,
158    pub devices: StoreDeviceStateRef,
159}
160
161impl StoreSnapshotState {
162    fn validate(
163        &self,
164        store_root_hash: ObjectHash,
165        coverage: &CommitFrontier,
166    ) -> Result<(), StoreProtocolError> {
167        self.membership.validate_shape()?;
168        validate_store_device_state_ref(&self.devices)?;
169        if self.membership.recovery() != self.devices.recovery() {
170            return Err(StoreProtocolError::OwnerRecoveryMismatch);
171        }
172        if self.devices.frontier() != coverage {
173            return Err(StoreProtocolError::DeviceStateMismatch);
174        }
175        let _ = store_root_hash;
176        Ok(())
177    }
178}
179
180#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
181#[serde(deny_unknown_fields)]
182pub struct StoreAckExclusionState {
183    pub proposal_freezes: Vec<StoreDeviceProposalAck>,
184}
185
186#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
187#[serde(deny_unknown_fields)]
188pub struct StoreDeviceProposalAck {
189    pub proposal: StoreDeviceExclusionProposalRef,
190    pub target_cut: StoreHistoryCut,
191}
192
193impl StoreAck {
194    pub fn signed(
195        store_root_hash: ObjectHash,
196        sequence: u64,
197        assertion: StoreAckAssertion,
198        last_sync: String,
199        successor: SuccessorLink,
200        device_signer: &UserKeypair,
201    ) -> Result<Self, StoreProtocolError> {
202        validate_successor_sequence(sequence, &successor)?;
203        let StoreAckAssertion {
204            registration,
205            store_cut,
206            device_state,
207            snapshot,
208            exclusions,
209        } = assertion;
210        validate_ack_state(
211            store_root_hash,
212            &registration,
213            &store_cut,
214            &device_state,
215            &exclusions,
216        )?;
217        Ok(Signed::sign(
218            StoreAckBody {
219                store_root_hash,
220                registration,
221                sequence,
222                store_cut,
223                device_state,
224                snapshot,
225                exclusions,
226                last_sync,
227                successor,
228            },
229            device_signer,
230        ))
231    }
232
233    pub fn ack_hash(&self) -> ObjectHash {
234        self.hash()
235    }
236
237    pub fn semantic_hash_from_bytes(bytes: &[u8]) -> Result<ObjectHash, StoreProtocolError> {
238        let ack: Self = crate::objects::decode_protocol_object(bytes)?;
239        Ok(ack.ack_hash())
240    }
241
242    pub fn parse_at(
243        bytes: &[u8],
244        expected_store_root: &StoreRootRef,
245        expected: &StoreAckRef,
246        author: &StoreDeviceRegistration,
247    ) -> Result<Self, StoreProtocolError> {
248        let ack: Self = crate::objects::decode_protocol_object(bytes)?;
249        ack.require_version()?;
250        crate::objects::verify_store_root(
251            expected_store_root.store_root_hash,
252            ack.store_root_hash,
253        )?;
254        ack.registration.verify_registration(author)?;
255        if ack.registration != expected.registration {
256            return Err(StoreProtocolError::DeviceRegistrationRefMismatch {
257                device_id: expected.registration.device_id.to_string(),
258                expected: expected.registration.registration_hash,
259                actual: ack.registration.registration_hash,
260            });
261        }
262        if ack.sequence != expected.sequence {
263            return Err(StoreProtocolError::RelocatedSlot {
264                expected: ack_slot_prefix(&author.device_id.to_string(), expected.sequence),
265                actual: ack_slot_prefix(&author.device_id.to_string(), ack.sequence),
266            });
267        }
268        validate_successor_sequence(ack.sequence, &ack.successor)?;
269        validate_ack_state(
270            ack.store_root_hash,
271            &ack.registration,
272            &ack.store_cut,
273            &ack.device_state,
274            &ack.exclusions,
275        )?;
276        let activation = author
277            .store_acknowledgement_activation(&ack.registration)?
278            .activation_id();
279        if ack.successor.activation != activation {
280            return Err(StoreProtocolError::Malformed(
281                "Store acknowledgement successor uses another stream activation".to_string(),
282            ));
283        }
284        ack.verify_by(&author.device_signing_pubkey)?;
285        if ack.ack_hash() != expected.ack_hash {
286            return Err(StoreProtocolError::ObjectHashMismatch {
287                expected: expected.ack_hash,
288                actual: ack.ack_hash(),
289            });
290        }
291        Ok(ack)
292    }
293}
294
295#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
296#[serde(deny_unknown_fields)]
297pub struct SnapshotMetaBody {
298    pub store_root_hash: ObjectHash,
299    pub author_registration: StoreDeviceRegistrationRef,
300    pub generation: u64,
301    pub predecessor: Option<StoreSnapshotRef>,
302    pub image: SnapshotImageRef,
303    /// The membership objects a reader needs to reach `state.membership`,
304    /// published beside the image. A joining device opens the Store keyring out
305    /// of the membership chain, so it cannot read anything the keyring
306    /// protects — this reference is what lets it take the chain in one read
307    /// instead of two round trips per membership change.
308    pub membership_rollup: MembershipRollupRef,
309    pub coverage: CommitFrontier,
310    pub state: StoreSnapshotState,
311    pub history_summary: RetainedVerifiedMergeHistorySummary,
312    pub schema_version: u32,
313    pub created_at: String,
314    pub successor: SnapshotSuccessorLink,
315}
316
317impl SignedBody for SnapshotMetaBody {
318    const DOMAIN: &'static [u8] = SNAPSHOT_DOMAIN;
319}
320
321pub type SnapshotMeta = Signed<SnapshotMetaBody>;
322
323impl RetainedVerifiedMergeHistorySummary {
324    fn validate(
325        &self,
326        store_root_hash: ObjectHash,
327        coverage: &CommitFrontier,
328        state: &StoreSnapshotState,
329    ) -> Result<(), StoreProtocolError> {
330        self.validate_snapshot_baseline()?;
331        if self.store_root_hash != store_root_hash
332            || self.frontier()? != coverage.0
333            || self.post_state != state.devices
334        {
335            return Err(StoreProtocolError::DeviceStateMismatch);
336        }
337        Ok(())
338    }
339}
340
341#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
342#[serde(deny_unknown_fields)]
343pub struct SnapshotImageRef {
344    pub image_hash: ObjectHash,
345    pub object: ExactObjectRef,
346}
347
348#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
349#[serde(deny_unknown_fields)]
350pub struct StoreSnapshotRef {
351    pub generation: u64,
352    pub snapshot_hash: ObjectHash,
353    pub object: ExactObjectRef,
354}
355
356#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
357#[serde(deny_unknown_fields)]
358pub struct SnapshotSuccessorLink {
359    pub activation: StreamActivationId,
360    pub predecessor: Option<StoreSnapshotRef>,
361    pub next_slot: ObjectSlot,
362}
363
364impl SnapshotMeta {
365    pub fn signed(
366        store_root_hash: ObjectHash,
367        author_registration: StoreDeviceRegistrationRef,
368        generation: u64,
369        predecessor: Option<StoreSnapshotRef>,
370        image: SnapshotImageRef,
371        membership_rollup: MembershipRollupRef,
372        coverage: CommitFrontier,
373        state: StoreSnapshotState,
374        history_summary: RetainedVerifiedMergeHistorySummary,
375        schema_version: u32,
376        created_at: String,
377        successor: SnapshotSuccessorLink,
378        device_signer: &UserKeypair,
379    ) -> Result<Self, StoreProtocolError> {
380        validate_snapshot_generation(generation, predecessor.as_ref())?;
381        validate_commit_frontier(&coverage)?;
382        state.validate(store_root_hash, &coverage)?;
383        history_summary.validate(store_root_hash, &coverage, &state)?;
384        Ok(Signed::sign(
385            SnapshotMetaBody {
386                store_root_hash,
387                author_registration,
388                generation,
389                predecessor,
390                image,
391                membership_rollup,
392                coverage,
393                state,
394                history_summary,
395                schema_version,
396                created_at,
397                successor,
398            },
399            device_signer,
400        ))
401    }
402
403    pub fn snapshot_hash(&self) -> ObjectHash {
404        self.hash()
405    }
406
407    pub fn semantic_hash_from_bytes(bytes: &[u8]) -> Result<ObjectHash, StoreProtocolError> {
408        let meta: Self = crate::objects::decode_protocol_object(bytes)?;
409        Ok(meta.snapshot_hash())
410    }
411
412    pub fn parse_at(
413        bytes: &[u8],
414        expected_store_root_hash: ObjectHash,
415        expected: &StoreSnapshotRef,
416        author: &StoreDeviceRegistration,
417    ) -> Result<Self, StoreProtocolError> {
418        let meta: Self = crate::objects::decode_protocol_object(bytes)?;
419        meta.require_version()?;
420        crate::objects::verify_store_root(expected_store_root_hash, meta.store_root_hash)?;
421        meta.author_registration.verify_registration(author)?;
422        if meta.generation != expected.generation {
423            return Err(StoreProtocolError::RelocatedSlot {
424                expected: snapshot_semantic_prefix(
425                    &author.device_id.to_string(),
426                    expected.snapshot_hash,
427                ),
428                actual: snapshot_semantic_prefix(
429                    &author.device_id.to_string(),
430                    meta.snapshot_hash(),
431                ),
432            });
433        }
434        validate_snapshot_generation(meta.generation, meta.predecessor.as_ref())?;
435        validate_commit_frontier(&meta.coverage)?;
436        meta.state
437            .validate(expected_store_root_hash, &meta.coverage)?;
438        meta.history_summary
439            .validate(expected_store_root_hash, &meta.coverage, &meta.state)?;
440        meta.verify_by(&author.device_signing_pubkey)?;
441        let actual = meta.snapshot_hash();
442        if actual != expected.snapshot_hash {
443            return Err(StoreProtocolError::ObjectHashMismatch {
444                expected: expected.snapshot_hash,
445                actual,
446            });
447        }
448        Ok(meta)
449    }
450
451    pub fn parse_stream_entry_at(
452        bytes: &[u8],
453        expected_store_root: &StoreRootRef,
454        expected_registration: &StoreDeviceRegistrationRef,
455        author: &StoreDeviceRegistration,
456        expected: &StoreSnapshotRef,
457    ) -> Result<Self, StoreProtocolError> {
458        let meta = Self::parse_at(bytes, expected_store_root.store_root_hash, expected, author)?;
459        let next_generation = expected.generation.checked_add(1).ok_or_else(|| {
460            StoreProtocolError::Malformed("Store snapshot generation overflow".to_string())
461        })?;
462        let activation = author
463            .store_snapshot_activation(expected_registration)?
464            .activation_id();
465        if meta.author_registration != *expected_registration
466            || meta.successor.activation != activation
467            || meta.successor.predecessor != meta.predecessor
468            || meta.successor.next_slot.logical_key()
469                != format!(
470                    "{}.json",
471                    snapshot_slot_prefix(&author.device_id.to_string(), next_generation)
472                )
473        {
474            return Err(StoreProtocolError::Malformed(
475                "Store snapshot metadata is outside its activated exact stream".to_string(),
476            ));
477        }
478        Ok(meta)
479    }
480}
481
482fn validate_snapshot_generation(
483    generation: u64,
484    predecessor: Option<&StoreSnapshotRef>,
485) -> Result<(), StoreProtocolError> {
486    match (generation, predecessor) {
487        (0, None) => Ok(()),
488        (0, Some(_)) | (_, None) => Err(StoreProtocolError::Malformed(
489            "Store snapshot generation and predecessor disagree".to_string(),
490        )),
491        (generation, Some(predecessor)) => {
492            let expected = predecessor.generation.checked_add(1).ok_or_else(|| {
493                StoreProtocolError::Malformed("Store snapshot generation overflow".to_string())
494            })?;
495            if generation != expected {
496                return Err(StoreProtocolError::Malformed(
497                    "Store snapshot generation does not follow its predecessor".to_string(),
498                ));
499            }
500            Ok(())
501        }
502    }
503}
coven_protocol/store_commit/ack_snapshot.rs

coven_protocol/store_commit/
ack_snapshot.rs
