coven_database/store/store_session/
acknowledgements.rs

1use super::*;
2use crate::store_ack_records::{
3    load_expected_outbound_store_ack_on, set_outbound_store_ack_activation_on,
4};
5use coven_protocol::remote_object::CandidateNonactivation;
6
7impl StoreSession<'_> {
8    fn prepare_acknowledgement_activation(
9        &mut self,
10        expected: &StoreAckRef,
11        candidate: PreparedStoreOperationCommit,
12    ) -> Result<(), DbError> {
13        let authority = self.local_store_authority()?;
14        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
15        let outbound = load_expected_outbound_store_ack_on(
16            &tx,
17            &authority,
18            expected,
19            "prepared activation names a different Store acknowledgement",
20        )?;
21        match outbound.activation {
22            OutboundStoreAckActivation::AwaitingCandidate => {}
23            OutboundStoreAckActivation::Prepared(existing)
24                if existing.reference == candidate.reference =>
25            {
26                return Ok(());
27            }
28            OutboundStoreAckActivation::Prepared(_)
29            | OutboundStoreAckActivation::Nonactivating(_) => {
30                return Err(DbError::Message(
31                    "Store acknowledgement already has a different activation candidate"
32                        .to_string(),
33                ));
34            }
35        }
36        for remote in candidate
37            .acknowledgement_remote_objects(&outbound.ack)
38            .map_err(DbError::from)?
39        {
40            persist_exact_remote_object_on(
41                &tx,
42                self.store_dir,
43                &remote,
44                "Merge Store acknowledgement activation object",
45            )?;
46        }
47        for circle in &outbound.circle_acknowledgements {
48            for remote in candidate
49                .circle_acknowledgement_remote_objects(&circle.ack)
50                .map_err(DbError::from)?
51            {
52                persist_exact_remote_object_on(
53                    &tx,
54                    self.store_dir,
55                    &remote,
56                    "Merge Circle acknowledgement activation object",
57                )?;
58            }
59        }
60        set_outbound_store_ack_activation_on(
61            &tx,
62            expected,
63            &OutboundStoreAckActivation::Prepared(candidate),
64            "outbound Store acknowledgement disappeared during activation preparation",
65        )?;
66        tx.commit().map_err(DbError::from)
67    }
68
69    fn begin_acknowledgement_nonactivation(
70        &mut self,
71        expected: &StoreAckRef,
72        verified_candidate: &StoreBatchCommitRef,
73        nonactivation: CandidateNonactivation,
74    ) -> Result<(), DbError> {
75        let authority = self.local_store_authority()?;
76        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
77        let outbound = load_expected_outbound_store_ack_on(
78            &tx,
79            &authority,
80            expected,
81            "nonactivation names another Store acknowledgement",
82        )?;
83        let (candidate, already_nonactivating) = match outbound.activation {
84            OutboundStoreAckActivation::Prepared(candidate) => (candidate, false),
85            OutboundStoreAckActivation::Nonactivating(candidate) => (candidate, true),
86            OutboundStoreAckActivation::AwaitingCandidate => {
87                return Err(DbError::Message(
88                    "Store acknowledgement has no prepared Merge activation candidate".to_string(),
89                ));
90            }
91        };
92        if &candidate.reference != verified_candidate {
93            return Err(DbError::Message(
94                "verified nonactivation names another Store acknowledgement candidate".to_string(),
95            ));
96        }
97        if nonactivation.candidate().canonical_signed_bytes != candidate.commit.to_bytes() {
98            return Err(DbError::Message(
99                "verified nonactivation bytes differ from the Store acknowledgement candidate"
100                    .to_string(),
101            ));
102        }
103        let head = candidate.head_ref();
104        if already_nonactivating {
105            let commit = load_remote_object_on(&tx, remote_object_id(&candidate.reference.object))?;
106            if commit
107                .candidate_nonactivation_proof(&candidate.reference)
108                .map_err(DbError::from)?
109                != Some(nonactivation.proof())
110            {
111                return Err(DbError::Message(
112                    "nonactivating Merge acknowledgement commit carries a different durable proof"
113                        .to_string(),
114                ));
115            }
116            let inert =
117                load_protocol_inert_object_on(&tx, remote_object_id(&outbound.reference.object))?;
118            if inert
119                .candidate_nonactivation_proof(&candidate.reference)
120                .map_err(DbError::from)?
121                != Some(nonactivation.proof())
122            {
123                return Err(DbError::Message(
124                    "nonactivating Merge acknowledgement carries a different durable proof"
125                        .to_string(),
126                ));
127            }
128            let head_remote = load_remote_object_on(&tx, remote_object_id(&head.object))?;
129            if head_remote
130                .candidate_nonactivation_proof(&candidate.reference)
131                .map_err(DbError::from)?
132                != Some(nonactivation.proof())
133            {
134                return Err(DbError::Message(
135                    "nonactivating Merge acknowledgement head carries a different durable proof"
136                        .to_string(),
137                ));
138            }
139            return Ok(());
140        }
141        if begin_remote_candidate_nonactivation_on(
142            &tx,
143            remote_object_id(&outbound.reference.object),
144            nonactivation.clone(),
145        )?
146        .is_some()
147        {
148            return Err(DbError::Message(
149                "Store acknowledgement became an exact cleanup target".to_string(),
150            ));
151        }
152        if begin_remote_candidate_nonactivation_on(
153            &tx,
154            remote_object_id(&head.object),
155            nonactivation.clone(),
156        )?
157        .is_some()
158        {
159            return Err(DbError::Message(
160                "Store activation head became an exact cleanup target".to_string(),
161            ));
162        }
163        if begin_remote_candidate_nonactivation_on(
164            &tx,
165            remote_object_id(&candidate.reference.object),
166            nonactivation,
167        )?
168        .is_none()
169        {
170            return Err(DbError::Message(
171                "losing Store acknowledgement commit has no exact cleanup target".to_string(),
172            ));
173        }
174        set_outbound_store_ack_activation_on(
175            &tx,
176            expected,
177            &OutboundStoreAckActivation::Nonactivating(candidate),
178            "outbound Store acknowledgement disappeared during nonactivation",
179        )?;
180        tx.commit().map_err(DbError::from)
181    }
182
183    fn adopt_acknowledgement_head(
184        &mut self,
185        expected: &StoreAckRef,
186        winner: StoreDeviceHead,
187        winner_prepared: PreparedExactObject,
188    ) -> Result<(), DbError> {
189        let authority = self.local_store_authority()?;
190        let outbound = load_expected_outbound_store_ack_on(
191            self.conn,
192            &authority,
193            expected,
194            "alternate Merge head names another Store acknowledgement",
195        )?;
196        let OutboundStoreAckActivation::Prepared(candidate) = outbound.activation else {
197            return Err(DbError::Message(
198                "Store acknowledgement has no prepared Merge candidate".to_string(),
199            ));
200        };
201        let registration = self.activated_registration(&candidate.commit.author_registration)?;
202        let root = &registration.value().store_root;
203        let verified = StoreDeviceHead::parse_at(
204            &winner.to_bytes(),
205            root.store_root_hash,
206            registration.value(),
207            &candidate.reference,
208        )
209        .map_err(|error| DbError::context("verify alternate Merge head", error))?;
210        if verified != winner {
211            return Err(DbError::Message(
212                "alternate Merge head changed during exact verification".to_string(),
213            ));
214        }
215        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
216        let current = candidate.head_ref();
217        replace_prepared_merge_head_remote_on(
218            &tx,
219            self.store_dir,
220            &current.object,
221            &winner,
222            winner_prepared.reference(),
223            &candidate.reference,
224        )?;
225        let mut candidate = candidate;
226        candidate
227            .adopt_merge_head(winner, winner_prepared.reference().clone())
228            .map_err(DbError::from)?;
229        set_outbound_store_ack_activation_on(
230            &tx,
231            expected,
232            &OutboundStoreAckActivation::Prepared(candidate),
233            "outbound Store acknowledgement disappeared during head adoption",
234        )?;
235        tx.commit().map_err(DbError::from)
236    }
237
238    fn acknowledgement_cleanup_target(
239        &mut self,
240        expected: &StoreAckRef,
241    ) -> Result<Option<CandidateCleanupObject>, DbError> {
242        let authority = self.local_store_authority()?;
243        let conn = self.conn;
244        let outbound = load_expected_outbound_store_ack_on(
245            conn,
246            &authority,
247            expected,
248            "Store acknowledgement cleanup names another exact object",
249        )?;
250        let OutboundStoreAckActivation::Nonactivating(candidate) = outbound.activation else {
251            return Err(DbError::Message(
252                "Store acknowledgement activation is not nonactivating Merge".to_string(),
253            ));
254        };
255        Ok(super::candidate_records::candidate_cleanup_targets_on(
256            conn,
257            &candidate.reference,
258            std::slice::from_ref(&candidate.reference.object),
259        )?
260        .into_iter()
261        .next())
262    }
263
264    fn complete_nonactivating_acknowledgement(
265        &mut self,
266        expected: &StoreAckRef,
267    ) -> Result<(), DbError> {
268        let authority = self.local_store_authority()?;
269        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
270        let outbound = load_expected_outbound_store_ack_on(
271            &tx,
272            &authority,
273            expected,
274            "Store acknowledgement completion names another exact object",
275        )?;
276        let OutboundStoreAckActivation::Nonactivating(candidate) = &outbound.activation else {
277            return Err(DbError::Message(
278                "Store acknowledgement activation is not nonactivating Merge".to_string(),
279            ));
280        };
281        let head = candidate.head_ref();
282        if !super::candidate_records::candidate_cleanup_targets_on(
283            &tx,
284            &candidate.reference,
285            &[candidate.reference.object.clone(), head.object.clone()],
286        )?
287        .is_empty()
288        {
289            return Err(DbError::Message(
290                "losing Store acknowledgement cleanup is incomplete".to_string(),
291            ));
292        }
293        let commit_id = remote_object_id(&candidate.reference.object);
294        let commit = load_remote_object_on(&tx, commit_id)?;
295        let proof = commit
296            .candidate_nonactivation_proof(&candidate.reference)
297            .map_err(DbError::from)?
298            .ok_or_else(|| {
299                DbError::Message("losing Store acknowledgement commit lacks its proof".to_string())
300            })?;
301        if !matches!(proof, CandidateNonactivationProof::MergeWinner { .. }) {
302            return Err(DbError::Message(
303                "nonactivating Merge acknowledgement carries another proof".to_string(),
304            ));
305        }
306        let head_id = remote_object_id(&head.object);
307        let head_remote = load_remote_object_on(&tx, head_id)?;
308        if head_remote
309            .candidate_nonactivation_proof(&candidate.reference)
310            .map_err(DbError::from)?
311            != Some(proof)
312        {
313            return Err(DbError::Message(
314                "losing Store acknowledgement head carries a different proof".to_string(),
315            ));
316        }
317        let inert =
318            load_protocol_inert_object_on(&tx, remote_object_id(&outbound.reference.object))?;
319        if inert
320            .candidate_nonactivation_proof(&candidate.reference)
321            .map_err(DbError::from)?
322            != Some(proof)
323        {
324            return Err(DbError::Message(
325                "protocol-inert acknowledgement lacks its candidate proof".to_string(),
326            ));
327        }
328        super::candidate_records::delete_remote_objects_on(
329            &tx,
330            [commit_id, head_id],
331            "nonactivating acknowledgement",
332        )?;
333        // A losing acknowledgement activated no commit, so the standing state
334        // names none: the next cycle compares its assertion against a history
335        // this device added nothing to.
336        finish_outbound_store_ack_on(
337            &tx,
338            expected,
339            &outbound.ack.value.successor.next_slot,
340            &coven_protocol::store_commit::StandingStoreAck {
341                assertion: outbound.ack.value.assertion(),
342                activating_commit: None,
343            },
344        )?;
345        tx.commit().map_err(DbError::from)
346    }
347}
348
349impl StoreDatabase {
350    pub async fn prepare_acknowledgement_activation(
351        &self,
352        expected: StoreAckRef,
353        candidate: PreparedStoreOperationCommit,
354    ) -> Result<(), DbError> {
355        self.call_store(move |session| {
356            session.prepare_acknowledgement_activation(&expected, candidate)
357        })
358        .await
359    }
360
361    pub async fn begin_acknowledgement_nonactivation(
362        &self,
363        expected: StoreAckRef,
364        nonactivation: VerifiedCandidateNonactivation,
365    ) -> Result<(), DbError> {
366        let verified_candidate = nonactivation.candidate_reference().map_err(DbError::from)?;
367        if !matches!(
368            nonactivation.proof(),
369            CandidateNonactivationProof::MergeWinner { .. }
370        ) {
371            return Err(DbError::Message(
372                "Merge acknowledgement requires a Merge-winner nonactivation proof".to_string(),
373            ));
374        }
375        let nonactivation = nonactivation.into_durable();
376        self.call_store(move |session| {
377            session.begin_acknowledgement_nonactivation(
378                &expected,
379                &verified_candidate,
380                nonactivation,
381            )
382        })
383        .await
384    }
385
386    pub async fn adopt_acknowledgement_head(
387        &self,
388        expected: StoreAckRef,
389        winner: StoreDeviceHead,
390        winner_prepared: PreparedExactObject,
391    ) -> Result<(), DbError> {
392        self.call_store(move |session| {
393            session.adopt_acknowledgement_head(&expected, winner, winner_prepared)
394        })
395        .await
396    }
397
398    pub async fn acknowledgement_cleanup_target(
399        &self,
400        expected: StoreAckRef,
401    ) -> Result<Option<CandidateCleanupObject>, DbError> {
402        self.call_store(move |session| session.acknowledgement_cleanup_target(&expected))
403            .await
404    }
405
406    pub async fn complete_nonactivating_acknowledgement(
407        &self,
408        expected: StoreAckRef,
409    ) -> Result<(), DbError> {
410        self.call_store(move |session| session.complete_nonactivating_acknowledgement(&expected))
411            .await
412    }
413}
coven_database/store/store_session/acknowledgements.rs

coven_database/store/store_session/
acknowledgements.rs
