coven_database/store/store_session/
owner_promotion.rs

1use std::collections::BTreeSet;
2
3use crate::{persist_exact_remote_object_on, DbError};
4
5use super::{StoreDatabase, StoreSession};
6
7impl StoreSession<'_> {
8    fn begin_owner_promotion_journal(
9        &self,
10        journal_key: &str,
11        target_key: &str,
12        value: &str,
13    ) -> Result<coven_protocol::owner_promotion_journal::OwnerPromotionJournal, DbError> {
14        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
15        tx.execute(
16            "INSERT OR IGNORE INTO protocol_state (key, value) VALUES (?1, ?2)",
17            (journal_key, value),
18        )
19        .map_err(DbError::from)?;
20        tx.execute(
21            "INSERT OR IGNORE INTO protocol_state (key, value) VALUES (?1, ?2)",
22            (target_key, value),
23        )
24        .map_err(DbError::from)?;
25        let by_id = crate::required_protocol_state_on(&tx, journal_key)?;
26        let by_target = crate::required_protocol_state_on(&tx, target_key)?;
27        if by_id != by_target {
28            return Err(DbError::Message(
29                "Owner-promotion id and target journals disagree".to_string(),
30            ));
31        }
32        tx.commit().map_err(DbError::from)?;
33        serde_json::from_str(&by_id)
34            .map_err(|error| DbError::context("parse begun Owner-promotion journal", error))
35    }
36
37    fn begin_owner_promotion_acceptance_journal(
38        &self,
39        journal_key: &str,
40        value: &str,
41    ) -> Result<coven_protocol::owner_promotion_journal::OwnerPromotionJournal, DbError> {
42        self.conn
43            .execute(
44                "INSERT OR IGNORE INTO protocol_state (key, value) VALUES (?1, ?2)",
45                (journal_key, value),
46            )
47            .map_err(DbError::from)?;
48        let actual = crate::required_protocol_state_on(self.conn, journal_key)?;
49        if actual != value {
50            return Err(DbError::Message(
51                "Owner-promotion id is already bound to different candidate acceptance".to_string(),
52            ));
53        }
54        serde_json::from_str(&actual).map_err(|error| {
55            DbError::context("parse begun Owner-promotion candidate acceptance", error)
56        })
57    }
58
59    fn advance_owner_promotion_journal(
60        &self,
61        transition: coven_protocol::owner_promotion_journal::OwnerPromotionJournalTransition,
62    ) -> Result<(), DbError> {
63        let (journal_key, target_key, previous_value, next_value, remote_objects) =
64            transition.into_values();
65        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
66        crate::store::store_session::StoreTransaction::new(&tx, self.store_dir)
67            .advance_owner_promotion_journal(
68                journal_key,
69                target_key,
70                previous_value,
71                next_value,
72                remote_objects,
73            )?;
74        tx.commit().map_err(DbError::from)
75    }
76
77    fn end_nonactivated_owner_promotion_candidate(
78        &self,
79        transition: coven_protocol::owner_promotion_journal::OwnerPromotionJournalTransition,
80        candidate: coven_protocol::store_commit::StoreBatchCommitRef,
81        objects: Vec<coven_protocol::objects::ExactObjectRef>,
82        nonactivation: coven_protocol::remote_object::CandidateNonactivation,
83    ) -> Result<Vec<super::candidate_records::CandidateCleanupObject>, DbError> {
84        let (journal_key, target_key, previous_value, next_value, remote_objects) =
85            transition.into_values();
86        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
87        let cleanup = super::candidate_records::begin_candidate_nonactivation_targets_on(
88            &tx,
89            &candidate,
90            &objects,
91            &nonactivation,
92        )?;
93        crate::store::store_session::StoreTransaction::new(&tx, self.store_dir)
94            .advance_owner_promotion_journal(
95                journal_key,
96                target_key,
97                previous_value,
98                next_value,
99                remote_objects,
100            )?;
101        tx.commit().map_err(DbError::from)?;
102        Ok(cleanup)
103    }
104
105    fn owner_promotion_candidate_cleanup_targets(
106        &self,
107        candidate: &coven_protocol::store_commit::StoreBatchCommitRef,
108        objects: &[coven_protocol::objects::ExactObjectRef],
109    ) -> Result<Vec<super::candidate_records::CandidateCleanupObject>, DbError> {
110        super::candidate_records::candidate_cleanup_targets_on(self.conn, candidate, objects)
111    }
112
113    fn replace_failed_owner_promotion_journal(
114        &self,
115        replacement: coven_protocol::owner_promotion_journal::OwnerPromotionJournal,
116        target_key: String,
117        replacement_key: String,
118        previous_value: String,
119        replacement_value: String,
120    ) -> Result<coven_protocol::owner_promotion_journal::OwnerPromotionJournal, DbError> {
121        let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
122        let inserted = tx
123            .execute(
124                "INSERT OR IGNORE INTO protocol_state (key, value) VALUES (?1, ?2)",
125                (&replacement_key, &replacement_value),
126            )
127            .map_err(DbError::from)?;
128        if inserted != 1 {
129            return Err(DbError::Message(
130                "fresh Owner-promotion retry identity is already present".to_string(),
131            ));
132        }
133        let replaced = tx
134            .execute(
135                "UPDATE protocol_state SET value = ?1 WHERE key = ?2 AND value = ?3",
136                (&replacement_value, &target_key, &previous_value),
137            )
138            .map_err(DbError::from)?;
139        if replaced != 1 {
140            return Err(DbError::Message(
141                "Owner-promotion retry lost its exact failed target attempt".to_string(),
142            ));
143        }
144        tx.commit().map_err(DbError::from)?;
145        Ok(replacement)
146    }
147}
148
149impl StoreDatabase {
150    pub async fn load_owner_promotion_journal(
151        &self,
152        promotion_id: coven_protocol::store_commit::OwnerPromotionId,
153    ) -> Result<Option<coven_protocol::owner_promotion_journal::OwnerPromotionJournal>, DbError>
154    {
155        let key = format!("owner_promotion/{promotion_id}");
156        self.call_store(move |session| {
157            session
158                .protocol_state(&key)?
159                .map(|value| {
160                    let journal: coven_protocol::owner_promotion_journal::OwnerPromotionJournal =
161                        serde_json::from_str(&value).map_err(|error| {
162                            DbError::context("parse Owner-promotion journal", error)
163                        })?;
164                    journal.validate_id(promotion_id).map_err(DbError::from)?;
165                    Ok(journal)
166                })
167                .transpose()
168        })
169        .await
170    }
171
172    pub async fn load_owner_promotion_target(
173        &self,
174        key: String,
175    ) -> Result<Option<coven_protocol::owner_promotion_journal::OwnerPromotionJournal>, DbError>
176    {
177        self.call_store(move |session| {
178            let value = session.protocol_state(&key)?;
179            let Some(value) = value else {
180                return Ok(None);
181            };
182            let journal: coven_protocol::owner_promotion_journal::OwnerPromotionJournal =
183                serde_json::from_str(&value).map_err(|error| {
184                    DbError::context("parse Owner-promotion target journal", error)
185                })?;
186            journal.validate_target_key(&key).map_err(DbError::from)?;
187            let journal_key = format!("owner_promotion/{}", journal.promotion_id());
188            let by_id = session.protocol_state(&journal_key)?;
189            if by_id.as_deref() != Some(value.as_str()) {
190                return Err(DbError::Message(
191                    "Owner-promotion target and id journals disagree".to_string(),
192                ));
193            }
194            Ok(Some(journal))
195        })
196        .await
197    }
198
199    pub async fn begin_owner_promotion_journal(
200        &self,
201        target_key: String,
202        journal: coven_protocol::owner_promotion_journal::OwnerPromotionJournal,
203    ) -> Result<coven_protocol::owner_promotion_journal::OwnerPromotionJournal, DbError> {
204        journal.validate_begin().map_err(DbError::from)?;
205        if journal.target_state_key().map_err(DbError::from)? != target_key {
206            return Err(DbError::Message(
207                "Owner-promotion target index differs from its journal target".to_string(),
208            ));
209        }
210        let journal_key = format!("owner_promotion/{}", journal.promotion_id());
211        let value = serde_json::to_string(&journal)
212            .map_err(|error| DbError::context("serialize Owner-promotion journal", error))?;
213        self.call_store(move |session| {
214            session.begin_owner_promotion_journal(&journal_key, &target_key, &value)
215        })
216        .await
217    }
218
219    pub async fn begin_owner_promotion_acceptance_journal(
220        &self,
221        journal: coven_protocol::owner_promotion_journal::OwnerPromotionJournal,
222    ) -> Result<coven_protocol::owner_promotion_journal::OwnerPromotionJournal, DbError> {
223        journal.validate_acceptance_begin().map_err(DbError::from)?;
224        let journal_key = format!("owner_promotion/{}", journal.promotion_id());
225        let value = serde_json::to_string(&journal).map_err(|error| {
226            DbError::context("serialize Owner-promotion candidate acceptance", error)
227        })?;
228        self.call_store(move |session| {
229            session.begin_owner_promotion_acceptance_journal(&journal_key, &value)
230        })
231        .await
232    }
233
234    pub async fn advance_owner_promotion_journal(
235        &self,
236        transition: coven_protocol::owner_promotion_journal::OwnerPromotionJournalTransition,
237    ) -> Result<(), DbError> {
238        self.call_store(move |session| session.advance_owner_promotion_journal(transition))
239            .await
240    }
241
242    /// End a promotion whose Store candidate lost its stream position: record the
243    /// nonactivation against every object that candidate published and advance the
244    /// journal onto its stale successor in one transaction, returning the objects
245    /// to delete. The candidate publishes its membership entry and head before the
246    /// Store head that decides the position, so those sit in create-once slots the
247    /// promoter's next attempt composes into; leaving them there would refuse every
248    /// later membership publication on that stream.
249    pub async fn end_nonactivated_owner_promotion_candidate(
250        &self,
251        transition: coven_protocol::owner_promotion_journal::OwnerPromotionJournalTransition,
252        candidate: coven_protocol::store_commit::StoreBatchCommitRef,
253        objects: Vec<coven_protocol::objects::ExactObjectRef>,
254        nonactivation: coven_protocol::remote_object::VerifiedCandidateNonactivation,
255    ) -> Result<Vec<super::candidate_records::CandidateCleanupObject>, DbError> {
256        if nonactivation.candidate_reference().map_err(DbError::from)? != candidate {
257            return Err(DbError::Message(
258                "verified nonactivation names another Owner-promotion candidate".to_string(),
259            ));
260        }
261        let nonactivation = nonactivation.into_durable();
262        self.call_store(move |session| {
263            session.end_nonactivated_owner_promotion_candidate(
264                transition,
265                candidate,
266                objects,
267                nonactivation,
268            )
269        })
270        .await
271    }
272
273    /// The published objects of a promotion candidate that already lost, still
274    /// awaiting deletion. An interrupted cleanup resumes through this: the stale
275    /// journal names the candidate, and each object's durable state says whether it
276    /// is still there.
277    pub async fn owner_promotion_candidate_cleanup_targets(
278        &self,
279        candidate: coven_protocol::store_commit::StoreBatchCommitRef,
280        objects: Vec<coven_protocol::objects::ExactObjectRef>,
281    ) -> Result<Vec<super::candidate_records::CandidateCleanupObject>, DbError> {
282        self.call_store(move |session| {
283            session.owner_promotion_candidate_cleanup_targets(&candidate, &objects)
284        })
285        .await
286    }
287
288    pub async fn replace_failed_owner_promotion_journal(
289        &self,
290        previous: coven_protocol::owner_promotion_journal::OwnerPromotionJournal,
291        replacement: coven_protocol::owner_promotion_journal::OwnerPromotionJournal,
292    ) -> Result<coven_protocol::owner_promotion_journal::OwnerPromotionJournal, DbError> {
293        previous
294            .validate_failed_attempt_replacement(&replacement)
295            .map_err(DbError::from)?;
296        let target_key = previous.target_state_key().map_err(DbError::from)?;
297        if replacement.target_state_key().map_err(DbError::from)? != target_key {
298            return Err(DbError::Message(
299                "Owner-promotion retry target differs from its failed attempt".to_string(),
300            ));
301        }
302        let replacement_key = format!("owner_promotion/{}", replacement.promotion_id());
303        let previous_value = serde_json::to_string(&previous)
304            .map_err(|error| DbError::context("serialize failed Owner-promotion journal", error))?;
305        let replacement_value = serde_json::to_string(&replacement).map_err(|error| {
306            DbError::context("serialize replacement Owner-promotion journal", error)
307        })?;
308        self.call_store(move |session| {
309            session.replace_failed_owner_promotion_journal(
310                replacement,
311                target_key,
312                replacement_key,
313                previous_value,
314                replacement_value,
315            )
316        })
317        .await
318    }
319}
320
321#[allow(clippy::too_many_arguments)]
322pub(super) fn advance_owner_promotion_journal_on(
323    tx: &rusqlite::Transaction<'_>,
324    store_dir: &coven_foundation::store_dir::StoreDir,
325    journal_key: String,
326    target_key: String,
327    previous_value: String,
328    next_value: String,
329    remote_objects: Vec<coven_protocol::remote_object::ClosedRemoteObject>,
330) -> Result<(), DbError> {
331    let mut object_ids = BTreeSet::new();
332    for remote in &remote_objects {
333        if !object_ids.insert(remote.object_id()) {
334            return Err(DbError::Message(
335                "Owner-promotion journal repeats a remote object".to_string(),
336            ));
337        }
338        persist_exact_remote_object_on(tx, store_dir, remote, "Owner-promotion candidate object")?;
339    }
340    let by_id = tx
341        .execute(
342            "UPDATE protocol_state SET value = ?1 WHERE key = ?2 AND value = ?3",
343            (&next_value, &journal_key, &previous_value),
344        )
345        .map_err(DbError::from)?;
346    let by_target = tx
347        .execute(
348            "UPDATE protocol_state SET value = ?1 WHERE key = ?2 AND value = ?3",
349            (&next_value, &target_key, &previous_value),
350        )
351        .map_err(DbError::from)?;
352    if by_id != 1 || by_target != 1 {
353        return Err(DbError::Message(
354            "Owner-promotion journal advance lost its exact predecessor".to_string(),
355        ));
356    }
357    Ok(())
358}
coven_database/store/store_session/owner_promotion.rs

coven_database/store/store_session/
owner_promotion.rs
