coven_protocol/store_commit/
ack_snapshot.rs

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

coven_protocol/store_commit/
ack_snapshot.rs
