1mod history;
4pub(crate) use history::DeviceExclusionHistory;
5
6use coven_protocol::device_exclusion_journal::{
7 DurableStoreDeviceExclusionObject, DurableStoreDeviceExclusionOperation,
8 StoreDeviceExclusionCompletion, StoreDeviceExclusionJournalError,
9};
10
11use super::{AuthorizedWriterOperation, StoreError};
12use crate::sync::store::commit_publication::operation::commit_plan::{
13 PreparedStoreOperationCommit, StoreOperationBatch, StoreOperationPublicationOutcome,
14};
15use crate::sync::store::commit_verification::merge_history::MergeHistoryVerifier;
16use coven_database::DbError;
17use coven_database::StoreDatabase;
18use coven_protocol::objects::{ProtocolObjectContext, ProtocolObjectDomain};
19use coven_protocol::store_commit::{
20 device_exclusion_outcome_semantic_prefix, device_exclusion_proposal_semantic_prefix,
21 ObjectHash, StoreBatchCommitRef, StoreDeviceExclusionOutcome, StoreDeviceExclusionOutcomeRef,
22 StoreDeviceExclusionProof, StoreDeviceExclusionProposal, StoreDeviceExclusionProposalId,
23 StoreDeviceExclusionProposalRef, StoreDeviceProposalState, StoreDeviceStatus, StoreHistoryCut,
24 StoreProtocolError,
25};
26use coven_storage::CloudSyncObjectStorage;
27
28#[derive(Debug, Clone, PartialEq, Eq)]
29pub enum StoreDeviceExclusionResult {
30 ProposalActivated {
31 proposal: StoreDeviceExclusionProposalRef,
32 commit: StoreBatchCommitRef,
33 },
34 OutcomeActivated {
35 outcome: StoreDeviceExclusionOutcomeRef,
36 commit: StoreBatchCommitRef,
37 },
38 OutcomeSlotOccupied {
39 intended: StoreDeviceExclusionOutcomeRef,
40 winner: StoreDeviceExclusionOutcomeRef,
41 },
42 CandidateNonactivated {
43 object_hash: ObjectHash,
44 candidate: StoreBatchCommitRef,
45 },
46}
47
48#[cfg(any(test, feature = "test-utils"))]
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub struct StoreDeviceExclusionOperationInfo {
51 pub operation_id: ObjectHash,
52 pub status: StoreDeviceExclusionOperationStatus,
53}
54
55#[cfg(any(test, feature = "test-utils"))]
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub enum StoreDeviceExclusionOperationStatus {
58 Pending,
59 Completed(StoreDeviceExclusionResult),
60}
61
62#[derive(Debug, thiserror::Error)]
63pub enum StoreDeviceExclusionError {
64 #[error("Store-device exclusion operation {0} remains active")]
65 OperationActive(ObjectHash),
66 #[error("the local Store device has no active Owner authority")]
67 OwnerAuthorityRequired,
68 #[error("the target Store device is not active at the exact predecessor state")]
69 TargetNotActive,
70 #[error("the active Owner device cannot exclude its own registration")]
71 CannotExcludeLocalDevice,
72 #[error("Store-device exclusion database state: {0}")]
73 Database(#[from] DbError),
74 #[error("Store-device exclusion object: {0}")]
75 Object(#[from] coven_protocol::objects::StoreObjectError),
76 #[error("Store-device exclusion protocol: {0}")]
77 Protocol(#[from] StoreProtocolError),
78 #[error("Store-device exclusion JSON: {0}")]
79 Json(#[from] serde_json::Error),
80 #[error("Store-device exclusion publication: {0}")]
81 Outbound(#[from] StoreError),
82 #[error("Store-device exclusion storage: {0}")]
83 Storage(#[from] coven_protocol::objects::StorageError),
84 #[error("Store-device exclusion journal: {0}")]
85 Journal(#[from] StoreDeviceExclusionJournalError),
86 #[error("Store-device exclusion state is invalid: {0}")]
87 InvalidState(String),
88}
89
90pub(crate) async fn propose_for_device(
92 database: &StoreDatabase,
93 writer: &mut AuthorizedWriterOperation<'_>,
94 device_id: coven_protocol::store_commit::StoreDeviceId,
95) -> Result<StoreDeviceExclusionProposalRef, StoreDeviceExclusionError> {
96 let target = database
97 .activated_store_device_registration_for_device(device_id)
98 .await?
99 .ok_or(StoreDeviceExclusionError::TargetNotActive)?;
100 match writer
101 .device_exclusion()
102 .propose(target.reference())
103 .await?
104 {
105 StoreDeviceExclusionResult::ProposalActivated { proposal, .. } => Ok(proposal),
106 other => Err(StoreDeviceExclusionError::InvalidState(format!(
107 "proposal did not activate: {other:?}"
108 ))),
109 }
110}
111
112pub(crate) async fn cancel_proposal(
113 writer: &mut AuthorizedWriterOperation<'_>,
114 proposal: &StoreDeviceExclusionProposalRef,
115) -> Result<(), StoreDeviceExclusionError> {
116 match writer.device_exclusion().cancel(proposal).await? {
117 StoreDeviceExclusionResult::OutcomeActivated { .. } => Ok(()),
118 other => Err(StoreDeviceExclusionError::InvalidState(format!(
119 "cancellation did not activate: {other:?}"
120 ))),
121 }
122}
123
124pub(crate) async fn finalize_proposal(
125 writer: &mut AuthorizedWriterOperation<'_>,
126 proposal: &StoreDeviceExclusionProposalRef,
127) -> Result<(), StoreDeviceExclusionError> {
128 match writer.device_exclusion().exclude(proposal).await? {
129 StoreDeviceExclusionResult::OutcomeActivated { .. } => Ok(()),
130 other => Err(StoreDeviceExclusionError::InvalidState(format!(
131 "exclusion did not activate: {other:?}"
132 ))),
133 }
134}
135
136#[cfg(any(test, feature = "test-utils"))]
137pub(crate) async fn operations_for_test(
138 database: &StoreDatabase,
139) -> Result<Vec<StoreDeviceExclusionOperationInfo>, StoreDeviceExclusionError> {
140 database
141 .outbound_store_device_exclusion_operations()
142 .await?
143 .into_iter()
144 .map(|operation| {
145 let operation_id = operation.operation_id();
146 let status = if operation.is_completed() {
147 StoreDeviceExclusionOperationStatus::Completed(completion_result(&operation)?)
148 } else {
149 StoreDeviceExclusionOperationStatus::Pending
150 };
151 Ok(StoreDeviceExclusionOperationInfo {
152 operation_id,
153 status,
154 })
155 })
156 .collect()
157}
158
159#[cfg(any(test, feature = "test-utils"))]
166pub(crate) async fn stage_uploaded_proposal_for_test(
167 database: &StoreDatabase,
168 writer: &mut AuthorizedWriterOperation<'_>,
169) -> Result<StoreDeviceExclusionProposalRef, StoreDeviceExclusionError> {
170 let plan = Box::new(writer.prepare_plan().await?);
171 let target = plan.local_registration_reference_for_test();
172 let proposal_id = StoreDeviceExclusionProposalId::from_hash(ObjectHash::digest(
173 b"restart exclusion proposal",
174 ));
175 let mut exclusion = writer.device_exclusion();
176 let durable = exclusion.stage_proposal(plan, &target, proposal_id).await?;
177 let DurableStoreDeviceExclusionObject::Proposal { reference, .. } = durable.object() else {
178 return Err(StoreDeviceExclusionError::InvalidState(
179 "staged exclusion operation is not a proposal".to_string(),
180 ));
181 };
182 let reference = reference.clone();
183 exclusion.create_exact_object(&durable).await?;
184 database
185 .mark_store_device_exclusion_authority_uploaded(durable)
186 .await?;
187 Ok(reference)
188}
189
190pub(crate) struct AuthorizedDeviceExclusion<'operation, 'storage> {
191 writer: &'operation mut AuthorizedWriterOperation<'storage>,
192 database: StoreDatabase,
193 storage: std::sync::Arc<dyn CloudSyncObjectStorage>,
194}
195
196impl<'operation, 'storage> AuthorizedDeviceExclusion<'operation, 'storage> {
197 pub(crate) fn new(
198 writer: &'operation mut AuthorizedWriterOperation<'storage>,
199 database: StoreDatabase,
200 storage: std::sync::Arc<dyn CloudSyncObjectStorage>,
201 ) -> Self {
202 Self {
203 writer,
204 database,
205 storage,
206 }
207 }
208
209 async fn create_exact_object(
210 &self,
211 operation: &DurableStoreDeviceExclusionOperation,
212 ) -> Result<(), StoreDeviceExclusionJournalError> {
213 let context = operation.object().context();
214 let prefix = operation.object().semantic_prefix()?;
215 self.storage
216 .create_verified_protocol_object(
217 &context,
218 operation.object().prepared(),
219 prefix,
220 &operation.object().semantic_bytes(),
221 )
222 .await
223 .map_err(StoreDeviceExclusionJournalError::Storage)
224 }
225
226 pub(crate) async fn resume(
227 &mut self,
228 ) -> Result<Option<StoreDeviceExclusionResult>, StoreDeviceExclusionError> {
229 let database = self.database.clone();
230 let _lock = database.device_exclusion_permit().await;
231 let Some(operation) = database.active_outbound_store_device_exclusion().await? else {
232 return Ok(None);
233 };
234 self.drive(Box::new(operation)).await.map(Some)
235 }
236
237 async fn reject_active_operation(&self) -> Result<(), StoreDeviceExclusionError> {
238 if let Some(operation) = self
239 .database
240 .active_outbound_store_device_exclusion()
241 .await?
242 {
243 return Err(StoreDeviceExclusionError::OperationActive(
244 operation.operation_id(),
245 ));
246 }
247 Ok(())
248 }
249
250 pub(crate) async fn propose(
251 &mut self,
252 target: &coven_protocol::store_commit::StoreDeviceRegistrationRef,
253 ) -> Result<StoreDeviceExclusionResult, StoreDeviceExclusionError> {
254 let database = self.database.clone();
255 let _lock = database.device_exclusion_permit().await;
256 self.reject_active_operation().await?;
257 let durable = self.prepare_proposal(target).await?;
258 self.drive(Box::new(durable)).await
259 }
260
261 pub(crate) async fn cancel(
262 &mut self,
263 proposal: &StoreDeviceExclusionProposalRef,
264 ) -> Result<StoreDeviceExclusionResult, StoreDeviceExclusionError> {
265 self.publish_outcome(proposal, OutcomeIntent::Cancel).await
266 }
267
268 pub(crate) async fn exclude(
269 &mut self,
270 proposal: &StoreDeviceExclusionProposalRef,
271 ) -> Result<StoreDeviceExclusionResult, StoreDeviceExclusionError> {
272 self.publish_outcome(proposal, OutcomeIntent::Exclude).await
273 }
274
275 async fn prepare_proposal(
276 &mut self,
277 target: &coven_protocol::store_commit::StoreDeviceRegistrationRef,
278 ) -> Result<DurableStoreDeviceExclusionOperation, StoreDeviceExclusionError> {
279 let database = self.database.clone();
280 let plan = Box::new(self.writer.prepare_plan().await?);
281 if plan.is_local_registration(target) {
282 return Err(StoreDeviceExclusionError::CannotExcludeLocalDevice);
283 }
284 let state = Box::new(
285 database
286 .resolved_store_device_state(plan.device_state())
287 .await?,
288 );
289 require_active_target(&state, target)?;
290 let proposal_id = StoreDeviceExclusionProposalId::from_hash(ObjectHash::digest(
291 database.new_store_write_id().as_str().as_bytes(),
292 ));
293 self.stage_proposal(plan, target, proposal_id).await
294 }
295
296 async fn stage_proposal(
301 &mut self,
302 plan: Box<crate::sync::store::commit_publication::operation::commit_plan::StoreOperationCommitPlan>,
303 target: &coven_protocol::store_commit::StoreDeviceRegistrationRef,
304 proposal_id: StoreDeviceExclusionProposalId,
305 ) -> Result<DurableStoreDeviceExclusionOperation, StoreDeviceExclusionError> {
306 let database = self.database.clone();
307 let target_registration = database
308 .activated_store_device_registration(target.clone())
309 .await?;
310 let owner_grant = plan
311 .owner_grant()
312 .cloned()
313 .ok_or(StoreDeviceExclusionError::OwnerAuthorityRequired)?;
314 let outcome_prefix =
315 device_exclusion_outcome_semantic_prefix(target.device_id, proposal_id);
316 let outcome_context = ProtocolObjectContext::signed_plaintext(
317 plan.root().store_root_hash,
318 ProtocolObjectDomain::StoreDeviceExclusionOutcome,
319 );
320 let outcome_slot = self
321 .storage
322 .allocate_protocol_slot(&outcome_context, &outcome_prefix, ".json")
323 .await?;
324 let proposal = plan.sign_device_exclusion_proposal(
325 proposal_id,
326 target.clone(),
327 target_registration.value(),
328 outcome_slot,
329 owner_grant,
330 )?;
331 let proposal_prefix = device_exclusion_proposal_semantic_prefix(
332 target.device_id,
333 proposal_id,
334 proposal.proposal_hash(),
335 );
336 let proposal_context = ProtocolObjectContext::signed_plaintext(
337 plan.root().store_root_hash,
338 ProtocolObjectDomain::StoreDeviceExclusionProposal,
339 );
340 let proposal_slot = self
341 .storage
342 .allocate_protocol_slot(&proposal_context, &proposal_prefix, ".json")
343 .await?;
344 let prepared = self.storage.prepare_protocol_object(
345 &proposal_context,
346 proposal_slot,
347 &proposal_prefix,
348 proposal.to_bytes(),
349 )?;
350 let reference = StoreDeviceExclusionProposalRef::from_proposal(
351 &proposal,
352 prepared.reference().clone(),
353 )?;
354 let retained = plan.retain_device_exclusion_proposal(
355 reference.clone(),
356 &proposal,
357 target_registration.value(),
358 )?;
359 let candidate = Box::pin(self.writer.prepare_candidate(
360 *plan,
361 StoreOperationBatch::DeviceExclusionProposal(retained),
362 ))
363 .await?;
364 let operation = DurableStoreDeviceExclusionOperation::prepared(
365 DurableStoreDeviceExclusionObject::Proposal {
366 reference,
367 value: proposal,
368 prepared,
369 },
370 candidate,
371 )?;
372 let durable = Box::pin(database.begin_outbound_store_device_exclusion(operation)).await?;
373 #[cfg(any(test, feature = "test-utils"))]
374 database
375 .reach_test_point(
376 coven_database::DatabaseTestPoint::StoreDeviceExclusionCandidateStaged,
377 )
378 .await;
379 Ok(durable)
380 }
381
382 async fn publish_outcome(
383 &mut self,
384 proposal_ref: &StoreDeviceExclusionProposalRef,
385 intent: OutcomeIntent,
386 ) -> Result<StoreDeviceExclusionResult, StoreDeviceExclusionError> {
387 let database = self.database.clone();
388 let _lock = database.device_exclusion_permit().await;
389 self.reject_active_operation().await?;
390 let durable = self.prepare_outcome(proposal_ref, intent).await?;
391 self.drive(Box::new(durable)).await
392 }
393
394 async fn prepare_outcome(
395 &mut self,
396 proposal_ref: &StoreDeviceExclusionProposalRef,
397 intent: OutcomeIntent,
398 ) -> Result<DurableStoreDeviceExclusionOperation, StoreDeviceExclusionError> {
399 let database = self.database.clone();
400 let plan = self.writer.prepare_plan().await?;
401 let owner_grant = plan
402 .owner_grant()
403 .cloned()
404 .ok_or(StoreDeviceExclusionError::OwnerAuthorityRequired)?;
405 let proposal = self
406 .writer
407 .device_exclusion_history()
408 .load_proposal(proposal_ref)
409 .await?;
410 let state = database
411 .resolved_store_device_state(plan.device_state())
412 .await?;
413 require_pending_proposal(&state, proposal_ref)?;
414 let outcome = match intent {
415 OutcomeIntent::Cancel => {
416 StoreDeviceExclusionOutcome::Cancelled(plan.sign_device_exclusion_cancellation(
417 proposal_ref.clone(),
418 &proposal.object.value,
419 owner_grant,
420 )?)
421 }
422 OutcomeIntent::Exclude => {
423 let proof = self
424 .build_exclusion_proof(proposal_ref, &proposal.object.value)
425 .await?;
426 StoreDeviceExclusionOutcome::Excluded(plan.sign_device_exclusion(
427 proposal_ref.clone(),
428 &proposal.object.value,
429 proposal_ref.target.clone(),
430 &proposal.target,
431 proof,
432 owner_grant,
433 )?)
434 }
435 };
436 let prefix = device_exclusion_outcome_semantic_prefix(
437 proposal_ref.target.device_id,
438 proposal_ref.proposal_id,
439 );
440 let context = ProtocolObjectContext::signed_plaintext(
441 plan.root().store_root_hash,
442 ProtocolObjectDomain::StoreDeviceExclusionOutcome,
443 );
444 let prepared = self.storage.prepare_protocol_object(
445 &context,
446 proposal.object.value.outcome_slot.clone(),
447 &prefix,
448 outcome.to_bytes(),
449 )?;
450 let reference = StoreDeviceExclusionOutcomeRef::from_outcome(
451 &outcome,
452 &proposal.object.value,
453 prepared.reference().clone(),
454 )?;
455 let retained_proposal =
456 coven_protocol::store_commit::RetainedStoreDeviceExclusionProposal::from_verified(
457 &proposal,
458 );
459 let retained =
460 plan.retain_device_exclusion_outcome(&reference, retained_proposal, &outcome)?;
461 let candidate = Box::pin(
462 self.writer
463 .prepare_candidate(plan, StoreOperationBatch::DeviceExclusionOutcome(retained)),
464 )
465 .await?;
466 let operation = DurableStoreDeviceExclusionOperation::prepared(
467 DurableStoreDeviceExclusionObject::Outcome {
468 reference,
469 value: outcome,
470 prepared,
471 },
472 candidate,
473 )?;
474 let durable = Box::pin(database.begin_outbound_store_device_exclusion(operation)).await?;
475 #[cfg(any(test, feature = "test-utils"))]
476 database
477 .reach_test_point(
478 coven_database::DatabaseTestPoint::StoreDeviceExclusionCandidateStaged,
479 )
480 .await;
481 Ok(durable)
482 }
483
484 async fn drive(
485 &mut self,
486 mut operation: Box<DurableStoreDeviceExclusionOperation>,
487 ) -> Result<StoreDeviceExclusionResult, StoreDeviceExclusionError> {
488 loop {
489 if let Some(result) = self.resume_candidate(&mut operation).await? {
490 return Ok(result);
491 }
492 if let Some(result) = self.ensure_authority_uploaded(operation.as_ref()).await? {
493 return Ok(result);
494 }
495 match self.publish_candidate(&mut operation).await? {
496 DeviceExclusionPublicationProgress::Completed(result) => return Ok(result),
497 DeviceExclusionPublicationProgress::Continue => {}
498 DeviceExclusionPublicationProgress::ReplacementRequired(proof) => {
499 self.replace_candidate(&mut operation, proof).await?;
500 }
501 }
502 }
503 }
504
505 async fn publish_candidate(
506 &mut self,
507 operation: &mut Box<DurableStoreDeviceExclusionOperation>,
508 ) -> Result<DeviceExclusionPublicationProgress, StoreDeviceExclusionError> {
509 let database = self.database.clone();
510 let candidate = operation.candidate().cloned().ok_or_else(|| {
511 StoreDeviceExclusionError::InvalidState(
512 "active exclusion operation has no activation candidate".to_string(),
513 )
514 })?;
515 let publication = Box::new(
516 self.writer
517 .publish_prepared(Box::new(candidate), None, None)
518 .await?,
519 );
520 match publication.as_ref() {
521 StoreOperationPublicationOutcome::Activated(_) => {
522 **operation = Box::pin(
523 database.complete_outbound_store_device_exclusion_activation(
524 operation.as_ref().clone(),
525 ),
526 )
527 .await?;
528 completion_result(operation.as_ref())
529 .map(DeviceExclusionPublicationProgress::Completed)
530 }
531 StoreOperationPublicationOutcome::RepreparedCandidate(candidate) => {
532 **operation = Box::pin(database.replace_outbound_store_device_exclusion_candidate(
533 operation.as_ref().clone(),
534 candidate.as_ref().clone(),
535 ))
536 .await?;
537 Ok(DeviceExclusionPublicationProgress::Continue)
538 }
539 StoreOperationPublicationOutcome::NonactivatedCandidate {
540 candidate,
541 nonactivation,
542 } => {
543 if operation.candidate() != Some(candidate.as_ref()) {
544 return Err(StoreDeviceExclusionError::InvalidState(
545 "publication conflict names another exclusion candidate".to_string(),
546 ));
547 }
548 if matches!(
549 operation.object(),
550 DurableStoreDeviceExclusionObject::Outcome { .. }
551 ) {
552 return Ok(DeviceExclusionPublicationProgress::ReplacementRequired(
553 nonactivation.as_ref().clone(),
554 ));
555 } else {
556 **operation = Box::pin(
557 database.begin_outbound_store_device_exclusion_nonactivation(
558 operation.as_ref().clone(),
559 nonactivation.as_ref().clone(),
560 ),
561 )
562 .await?;
563 }
564 Ok(DeviceExclusionPublicationProgress::Continue)
565 }
566 StoreOperationPublicationOutcome::Reprepared
567 | StoreOperationPublicationOutcome::Nonactivated(_) => {
568 Err(StoreDeviceExclusionError::InvalidState(
569 "exclusion publication entered acknowledgement-only conflict state".to_string(),
570 ))
571 }
572 }
573 }
574
575 async fn replace_candidate(
576 &mut self,
577 operation: &mut Box<DurableStoreDeviceExclusionOperation>,
578 nonactivation: coven_protocol::remote_object::VerifiedCandidateNonactivation,
579 ) -> Result<(), StoreDeviceExclusionError> {
580 let database = self.database.clone();
581 let replacement = self
582 .prepare_replacement_candidate(operation.object())
583 .await?;
584 **operation = Box::pin(database.begin_outbound_store_device_exclusion_replacement(
585 operation.as_ref().clone(),
586 replacement,
587 nonactivation,
588 ))
589 .await?;
590 Ok(())
591 }
592
593 async fn ensure_authority_uploaded(
594 &mut self,
595 operation: &DurableStoreDeviceExclusionOperation,
596 ) -> Result<Option<StoreDeviceExclusionResult>, StoreDeviceExclusionError> {
597 let database = self.database.clone();
598 match Box::pin(self.create_exact_object(operation)).await {
599 Ok(()) => {}
600 Err(StoreDeviceExclusionJournalError::Storage(
601 coven_protocol::objects::StorageError::SlotCollision(_),
602 )) => {
603 if let Some(completed) = self.resolve_object_collision(operation.clone()).await? {
604 return completion_result(&completed).map(Some);
605 }
606 }
607 Err(error) => return Err(error.into()),
608 }
609 Box::pin(database.mark_store_device_exclusion_authority_uploaded(operation.clone()))
610 .await?;
611 Ok(None)
612 }
613
614 async fn resume_candidate(
615 &mut self,
616 operation: &mut Box<DurableStoreDeviceExclusionOperation>,
617 ) -> Result<Option<StoreDeviceExclusionResult>, StoreDeviceExclusionError> {
618 let database = self.database.clone();
619 match operation.as_ref() {
620 DurableStoreDeviceExclusionOperation::CandidateNonactivating { .. } => {
621 let targets = Box::pin(
622 database.nonactivating_store_device_exclusion_cleanup_targets(
623 operation.as_ref().clone(),
624 ),
625 )
626 .await?;
627 crate::sync::store::authorization::delete_candidate_cleanup_targets::<
628 StoreDeviceExclusionError,
629 >(self.storage.as_ref(), &database, targets)
630 .await?;
631 **operation = Box::pin(
632 database
633 .complete_nonactivating_store_device_exclusion(operation.as_ref().clone()),
634 )
635 .await?;
636 completion_result(operation.as_ref()).map(Some)
637 }
638 DurableStoreDeviceExclusionOperation::ReplacingCandidate { .. } => {
639 let targets = Box::pin(
640 database.nonactivating_store_device_exclusion_cleanup_targets(
641 operation.as_ref().clone(),
642 ),
643 )
644 .await?;
645 crate::sync::store::authorization::delete_candidate_cleanup_targets::<
646 StoreDeviceExclusionError,
647 >(self.storage.as_ref(), &database, targets)
648 .await?;
649 **operation = Box::pin(
650 database.complete_store_device_exclusion_replacement_cleanup(
651 operation.as_ref().clone(),
652 ),
653 )
654 .await?;
655 Ok(None)
656 }
657 DurableStoreDeviceExclusionOperation::CandidatePrepared { candidate, .. } => {
658 let reference = candidate.reference.clone();
659 let stream = reference.coord.stream_id.to_string();
660 if database
661 .exact_materialized_ref(&stream, reference.coord.sequence())
662 .await?
663 == Some(reference)
664 {
665 **operation = Box::pin(
666 database.complete_outbound_store_device_exclusion_activation(
667 operation.as_ref().clone(),
668 ),
669 )
670 .await?;
671 completion_result(operation.as_ref()).map(Some)
672 } else {
673 Ok(None)
674 }
675 }
676 DurableStoreDeviceExclusionOperation::Completed(_) => {
677 completion_result(operation.as_ref()).map(Some)
678 }
679 }
680 }
681
682 async fn prepare_replacement_candidate(
683 &mut self,
684 object: &DurableStoreDeviceExclusionObject,
685 ) -> Result<PreparedStoreOperationCommit, StoreDeviceExclusionError> {
686 let database = self.database.clone();
687 let DurableStoreDeviceExclusionObject::Outcome {
688 reference, value, ..
689 } = object
690 else {
691 return Err(StoreDeviceExclusionError::InvalidState(
692 "only an exclusion outcome can acquire a replacement candidate".to_string(),
693 ));
694 };
695 let plan = self.writer.prepare_plan().await?;
696 let state = database
697 .resolved_store_device_state(plan.device_state())
698 .await?;
699 require_pending_proposal(&state, reference.proposal())?;
700 let proposal = self
701 .writer
702 .device_exclusion_history()
703 .load_proposal(reference.proposal())
704 .await?;
705 let retained = plan.retain_device_exclusion_outcome(
706 reference,
707 coven_protocol::store_commit::RetainedStoreDeviceExclusionProposal::from_verified(
708 &proposal,
709 ),
710 value,
711 )?;
712 Box::pin(
713 self.writer
714 .prepare_candidate(plan, StoreOperationBatch::DeviceExclusionOutcome(retained)),
715 )
716 .await
717 .map_err(StoreDeviceExclusionError::from)
718 }
719
720 async fn resolve_object_collision(
721 &mut self,
722 operation: DurableStoreDeviceExclusionOperation,
723 ) -> Result<Option<DurableStoreDeviceExclusionOperation>, StoreDeviceExclusionError> {
724 let database = self.database.clone();
725 let intended = operation.object();
726 let (bytes, prepared) = self
727 .storage
728 .read_prepared_protocol_slot(
729 &intended.context(),
730 intended.object().slot(),
731 intended.semantic_prefix()?,
732 )
733 .await?;
734 if bytes == intended.semantic_bytes() {
735 if prepared.reference() != intended.object() {
736 return Err(StoreDeviceExclusionError::InvalidState(
737 "identical exclusion bytes produced a different exact object reference"
738 .to_string(),
739 ));
740 }
741 return Ok(None);
742 }
743 let DurableStoreDeviceExclusionObject::Outcome {
744 reference: intended_ref,
745 ..
746 } = intended
747 else {
748 return Err(StoreDeviceExclusionError::InvalidState(
749 "proposal hash slot contains different signed bytes".to_string(),
750 ));
751 };
752 let proposal = self
753 .writer
754 .device_exclusion_history()
755 .load_proposal(intended_ref.proposal())
756 .await?;
757 let unverified: StoreDeviceExclusionOutcome = serde_json::from_slice(&bytes)?;
758 let winner_ref = StoreDeviceExclusionOutcomeRef::from_outcome(
759 &unverified,
760 &proposal.object.value,
761 prepared.reference().clone(),
762 )?;
763 let winner = self
764 .writer
765 .device_exclusion_history()
766 .load_outcome(&winner_ref, &proposal)
767 .await?;
768 if winner.object.value != unverified || winner.object.bytes != bytes {
769 return Err(StoreDeviceExclusionError::InvalidState(
770 "occupied exclusion outcome changed during exact verification".to_string(),
771 ));
772 }
773 let completed = Box::pin(database.complete_outbound_store_device_exclusion_slot_loss(
774 operation,
775 DurableStoreDeviceExclusionObject::Outcome {
776 reference: winner_ref,
777 value: unverified,
778 prepared,
779 },
780 ))
781 .await?;
782 Ok(Some(completed))
783 }
784
785 async fn build_exclusion_proof(
786 &mut self,
787 proposal_ref: &StoreDeviceExclusionProposalRef,
788 proposal: &StoreDeviceExclusionProposal,
789 ) -> Result<StoreDeviceExclusionProof, StoreDeviceExclusionError> {
790 let database = self.database.clone();
791 let frozen = database
792 .resolved_store_device_state(&proposal.frozen_device_state)
793 .await?;
794 let mut acknowledgements = Vec::new();
795 let mut cutoff: Option<StoreHistoryCut> = None;
796 for record in frozen.devices.values() {
797 if record.registration == proposal.target
798 || !matches!(record.status, StoreDeviceStatus::Active)
799 {
800 continue;
801 }
802 let reference = database
803 .activated_store_ack(&record.registration)
804 .await?
805 .ok_or_else(|| {
806 StoreDeviceExclusionError::InvalidState(format!(
807 "registration {} has not acknowledged exclusion proposal {}",
808 record.registration.device_id, proposal_ref.proposal_id
809 ))
810 })?
811 .reference;
812 let registration = database
813 .activated_store_device_registration(record.registration.clone())
814 .await?;
815 let acknowledgement = self
816 .writer
817 .device_exclusion_history()
818 .load_acknowledgement(&reference, registration.value())
819 .await?;
820 let proposal_freezes = &acknowledgement.exclusions.proposal_freezes;
821 let freeze = proposal_freezes
822 .iter()
823 .find(|freeze| freeze.proposal == *proposal_ref)
824 .ok_or_else(|| {
825 StoreDeviceExclusionError::InvalidState(format!(
826 "registration {} acknowledgement omits exclusion proposal {}",
827 record.registration.device_id, proposal_ref.proposal_id
828 ))
829 })?;
830 cutoff = Some(match cutoff {
831 Some(current) => current.join(freeze.target_cut.clone())?,
832 None => freeze.target_cut.clone(),
833 });
834 acknowledgements.push(reference);
835 }
836 acknowledgements.sort();
837 let cutoff = cutoff.ok_or_else(|| {
838 StoreDeviceExclusionError::InvalidState(
839 "Merge exclusion has no remaining active-device acknowledgement".to_string(),
840 )
841 })?;
842 Ok(StoreDeviceExclusionProof {
843 frozen_device_state: proposal.frozen_device_state.clone(),
844 remaining_device_acks: acknowledgements,
845 cutoff,
846 })
847 }
848}
849
850#[derive(Clone, Copy)]
851enum OutcomeIntent {
852 Exclude,
853 Cancel,
854}
855
856enum DeviceExclusionPublicationProgress {
857 Completed(StoreDeviceExclusionResult),
858 Continue,
859 ReplacementRequired(coven_protocol::remote_object::VerifiedCandidateNonactivation),
860}
861
862fn completion_result(
863 operation: &DurableStoreDeviceExclusionOperation,
864) -> Result<StoreDeviceExclusionResult, StoreDeviceExclusionError> {
865 let DurableStoreDeviceExclusionOperation::Completed(completion) = operation else {
866 return Err(StoreDeviceExclusionError::InvalidState(
867 "Store-device exclusion operation is not complete".to_string(),
868 ));
869 };
870 Ok(match completion {
871 StoreDeviceExclusionCompletion::Activated { object, candidate } => match object {
872 DurableStoreDeviceExclusionObject::Proposal { reference, .. } => {
873 StoreDeviceExclusionResult::ProposalActivated {
874 proposal: reference.clone(),
875 commit: candidate.reference.clone(),
876 }
877 }
878 DurableStoreDeviceExclusionObject::Outcome { reference, .. } => {
879 StoreDeviceExclusionResult::OutcomeActivated {
880 outcome: reference.clone(),
881 commit: candidate.reference.clone(),
882 }
883 }
884 },
885 StoreDeviceExclusionCompletion::OutcomeSlotOccupied { intended, winner } => {
886 let (
887 DurableStoreDeviceExclusionObject::Outcome {
888 reference: intended,
889 ..
890 },
891 DurableStoreDeviceExclusionObject::Outcome {
892 reference: winner, ..
893 },
894 ) = (intended, winner)
895 else {
896 return Err(StoreDeviceExclusionError::InvalidState(
897 "outcome-slot completion contains a non-outcome object".to_string(),
898 ));
899 };
900 StoreDeviceExclusionResult::OutcomeSlotOccupied {
901 intended: intended.clone(),
902 winner: winner.clone(),
903 }
904 }
905 StoreDeviceExclusionCompletion::CandidateNonactivated {
906 object, candidate, ..
907 } => StoreDeviceExclusionResult::CandidateNonactivated {
908 object_hash: object.operation_id(),
909 candidate: candidate.reference.clone(),
910 },
911 })
912}
913
914fn require_active_target(
915 state: &coven_protocol::store_commit::ResolvedStoreDeviceState,
916 target: &coven_protocol::store_commit::StoreDeviceRegistrationRef,
917) -> Result<(), StoreDeviceExclusionError> {
918 if !matches!(
919 state.devices.get(&target.device_id),
920 Some(record)
921 if record.registration == *target && matches!(record.status, StoreDeviceStatus::Active)
922 ) {
923 return Err(StoreDeviceExclusionError::TargetNotActive);
924 }
925 Ok(())
926}
927
928fn require_pending_proposal(
929 state: &coven_protocol::store_commit::ResolvedStoreDeviceState,
930 proposal: &StoreDeviceExclusionProposalRef,
931) -> Result<(), StoreDeviceExclusionError> {
932 require_active_target(state, &proposal.target)?;
933 if !matches!(
934 state.devices
935 .get(&proposal.target.device_id)
936 .and_then(|record| record.proposals.get(&proposal.proposal_id)),
937 Some(StoreDeviceProposalState::Pending { proposal: current }) if current == proposal
938 ) {
939 return Err(StoreDeviceExclusionError::InvalidState(
940 "exclusion proposal is not pending at the exact candidate predecessor".to_string(),
941 ));
942 }
943 Ok(())
944}
945
946#[cfg(test)]
947mod tests;