coven_database/
operation_models.rs

1use super::*;
2
3#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
4#[serde(rename_all = "snake_case", deny_unknown_fields)]
5pub enum ActiveStorePublicationOwner {
6    StoreWrite(WriteId),
7    StoreAcknowledgement,
8    DeviceJoin(coven_protocol::store_commit::DeviceJoinAttemptId),
9    DeviceExclusion(ObjectHash),
10    MembershipMutation,
11    OwnerPromotion(coven_protocol::store_commit::OwnerPromotionId),
12    Reclaim(ObjectHash),
13    CircleOperation(coven_protocol::circle::CircleOperationId),
14    OwnerRecovery,
15    Snapshot,
16}
17
18#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
19#[serde(rename_all = "snake_case", deny_unknown_fields)]
20pub enum ActiveStorePublicationAttempt {
21    AwaitingPreparation {
22        write_id: WriteId,
23        author_registration: StoreDeviceRegistrationRef,
24        coord: StoreCommitCoord,
25    },
26    Discarding {
27        write_id: WriteId,
28        author_registration: StoreDeviceRegistrationRef,
29        coord: StoreCommitCoord,
30    },
31    Commit {
32        write_id: WriteId,
33        author_registration: StoreDeviceRegistrationRef,
34        coord: StoreCommitCoord,
35        publication: coven_protocol::prepared_commit::PreparedStorePublication,
36    },
37    MembershipAbandonment {
38        candidate: Box<coven_protocol::prepared_commit::PreparedStoreOperationCommit>,
39    },
40    CompletingCoveredWrite {
41        write_id: WriteId,
42        position: coven_protocol::write::SnapshotCoveredPosition,
43        publication: coven_protocol::prepared_commit::PreparedStorePublication,
44    },
45    Snapshot {
46        publication: coven_protocol::prepared_commit::PreparedStorePublication,
47        retired_objects: Vec<coven_protocol::objects::ExactObjectRef>,
48    },
49    SnapshotSuperseded {
50        publication: coven_protocol::prepared_commit::PreparedStorePublication,
51        snapshot: PublishedStoreSnapshot,
52        retired_objects: Vec<coven_protocol::objects::ExactObjectRef>,
53    },
54}
55
56#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
57#[serde(deny_unknown_fields)]
58pub struct RetiredStoreCandidate {
59    pub nonactivation: coven_protocol::remote_object::CandidateNonactivation,
60    pub inputs: RetiredStoreCandidateInputs,
61    pub publications: Vec<coven_protocol::store_commit::StorePublicationRef>,
62}
63
64#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
65#[serde(rename_all = "snake_case", deny_unknown_fields)]
66pub enum RetiredStoreCandidateInputs {
67    Write(Vec<PreparedAudienceBlob>),
68    Acknowledgement(coven_protocol::store_commit::RetainedVerifiedActivatedAck),
69    Membership(coven_protocol::membership_mutation::PreparedMembershipPublication),
70}
71
72impl RetiredStoreCandidate {
73    pub(crate) fn blobs(&self) -> &[PreparedAudienceBlob] {
74        match &self.inputs {
75            RetiredStoreCandidateInputs::Write(blobs) => blobs,
76            RetiredStoreCandidateInputs::Acknowledgement(_)
77            | RetiredStoreCandidateInputs::Membership(_) => &[],
78        }
79    }
80
81    pub(crate) fn candidate(&self) -> Result<StoreBatchCommitRef, DbError> {
82        self.nonactivation.reference().map_err(DbError::from)
83    }
84
85    pub(crate) fn objects(&self) -> Result<Vec<coven_protocol::objects::ExactObjectRef>, DbError> {
86        self.nonactivation.validate()?;
87        let commit: coven_protocol::store_commit::StoreBatchCommit =
88            serde_json::from_slice(&self.nonactivation.candidate().canonical_signed_bytes)
89                .map_err(|error| DbError::context("retired Store write candidate", error))?;
90        let mut objects = crate::candidate_graph_exact_objects(&commit)?
91            .into_iter()
92            .collect::<std::collections::BTreeSet<_>>();
93        objects.insert(self.nonactivation.candidate().object.clone());
94        objects.extend(self.blobs().iter().map(|blob| blob.blob().object().clone()));
95        if let RetiredStoreCandidateInputs::Membership(publication) = &self.inputs {
96            objects.extend(publication.candidate_object_refs(&commit, &self.candidate()?)?);
97        }
98        if let RetiredStoreCandidateInputs::Acknowledgement(proof) = &self.inputs {
99            proof.validate_predecessors()?;
100            if commit.acknowledgement() != Some(&proof.acknowledgement.0)
101                || proof.activating_commit != self.candidate()?
102            {
103                return Err(DbError::Message(
104                    "retired acknowledgement proof names another candidate".into(),
105                ));
106            }
107            objects.extend(commit.retained_operation_objects()?);
108            objects.extend(
109                proof
110                    .predecessors
111                    .iter()
112                    .map(|(reference, _)| reference.object.clone()),
113            );
114        }
115        Ok(objects.into_iter().collect())
116    }
117}
118
119/// The exact conditional Store-publication attempt owned by one local
120/// operation.
121#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
122#[serde(deny_unknown_fields)]
123pub struct ActiveStorePublication {
124    owner: ActiveStorePublicationOwner,
125    attempt: ActiveStorePublicationAttempt,
126    /// An exact entry whose publication position was won by another entry.
127    /// Kept until its remote bytes have been deleted, independently of whether
128    /// the same logical write succeeds under the replacement attempt.
129    superseded_entry: Option<coven_protocol::store_commit::StorePublicationRef>,
130    retired_candidates: Vec<RetiredStoreCandidate>,
131}
132
133impl ActiveStorePublication {
134    pub fn for_commit(
135        owner: ActiveStorePublicationOwner,
136        candidate: &coven_protocol::prepared_commit::PreparedStoreOperationCommit,
137    ) -> Result<Self, DbError> {
138        Self::commit(
139            owner,
140            candidate.commit.write_id.clone(),
141            candidate.commit.author_registration.clone(),
142            candidate.reference.coord.clone(),
143            candidate.publication.clone(),
144        )
145    }
146
147    pub fn commit(
148        owner: ActiveStorePublicationOwner,
149        write_id: WriteId,
150        author_registration: StoreDeviceRegistrationRef,
151        coord: StoreCommitCoord,
152        publication: coven_protocol::prepared_commit::PreparedStorePublication,
153    ) -> Result<Self, DbError> {
154        if matches!(&owner, ActiveStorePublicationOwner::StoreWrite(owner) if owner != &write_id) {
155            return Err(DbError::Message(
156                "active Store-write owner differs from its logical write".to_string(),
157            ));
158        }
159        let attempt = ActiveStorePublicationAttempt::Commit {
160            write_id,
161            author_registration,
162            coord,
163            publication,
164        };
165        Self::validated(owner, attempt)
166    }
167
168    pub fn snapshot(
169        publication: coven_protocol::prepared_commit::PreparedStorePublication,
170    ) -> Result<Self, DbError> {
171        let owner = ActiveStorePublicationOwner::Snapshot;
172        let attempt = ActiveStorePublicationAttempt::Snapshot {
173            publication,
174            retired_objects: Vec::new(),
175        };
176        Self::validated(owner, attempt)
177    }
178
179    fn validated(
180        owner: ActiveStorePublicationOwner,
181        attempt: ActiveStorePublicationAttempt,
182    ) -> Result<Self, DbError> {
183        let publication = match &attempt {
184            ActiveStorePublicationAttempt::AwaitingPreparation { .. }
185            | ActiveStorePublicationAttempt::Discarding { .. }
186            | ActiveStorePublicationAttempt::CompletingCoveredWrite { .. }
187            | ActiveStorePublicationAttempt::SnapshotSuperseded { .. } => {
188                return Err(DbError::Message(
189                    "an awaiting reservation has no prepared publication".to_string(),
190                ));
191            }
192            ActiveStorePublicationAttempt::Commit { publication, .. }
193            | ActiveStorePublicationAttempt::Snapshot { publication, .. } => publication,
194            ActiveStorePublicationAttempt::MembershipAbandonment { candidate } => {
195                &candidate.publication
196            }
197        };
198        let payload_matches = match (&attempt, &publication.entry.payload) {
199            (
200                ActiveStorePublicationAttempt::Commit {
201                    author_registration,
202                    coord,
203                    ..
204                },
205                coven_protocol::store_commit::StorePublicationPayload::Commit(reference),
206            ) => {
207                &publication.entry.author_registration == author_registration
208                    && &reference.coord == coord
209            }
210            (
211                ActiveStorePublicationAttempt::Snapshot { .. },
212                coven_protocol::store_commit::StorePublicationPayload::Snapshot(_),
213            ) => true,
214            (
215                ActiveStorePublicationAttempt::MembershipAbandonment { candidate },
216                coven_protocol::store_commit::StorePublicationPayload::Commit(reference),
217            ) => {
218                owner == ActiveStorePublicationOwner::MembershipMutation
219                    && reference == &candidate.reference
220                    && publication.entry.author_registration == candidate.commit.author_registration
221                    && !candidate.commit.abandoned_candidates().is_empty()
222            }
223            _ => false,
224        };
225        if !payload_matches {
226            return Err(DbError::Message(
227                "active Store publication source differs from its exact attempt".to_string(),
228            ));
229        }
230        publication
231            .reference()
232            .map_err(|error| DbError::context("active Store publication attempt", error))?;
233        Ok(Self {
234            owner,
235            attempt,
236            superseded_entry: None,
237            retired_candidates: Vec::new(),
238        })
239    }
240
241    pub fn owner(&self) -> &ActiveStorePublicationOwner {
242        &self.owner
243    }
244
245    pub(crate) fn author_registration(&self) -> &StoreDeviceRegistrationRef {
246        match &self.attempt {
247            ActiveStorePublicationAttempt::AwaitingPreparation {
248                author_registration,
249                ..
250            }
251            | ActiveStorePublicationAttempt::Discarding {
252                author_registration,
253                ..
254            }
255            | ActiveStorePublicationAttempt::Commit {
256                author_registration,
257                ..
258            } => author_registration,
259            ActiveStorePublicationAttempt::MembershipAbandonment { candidate } => {
260                &candidate.commit.author_registration
261            }
262            ActiveStorePublicationAttempt::CompletingCoveredWrite { position, .. } => {
263                &position.author_registration
264            }
265            ActiveStorePublicationAttempt::Snapshot { publication, .. }
266            | ActiveStorePublicationAttempt::SnapshotSuperseded { publication, .. } => {
267                &publication.entry.author_registration
268            }
269        }
270    }
271
272    pub fn attempt(
273        &self,
274    ) -> Result<&coven_protocol::prepared_commit::PreparedStorePublication, DbError> {
275        match &self.attempt {
276            ActiveStorePublicationAttempt::AwaitingPreparation { .. } => Err(DbError::Message(
277                "reserved Store write is awaiting candidate preparation".to_string(),
278            )),
279            ActiveStorePublicationAttempt::Discarding { .. } => Err(DbError::Message(
280                "reserved Store write is being discarded".to_string(),
281            )),
282            ActiveStorePublicationAttempt::Commit { publication, .. }
283            | ActiveStorePublicationAttempt::Snapshot { publication, .. }
284            | ActiveStorePublicationAttempt::SnapshotSuperseded { publication, .. }
285            | ActiveStorePublicationAttempt::CompletingCoveredWrite { publication, .. } => {
286                Ok(publication)
287            }
288            ActiveStorePublicationAttempt::MembershipAbandonment { candidate } => {
289                Ok(&candidate.publication)
290            }
291        }
292    }
293
294    pub fn covered_write_position(
295        &self,
296    ) -> Option<&coven_protocol::write::SnapshotCoveredPosition> {
297        match &self.attempt {
298            ActiveStorePublicationAttempt::CompletingCoveredWrite { position, .. } => {
299                Some(position)
300            }
301            _ => None,
302        }
303    }
304
305    pub(crate) fn begin_covered_write_completion(
306        &self,
307        position: coven_protocol::write::SnapshotCoveredPosition,
308    ) -> Result<Self, DbError> {
309        let ActiveStorePublicationAttempt::Commit {
310            write_id,
311            author_registration,
312            coord,
313            publication,
314        } = &self.attempt
315        else {
316            return Err(DbError::Message(
317                "covered completion requires a prepared Store write".into(),
318            ));
319        };
320        if self.owner != ActiveStorePublicationOwner::StoreWrite(write_id.clone())
321            || position.author_registration != *author_registration
322            || position.coord != *coord
323            || !self.retired_candidates.is_empty()
324        {
325            return Err(DbError::Message(
326                "covered completion differs from its reserved write or retains candidate cleanup"
327                    .into(),
328            ));
329        }
330        let mut completed = self.clone();
331        completed.attempt = ActiveStorePublicationAttempt::CompletingCoveredWrite {
332            write_id: write_id.clone(),
333            position,
334            publication: publication.clone(),
335        };
336        Ok(completed)
337    }
338
339    pub fn is_awaiting_preparation(&self) -> bool {
340        matches!(
341            self.attempt,
342            ActiveStorePublicationAttempt::AwaitingPreparation { .. }
343        )
344    }
345
346    pub fn is_discarding(&self) -> bool {
347        matches!(
348            self.attempt,
349            ActiveStorePublicationAttempt::Discarding { .. }
350        )
351    }
352
353    pub(crate) fn begin_discard(&self) -> Result<Self, DbError> {
354        let ActiveStorePublicationAttempt::AwaitingPreparation {
355            write_id,
356            author_registration,
357            coord,
358        } = &self.attempt
359        else {
360            return Err(DbError::Message(
361                "discard requires a write whose old candidate has been retired".to_string(),
362            ));
363        };
364        if self.owner != ActiveStorePublicationOwner::StoreWrite(write_id.clone()) {
365            return Err(DbError::Message(
366                "discard reservation differs from its Store-write owner".to_string(),
367            ));
368        }
369        let mut discarded = self.clone();
370        discarded.attempt = ActiveStorePublicationAttempt::Discarding {
371            write_id: write_id.clone(),
372            author_registration: author_registration.clone(),
373            coord: coord.clone(),
374        };
375        Ok(discarded)
376    }
377
378    pub fn retired_candidates(&self) -> &[RetiredStoreCandidate] {
379        &self.retired_candidates
380    }
381
382    pub fn retired_snapshot_objects(&self) -> &[coven_protocol::objects::ExactObjectRef] {
383        match &self.attempt {
384            ActiveStorePublicationAttempt::Snapshot {
385                retired_objects, ..
386            }
387            | ActiveStorePublicationAttempt::SnapshotSuperseded {
388                retired_objects, ..
389            } => retired_objects,
390            _ => &[],
391        }
392    }
393
394    pub(crate) fn retain_snapshot_cleanup(
395        &mut self,
396        objects: Vec<coven_protocol::objects::ExactObjectRef>,
397    ) -> Result<(), DbError> {
398        let ActiveStorePublicationAttempt::Snapshot {
399            retired_objects, ..
400        } = &mut self.attempt
401        else {
402            return Err(DbError::Message(
403                "snapshot cleanup has another publication owner".into(),
404            ));
405        };
406        if !retired_objects.is_empty() {
407            return Err(DbError::Message(
408                "snapshot cleanup is already pending".into(),
409            ));
410        }
411        *retired_objects = objects;
412        Ok(())
413    }
414
415    pub(crate) fn complete_snapshot_cleanup(&mut self) -> Result<(), DbError> {
416        let retired_objects = match &mut self.attempt {
417            ActiveStorePublicationAttempt::Snapshot {
418                retired_objects, ..
419            }
420            | ActiveStorePublicationAttempt::SnapshotSuperseded {
421                retired_objects, ..
422            } => retired_objects,
423            _ => {
424                return Err(DbError::Message(
425                    "snapshot cleanup has another publication owner".into(),
426                ));
427            }
428        };
429        if retired_objects.is_empty() {
430            return Err(DbError::Message("snapshot has no pending cleanup".into()));
431        }
432        retired_objects.clear();
433        Ok(())
434    }
435
436    pub fn superseding_snapshot(&self) -> Option<&PublishedStoreSnapshot> {
437        match &self.attempt {
438            ActiveStorePublicationAttempt::SnapshotSuperseded { snapshot, .. } => Some(snapshot),
439            _ => None,
440        }
441    }
442
443    pub(crate) fn supersede_snapshot(
444        &self,
445        snapshot: PublishedStoreSnapshot,
446        retired_objects: Vec<coven_protocol::objects::ExactObjectRef>,
447    ) -> Result<Self, DbError> {
448        let ActiveStorePublicationAttempt::Snapshot {
449            publication,
450            retired_objects: pending,
451        } = &self.attempt
452        else {
453            return Err(DbError::Message(
454                "only a pending snapshot request can be superseded".into(),
455            ));
456        };
457        if !pending.is_empty() || self.owner != ActiveStorePublicationOwner::Snapshot {
458            return Err(DbError::Message(
459                "snapshot supersession has unfinished candidate cleanup".into(),
460            ));
461        }
462        let mut superseded = self.clone();
463        superseded.attempt = ActiveStorePublicationAttempt::SnapshotSuperseded {
464            publication: publication.clone(),
465            snapshot,
466            retired_objects,
467        };
468        Ok(superseded)
469    }
470
471    pub(crate) fn await_preparation(
472        &self,
473        cleanup: RetiredStoreCandidate,
474    ) -> Result<Self, DbError> {
475        if self.covered_write_position().is_some() {
476            return Err(DbError::Message(
477                "covered write completion cannot return to preparation".into(),
478            ));
479        }
480        let Some((write_id, author_registration, coord)) = self.commit_reservation() else {
481            return Err(DbError::Message(
482                "snapshot publication cannot reserve a Store write".to_string(),
483            ));
484        };
485        let candidate = cleanup.candidate()?;
486        let owner_matches = match (&self.owner, &cleanup.inputs) {
487            (
488                ActiveStorePublicationOwner::StoreWrite(owner),
489                RetiredStoreCandidateInputs::Write(_),
490            ) => owner == write_id,
491            (
492                ActiveStorePublicationOwner::MembershipMutation,
493                RetiredStoreCandidateInputs::Membership(_),
494            ) => matches!(
495                cleanup.nonactivation.proof(),
496                coven_protocol::remote_object::CandidateNonactivationProof::AuthorityRetirement { .. }
497            ) && self.retired_candidates.is_empty(),
498            _ => false,
499        };
500        if !owner_matches
501            || candidate.coord != *coord
502            || self.attempt()?.entry.payload
503                != coven_protocol::store_commit::StorePublicationPayload::Commit(candidate.clone())
504        {
505            return Err(DbError::Message(
506                "replacement cleanup differs from the reserved Store write".to_string(),
507            ));
508        }
509        cleanup.objects()?;
510        let mut retired_candidates = self.retired_candidates.clone();
511        retired_candidates.push(cleanup);
512        Ok(Self {
513            owner: self.owner.clone(),
514            attempt: ActiveStorePublicationAttempt::AwaitingPreparation {
515                write_id: write_id.clone(),
516                author_registration: author_registration.clone(),
517                coord: coord.clone(),
518            },
519            superseded_entry: None,
520            retired_candidates,
521        })
522    }
523
524    pub(crate) fn complete_retired_candidate_cleanup(&mut self) -> Result<(), DbError> {
525        if self.retired_candidates.is_empty() {
526            return Err(DbError::Message(
527                "reserved write has no retired candidate cleanup".to_string(),
528            ));
529        }
530        self.retired_candidates.clear();
531        Ok(())
532    }
533
534    pub(crate) fn continue_membership_after_abandonment(
535        &self,
536        cleanup: RetiredStoreCandidate,
537    ) -> Result<Self, DbError> {
538        let Some((write_id, author_registration, coord)) = self.commit_reservation() else {
539            return Err(DbError::Message(
540                "membership abandonment has no reserved author position".into(),
541            ));
542        };
543        if self.owner != ActiveStorePublicationOwner::MembershipMutation
544            || !matches!(
545                self.attempt,
546                ActiveStorePublicationAttempt::MembershipAbandonment { .. }
547            )
548            || cleanup.candidate()?.coord != *coord
549            || !matches!(cleanup.inputs, RetiredStoreCandidateInputs::Membership(_))
550            || !self.retired_candidates.is_empty()
551            || self.superseded_entry.is_some()
552        {
553            return Err(DbError::Message(
554                "membership abandonment differs from its reserved mutation".into(),
555            ));
556        }
557        cleanup.objects()?;
558        let sequence = coord.sequence.checked_add(1).ok_or_else(|| {
559            DbError::Message("membership continuation exhausts its author sequence".into())
560        })?;
561        Ok(Self {
562            owner: self.owner.clone(),
563            attempt: ActiveStorePublicationAttempt::AwaitingPreparation {
564                write_id: write_id.clone(),
565                author_registration: author_registration.clone(),
566                coord: StoreCommitCoord {
567                    stream_id: coord.stream_id,
568                    sequence,
569                },
570            },
571            superseded_entry: None,
572            retired_candidates: vec![cleanup],
573        })
574    }
575
576    pub fn membership_abandonment(
577        &self,
578    ) -> Option<&coven_protocol::prepared_commit::PreparedStoreOperationCommit> {
579        match &self.attempt {
580            ActiveStorePublicationAttempt::MembershipAbandonment { candidate } => Some(candidate),
581            _ => None,
582        }
583    }
584
585    pub(crate) fn begin_membership_abandonment(
586        &self,
587        candidate: coven_protocol::prepared_commit::PreparedStoreOperationCommit,
588    ) -> Result<Self, DbError> {
589        candidate.validate_closed_shape()?;
590        if self.owner != ActiveStorePublicationOwner::MembershipMutation
591            || !matches!(self.attempt, ActiveStorePublicationAttempt::Commit { .. })
592            || self.commit_reservation()
593                != Some((
594                    &candidate.commit.write_id,
595                    &candidate.commit.author_registration,
596                    &candidate.reference.coord,
597                ))
598            || self.superseded_entry.is_some()
599            || !self.retired_candidates.is_empty()
600        {
601            return Err(DbError::Message(
602                "membership abandonment changes its operation or retains earlier cleanup".into(),
603            ));
604        }
605        Self::validated(
606            self.owner.clone(),
607            ActiveStorePublicationAttempt::MembershipAbandonment {
608                candidate: Box::new(candidate),
609            },
610        )
611    }
612
613    pub(crate) fn replace_acknowledgement_candidate(
614        &self,
615        candidate: &coven_protocol::prepared_commit::PreparedStoreOperationCommit,
616        cleanup: RetiredStoreCandidate,
617    ) -> Result<Self, DbError> {
618        if self.owner != ActiveStorePublicationOwner::StoreAcknowledgement
619            || self.commit_reservation()
620                != Some((
621                    &candidate.commit.write_id,
622                    &candidate.commit.author_registration,
623                    &candidate.reference.coord,
624                ))
625            || self.attempt()?.entry.payload
626                != coven_protocol::store_commit::StorePublicationPayload::Commit(
627                    cleanup.candidate()?,
628                )
629            || !matches!(
630                cleanup.inputs,
631                RetiredStoreCandidateInputs::Acknowledgement(_)
632            )
633        {
634            return Err(DbError::Message(
635                "acknowledgement replacement changes its reserved operation".into(),
636            ));
637        }
638        cleanup.objects()?;
639        let mut replacement = self.replace_attempt(candidate.publication.clone())?;
640        replacement.retired_candidates.push(cleanup);
641        Ok(replacement)
642    }
643
644    pub fn commit_reservation(
645        &self,
646    ) -> Option<(&WriteId, &StoreDeviceRegistrationRef, &StoreCommitCoord)> {
647        match &self.attempt {
648            ActiveStorePublicationAttempt::AwaitingPreparation {
649                write_id,
650                author_registration,
651                coord,
652                ..
653            }
654            | ActiveStorePublicationAttempt::Discarding {
655                write_id,
656                author_registration,
657                coord,
658            }
659            | ActiveStorePublicationAttempt::Commit {
660                write_id,
661                author_registration,
662                coord,
663                ..
664            } => Some((write_id, author_registration, coord)),
665            ActiveStorePublicationAttempt::MembershipAbandonment { candidate } => Some((
666                &candidate.commit.write_id,
667                &candidate.commit.author_registration,
668                &candidate.reference.coord,
669            )),
670            ActiveStorePublicationAttempt::CompletingCoveredWrite {
671                write_id, position, ..
672            } => Some((write_id, &position.author_registration, &position.coord)),
673            ActiveStorePublicationAttempt::Snapshot { .. }
674            | ActiveStorePublicationAttempt::SnapshotSuperseded { .. } => None,
675        }
676    }
677
678    pub fn superseded_entry(&self) -> Option<&coven_protocol::store_commit::StorePublicationRef> {
679        self.superseded_entry.as_ref()
680    }
681
682    pub fn same_commit_reservation(&self, other: &Self) -> bool {
683        self.commit_reservation().is_some()
684            && self.owner == other.owner
685            && self.commit_reservation() == other.commit_reservation()
686            && matches!((self.attempt(), other.attempt()), (Ok(left), Ok(right)) if left.entry.payload == right.entry.payload)
687    }
688
689    pub(crate) fn retain_superseded_entry(
690        &mut self,
691        reference: coven_protocol::store_commit::StorePublicationRef,
692    ) -> Result<(), DbError> {
693        if self.superseded_entry.is_some() || reference == self.attempt()?.reference()? {
694            return Err(DbError::Message(
695                "Store publication already owns superseded entry cleanup".to_string(),
696            ));
697        }
698        self.superseded_entry = Some(reference);
699        Ok(())
700    }
701
702    pub(crate) fn complete_superseded_entry_cleanup(&mut self) -> Result<(), DbError> {
703        if self.superseded_entry.take().is_none() {
704            return Err(DbError::Message(
705                "Store publication has no superseded entry cleanup".to_string(),
706            ));
707        }
708        Ok(())
709    }
710
711    pub fn replace_attempt(
712        &self,
713        attempt: coven_protocol::prepared_commit::PreparedStorePublication,
714    ) -> Result<Self, DbError> {
715        if self.superseded_entry.is_some() || !self.retired_snapshot_objects().is_empty() {
716            return Err(DbError::Message(
717                "Store publication must finish superseded entry cleanup before another replacement"
718                    .to_string(),
719            ));
720        }
721        let replacement = match &self.attempt {
722            ActiveStorePublicationAttempt::Discarding { .. }
723            | ActiveStorePublicationAttempt::CompletingCoveredWrite { .. }
724            | ActiveStorePublicationAttempt::SnapshotSuperseded { .. } => {
725                return Err(DbError::Message(
726                    "terminal operation cannot prepare another publication candidate".to_string(),
727                ));
728            }
729            ActiveStorePublicationAttempt::AwaitingPreparation {
730                write_id,
731                author_registration,
732                coord,
733            } => ActiveStorePublicationAttempt::Commit {
734                write_id: write_id.clone(),
735                author_registration: author_registration.clone(),
736                coord: coord.clone(),
737                publication: attempt,
738            },
739            ActiveStorePublicationAttempt::Commit {
740                write_id,
741                author_registration,
742                coord,
743                ..
744            } => ActiveStorePublicationAttempt::Commit {
745                write_id: write_id.clone(),
746                author_registration: author_registration.clone(),
747                coord: coord.clone(),
748                publication: attempt,
749            },
750            ActiveStorePublicationAttempt::Snapshot { .. } => {
751                ActiveStorePublicationAttempt::Snapshot {
752                    publication: attempt,
753                    retired_objects: Vec::new(),
754                }
755            }
756            ActiveStorePublicationAttempt::MembershipAbandonment { candidate } => {
757                let mut candidate = candidate.clone();
758                candidate.publication = attempt;
759                candidate.validate_closed_shape()?;
760                ActiveStorePublicationAttempt::MembershipAbandonment { candidate }
761            }
762        };
763        let mut replacement = Self::validated(self.owner.clone(), replacement)?;
764        replacement.retired_candidates = self.retired_candidates.clone();
765        Ok(replacement)
766    }
767}
768
769#[derive(Debug, Clone)]
770pub struct DurableDeviceRegistration {
771    pub device_id: coven_protocol::store_commit::StoreDeviceId,
772    pub registration_hash: ObjectHash,
773    pub registration_bytes: Vec<u8>,
774    pub prepared: PreparedExactObject,
775    pub initial_ack_ref: StoreAckRef,
776    pub initial_ack: ExactProtocolObject<StoreAck>,
777    pub state: LocalDeviceRegistrationState,
778}
779
780/// The exact Owner-recovery commit and publication staged before either is
781/// published. A retry reopens these same stored bytes instead of resealing the
782/// semantic objects into different exact identities.
783pub struct OwnerRecoveryPublication {
784    pub commit: ExactProtocolObject<coven_protocol::store_commit::VerifiedStoreBatchCommit>,
785    pub publication: coven_protocol::prepared_commit::PreparedStorePublication,
786    pub history_evidence: coven_protocol::store_commit::RetainedMergeCommitEvidence,
787}
788
789impl OwnerRecoveryPublication {
790    pub fn membership_publication(
791        &self,
792    ) -> Result<coven_protocol::membership_mutation::PreparedMembershipPublication, DbError> {
793        let proof = self
794            .history_evidence
795            .membership_proof
796            .as_ref()
797            .ok_or_else(|| {
798                DbError::Message("Owner recovery lacks its exact membership activation".into())
799            })?;
800        let publication = coven_protocol::membership_mutation::PreparedMembershipPublication {
801            entry: proof.entry_value.clone(),
802            entry_ref: proof.entry.clone(),
803            head: proof.head_value.clone(),
804            head_ref: proof.head.clone(),
805        };
806        publication.validate().map_err(|error| {
807            DbError::context(
808                "Owner recovery membership publication",
809                coven_protocol::prepared_commit::PreparedCommitError::from(error),
810            )
811        })?;
812        Ok(publication)
813    }
814
815    pub fn remote_objects(
816        &self,
817    ) -> Result<Vec<coven_protocol::remote_object::ClosedRemoteObject>, DbError> {
818        let mut objects = self
819            .membership_publication()?
820            .candidate_remote_objects(self.commit.value.value(), self.commit.value.reference())?;
821        objects.push(
822            coven_protocol::remote_object::RemoteObjectRecord::candidate_commit(
823                self.commit.value.reference().clone(),
824                &self.commit.bytes,
825                self.commit.prepared.stored_bytes(),
826            )?,
827        );
828        Ok(objects)
829    }
830}
831
832#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
833#[serde(rename_all = "snake_case", deny_unknown_fields)]
834pub enum LocalDeviceRegistrationState {
835    Prepared,
836    RegistrationPublished,
837    RegistrationActivated {
838        authority: coven_protocol::store_commit::StoreDeviceRegistrationActivation,
839    },
840    Created,
841    Activated {
842        authority: coven_protocol::store_commit::StoreDeviceRegistrationActivation,
843    },
844}
845
846pub type PreparedLocalDeviceRegistrationRow =
847    (String, String, Vec<u8>, String, String, Vec<u8>, String);
848pub type LocalDeviceRegistrationJournalRow = (
849    String,
850    String,
851    Vec<u8>,
852    String,
853    String,
854    Vec<u8>,
855    String,
856    String,
857);
858
859impl DurableDeviceRegistration {
860    pub fn is_activated(&self) -> bool {
861        matches!(self.state, LocalDeviceRegistrationState::Activated { .. })
862    }
863}
864
865#[derive(Debug, Clone)]
866pub struct DurableMembershipMutation {
867    pub intent_hash: ObjectHash,
868    pub plan_bytes: Vec<u8>,
869    pub progress_bytes: Vec<u8>,
870}
871
872pub struct DurableSnapshotPublication {
873    pub reference: StoreSnapshotRef,
874    pub meta: ExactProtocolObject<SnapshotMeta>,
875    pub publication: coven_protocol::prepared_commit::PreparedStorePublication,
876    /// The membership rollup the metadata names. Staged with the snapshot and
877    /// uploaded before it, so a published snapshot never names a rollup that is
878    /// not at the provider.
879    pub rollup: ExactProtocolObject<coven_protocol::store_commit::MembershipRollup>,
880    pub image: PreparedProtocolObject<Vec<u8>>,
881    pub blobs: Vec<PreparedSnapshotBlob>,
882}
883
884pub enum StoreSnapshotPublicationStage {
885    Initial,
886    Replacing {
887        previous: StoreSnapshotRef,
888        accepted: crate::AcceptedStorePublicationInterval,
889    },
890}
891
892#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
893#[serde(deny_unknown_fields)]
894pub struct PreparedSnapshotBlob {
895    pub bindings: Vec<RowBlobLocatorBinding>,
896    pub authority: coven_protocol::audience_package::PackageAudience,
897    pub remote: RemoteObjectRecord,
898}
899
900#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
901#[serde(deny_unknown_fields)]
902pub struct PublishedStoreSnapshot {
903    pub reference: StoreSnapshotRef,
904    pub meta: SnapshotMeta,
905}
906
907pub struct DurableCircleSnapshotPublication {
908    pub reference: coven_protocol::store_commit::CircleSnapshotRef,
909    pub meta: ExactProtocolObject<coven_protocol::store_commit::CircleSnapshotMeta>,
910    pub image: PreparedProtocolObject<Vec<u8>>,
911}
912
913#[derive(Debug, Clone)]
914pub struct PublishedCircleSnapshot {
915    pub reference: coven_protocol::store_commit::CircleSnapshotRef,
916    pub successor_slot: coven_protocol::objects::ObjectSlot,
917    pub cut: coven_protocol::store_commit::CommitFrontier,
918}
coven_database/operation_models.rs

coven_database/
operation_models.rs
