coven_database/store/store_session/owner_promotion/
retirement.rs

1use super::*;
2use crate::store::store_session::{active_store_publication, candidate_records, StoreTransaction};
3use coven_protocol::membership::MembershipChain;
4use coven_protocol::objects::ExactObjectRef;
5use coven_protocol::owner_promotion_journal::{
6    OwnerPromotionJournal, OwnerPromotionJournalState, OwnerPromotionStaleEvidence,
7};
8use coven_protocol::remote_object::{remote_object_id, CandidateNonactivation};
9use coven_protocol::store_commit::{
10    OwnerPromotionStaleReason, StoreBatchCommitRef, StorePublicationRef,
11};
12
13fn require_journal_id_on(
14    conn: &rusqlite::Connection,
15    expected: &OwnerPromotionJournal,
16) -> Result<String, DbError> {
17    expected.validate_id(expected.promotion_id)?;
18    let encoded = serde_json::to_string(expected)
19        .map_err(|error| DbError::context("serialize promotion retirement", error))?;
20    let key = format!("owner_promotion/{}", expected.promotion_id);
21    if crate::required_protocol_state_on(conn, &key)? != encoded {
22        return Err(DbError::Message(
23            "promotion retirement lost its exact journal".into(),
24        ));
25    }
26    Ok(encoded)
27}
28
29fn require_target_on(
30    conn: &rusqlite::Connection,
31    expected: &OwnerPromotionJournal,
32    encoded: &str,
33) -> Result<(), DbError> {
34    if crate::required_protocol_state_on(conn, &expected.target_state_key()?)? != encoded {
35        return Err(DbError::Message(
36            "promotion retirement lost its current target attempt".into(),
37        ));
38    }
39    Ok(())
40}
41
42fn retirement_parts(
43    journal: &OwnerPromotionJournal,
44) -> Result<(&CandidateNonactivation, Vec<ExactObjectRef>), DbError> {
45    match &journal.state {
46        OwnerPromotionJournalState::Nonactivated { nonactivation, .. } => Ok((
47            nonactivation,
48            vec![nonactivation.candidate().object.clone()],
49        )),
50        OwnerPromotionJournalState::Stale { evidence, .. } => match evidence.as_ref() {
51            OwnerPromotionStaleEvidence::Candidate {
52                nonactivation,
53                candidate,
54            } => Ok((
55                nonactivation,
56                candidate
57                    .prepared_membership_publication()?
58                    .candidate_object_refs(&candidate.commit, &candidate.reference)?,
59            )),
60            OwnerPromotionStaleEvidence::BeforePublication => Err(DbError::Message(
61                "unprepared promotion has no candidate cleanup".into(),
62            )),
63        },
64        _ => Err(DbError::Message(
65            "promotion has no durable nonactivation decision".into(),
66        )),
67    }
68}
69
70fn require_reservation(
71    active: &ActiveStorePublication,
72    promotion: &OwnerPromotionJournal,
73    candidate: &StoreBatchCommitRef,
74) -> Result<(), DbError> {
75    if active.owner() != &ActiveStorePublicationOwner::OwnerPromotion(promotion.promotion_id)
76        || active.attempt()?.entry.payload
77            != coven_protocol::store_commit::StorePublicationPayload::Commit(candidate.clone())
78        || !active.retired_candidates().is_empty()
79    {
80        return Err(DbError::Message(
81            "promotion retirement differs from its reserved candidate".into(),
82        ));
83    }
84    Ok(())
85}
86
87fn pending_retirement_on(
88    conn: &rusqlite::Connection,
89    journal: &OwnerPromotionJournal,
90) -> Result<Option<ActiveStorePublication>, DbError> {
91    let encoded = require_journal_id_on(conn, journal)?;
92    let (proof, _) = retirement_parts(journal)?;
93    let Some(active) = active_store_publication::load_active_store_publication_on(conn)? else {
94        return Ok(None);
95    };
96    if active.owner() != &ActiveStorePublicationOwner::OwnerPromotion(journal.promotion_id) {
97        // A completed terminal journal survives subsequent independent operations.
98        return Ok(None);
99    }
100    require_target_on(conn, journal, &encoded)?;
101    require_reservation(&active, journal, &proof.reference()?)?;
102    Ok(Some(active))
103}
104
105impl StoreSession<'_> {
106    fn retire_owner_promotion_candidate_authority(
107        &mut self,
108        expected: OwnerPromotionJournal,
109        membership: MembershipChain,
110        publication: StorePublicationRef,
111    ) -> Result<OwnerPromotionJournal, DbError> {
112        let candidate = match &expected.state {
113            OwnerPromotionJournalState::RequestPrepared { candidate, .. }
114            | OwnerPromotionJournalState::MergeHeadPrepared { candidate, .. } => candidate,
115            _ => {
116                return Err(DbError::Message(
117                    "promotion retirement requires an unaccepted prepared candidate".into(),
118                ))
119            }
120        };
121        let tx = self.conn.unchecked_transaction()?;
122        let encoded = require_journal_id_on(&tx, &expected)?;
123        require_target_on(&tx, &expected, &encoded)?;
124        let active = active_store_publication::load_active_store_publication_on(&tx)?
125            .ok_or_else(|| DbError::Message("promotion retirement lost its reservation".into()))?;
126        require_reservation(&active, &expected, &candidate.reference)?;
127        if active.commit_reservation()
128            != Some((
129                &candidate.commit.write_id,
130                &candidate.commit.author_registration,
131                &candidate.reference.coord,
132            ))
133        {
134            return Err(DbError::Message(
135                "promotion retirement differs from its logical write".into(),
136            ));
137        }
138        crate::remote_object_records::validate_remote_object_on(
139            &tx,
140            remote_object_id(&candidate.reference.object),
141            &candidate.reference.object,
142            &candidate.commit.to_bytes(),
143        )?;
144        let nonactivation = StoreTransaction::new(&tx, self.store_dir)
145            .candidate_grant_nonactivation(
146                self.verified_store_authority,
147                &membership,
148                &candidate.reference,
149                &candidate.commit,
150                &publication,
151            )?;
152        let state = match &expected.state {
153            OwnerPromotionJournalState::RequestPrepared { request, .. } => {
154                OwnerPromotionJournalState::Nonactivated {
155                    request: request.clone(),
156                    nonactivation,
157                }
158            }
159            OwnerPromotionJournalState::MergeHeadPrepared {
160                acceptance,
161                candidate,
162                ..
163            } => OwnerPromotionJournalState::Stale {
164                acceptance: acceptance.clone(),
165                reason: OwnerPromotionStaleReason::MergeActivationRejected,
166                evidence: Box::new(OwnerPromotionStaleEvidence::Candidate {
167                    nonactivation,
168                    candidate: candidate.clone(),
169                }),
170            },
171            _ => unreachable!("prepared promotion was checked"),
172        };
173        let next = OwnerPromotionJournal {
174            promotion_id: expected.promotion_id,
175            target: expected.target.clone(),
176            state,
177        };
178        let (previous, _) = expected.into_predecessor()?;
179        let transition = previous.transition_to(&next)?;
180        let (proof, objects) = retirement_parts(&next)?;
181        candidate_records::begin_candidate_nonactivation_targets_on(
182            &tx,
183            &proof.reference()?,
184            &objects,
185            proof,
186        )?;
187        let (journal_key, target_key, before, after, _) = transition.into_values();
188        replace_owner_promotion_journal_on(&tx, &journal_key, &target_key, &before, &after)?;
189        tx.commit()?;
190        Ok(next)
191    }
192
193    fn owner_promotion_retirement_targets(
194        &self,
195        expected: &OwnerPromotionJournal,
196    ) -> Result<Vec<crate::CandidateCleanupObject>, DbError> {
197        let Some(active) = pending_retirement_on(self.conn, expected)? else {
198            return Ok(Vec::new());
199        };
200        let (proof, objects) = retirement_parts(expected)?;
201        let mut targets = candidate_records::candidate_cleanup_targets_on(
202            self.conn,
203            &proof.reference()?,
204            &objects,
205        )?;
206        targets.push(crate::CandidateCleanupObject {
207            object: active.attempt()?.entry_object.clone(),
208        });
209        if let Some(previous) = active.superseded_entry() {
210            targets.push(crate::CandidateCleanupObject {
211                object: previous.object.clone(),
212            });
213        }
214        targets.sort_by(|a, b| a.object.cmp(&b.object));
215        targets.dedup_by(|a, b| a.object == b.object);
216        Ok(targets)
217    }
218
219    fn complete_owner_promotion_retirement(
220        &self,
221        expected: OwnerPromotionJournal,
222    ) -> Result<(), DbError> {
223        let tx = self.conn.unchecked_transaction()?;
224        let Some(active) = pending_retirement_on(&tx, &expected)? else {
225            return Ok(());
226        };
227        let (proof, objects) = retirement_parts(&expected)?;
228        let candidate = proof.reference()?;
229        let targets = candidate_records::candidate_cleanup_targets_on(&tx, &candidate, &objects)?;
230        let mut removed = Vec::new();
231        for target in targets {
232            let id = remote_object_id(&target.object);
233            let mut remote = crate::load_remote_object_on(&tx, id)?;
234            remote.mark_absent_verified()?;
235            crate::update_remote_object_on(&tx, id, &remote)?;
236            removed.push(id);
237        }
238        candidate_records::require_candidate_cleanup_complete_on(
239            &tx,
240            &candidate,
241            &objects,
242            "promotion cleanup is incomplete",
243        )?;
244        candidate_records::delete_remote_objects_on(&tx, removed, "retired promotion")?;
245        let mut completed = active.clone();
246        if completed.superseded_entry().is_some() {
247            completed.complete_superseded_entry_cleanup()?;
248            active_store_publication::update_active_store_publication_on(&tx, &active, &completed)?;
249        }
250        active_store_publication::clear_active_store_publication_on(&tx, &completed)?;
251        tx.commit()?;
252        Ok(())
253    }
254}
255
256impl StoreDatabase {
257    pub async fn retire_owner_promotion_candidate_authority(
258        &self,
259        expected: OwnerPromotionJournal,
260        membership: MembershipChain,
261        publication: StorePublicationRef,
262    ) -> Result<OwnerPromotionJournal, DbError> {
263        self.call_store(move |session| {
264            session.retire_owner_promotion_candidate_authority(expected, membership, publication)
265        })
266        .await
267    }
268
269    pub async fn owner_promotion_retirement_targets(
270        &self,
271        expected: OwnerPromotionJournal,
272    ) -> Result<(OwnerPromotionJournal, Vec<crate::CandidateCleanupObject>), DbError> {
273        self.call_store(move |session| {
274            let targets = session.owner_promotion_retirement_targets(&expected)?;
275            Ok((expected, targets))
276        })
277        .await
278    }
279
280    pub async fn complete_owner_promotion_retirement(
281        &self,
282        expected: OwnerPromotionJournal,
283    ) -> Result<(), DbError> {
284        self.call_store(move |session| session.complete_owner_promotion_retirement(expected))
285            .await
286    }
287}
coven_database/store/store_session/owner_promotion/retirement.rs

coven_database/store/store_session/owner_promotion/
retirement.rs
