coven_replication/sync/store/commit_publication/operation/commit_plan/
plan.rs

1use super::*;
2
3pub(crate) enum StoreOperationBatch {
4    Acknowledgement {
5        reference: super::store_commit::StoreAckRef,
6        value: super::store_commit::StoreAck,
7        circle_acknowledgements: Vec<CircleAckActivation>,
8    },
9
10    ProviderAccessGrant(super::provider::StoreMemberProviderAccessGrantRef),
11    Attempt(coven_protocol::store_commit::DeviceJoinAttemptId),
12    SamePrincipalDeviceJoin {
13        attempt_id: coven_protocol::store_commit::DeviceJoinAttemptId,
14        registration: Box<ActivatedStoreDeviceRegistration>,
15    },
16    Abandonment(coven_protocol::store_commit::DeviceJoinAbandonmentRef),
17    JoinActivation {
18        registration: Box<ActivatedStoreDeviceRegistration>,
19    },
20    DeviceExclusionProposal(super::store_commit::RetainedStoreDeviceExclusionProposal),
21    DeviceExclusionOutcome(super::store_commit::RetainedStoreDeviceExclusionOutcome),
22    ReclaimAuthorization(Box<coven_protocol::reclaim::ReclaimAuthorizationRef>),
23    ReclaimReceipt(Box<coven_protocol::reclaim::ReclaimReceiptRef>),
24    OwnerPromotionRequest(super::store_commit::OwnerPromotionRequest),
25    MergeMembershipActivation {
26        transition: super::membership::MergeMembershipHeadTransition,
27        stream_activations: Vec<super::store_commit::StreamActivation>,
28    },
29}
30
31pub struct StoreOperationPlanCommon {
32    /// This device's turn to author its own next Store commit, taken when the
33    /// position this plan's order extends was read. A plan is the live claim on
34    /// that position: hold it until the commit has published its head, or until
35    /// the candidate is durably persisted for a later publisher to activate.
36    _authorship: coven_database::OwnStreamAuthorship,
37    writer: std::sync::Arc<LocalStoreWriter>,
38    root: StoreRootRef,
39    coord: StoreCommitCoord,
40    order: StoreCommitOrder,
41    membership_state: super::circle_control::StoreMembershipStateRef,
42    device_state: super::store_commit::StoreDeviceStateRef,
43    membership_authority: StoreOperationMembershipAuthority,
44    owner_grant: Option<super::membership::MembershipGrantId>,
45}
46
47pub struct StoreOperationCommitPlan {
48    common: StoreOperationPlanCommon,
49    membership: MembershipChain,
50    predecessor_state: super::store_commit::ResolvedStoreDeviceState,
51}
52
53impl std::ops::Deref for StoreOperationCommitPlan {
54    type Target = StoreOperationPlanCommon;
55
56    fn deref(&self) -> &Self::Target {
57        &self.common
58    }
59}
60
61impl StoreOperationPlanCommon {
62    #[allow(clippy::too_many_arguments)]
63    pub(crate) fn new(
64        authorship: coven_database::OwnStreamAuthorship,
65        writer: std::sync::Arc<LocalStoreWriter>,
66        root: StoreRootRef,
67        coord: StoreCommitCoord,
68        order: StoreCommitOrder,
69        membership_state: super::circle_control::StoreMembershipStateRef,
70        device_state: super::store_commit::StoreDeviceStateRef,
71        membership_authority: StoreOperationMembershipAuthority,
72        owner_grant: Option<super::membership::MembershipGrantId>,
73    ) -> Self {
74        Self {
75            _authorship: authorship,
76            writer,
77            root,
78            coord,
79            order,
80            membership_state,
81            device_state,
82            membership_authority,
83            owner_grant,
84        }
85    }
86
87    pub(crate) fn validate_acknowledgement(
88        &self,
89        acknowledgement: &super::store_commit::StoreAck,
90    ) -> Result<(), StoreError> {
91        let predecessor_cut = self.order.predecessor_cut().map_err(StoreError::from)?;
92        if !self
93            .writer
94            .is_authored_by_registration(&acknowledgement.registration)
95            || acknowledgement.store_cut != predecessor_cut
96            || acknowledgement.device_state != self.device_state
97        {
98            return Err(StoreError::InvalidOutbound(
99                "Store acknowledgement differs from its operation commit predecessor".to_string(),
100            ));
101        }
102        Ok(())
103    }
104
105    fn sign_batch(
106        &self,
107        write_id: coven_protocol::write::WriteId,
108        batch: StoreOperationBatch,
109    ) -> Result<(StoreBatchCommit, Option<ActivatedStoreDeviceRegistration>), StoreError> {
110        self.writer.sign_operation_batch(
111            write_id,
112            StoreOperationSigningContext {
113                root: self.root.clone(),
114                coord: self.coord.clone(),
115                order: self.order.clone(),
116                membership_state: self.membership_state.clone(),
117                device_state: self.device_state.clone(),
118                membership_authority: self.membership_authority.clone(),
119            },
120            batch,
121        )
122    }
123}
124
125impl StoreOperationCommitPlan {
126    pub(crate) fn new(
127        common: StoreOperationPlanCommon,
128        membership: MembershipChain,
129        predecessor_state: super::store_commit::ResolvedStoreDeviceState,
130    ) -> Self {
131        Self {
132            common,
133            membership,
134            predecessor_state,
135        }
136    }
137
138    pub(crate) fn membership(&self) -> &MembershipChain {
139        &self.membership
140    }
141
142    pub(crate) fn predecessor_state(&self) -> &super::store_commit::ResolvedStoreDeviceState {
143        &self.predecessor_state
144    }
145
146    pub(crate) fn sign_batch(
147        &self,
148        write_id: coven_protocol::write::WriteId,
149        batch: StoreOperationBatch,
150    ) -> Result<(StoreBatchCommit, Option<ActivatedStoreDeviceRegistration>), StoreError> {
151        self.common.sign_batch(write_id, batch)
152    }
153
154    pub(crate) fn validate_acknowledgement(
155        &self,
156        acknowledgement: &super::store_commit::StoreAck,
157    ) -> Result<(), StoreError> {
158        self.common.validate_acknowledgement(acknowledgement)
159    }
160
161    #[allow(clippy::too_many_arguments)]
162    pub(crate) fn sign_owner_promotion_request(
163        &self,
164        promotion_id: super::store_commit::OwnerPromotionId,
165        member_registration: super::store_commit::StoreDeviceRegistrationRef,
166        member_pubkey: String,
167        member_grant: super::membership::MembershipGrantId,
168        finalization: super::store_commit::OwnerPromotionFinalization,
169    ) -> Result<super::store_commit::OwnerPromotionRequest, StoreError> {
170        let promoter_owner_grant = self.owner_grant.clone().ok_or_else(|| {
171            StoreError::InvalidOutbound(
172                "Owner-promotion request author has no active Owner grant".to_string(),
173            )
174        })?;
175        self.writer.sign_owner_promotion_request(
176            promotion_id,
177            &self.root,
178            promoter_owner_grant,
179            member_pubkey,
180            member_grant,
181            member_registration,
182            self.membership_state.clone(),
183            self.device_state.clone(),
184            finalization,
185        )
186    }
187
188    pub(crate) fn predecessor_cut(&self) -> Result<StoreHistoryCut, StoreError> {
189        self.order.predecessor_cut().map_err(StoreError::from)
190    }
191
192    pub(crate) fn membership_state(&self) -> &super::circle_control::StoreMembershipStateRef {
193        &self.membership_state
194    }
195
196    #[cfg(test)]
197    pub(crate) fn membership_authority(&self) -> &StoreOperationMembershipAuthority {
198        &self.membership_authority
199    }
200
201    pub(crate) fn device_state(&self) -> &super::store_commit::StoreDeviceStateRef {
202        &self.device_state
203    }
204
205    pub(crate) fn root(&self) -> &StoreRootRef {
206        &self.root
207    }
208
209    pub(crate) fn coord(&self) -> &StoreCommitCoord {
210        &self.coord
211    }
212
213    pub(crate) fn device_id(&self) -> &super::store_commit::StoreDeviceId {
214        self.writer.device_id()
215    }
216
217    pub(crate) fn author_pubkey(&self) -> String {
218        self.writer.author_pubkey()
219    }
220
221    pub(crate) fn is_local_registration(
222        &self,
223        registration: &super::store_commit::StoreDeviceRegistrationRef,
224    ) -> bool {
225        self.writer.is_authored_by_registration(registration)
226    }
227
228    pub(crate) fn retain_device_exclusion_proposal(
229        &self,
230        reference: super::store_commit::StoreDeviceExclusionProposalRef,
231        proposal: &super::store_commit::StoreDeviceExclusionProposal,
232        target: &super::store_commit::StoreDeviceRegistration,
233    ) -> Result<super::store_commit::RetainedStoreDeviceExclusionProposal, StoreError> {
234        self.writer
235            .retain_device_exclusion_proposal(reference, proposal, target)
236            .map_err(StoreError::from)
237    }
238
239    pub(crate) fn retain_device_exclusion_outcome(
240        &self,
241        reference: &super::store_commit::StoreDeviceExclusionOutcomeRef,
242        proposal: super::store_commit::RetainedStoreDeviceExclusionProposal,
243        outcome: &super::store_commit::StoreDeviceExclusionOutcome,
244    ) -> Result<super::store_commit::RetainedStoreDeviceExclusionOutcome, StoreError> {
245        self.writer
246            .retain_device_exclusion_outcome(reference, proposal, outcome)
247            .map_err(StoreError::from)
248    }
249
250    pub(crate) fn announcement_activation_id(
251        &self,
252    ) -> Result<super::store_commit::StreamActivationId, StoreError> {
253        self.writer
254            .announcement_activation_id()
255            .map_err(StoreError::from)
256    }
257
258    pub(crate) fn verify_prepared_commit(
259        &self,
260        bytes: &[u8],
261        object: coven_protocol::objects::ExactObjectRef,
262    ) -> Result<super::store_commit::VerifiedStoreBatchCommit, StoreError> {
263        self.writer
264            .verify_prepared_commit(bytes, self.root.store_root_hash, self.coord.clone(), object)
265            .map_err(StoreError::from)
266    }
267
268    pub(crate) async fn retain_acknowledgement(
269        &self,
270        history: &AuthorizedStoreHistory<'_>,
271        activating_commit: &super::store_commit::StoreBatchCommitRef,
272        activating_commit_value: &super::store_commit::StoreBatchCommit,
273        reference: super::store_commit::StoreAckRef,
274        value: super::store_commit::StoreAck,
275    ) -> Result<super::store_commit::RetainedVerifiedActivatedAck, pull::StorePullError> {
276        self.writer
277            .retain_acknowledgement(
278                history,
279                activating_commit,
280                activating_commit_value,
281                reference,
282                value,
283            )
284            .await
285    }
286
287    pub(crate) fn owner_grant(&self) -> Option<&super::membership::MembershipGrantId> {
288        self.owner_grant.as_ref()
289    }
290
291    pub(crate) fn effective_provider_admin_grant(
292        &self,
293        state: &coven_protocol::provider::ProviderAdminState,
294    ) -> Option<coven_protocol::provider::ProviderAdminGrantId> {
295        self.writer.effective_provider_admin_grant(state)
296    }
297
298    pub(crate) fn sign_reclaim_evidence(
299        &self,
300        claim: coven_protocol::reclaim::ReclaimClaim,
301    ) -> Result<coven_protocol::reclaim::ReclaimEvidence, StoreError> {
302        self.writer
303            .sign_reclaim_evidence(self.root.store_root_hash, claim)
304            .map_err(StoreError::from)
305    }
306
307    pub(crate) fn sign_reclaim_authorization(
308        &self,
309        target: coven_protocol::reclaim::ReclaimTarget,
310        evidence: coven_protocol::reclaim::ReclaimEvidenceRef,
311        authority: coven_protocol::reclaim::StoreReclaimAuthority,
312    ) -> coven_protocol::reclaim::ReclaimAuthorization {
313        self.writer.sign_reclaim_authorization(
314            self.root.store_root_hash,
315            target,
316            evidence,
317            authority,
318        )
319    }
320
321    #[allow(clippy::too_many_arguments)]
322    pub(crate) fn sign_device_exclusion_proposal(
323        &self,
324        proposal_id: super::store_commit::StoreDeviceExclusionProposalId,
325        target: super::store_commit::StoreDeviceRegistrationRef,
326        target_registration: &super::store_commit::StoreDeviceRegistration,
327        outcome_slot: coven_protocol::objects::ObjectSlot,
328        owner_grant: super::membership::MembershipGrantId,
329    ) -> Result<super::store_commit::StoreDeviceExclusionProposal, StoreError> {
330        self.writer.sign_device_exclusion_proposal(
331            self.root.store_root_hash,
332            proposal_id,
333            target,
334            target_registration,
335            self.device_state.clone(),
336            outcome_slot,
337            owner_grant,
338        )
339    }
340
341    pub(crate) fn sign_device_exclusion_cancellation(
342        &self,
343        proposal: super::store_commit::StoreDeviceExclusionProposalRef,
344        proposal_value: &super::store_commit::StoreDeviceExclusionProposal,
345        owner_grant: super::membership::MembershipGrantId,
346    ) -> Result<super::store_commit::StoreDeviceExclusionCancellation, StoreError> {
347        self.writer
348            .sign_device_exclusion_cancellation(proposal, proposal_value, owner_grant)
349    }
350
351    #[allow(clippy::too_many_arguments)]
352    pub(crate) fn sign_device_exclusion(
353        &self,
354        proposal: super::store_commit::StoreDeviceExclusionProposalRef,
355        proposal_value: &super::store_commit::StoreDeviceExclusionProposal,
356        target: super::store_commit::StoreDeviceRegistrationRef,
357        target_registration: &super::store_commit::StoreDeviceRegistration,
358        proof: super::store_commit::StoreDeviceExclusionProof,
359        owner_grant: super::membership::MembershipGrantId,
360    ) -> Result<super::store_commit::StoreDeviceExclusion, StoreError> {
361        self.writer.sign_device_exclusion(
362            proposal,
363            proposal_value,
364            target,
365            target_registration,
366            proof,
367            owner_grant,
368        )
369    }
370
371    pub(crate) fn sign_device_head(
372        &self,
373        commit: super::store_commit::StoreBatchCommitRef,
374        successor: super::store_commit::SuccessorLink,
375    ) -> Result<super::store_commit::StoreDeviceHead, StoreError> {
376        self.writer
377            .sign_device_head(self.root.store_root_hash, commit, successor)
378    }
379
380    pub(crate) fn sign_reclaim_receipt(
381        &self,
382        authorization: coven_protocol::reclaim::ReclaimAuthorizationRef,
383        provider_admin_grant: coven_protocol::provider::ProviderAdminGrantId,
384    ) -> Result<coven_protocol::reclaim::ReclaimReceipt, StoreError> {
385        self.writer.sign_reclaim_receipt(
386            self.root.store_root_hash,
387            authorization,
388            self.membership_state.clone(),
389            provider_admin_grant,
390        )
391    }
392
393    #[cfg(any(test, feature = "test-utils"))]
394    pub(crate) fn local_registration_reference_for_test(
395        &self,
396    ) -> super::store_commit::StoreDeviceRegistrationRef {
397        self.writer.registration_reference_for_test()
398    }
399}
coven_replication/sync/store/commit_publication/operation/commit_plan/plan.rs

coven_replication/sync/store/commit_publication/operation/commit_plan/
plan.rs
