1mod retirement;
2
3use std::collections::BTreeSet;
4
5use crate::{
6 persist_exact_remote_object_on, ActiveStorePublication, ActiveStorePublicationOwner, DbError,
7};
8
9use super::{StoreDatabase, StoreSession};
10
11impl StoreSession<'_> {
12 fn begin_owner_promotion_journal(
13 &self,
14 journal_key: &str,
15 target_key: &str,
16 value: &str,
17 ) -> Result<coven_protocol::owner_promotion_journal::OwnerPromotionJournal, DbError> {
18 let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
19 tx.execute(
20 "INSERT OR IGNORE INTO protocol_state (key, value) VALUES (?1, ?2)",
21 (journal_key, value),
22 )
23 .map_err(DbError::from)?;
24 tx.execute(
25 "INSERT OR IGNORE INTO protocol_state (key, value) VALUES (?1, ?2)",
26 (target_key, value),
27 )
28 .map_err(DbError::from)?;
29 let by_id = crate::required_protocol_state_on(&tx, journal_key)?;
30 let by_target = crate::required_protocol_state_on(&tx, target_key)?;
31 if by_id != by_target {
32 return Err(DbError::Message(
33 "Owner-promotion id and target journals disagree".to_string(),
34 ));
35 }
36 tx.commit().map_err(DbError::from)?;
37 serde_json::from_str(&by_id)
38 .map_err(|error| DbError::context("parse begun Owner-promotion journal", error))
39 }
40
41 fn begin_owner_promotion_acceptance_journal(
42 &self,
43 journal_key: &str,
44 value: &str,
45 ) -> Result<coven_protocol::owner_promotion_journal::OwnerPromotionJournal, DbError> {
46 self.conn
47 .execute(
48 "INSERT OR IGNORE INTO protocol_state (key, value) VALUES (?1, ?2)",
49 (journal_key, value),
50 )
51 .map_err(DbError::from)?;
52 let actual = crate::required_protocol_state_on(self.conn, journal_key)?;
53 if actual != value {
54 return Err(DbError::Message(
55 "Owner-promotion id is already bound to different candidate acceptance".to_string(),
56 ));
57 }
58 serde_json::from_str(&actual).map_err(|error| {
59 DbError::context("parse begun Owner-promotion candidate acceptance", error)
60 })
61 }
62
63 fn advance_owner_promotion_journal(
64 &self,
65 transition: coven_protocol::owner_promotion_journal::OwnerPromotionJournalTransition,
66 accepted_request: Option<crate::AcceptedStoreCommitEvidence>,
67 ) -> Result<(), DbError> {
68 let (journal_key, target_key, previous_value, next_value, remote_objects) =
69 transition.into_values();
70 let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
71 advance_owner_promotion_journal_on(
72 &tx,
73 self.store_dir,
74 journal_key,
75 target_key,
76 previous_value,
77 next_value,
78 remote_objects,
79 accepted_request.as_ref(),
80 )?;
81 tx.commit().map_err(DbError::from)
82 }
83
84 fn replace_failed_owner_promotion_journal(
85 &self,
86 replacement: coven_protocol::owner_promotion_journal::OwnerPromotionJournal,
87 target_key: String,
88 replacement_key: String,
89 previous_value: String,
90 replacement_value: String,
91 ) -> Result<coven_protocol::owner_promotion_journal::OwnerPromotionJournal, DbError> {
92 let tx = self.conn.unchecked_transaction().map_err(DbError::from)?;
93 let previous: coven_protocol::owner_promotion_journal::OwnerPromotionJournal =
94 serde_json::from_str(&previous_value)
95 .map_err(|error| DbError::context("parse failed promotion", error))?;
96 if super::active_store_publication::load_active_store_publication_on(&tx)?.is_some_and(
97 |active| {
98 active.owner()
99 == &ActiveStorePublicationOwner::OwnerPromotion(previous.promotion_id)
100 },
101 ) {
102 return Err(DbError::Message(
103 "failed promotion still owns candidate cleanup".into(),
104 ));
105 }
106 let inserted = tx
107 .execute(
108 "INSERT OR IGNORE INTO protocol_state (key, value) VALUES (?1, ?2)",
109 (&replacement_key, &replacement_value),
110 )
111 .map_err(DbError::from)?;
112 if inserted != 1 {
113 return Err(DbError::Message(
114 "fresh Owner-promotion retry identity is already present".to_string(),
115 ));
116 }
117 let replaced = tx
118 .execute(
119 "UPDATE protocol_state SET value = ?1 WHERE key = ?2 AND value = ?3",
120 (&replacement_value, &target_key, &previous_value),
121 )
122 .map_err(DbError::from)?;
123 if replaced != 1 {
124 return Err(DbError::Message(
125 "Owner-promotion retry lost its exact failed target attempt".to_string(),
126 ));
127 }
128 tx.commit().map_err(DbError::from)?;
129 Ok(replacement)
130 }
131}
132
133impl StoreDatabase {
134 pub async fn load_owner_promotion_journal(
135 &self,
136 promotion_id: coven_protocol::store_commit::OwnerPromotionId,
137 ) -> Result<Option<coven_protocol::owner_promotion_journal::OwnerPromotionJournal>, DbError>
138 {
139 let key = format!("owner_promotion/{promotion_id}");
140 self.call_store(move |session| {
141 session
142 .protocol_state(&key)?
143 .map(|value| {
144 let journal: coven_protocol::owner_promotion_journal::OwnerPromotionJournal =
145 serde_json::from_str(&value).map_err(|error| {
146 DbError::context("parse Owner-promotion journal", error)
147 })?;
148 journal.validate_id(promotion_id).map_err(DbError::from)?;
149 Ok(journal)
150 })
151 .transpose()
152 })
153 .await
154 }
155
156 pub async fn load_owner_promotion_target(
157 &self,
158 key: String,
159 ) -> Result<Option<coven_protocol::owner_promotion_journal::OwnerPromotionJournal>, DbError>
160 {
161 self.call_store(move |session| {
162 let value = session.protocol_state(&key)?;
163 let Some(value) = value else {
164 return Ok(None);
165 };
166 let journal: coven_protocol::owner_promotion_journal::OwnerPromotionJournal =
167 serde_json::from_str(&value).map_err(|error| {
168 DbError::context("parse Owner-promotion target journal", error)
169 })?;
170 journal.validate_target_key(&key).map_err(DbError::from)?;
171 let journal_key = format!("owner_promotion/{}", journal.promotion_id());
172 let by_id = session.protocol_state(&journal_key)?;
173 if by_id.as_deref() != Some(value.as_str()) {
174 return Err(DbError::Message(
175 "Owner-promotion target and id journals disagree".to_string(),
176 ));
177 }
178 Ok(Some(journal))
179 })
180 .await
181 }
182
183 pub async fn begin_owner_promotion_journal(
184 &self,
185 target_key: String,
186 journal: coven_protocol::owner_promotion_journal::OwnerPromotionJournal,
187 ) -> Result<coven_protocol::owner_promotion_journal::OwnerPromotionJournal, DbError> {
188 journal.validate_begin().map_err(DbError::from)?;
189 if journal.target_state_key().map_err(DbError::from)? != target_key {
190 return Err(DbError::Message(
191 "Owner-promotion target index differs from its journal target".to_string(),
192 ));
193 }
194 let journal_key = format!("owner_promotion/{}", journal.promotion_id());
195 let value = serde_json::to_string(&journal)
196 .map_err(|error| DbError::context("serialize Owner-promotion journal", error))?;
197 self.call_store(move |session| {
198 session.begin_owner_promotion_journal(&journal_key, &target_key, &value)
199 })
200 .await
201 }
202
203 pub async fn begin_owner_promotion_acceptance_journal(
204 &self,
205 journal: coven_protocol::owner_promotion_journal::OwnerPromotionJournal,
206 ) -> Result<coven_protocol::owner_promotion_journal::OwnerPromotionJournal, DbError> {
207 journal.validate_acceptance_begin().map_err(DbError::from)?;
208 let journal_key = format!("owner_promotion/{}", journal.promotion_id());
209 let value = serde_json::to_string(&journal).map_err(|error| {
210 DbError::context("serialize Owner-promotion candidate acceptance", error)
211 })?;
212 self.call_store(move |session| {
213 session.begin_owner_promotion_acceptance_journal(&journal_key, &value)
214 })
215 .await
216 }
217
218 pub async fn advance_owner_promotion_journal(
219 &self,
220 transition: coven_protocol::owner_promotion_journal::OwnerPromotionJournalTransition,
221 ) -> Result<(), DbError> {
222 self.call_store(move |session| session.advance_owner_promotion_journal(transition, None))
223 .await
224 }
225
226 pub async fn advance_accepted_owner_promotion_request(
229 &self,
230 transition: coven_protocol::owner_promotion_journal::OwnerPromotionJournalTransition,
231 acceptance: crate::AcceptedStoreCommitEvidence,
232 ) -> Result<(), DbError> {
233 self.call_store(move |session| {
234 session.advance_owner_promotion_journal(transition, Some(acceptance))
235 })
236 .await
237 }
238
239 pub async fn replace_failed_owner_promotion_journal(
240 &self,
241 previous: coven_protocol::owner_promotion_journal::OwnerPromotionJournal,
242 replacement: coven_protocol::owner_promotion_journal::OwnerPromotionJournal,
243 ) -> Result<coven_protocol::owner_promotion_journal::OwnerPromotionJournal, DbError> {
244 previous
245 .validate_failed_attempt_replacement(&replacement)
246 .map_err(DbError::from)?;
247 let target_key = previous.target_state_key().map_err(DbError::from)?;
248 if replacement.target_state_key().map_err(DbError::from)? != target_key {
249 return Err(DbError::Message(
250 "Owner-promotion retry target differs from its failed attempt".to_string(),
251 ));
252 }
253 let replacement_key = format!("owner_promotion/{}", replacement.promotion_id());
254 let previous_value = serde_json::to_string(&previous)
255 .map_err(|error| DbError::context("serialize failed Owner-promotion journal", error))?;
256 let replacement_value = serde_json::to_string(&replacement).map_err(|error| {
257 DbError::context("serialize replacement Owner-promotion journal", error)
258 })?;
259 self.call_store(move |session| {
260 session.replace_failed_owner_promotion_journal(
261 replacement,
262 target_key,
263 replacement_key,
264 previous_value,
265 replacement_value,
266 )
267 })
268 .await
269 }
270}
271
272#[allow(clippy::too_many_arguments)]
273pub(super) fn advance_owner_promotion_journal_on(
274 tx: &rusqlite::Transaction<'_>,
275 store_dir: &coven_foundation::store_dir::StoreDir,
276 journal_key: String,
277 target_key: String,
278 previous_value: String,
279 next_value: String,
280 remote_objects: Vec<coven_protocol::remote_object::ClosedRemoteObject>,
281 accepted_request: Option<&crate::AcceptedStoreCommitEvidence>,
282) -> Result<(), DbError> {
283 use coven_protocol::owner_promotion_journal::{
284 OwnerPromotionJournal, OwnerPromotionJournalState,
285 };
286
287 let previous: OwnerPromotionJournal = serde_json::from_str(&previous_value)
288 .map_err(|error| DbError::context("parse prior Owner-promotion journal", error))?;
289 let next: OwnerPromotionJournal = serde_json::from_str(&next_value)
290 .map_err(|error| DbError::context("parse successor Owner-promotion journal", error))?;
291 if previous.promotion_id() != next.promotion_id() {
292 return Err(DbError::Message(
293 "Owner-promotion journal advance changes promotion identity".to_string(),
294 ));
295 }
296 if matches!(&next.state, OwnerPromotionJournalState::Nonactivated { .. })
297 || matches!(&next.state, OwnerPromotionJournalState::Stale { evidence, .. }
298 if matches!(evidence.as_ref(), coven_protocol::owner_promotion_journal::OwnerPromotionStaleEvidence::Candidate { .. }))
299 {
300 return Err(DbError::Message(
301 "promotion nonactivation requires accepted authority and owned cleanup".into(),
302 ));
303 }
304 match (&previous.state, &next.state, accepted_request) {
305 (
306 OwnerPromotionJournalState::RequestPrepared { candidate, .. },
307 OwnerPromotionJournalState::RequestAccepted { publication, .. },
308 Some(accepted),
309 ) => {
310 let exact = accepted.exact_publication().ok_or_else(|| {
311 DbError::Message(
312 "Owner-promotion request lacks its exact accepted publication receipt".into(),
313 )
314 })?;
315 if accepted.commit_ref() != &candidate.reference
316 || accepted.commit_ref() != &publication.value.commit
317 || exact.reference() != &publication.value.publication
318 {
319 return Err(DbError::Message(
320 "Owner-promotion request differs from its accepted publication evidence".into(),
321 ));
322 }
323 }
324 (
325 OwnerPromotionJournalState::RequestPrepared { .. },
326 OwnerPromotionJournalState::RequestAccepted { .. },
327 None,
328 ) => {
329 return Err(DbError::Message(
330 "Owner-promotion request advancement requires accepted publication evidence".into(),
331 ));
332 }
333 (_, _, Some(_)) => {
334 return Err(DbError::Message(
335 "accepted request evidence cannot authorize another promotion transition".into(),
336 ));
337 }
338 (_, _, None) => {}
339 }
340 let previous_candidate = match &previous.state {
341 OwnerPromotionJournalState::RequestPrepared { candidate, .. }
342 | OwnerPromotionJournalState::RequestAccepted { candidate, .. }
343 | OwnerPromotionJournalState::MergeHeadPrepared { candidate, .. } => {
344 Some(candidate.as_ref())
345 }
346 _ => None,
347 };
348 let next_candidate = match &next.state {
349 OwnerPromotionJournalState::RequestPrepared { candidate, .. }
350 | OwnerPromotionJournalState::RequestAccepted { candidate, .. }
351 | OwnerPromotionJournalState::MergeHeadPrepared { candidate, .. } => {
352 Some(candidate.as_ref())
353 }
354 _ => None,
355 };
356 let owner = ActiveStorePublicationOwner::OwnerPromotion(next.promotion_id());
357 if let Some(candidate) = next_candidate {
358 let active = ActiveStorePublication::for_commit(owner.clone(), candidate)?;
359 match super::active_store_publication::load_active_store_publication_on(tx)? {
360 Some(existing)
361 if previous_candidate.is_some() && existing.same_commit_reservation(&active) => {}
362 Some(existing) => {
363 return Err(DbError::Message(format!(
364 "Owner-promotion publication is occupied by {:?}",
365 existing.owner()
366 )));
367 }
368 None if previous_candidate.is_none() => {
369 let claim = super::active_store_publication::claim_active_store_publication_on(
370 tx, &active,
371 )?;
372 if claim != super::active_store_publication::ActiveStorePublicationClaim::Acquired {
373 return Err(DbError::Message(
374 "Owner-promotion publication changed during journal advance".to_string(),
375 ));
376 }
377 }
378 None => {
379 return Err(DbError::Message(
380 "prepared Owner-promotion journal lost its active publication".to_string(),
381 ));
382 }
383 }
384 }
385 let mut object_ids = BTreeSet::new();
386 for remote in &remote_objects {
387 if !object_ids.insert(remote.object_id()) {
388 return Err(DbError::Message(
389 "Owner-promotion journal repeats a remote object".to_string(),
390 ));
391 }
392 persist_exact_remote_object_on(tx, store_dir, remote, "Owner-promotion candidate object")?;
393 }
394 if let (
395 OwnerPromotionJournalState::RequestAccepted {
396 candidate,
397 publication,
398 ..
399 },
400 OwnerPromotionJournalState::AwaitingAcceptance { .. },
401 ) = (&previous.state, &next.state)
402 {
403 let object_id = coven_protocol::remote_object::remote_object_id(&publication.object);
404 let remote = crate::load_remote_object_on(tx, object_id)?;
405 let expected = coven_protocol::remote_object::RemoteObjectRecord::prepared_owner_promotion_request_publication(
406 publication,
407 &candidate.commit,
408 )?;
409 if remote.object() != expected.object() || !remote.records_verified_upload() {
410 return Err(DbError::Message(
411 "Owner-promotion request result has not completed its exact upload".into(),
412 ));
413 }
414 let remote = remote.into_activated(&candidate.reference)?;
415 crate::update_remote_object_on(tx, object_id, &remote)?;
416 }
417 replace_owner_promotion_journal_on(
418 tx,
419 &journal_key,
420 &target_key,
421 &previous_value,
422 &next_value,
423 )?;
424 if let Some(candidate) = previous_candidate.filter(|_| next_candidate.is_none()) {
425 super::active_store_publication::clear_active_store_commit_for_owner_on(
426 tx,
427 &owner,
428 &candidate.reference,
429 )?;
430 }
431 Ok(())
432}
433
434fn replace_owner_promotion_journal_on(
435 tx: &rusqlite::Transaction<'_>,
436 journal_key: &str,
437 target_key: &str,
438 previous_value: &str,
439 next_value: &str,
440) -> Result<(), DbError> {
441 let by_id = tx
442 .execute(
443 "UPDATE protocol_state SET value = ?1 WHERE key = ?2 AND value = ?3",
444 (&next_value, &journal_key, &previous_value),
445 )
446 .map_err(DbError::from)?;
447 let by_target = tx
448 .execute(
449 "UPDATE protocol_state SET value = ?1 WHERE key = ?2 AND value = ?3",
450 (&next_value, &target_key, &previous_value),
451 )
452 .map_err(DbError::from)?;
453 if by_id != 1 || by_target != 1 {
454 return Err(DbError::Message(
455 "Owner-promotion journal advance lost its exact predecessor".to_string(),
456 ));
457 }
458 Ok(())
459}