1use alloc::boxed::Box;
2
3use crate::algebra::{ResourceVector, WideResourceVector};
4use crate::wire::{BindingEpoch, ConversationId, DeliverySeq, ParticipantId, ParticipantIndex};
5
6use super::{
7 ActiveBinding, CommittedDiedTerminal, ObserverProgressProjection,
8 claim_frontier::{ValidatedMarkerCandidate, ValidatedMarkerRecord},
9};
10
11#[derive(Clone, Copy, Debug, PartialEq, Eq)]
13pub struct ClosureDebt(WideResourceVector);
14
15impl ClosureDebt {
16 #[must_use]
18 pub const fn new(value: WideResourceVector) -> Option<Self> {
19 if value.is_zero() {
20 None
21 } else {
22 Some(Self(value))
23 }
24 }
25
26 #[must_use]
28 pub const fn value(self) -> WideResourceVector {
29 self.0
30 }
31}
32
33#[derive(Clone, Copy, Debug, PartialEq, Eq)]
35pub struct ObserverProjection {
36 through_seq: DeliverySeq,
37}
38
39impl ObserverProjection {
40 #[must_use]
42 pub const fn new(through_seq: DeliverySeq) -> Self {
43 Self { through_seq }
44 }
45
46 #[must_use]
48 pub const fn through_seq(self) -> DeliverySeq {
49 self.through_seq
50 }
51}
52
53#[derive(Clone, Copy, Debug, PartialEq, Eq)]
55pub struct PhysicalCompaction {
56 from_floor: DeliverySeq,
57 through_seq: DeliverySeq,
58}
59
60impl PhysicalCompaction {
61 #[must_use]
63 pub const fn new(from_floor: DeliverySeq, through_seq: DeliverySeq) -> Option<Self> {
64 if from_floor <= through_seq {
65 Some(Self {
66 from_floor,
67 through_seq,
68 })
69 } else {
70 None
71 }
72 }
73
74 #[must_use]
76 pub const fn from_floor(self) -> DeliverySeq {
77 self.from_floor
78 }
79
80 #[must_use]
82 pub const fn through_seq(self) -> DeliverySeq {
83 self.through_seq
84 }
85}
86
87#[allow(clippy::struct_field_names)]
108#[derive(Clone, Copy, Debug, PartialEq, Eq)]
109pub struct MarkerDelivery {
110 conversation_id: ConversationId,
111 participant_id: ParticipantId,
112 binding_epoch: BindingEpoch,
113 marker_delivery_seq: DeliverySeq,
114}
115
116impl MarkerDelivery {
117 #[must_use]
125 pub(super) const fn successor_from_validated_candidate(
126 candidate: ValidatedMarkerCandidate,
127 ) -> StoredEdge {
128 let conversation_id = candidate.conversation_id();
129 let participant_id = candidate.participant_id();
130 let marker_delivery_seq = candidate.delivery_seq();
131 let successor = match candidate.target_binding() {
132 super::FrontierBinding::Bound(binding_epoch) => StoredEdge::MarkerDelivery(Self {
133 conversation_id,
134 participant_id,
135 binding_epoch,
136 marker_delivery_seq,
137 }),
138 super::FrontierBinding::Detached(last_dead_binding_epoch) => {
139 StoredEdge::DetachedMarkerRelease(DetachedMarkerRelease {
140 participant_id,
141 marker_delivery_seq,
142 last_dead_binding_epoch,
143 })
144 }
145 };
146 candidate.consume();
147 successor
148 }
149
150 #[must_use]
152 pub(super) const fn from_validated_record(record: &ValidatedMarkerRecord) -> Self {
153 Self {
154 conversation_id: record.conversation_id(),
155 participant_id: record.participant_id(),
156 binding_epoch: record.binding_epoch(),
157 marker_delivery_seq: record.delivery_seq(),
158 }
159 }
160
161 #[must_use]
163 pub const fn conversation_id(self) -> ConversationId {
164 self.conversation_id
165 }
166
167 #[must_use]
169 pub const fn participant_id(self) -> ParticipantId {
170 self.participant_id
171 }
172
173 #[must_use]
175 pub const fn binding_epoch(self) -> BindingEpoch {
176 self.binding_epoch
177 }
178
179 #[must_use]
181 pub const fn marker_delivery_seq(self) -> DeliverySeq {
182 self.marker_delivery_seq
183 }
184}
185
186#[cfg(any(test, feature = "test-support"))]
187#[allow(clippy::expect_used, clippy::too_many_lines)]
188pub fn validated_marker_record_for_test(
189 conversation_id: crate::wire::ConversationId,
190 participant_id: ParticipantId,
191 target_binding: super::claim_frontier::FrontierBinding,
192 marker_delivery_seq: DeliverySeq,
193 cursor: DeliverySeq,
194) -> ValidatedMarkerRecord {
195 use alloc::{vec, vec::Vec};
196
197 use crate::outcome::CandidatePhase;
198
199 use super::{
200 AdmissionOrder, OrderClaims, OrderHigh, OrderLedger, RecoverySequenceReserve,
201 SequenceClaims, SequenceLedger,
202 claim_frontier::{
203 BindingTerminalOwner, ClaimFrontiers, ClaimFrontiersRestore, FrontierBinding,
204 FrontierParticipant, ImmutableSequenceCandidate, MarkerProvenance, MarkerRecordRequest,
205 MovableOrderClaim, MovableSequenceClaim, OrderClaimFrontierRestore, OrderDirectOwner,
206 RetainedCausalRecord, RetainedCausalRecordKind, SequenceClaimFrontierRestore,
207 SequenceDirectOwner, SequenceProductRangesRestore, TerminalProductRangeRestore,
208 },
209 };
210
211 let identity_slot_limit = participant_id
212 .checked_add(1)
213 .expect("test participant must fit a half-open identity domain");
214 let exit_seq = marker_delivery_seq
215 .checked_add(1)
216 .expect("test marker must leave an exit-claim suffix");
217 let binding_epoch = match target_binding {
218 FrontierBinding::Bound(epoch) | FrontierBinding::Detached(epoch) => epoch,
219 };
220 let terminal_owner = BindingTerminalOwner {
221 participant_index: participant_id,
222 binding_epoch,
223 };
224 let (sequence_claims, sequence_movable, products, order_claims, order_movable) =
225 match target_binding {
226 FrontierBinding::Bound(_) => {
227 let terminal_seq = exit_seq
228 .checked_add(1)
229 .expect("test marker must leave a terminal-claim suffix");
230 let product_seq = terminal_seq
231 .checked_add(1)
232 .expect("test marker must leave a terminal-product suffix");
233 (
234 SequenceClaims::new(1, 1, 0, RecoverySequenceReserve::None),
235 vec![
236 MovableSequenceClaim {
237 delivery_seq: exit_seq,
238 owner: SequenceDirectOwner::MembershipExit {
239 participant_index: participant_id,
240 },
241 },
242 MovableSequenceClaim {
243 delivery_seq: terminal_seq,
244 owner: SequenceDirectOwner::BindingTerminal(terminal_owner),
245 },
246 ],
247 SequenceProductRangesRestore {
248 live_times_terminal: vec![TerminalProductRangeRestore {
249 start: product_seq,
250 length: 1,
251 terminal: terminal_owner,
252 }],
253 live_times_replacement_terminal: None,
254 other_live_times_exit: vec![],
255 },
256 OrderClaims::new(1, 1, false, false)
257 .expect("bound test claims have no torn recovery pair"),
258 vec![
259 MovableOrderClaim {
260 transaction_order: 1,
261 owner: OrderDirectOwner::ActiveBindingTerminal(terminal_owner),
262 },
263 MovableOrderClaim {
264 transaction_order: 2,
265 owner: OrderDirectOwner::MembershipExit {
266 participant_index: participant_id,
267 },
268 },
269 ],
270 )
271 }
272 FrontierBinding::Detached(_) => (
273 SequenceClaims::new(1, 0, 0, RecoverySequenceReserve::None),
274 vec![MovableSequenceClaim {
275 delivery_seq: exit_seq,
276 owner: SequenceDirectOwner::MembershipExit {
277 participant_index: participant_id,
278 },
279 }],
280 SequenceProductRangesRestore::default(),
281 OrderClaims::new(0, 1, false, false)
282 .expect("detached test claims have no torn recovery pair"),
283 vec![MovableOrderClaim {
284 transaction_order: 1,
285 owner: OrderDirectOwner::MembershipExit {
286 participant_index: participant_id,
287 },
288 }],
289 ),
290 };
291 let sequence_ledger = SequenceLedger::try_new(marker_delivery_seq, sequence_claims)
292 .expect("test sequence frontier is within the numeric suffix");
293 let order_ledger = OrderLedger::try_new(OrderHigh::Allocated(0), order_claims)
294 .expect("test order frontier is within the numeric suffix");
295 let admission_order = AdmissionOrder::new(0, CandidatePhase::CompactionMarker, participant_id);
296 let mut prevalidated = ClaimFrontiers::prevalidate(
297 ClaimFrontiersRestore {
298 conversation_id,
299 active_identities: vec![FrontierParticipant::new(
300 participant_id,
301 cursor,
302 target_binding,
303 )],
304 identity_slot_limit,
305 retained_floor: u128::from(marker_delivery_seq),
306 retained_record_limit: 1,
307 retained_records: vec![RetainedCausalRecord {
308 delivery_seq: marker_delivery_seq,
309 admission_order,
310 kind: RetainedCausalRecordKind::CompactionMarker {
311 participant_index: participant_id,
312 provenance: MarkerProvenance::NonProductM,
313 },
314 }],
315 active_marker_anchors: vec![marker_delivery_seq],
316 historical_marker_deliveries: vec![],
317 historical_causal_facts: vec![],
318 sequence: SequenceClaimFrontierRestore {
319 movable_claims: sequence_movable,
320 immutable_candidates: Vec::<ImmutableSequenceCandidate>::new(),
321 products,
322 recovery: None,
323 },
324 order: OrderClaimFrontierRestore {
325 movable_claims: order_movable,
326 immutable_candidates: vec![],
327 recovery: None,
328 },
329 recovery_marker_delivery_seq: None,
330 },
331 sequence_ledger,
332 order_ledger,
333 )
334 .expect("complete test claim frontier must prevalidate");
335 let record = prevalidated
336 .take_marker_record(MarkerRecordRequest::planned(
337 participant_id,
338 marker_delivery_seq,
339 target_binding,
340 ))
341 .expect("prevalidated test frontier retains its exact marker");
342 if cursor >= marker_delivery_seq {
343 record.delivered_for_test()
344 } else {
345 record
346 }
347}
348
349#[cfg(any(test, feature = "test-support"))]
350pub fn marker_delivery_for_test(
351 participant_id: ParticipantId,
352 binding_epoch: BindingEpoch,
353 marker_delivery_seq: DeliverySeq,
354) -> Result<MarkerDelivery, super::storage::StorageRestoreError> {
355 let record = validated_marker_record_for_test(
356 1,
357 participant_id,
358 super::claim_frontier::FrontierBinding::Bound(binding_epoch),
359 marker_delivery_seq,
360 marker_delivery_seq.saturating_sub(1),
361 );
362 super::storage::MarkerDeliveryRestore {
363 participant_id,
364 binding_epoch,
365 marker_delivery_seq,
366 }
367 .restore_bound(1, record)
368}
369
370#[derive(Clone, Copy, Debug, PartialEq, Eq)]
388pub struct CursorProgressContinuous {
389 participant_id: ParticipantId,
390 binding_epoch: BindingEpoch,
391 through_seq: DeliverySeq,
392}
393
394impl CursorProgressContinuous {
395 #[cfg(test)]
397 #[must_use]
398 pub(crate) const fn new(
399 participant_id: ParticipantId,
400 binding_epoch: BindingEpoch,
401 through_seq: DeliverySeq,
402 ) -> Self {
403 Self {
404 participant_id,
405 binding_epoch,
406 through_seq,
407 }
408 }
409
410 #[must_use]
412 pub const fn participant_id(self) -> ParticipantId {
413 self.participant_id
414 }
415
416 #[must_use]
418 pub const fn binding_epoch(self) -> BindingEpoch {
419 self.binding_epoch
420 }
421
422 #[must_use]
424 pub const fn through_seq(self) -> DeliverySeq {
425 self.through_seq
426 }
427}
428
429#[derive(Clone, Copy, Debug, PartialEq, Eq)]
436pub struct CursorProgressMarker {
437 conversation_id: ConversationId,
438 participant_id: ParticipantId,
439 binding_epoch: BindingEpoch,
440 through_seq: DeliverySeq,
441 marker_delivery_seq: DeliverySeq,
442}
443
444impl CursorProgressMarker {
445 #[must_use]
447 pub const fn conversation_id(self) -> ConversationId {
448 self.conversation_id
449 }
450
451 #[must_use]
453 pub const fn participant_id(self) -> ParticipantId {
454 self.participant_id
455 }
456
457 #[must_use]
459 pub const fn binding_epoch(self) -> BindingEpoch {
460 self.binding_epoch
461 }
462
463 #[must_use]
465 pub const fn through_seq(self) -> DeliverySeq {
466 self.through_seq
467 }
468
469 #[must_use]
471 pub const fn marker_delivery_seq(self) -> DeliverySeq {
472 self.marker_delivery_seq
473 }
474}
475
476#[derive(Clone, Copy, Debug, PartialEq, Eq)]
492pub enum ParticipantCursorProgress {
493 Continuous(CursorProgressContinuous),
495 Marker(CursorProgressMarker),
497}
498
499impl ParticipantCursorProgress {
500 #[cfg(test)]
502 #[must_use]
503 pub(crate) const fn continuous(
504 participant_id: ParticipantId,
505 binding_epoch: BindingEpoch,
506 through_seq: DeliverySeq,
507 ) -> Self {
508 Self::Continuous(CursorProgressContinuous::new(
509 participant_id,
510 binding_epoch,
511 through_seq,
512 ))
513 }
514
515 pub(super) fn restore_continuous(
516 authority: OrdinaryBindingAuthority,
517 participant_id: ParticipantId,
518 binding_epoch: BindingEpoch,
519 through_seq: DeliverySeq,
520 ) -> Option<Self> {
521 if authority.binding.participant_id != participant_id
522 || authority.binding.binding_epoch != binding_epoch
523 || authority.through_seq != through_seq
524 {
525 return None;
526 }
527 Some(Self::Continuous(CursorProgressContinuous {
528 participant_id,
529 binding_epoch,
530 through_seq,
531 }))
532 }
533
534 #[must_use]
536 pub const fn participant_id(self) -> ParticipantId {
537 match self {
538 Self::Continuous(value) => value.participant_id,
539 Self::Marker(value) => value.participant_id,
540 }
541 }
542
543 #[must_use]
545 pub const fn binding_epoch(self) -> BindingEpoch {
546 match self {
547 Self::Continuous(value) => value.binding_epoch,
548 Self::Marker(value) => value.binding_epoch,
549 }
550 }
551
552 #[must_use]
554 pub const fn through_seq(self) -> DeliverySeq {
555 match self {
556 Self::Continuous(value) => value.through_seq,
557 Self::Marker(value) => value.through_seq,
558 }
559 }
560
561 #[must_use]
563 pub const fn marker_delivery_seq(self) -> Option<DeliverySeq> {
564 match self {
565 Self::Continuous(_) => None,
566 Self::Marker(value) => Some(value.marker_delivery_seq),
567 }
568 }
569}
570
571#[derive(Clone, Copy, Debug, PartialEq, Eq)]
576pub struct DetachedCredentialRecovery {
577 conversation_id: ConversationId,
578 participant_id: ParticipantId,
579 marker_delivery_seq: DeliverySeq,
580 prior_binding_epoch: BindingEpoch,
581}
582
583impl DetachedCredentialRecovery {
584 pub(super) const fn from_storage_description(
591 conversation_id: ConversationId,
592 participant_id: ParticipantId,
593 marker_delivery_seq: DeliverySeq,
594 prior_binding_epoch: BindingEpoch,
595 ) -> Self {
596 Self {
597 conversation_id,
598 participant_id,
599 marker_delivery_seq,
600 prior_binding_epoch,
601 }
602 }
603
604 #[must_use]
606 pub const fn conversation_id(self) -> ConversationId {
607 self.conversation_id
608 }
609
610 #[must_use]
612 pub const fn participant_id(self) -> ParticipantId {
613 self.participant_id
614 }
615
616 #[must_use]
618 pub const fn marker_delivery_seq(self) -> DeliverySeq {
619 self.marker_delivery_seq
620 }
621
622 #[must_use]
624 pub const fn prior_binding_epoch(self) -> BindingEpoch {
625 self.prior_binding_epoch
626 }
627}
628
629#[cfg(any(test, feature = "test-support"))]
630pub fn validated_marker_record_for_recovery_test(
631 recovery: DetachedCredentialRecovery,
632) -> ValidatedMarkerRecord {
633 validated_marker_record_for_test(
634 recovery.conversation_id(),
635 recovery.participant_id(),
636 super::claim_frontier::FrontierBinding::Detached(recovery.prior_binding_epoch()),
637 recovery.marker_delivery_seq(),
638 recovery.marker_delivery_seq(),
639 )
640}
641
642#[derive(Clone, Copy, Debug, PartialEq, Eq)]
647pub struct DetachedMarkerRelease {
648 participant_id: ParticipantId,
649 marker_delivery_seq: DeliverySeq,
650 last_dead_binding_epoch: BindingEpoch,
651}
652
653impl DetachedMarkerRelease {
654 #[must_use]
656 pub const fn participant_id(self) -> ParticipantId {
657 self.participant_id
658 }
659
660 #[must_use]
662 pub const fn marker_delivery_seq(self) -> DeliverySeq {
663 self.marker_delivery_seq
664 }
665
666 #[must_use]
668 pub const fn last_dead_binding_epoch(self) -> BindingEpoch {
669 self.last_dead_binding_epoch
670 }
671}
672
673#[derive(Clone, Copy, Debug, PartialEq, Eq)]
678pub struct DetachedCursorRelease {
679 participant_id: ParticipantId,
680 last_dead_binding_epoch: BindingEpoch,
681}
682
683impl DetachedCursorRelease {
684 #[must_use]
686 pub const fn participant_id(self) -> ParticipantId {
687 self.participant_id
688 }
689
690 #[must_use]
692 pub const fn last_dead_binding_epoch(self) -> BindingEpoch {
693 self.last_dead_binding_epoch
694 }
695}
696
697#[derive(Clone, Copy, Debug, PartialEq, Eq)]
699pub enum StoredEdge {
700 ObserverProjection(ObserverProjection),
702 PhysicalCompaction(PhysicalCompaction),
704 MarkerDelivery(MarkerDelivery),
706 ParticipantCursorProgress(ParticipantCursorProgress),
708 DetachedCredentialRecovery(DetachedCredentialRecovery),
710 DetachedMarkerRelease(DetachedMarkerRelease),
712 DetachedCursorRelease(DetachedCursorRelease),
714}
715
716#[derive(Clone, Copy, Debug, PartialEq, Eq)]
718pub enum ClosureState {
719 Clear,
721 Owed {
723 debt: ClosureDebt,
725 edge: StoredEdge,
727 },
728}
729
730#[derive(Clone, Copy, Debug, PartialEq, Eq)]
736pub struct OrdinaryDetachedAttachAdmission {
737 _private: (),
738}
739
740impl ClosureState {
741 pub const fn ordinary_detached_attach_admission(
748 self,
749 ) -> Result<OrdinaryDetachedAttachAdmission, Self> {
750 match self {
751 Self::Clear => Ok(OrdinaryDetachedAttachAdmission { _private: () }),
752 Self::Owed { .. } => Err(self),
753 }
754 }
755}
756
757#[derive(Clone, Copy, Debug, PartialEq, Eq)]
763pub struct DebtCompletion(ClosureState);
764
765impl DebtCompletion {
766 #[must_use]
768 pub const fn clear() -> Self {
769 Self(ClosureState::Clear)
770 }
771
772 #[must_use]
774 pub const fn observer_projection(debt: ClosureDebt, edge: ObserverProjection) -> Self {
775 Self(ClosureState::Owed {
776 debt,
777 edge: StoredEdge::ObserverProjection(edge),
778 })
779 }
780
781 #[must_use]
783 pub const fn physical_compaction(debt: ClosureDebt, edge: PhysicalCompaction) -> Self {
784 Self(ClosureState::Owed {
785 debt,
786 edge: StoredEdge::PhysicalCompaction(edge),
787 })
788 }
789
790 #[must_use]
792 pub const fn into_state(self) -> ClosureState {
793 self.0
794 }
795}
796
797#[derive(Clone, Copy, Debug, PartialEq, Eq)]
840pub struct OrdinaryBindingAuthority {
841 binding: ActiveBinding,
842 through_seq: DeliverySeq,
843}
844
845impl OrdinaryBindingAuthority {
846 pub(crate) const fn new(binding: ActiveBinding, through_seq: DeliverySeq) -> Self {
847 Self {
848 binding,
849 through_seq,
850 }
851 }
852
853 #[must_use]
855 pub const fn binding(self) -> ActiveBinding {
856 self.binding
857 }
858
859 #[must_use]
861 pub const fn through_seq(self) -> DeliverySeq {
862 self.through_seq
863 }
864
865 pub(crate) fn cursor_progressed(self, event: Event) -> Result<Self, Self> {
875 let EventKind::CursorProgressed {
876 participant_id,
877 binding_epoch,
878 progress:
879 CursorProgressEvent::Normal {
880 previous_cursor,
881 through_seq,
882 },
883 ..
884 } = event.0
885 else {
886 return Err(self);
887 };
888 self.participant_ack_progressed(
889 self.binding.conversation_id,
890 participant_id,
891 binding_epoch,
892 previous_cursor,
893 through_seq,
894 )
895 }
896
897 pub(crate) fn participant_ack_progressed(
899 self,
900 conversation_id: ConversationId,
901 participant_id: ParticipantId,
902 binding_epoch: BindingEpoch,
903 previous_cursor: DeliverySeq,
904 through_seq: DeliverySeq,
905 ) -> Result<Self, Self> {
906 if conversation_id != self.binding.conversation_id
907 || participant_id != self.binding.participant_id
908 || binding_epoch != self.binding.binding_epoch
909 || previous_cursor != self.through_seq
910 || through_seq <= previous_cursor
911 {
912 return Err(self);
913 }
914 Ok(Self {
915 through_seq,
916 ..self
917 })
918 }
919
920 pub(crate) fn binding_fate(
927 self,
928 terminal: CommittedDiedTerminal,
929 resulting_floor: DeliverySeq,
930 ) -> Result<OrdinaryBindingFate, Self> {
931 if terminal.participant_id() != self.binding.participant_id
932 || terminal.conversation_id() != self.binding.conversation_id
933 || terminal.binding_epoch() != self.binding.binding_epoch
934 {
935 return Err(self);
936 }
937 Ok(OrdinaryBindingFate {
938 conversation_id: self.binding.conversation_id,
939 through_seq: self.through_seq,
940 resulting_floor,
941 release: DetachedCursorRelease {
942 participant_id: self.binding.participant_id,
943 last_dead_binding_epoch: self.binding.binding_epoch,
944 },
945 })
946 }
947}
948
949#[derive(Clone, Copy, Debug, PartialEq, Eq)]
956pub struct OrdinaryBindingFate {
957 conversation_id: ConversationId,
958 through_seq: DeliverySeq,
959 resulting_floor: DeliverySeq,
960 release: DetachedCursorRelease,
961}
962
963impl OrdinaryBindingFate {
964 #[must_use]
966 pub const fn conversation_id(self) -> ConversationId {
967 self.conversation_id
968 }
969
970 #[must_use]
972 pub const fn through_seq(self) -> DeliverySeq {
973 self.through_seq
974 }
975
976 #[must_use]
978 pub const fn participant_id(self) -> ParticipantId {
979 self.release.participant_id
980 }
981
982 #[must_use]
984 pub const fn last_dead_binding_epoch(self) -> BindingEpoch {
985 self.release.last_dead_binding_epoch
986 }
987
988 #[must_use]
990 pub const fn resulting_floor(self) -> DeliverySeq {
991 self.resulting_floor
992 }
993 #[must_use]
995 pub const fn observer_progress_projection(&self) -> ObserverProgressProjection {
996 ObserverProgressProjection::new(self.conversation_id, self.resulting_floor)
997 }
998
999 #[must_use]
1001 pub const fn into_direct_state(self, debt: ClosureDebt) -> ClosureState {
1002 owed(debt, StoredEdge::DetachedCursorRelease(self.release))
1003 }
1004}
1005
1006#[derive(Debug, PartialEq, Eq)]
1012pub struct FencedAttachCommit {
1013 conversation_id: ConversationId,
1014 participant_id: ParticipantId,
1015 marker_delivery_seq: DeliverySeq,
1016 prior_binding_epoch: BindingEpoch,
1017 new_binding_epoch: BindingEpoch,
1018 next_state: ClosureState,
1019}
1020
1021impl FencedAttachCommit {
1022 #[must_use]
1024 pub const fn conversation_id(&self) -> ConversationId {
1025 self.conversation_id
1026 }
1027
1028 #[must_use]
1030 pub const fn participant_id(&self) -> ParticipantId {
1031 self.participant_id
1032 }
1033
1034 #[must_use]
1036 pub const fn marker_delivery_seq(&self) -> DeliverySeq {
1037 self.marker_delivery_seq
1038 }
1039
1040 #[must_use]
1042 pub const fn prior_binding_epoch(&self) -> BindingEpoch {
1043 self.prior_binding_epoch
1044 }
1045
1046 #[must_use]
1048 pub const fn new_binding_epoch(&self) -> BindingEpoch {
1049 self.new_binding_epoch
1050 }
1051
1052 #[must_use]
1054 pub const fn next_state(&self) -> ClosureState {
1055 self.next_state
1056 }
1057
1058 pub(super) fn restore_validated(
1063 recovery: DetachedCredentialRecovery,
1064 record_authority: &ValidatedMarkerRecord,
1065 debt: ClosureDebt,
1066 event: Event,
1067 successor: DebtCompletion,
1068 ) -> Option<Self> {
1069 if debt.value().is_zero()
1070 || record_authority.conversation_id() != recovery.conversation_id
1071 || record_authority.participant_id() != recovery.participant_id
1072 || record_authority.delivery_seq() != recovery.marker_delivery_seq
1073 || record_authority.target_binding()
1074 != super::FrontierBinding::Detached(recovery.prior_binding_epoch)
1075 || record_authority.occurrence()
1076 != super::claim_frontier::MarkerRecordOccurrence::Delivered
1077 {
1078 return None;
1079 }
1080 let EventKind::FencedRecoveryCommitted {
1081 participant_id,
1082 marker_delivery_seq,
1083 prior_binding_epoch,
1084 new_binding_epoch,
1085 ..
1086 } = event.0
1087 else {
1088 return None;
1089 };
1090 if participant_id != recovery.participant_id
1091 || marker_delivery_seq != recovery.marker_delivery_seq
1092 || prior_binding_epoch != recovery.prior_binding_epoch
1093 || !is_next_generation(prior_binding_epoch, new_binding_epoch)
1094 {
1095 return None;
1096 }
1097 Some(Self {
1098 conversation_id: recovery.conversation_id,
1099 participant_id,
1100 marker_delivery_seq,
1101 prior_binding_epoch,
1102 new_binding_epoch,
1103 next_state: successor.into_state(),
1104 })
1105 }
1106
1107 pub(super) fn recovered_binding_fate(
1120 self,
1121 event: Event,
1122 ) -> Result<RecoveredBindingFate, Box<Self>> {
1123 let EventKind::BindingFateObserved {
1124 participant_id,
1125 binding_epoch,
1126 resulting_floor,
1127 } = event.0
1128 else {
1129 return Err(Box::new(self));
1130 };
1131 if participant_id != self.participant_id || binding_epoch != self.new_binding_epoch {
1132 return Err(Box::new(self));
1133 }
1134 let ClosureState::Owed { debt, edge } = self.next_state else {
1135 return Err(Box::new(self));
1136 };
1137 let predecessor = match edge {
1138 StoredEdge::ObserverProjection(value) => {
1139 RecoveredStorageEdge::ObserverProjection(value)
1140 }
1141 StoredEdge::PhysicalCompaction(value) => {
1142 RecoveredStorageEdge::PhysicalCompaction(value)
1143 }
1144 _ => return Err(Box::new(self)),
1145 };
1146 Ok(RecoveredBindingFate {
1147 conversation_id: self.conversation_id,
1148 predecessor_debt: debt,
1149 predecessor,
1150 resulting_floor,
1151 release: DetachedCursorRelease {
1152 participant_id,
1153 last_dead_binding_epoch: binding_epoch,
1154 },
1155 })
1156 }
1157}
1158
1159#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1160enum RecoveredStorageEdge {
1161 ObserverProjection(ObserverProjection),
1162 PhysicalCompaction(PhysicalCompaction),
1163}
1164
1165impl RecoveredStorageEdge {
1166 const fn into_stored_edge(self) -> StoredEdge {
1167 match self {
1168 Self::ObserverProjection(value) => StoredEdge::ObserverProjection(value),
1169 Self::PhysicalCompaction(value) => StoredEdge::PhysicalCompaction(value),
1170 }
1171 }
1172}
1173
1174#[derive(Debug, PartialEq, Eq)]
1180pub struct RecoveredBindingFate {
1181 conversation_id: ConversationId,
1182 predecessor_debt: ClosureDebt,
1183 predecessor: RecoveredStorageEdge,
1184 resulting_floor: DeliverySeq,
1185 release: DetachedCursorRelease,
1186}
1187
1188impl RecoveredBindingFate {
1189 #[must_use]
1191 pub const fn conversation_id(&self) -> ConversationId {
1192 self.conversation_id
1193 }
1194
1195 #[must_use]
1197 pub const fn predecessor_state(&self) -> ClosureState {
1198 owed(self.predecessor_debt, self.predecessor.into_stored_edge())
1199 }
1200
1201 #[must_use]
1203 pub const fn participant_id(&self) -> ParticipantId {
1204 self.release.participant_id
1205 }
1206
1207 #[must_use]
1209 pub const fn last_dead_binding_epoch(&self) -> BindingEpoch {
1210 self.release.last_dead_binding_epoch
1211 }
1212
1213 #[must_use]
1215 pub const fn resulting_floor(&self) -> DeliverySeq {
1216 self.resulting_floor
1217 }
1218 #[must_use]
1220 pub const fn observer_progress_projection(&self) -> ObserverProgressProjection {
1221 ObserverProgressProjection::new(self.conversation_id, self.resulting_floor)
1222 }
1223}
1224
1225#[derive(Debug, PartialEq, Eq)]
1232pub struct PendingRecoveredCursorRelease {
1233 debt: ClosureDebt,
1234 predecessor: RecoveredStorageEdge,
1235 release: DetachedCursorRelease,
1236}
1237
1238impl PendingRecoveredCursorRelease {
1239 #[must_use]
1241 pub const fn current_state(&self) -> ClosureState {
1242 owed(self.debt, self.predecessor.into_stored_edge())
1243 }
1244
1245 #[must_use]
1247 pub const fn participant_id(&self) -> ParticipantId {
1248 self.release.participant_id
1249 }
1250
1251 #[must_use]
1253 pub const fn last_dead_binding_epoch(&self) -> BindingEpoch {
1254 self.release.last_dead_binding_epoch
1255 }
1256}
1257
1258#[derive(Debug, PartialEq, Eq)]
1260pub struct RecoveredCursorRelease {
1261 debt: ClosureDebt,
1262 release: DetachedCursorRelease,
1263}
1264
1265impl RecoveredCursorRelease {
1266 #[must_use]
1268 pub const fn debt(&self) -> ClosureDebt {
1269 self.debt
1270 }
1271
1272 #[must_use]
1274 pub const fn edge(&self) -> DetachedCursorRelease {
1275 self.release
1276 }
1277
1278 #[must_use]
1280 pub const fn into_state(self) -> ClosureState {
1281 owed(self.debt, StoredEdge::DetachedCursorRelease(self.release))
1282 }
1283}
1284
1285#[derive(Debug, PartialEq, Eq)]
1287pub enum RecoveredBindingFateTransition {
1288 PendingStorage(PendingRecoveredCursorRelease),
1290 DetachedCursorRelease(RecoveredCursorRelease),
1292}
1293
1294#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1295enum SuccessorUse {
1296 ObserverCompletion,
1297 ObserverMarkerAppend,
1298 ObserverLeave,
1299 PhysicalCompletion,
1300 PhysicalCover,
1301 CursorGreaterAck,
1302}
1303
1304#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1310pub struct ProjectionCompactionSuccessor {
1311 predecessor: StoredEdge,
1312 event: Event,
1313 use_kind: SuccessorUse,
1314 state: ClosureState,
1315}
1316
1317#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1324pub enum CursorFateSuccessor {
1325 DetachedCredentialRecovery(DetachedCredentialRecovery),
1327 DetachedCursorRelease(DetachedCursorRelease),
1329}
1330
1331impl CursorFateSuccessor {
1332 #[must_use]
1334 pub const fn into_stored_edge(self) -> StoredEdge {
1335 match self {
1336 Self::DetachedCredentialRecovery(value) => {
1337 StoredEdge::DetachedCredentialRecovery(value)
1338 }
1339 Self::DetachedCursorRelease(value) => StoredEdge::DetachedCursorRelease(value),
1340 }
1341 }
1342}
1343
1344#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1346pub enum DetachedAttachRefusal {
1347 RecoveryFence,
1349 MarkerNotDelivered,
1351 MarkerMismatch,
1353 DeliveredMarkerAwaitingAck,
1355 EpisodeChurnLimit,
1357 StaleAuthority,
1359 NoBinding,
1361}
1362
1363#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1364enum DetachedClaimTarget {
1365 CredentialRecovery {
1366 marker_delivery_seq: DeliverySeq,
1367 binding_epoch: BindingEpoch,
1368 },
1369 MarkerRelease {
1370 marker_delivery_seq: DeliverySeq,
1371 binding_epoch: BindingEpoch,
1372 },
1373 CursorRelease {
1374 binding_epoch: BindingEpoch,
1375 },
1376}
1377
1378#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1384pub struct KClaimBackedDetachedLeave {
1385 participant_id: ParticipantId,
1386 target: DetachedClaimTarget,
1387 actual_charge: ResourceVector,
1388}
1389
1390impl KClaimBackedDetachedLeave {
1391 #[must_use]
1393 pub const fn participant_id(self) -> ParticipantId {
1394 self.participant_id
1395 }
1396
1397 #[must_use]
1399 pub const fn actual_charge(self) -> ResourceVector {
1400 self.actual_charge
1401 }
1402}
1403
1404#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1413pub struct Event(EventKind);
1414
1415#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1416enum EventKind {
1417 ProjectionCompleted {
1418 through_seq: DeliverySeq,
1419 },
1420 CompactionCompleted {
1421 from_floor: DeliverySeq,
1422 through_seq: DeliverySeq,
1423 resulting_floor: DeliverySeq,
1424 },
1425 MarkerAppended {
1426 marker_delivery_seq: DeliverySeq,
1427 resulting_projection_through: DeliverySeq,
1428 },
1429 MarkerDelivered {
1430 participant_id: ParticipantId,
1431 binding_epoch: BindingEpoch,
1432 marker_delivery_seq: DeliverySeq,
1433 },
1434 CursorProgressed {
1435 participant_index: ParticipantIndex,
1436 participant_id: ParticipantId,
1437 binding_epoch: BindingEpoch,
1438 progress: CursorProgressEvent,
1439 resulting_floor: DeliverySeq,
1440 },
1441 BindingFateObserved {
1442 participant_id: ParticipantId,
1443 binding_epoch: BindingEpoch,
1444 resulting_floor: DeliverySeq,
1445 },
1446 LeaveCommitted {
1447 participant_id: ParticipantId,
1448 authority: LeaveAuthority,
1449 resulting_floor: DeliverySeq,
1450 },
1451 FencedRecoveryCommitted {
1452 participant_id: ParticipantId,
1453 marker_delivery_seq: DeliverySeq,
1454 prior_binding_epoch: BindingEpoch,
1455 new_binding_epoch: BindingEpoch,
1456 resulting_floor: DeliverySeq,
1457 },
1458}
1459
1460#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1461enum CursorProgressEvent {
1462 Normal {
1463 previous_cursor: DeliverySeq,
1464 through_seq: DeliverySeq,
1465 },
1466 Marker {
1467 marker_delivery_seq: DeliverySeq,
1468 },
1469}
1470
1471#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1472enum LeaveAuthority {
1473 Live(BindingEpoch),
1474 Detached,
1475}
1476
1477impl Event {
1478 #[must_use]
1480 pub const fn projection_completed(through_seq: DeliverySeq) -> Self {
1481 Self(EventKind::ProjectionCompleted { through_seq })
1482 }
1483
1484 #[must_use]
1486 pub const fn compaction_completed(
1487 from_floor: DeliverySeq,
1488 through_seq: DeliverySeq,
1489 resulting_floor: DeliverySeq,
1490 ) -> Option<Self> {
1491 if from_floor <= through_seq && resulting_floor > through_seq {
1492 Some(Self(EventKind::CompactionCompleted {
1493 from_floor,
1494 through_seq,
1495 resulting_floor,
1496 }))
1497 } else {
1498 None
1499 }
1500 }
1501
1502 #[must_use]
1504 pub const fn marker_appended(
1505 marker_delivery_seq: DeliverySeq,
1506 resulting_projection_through: DeliverySeq,
1507 ) -> Self {
1508 Self(EventKind::MarkerAppended {
1509 marker_delivery_seq,
1510 resulting_projection_through,
1511 })
1512 }
1513
1514 #[must_use]
1516 pub const fn marker_delivered(
1517 participant_id: ParticipantId,
1518 binding_epoch: BindingEpoch,
1519 marker_delivery_seq: DeliverySeq,
1520 ) -> Self {
1521 Self(EventKind::MarkerDelivered {
1522 participant_id,
1523 binding_epoch,
1524 marker_delivery_seq,
1525 })
1526 }
1527
1528 #[must_use]
1534 pub const fn cursor_progressed(
1535 participant_id: ParticipantId,
1536 binding_epoch: BindingEpoch,
1537 previous_cursor: DeliverySeq,
1538 through_seq: DeliverySeq,
1539 resulting_floor: DeliverySeq,
1540 ) -> Option<Self> {
1541 if through_seq > previous_cursor {
1542 Some(Self(EventKind::CursorProgressed {
1543 participant_index: participant_id,
1544 participant_id,
1545 binding_epoch,
1546 progress: CursorProgressEvent::Normal {
1547 previous_cursor,
1548 through_seq,
1549 },
1550 resulting_floor,
1551 }))
1552 } else {
1553 None
1554 }
1555 }
1556
1557 #[must_use]
1560 pub const fn marker_acknowledged(
1561 participant_id: ParticipantId,
1562 binding_epoch: BindingEpoch,
1563 marker_delivery_seq: DeliverySeq,
1564 resulting_floor: DeliverySeq,
1565 ) -> Self {
1566 Self(EventKind::CursorProgressed {
1567 participant_index: participant_id,
1568 participant_id,
1569 binding_epoch,
1570 progress: CursorProgressEvent::Marker {
1571 marker_delivery_seq,
1572 },
1573 resulting_floor,
1574 })
1575 }
1576
1577 #[must_use]
1579 pub const fn binding_fate_observed(
1580 participant_id: ParticipantId,
1581 binding_epoch: BindingEpoch,
1582 resulting_floor: DeliverySeq,
1583 ) -> Self {
1584 Self(EventKind::BindingFateObserved {
1585 participant_id,
1586 binding_epoch,
1587 resulting_floor,
1588 })
1589 }
1590
1591 #[must_use]
1593 pub const fn live_leave_committed(
1594 participant_id: ParticipantId,
1595 binding_epoch: BindingEpoch,
1596 resulting_floor: DeliverySeq,
1597 ) -> Self {
1598 Self(EventKind::LeaveCommitted {
1599 participant_id,
1600 authority: LeaveAuthority::Live(binding_epoch),
1601 resulting_floor,
1602 })
1603 }
1604
1605 #[must_use]
1607 pub const fn detached_leave_committed(
1608 participant_id: ParticipantId,
1609 resulting_floor: DeliverySeq,
1610 ) -> Self {
1611 Self(EventKind::LeaveCommitted {
1612 participant_id,
1613 authority: LeaveAuthority::Detached,
1614 resulting_floor,
1615 })
1616 }
1617
1618 #[must_use]
1620 pub const fn fenced_recovery_committed(
1621 participant_id: ParticipantId,
1622 marker_delivery_seq: DeliverySeq,
1623 prior_binding_epoch: BindingEpoch,
1624 new_binding_epoch: BindingEpoch,
1625 resulting_floor: DeliverySeq,
1626 ) -> Self {
1627 Self(EventKind::FencedRecoveryCommitted {
1628 participant_id,
1629 marker_delivery_seq,
1630 prior_binding_epoch,
1631 new_binding_epoch,
1632 resulting_floor,
1633 })
1634 }
1635}
1636
1637impl ObserverProjection {
1638 #[must_use]
1644 #[allow(
1645 dead_code,
1646 reason = "the crate-owned binding-fate operation invokes this sealed OP transition"
1647 )]
1648 pub const fn apply_ordinary_binding_fate(
1649 self,
1650 resulting_debt: ClosureDebt,
1651 authority: OrdinaryBindingFate,
1652 ) -> PendingRecoveredCursorRelease {
1653 PendingRecoveredCursorRelease {
1654 debt: resulting_debt,
1655 predecessor: RecoveredStorageEdge::ObserverProjection(self),
1656 release: authority.release,
1657 }
1658 }
1659
1660 pub fn apply_recovered_binding_fate(
1671 self,
1672 debt: ClosureDebt,
1673 resulting_debt: ClosureDebt,
1674 authority: RecoveredBindingFate,
1675 ) -> Result<RecoveredBindingFateTransition, RecoveredBindingFate> {
1676 if authority.predecessor != RecoveredStorageEdge::ObserverProjection(self)
1677 || authority.predecessor_debt != debt
1678 {
1679 return Err(authority);
1680 }
1681 Ok(RecoveredBindingFateTransition::PendingStorage(
1682 PendingRecoveredCursorRelease {
1683 debt: resulting_debt,
1684 predecessor: RecoveredStorageEdge::ObserverProjection(self),
1685 release: authority.release,
1686 },
1687 ))
1688 }
1689
1690 pub fn complete_after_recovered_binding_fate(
1697 self,
1698 event: Event,
1699 resulting_debt: Option<ClosureDebt>,
1700 pending: PendingRecoveredCursorRelease,
1701 ) -> Result<ClosureState, PendingRecoveredCursorRelease> {
1702 self.complete_after_binding_fate(event, resulting_debt, pending)
1703 }
1704
1705 pub fn complete_after_ordinary_binding_fate(
1712 self,
1713 event: Event,
1714 resulting_debt: Option<ClosureDebt>,
1715 pending: PendingRecoveredCursorRelease,
1716 ) -> Result<ClosureState, PendingRecoveredCursorRelease> {
1717 self.complete_after_binding_fate(event, resulting_debt, pending)
1718 }
1719
1720 pub(crate) fn complete_after_binding_fate(
1727 self,
1728 event: Event,
1729 resulting_debt: Option<ClosureDebt>,
1730 pending: PendingRecoveredCursorRelease,
1731 ) -> Result<ClosureState, PendingRecoveredCursorRelease> {
1732 if pending.predecessor != RecoveredStorageEdge::ObserverProjection(self)
1733 || projection_completion_boundary(self, event).is_none()
1734 {
1735 return Err(pending);
1736 }
1737 Ok(preserve_or_clear(
1738 resulting_debt,
1739 StoredEdge::DetachedCursorRelease(pending.release),
1740 ))
1741 }
1742
1743 #[must_use]
1745 pub const fn clear_after_completion(
1746 &self,
1747 event: &Event,
1748 ) -> Option<ProjectionCompactionSuccessor> {
1749 if projection_completion_boundary(*self, *event).is_some() {
1750 Some(ProjectionCompactionSuccessor {
1751 predecessor: StoredEdge::ObserverProjection(*self),
1752 event: *event,
1753 use_kind: SuccessorUse::ObserverCompletion,
1754 state: ClosureState::Clear,
1755 })
1756 } else {
1757 None
1758 }
1759 }
1760
1761 #[must_use]
1763 pub const fn strict_after_completion(
1764 &self,
1765 event: &Event,
1766 debt: ClosureDebt,
1767 edge: StoredEdge,
1768 successor_boundary: DeliverySeq,
1769 ) -> Option<ProjectionCompactionSuccessor> {
1770 let Some(completed_through) = projection_completion_boundary(*self, *event) else {
1771 return None;
1772 };
1773 if successor_boundary <= completed_through
1774 || !strict_edge_matches_boundary(edge, successor_boundary)
1775 {
1776 return None;
1777 }
1778 Some(ProjectionCompactionSuccessor {
1779 predecessor: StoredEdge::ObserverProjection(*self),
1780 event: *event,
1781 use_kind: SuccessorUse::ObserverCompletion,
1782 state: ClosureState::Owed { debt, edge },
1783 })
1784 }
1785
1786 pub fn complete(
1793 self,
1794 debt: ClosureDebt,
1795 event: Event,
1796 successor: ProjectionCompactionSuccessor,
1797 ) -> Result<ClosureState, ClosureState> {
1798 let original = owed(debt, StoredEdge::ObserverProjection(self));
1799 if successor.predecessor == StoredEdge::ObserverProjection(self)
1800 && successor.event == event
1801 && successor.use_kind == SuccessorUse::ObserverCompletion
1802 && projection_completion_boundary(self, event).is_some()
1803 {
1804 Ok(successor.state)
1805 } else {
1806 Err(original)
1807 }
1808 }
1809
1810 #[must_use]
1812 pub const fn later_projection_after_marker(
1813 &self,
1814 event: &Event,
1815 debt: ClosureDebt,
1816 successor: Self,
1817 ) -> Option<ProjectionCompactionSuccessor> {
1818 let EventKind::MarkerAppended {
1819 marker_delivery_seq,
1820 resulting_projection_through,
1821 } = event.0
1822 else {
1823 return None;
1824 };
1825 if marker_delivery_seq <= self.through_seq
1826 || resulting_projection_through < marker_delivery_seq
1827 || successor.through_seq != resulting_projection_through
1828 {
1829 return None;
1830 }
1831 Some(ProjectionCompactionSuccessor {
1832 predecessor: StoredEdge::ObserverProjection(*self),
1833 event: *event,
1834 use_kind: SuccessorUse::ObserverMarkerAppend,
1835 state: owed(debt, StoredEdge::ObserverProjection(successor)),
1836 })
1837 }
1838
1839 pub fn marker_appended(
1846 self,
1847 debt: ClosureDebt,
1848 event: Event,
1849 successor: ProjectionCompactionSuccessor,
1850 ) -> Result<ClosureState, ClosureState> {
1851 let original = owed(debt, StoredEdge::ObserverProjection(self));
1852 if successor.predecessor == StoredEdge::ObserverProjection(self)
1853 && successor.event == event
1854 && successor.use_kind == SuccessorUse::ObserverMarkerAppend
1855 {
1856 Ok(successor.state)
1857 } else {
1858 Err(original)
1859 }
1860 }
1861
1862 #[must_use]
1864 pub const fn later_projection_after_leave(
1865 &self,
1866 event: &Event,
1867 debt: ClosureDebt,
1868 successor: Self,
1869 ) -> Option<ProjectionCompactionSuccessor> {
1870 let EventKind::LeaveCommitted {
1871 resulting_floor, ..
1872 } = event.0
1873 else {
1874 return None;
1875 };
1876 if successor.through_seq <= self.through_seq || successor.through_seq < resulting_floor {
1877 return None;
1878 }
1879 Some(ProjectionCompactionSuccessor {
1880 predecessor: StoredEdge::ObserverProjection(*self),
1881 event: *event,
1882 use_kind: SuccessorUse::ObserverLeave,
1883 state: owed(debt, StoredEdge::ObserverProjection(successor)),
1884 })
1885 }
1886
1887 pub fn leave_with_later_projection(
1894 self,
1895 debt: ClosureDebt,
1896 event: Event,
1897 successor: ProjectionCompactionSuccessor,
1898 ) -> Result<ClosureState, ClosureState> {
1899 let original = owed(debt, StoredEdge::ObserverProjection(self));
1900 if successor.predecessor == StoredEdge::ObserverProjection(self)
1901 && successor.event == event
1902 && successor.use_kind == SuccessorUse::ObserverLeave
1903 {
1904 Ok(successor.state)
1905 } else {
1906 Err(original)
1907 }
1908 }
1909
1910 pub const fn independent_event(
1918 self,
1919 debt: ClosureDebt,
1920 event: Event,
1921 resulting_debt: Option<ClosureDebt>,
1922 ) -> Result<ClosureState, ClosureState> {
1923 let original = owed(debt, StoredEdge::ObserverProjection(self));
1924 if !matches!(
1925 event.0,
1926 EventKind::CursorProgressed { .. }
1927 | EventKind::BindingFateObserved { .. }
1928 | EventKind::LeaveCommitted { .. }
1929 ) {
1930 return Err(original);
1931 }
1932 Ok(preserve_or_clear(
1933 resulting_debt,
1934 StoredEdge::ObserverProjection(self),
1935 ))
1936 }
1937
1938 pub const fn charged_binding_change(
1945 self,
1946 debt: ClosureDebt,
1947 episode_churn_used: u64,
1948 delta_cycles: u64,
1949 episode_churn_limit: u64,
1950 resulting_debt: Option<ClosureDebt>,
1951 ) -> Result<ClosureState, (ClosureState, DetachedAttachRefusal)> {
1952 let original = owed(debt, StoredEdge::ObserverProjection(self));
1953 if !charged_churn_fits(episode_churn_used, delta_cycles, episode_churn_limit) {
1954 return Err((original, DetachedAttachRefusal::EpisodeChurnLimit));
1955 }
1956 Ok(preserve_or_clear(
1957 resulting_debt,
1958 StoredEdge::ObserverProjection(self),
1959 ))
1960 }
1961}
1962
1963impl PhysicalCompaction {
1964 #[must_use]
1966 #[allow(
1967 dead_code,
1968 reason = "the crate-owned binding-fate replay boundary invokes this sealed PC transition"
1969 )]
1970 pub(crate) const fn apply_ordinary_binding_fate(
1971 self,
1972 resulting_debt: ClosureDebt,
1973 authority: OrdinaryBindingFate,
1974 ) -> RecoveredBindingFateTransition {
1975 if authority.resulting_floor > self.through_seq {
1976 RecoveredBindingFateTransition::DetachedCursorRelease(RecoveredCursorRelease {
1977 debt: resulting_debt,
1978 release: authority.release,
1979 })
1980 } else {
1981 RecoveredBindingFateTransition::PendingStorage(PendingRecoveredCursorRelease {
1982 debt: resulting_debt,
1983 predecessor: RecoveredStorageEdge::PhysicalCompaction(self),
1984 release: authority.release,
1985 })
1986 }
1987 }
1988
1989 pub fn apply_recovered_binding_fate(
2000 self,
2001 debt: ClosureDebt,
2002 resulting_debt: ClosureDebt,
2003 authority: RecoveredBindingFate,
2004 ) -> Result<RecoveredBindingFateTransition, RecoveredBindingFate> {
2005 if authority.predecessor != RecoveredStorageEdge::PhysicalCompaction(self)
2006 || authority.predecessor_debt != debt
2007 {
2008 return Err(authority);
2009 }
2010 if authority.resulting_floor > self.through_seq {
2011 Ok(RecoveredBindingFateTransition::DetachedCursorRelease(
2012 RecoveredCursorRelease {
2013 debt: resulting_debt,
2014 release: authority.release,
2015 },
2016 ))
2017 } else {
2018 Ok(RecoveredBindingFateTransition::PendingStorage(
2019 PendingRecoveredCursorRelease {
2020 debt: resulting_debt,
2021 predecessor: RecoveredStorageEdge::PhysicalCompaction(self),
2022 release: authority.release,
2023 },
2024 ))
2025 }
2026 }
2027
2028 pub fn complete_after_recovered_binding_fate(
2035 self,
2036 event: Event,
2037 resulting_debt: Option<ClosureDebt>,
2038 pending: PendingRecoveredCursorRelease,
2039 ) -> Result<ClosureState, PendingRecoveredCursorRelease> {
2040 self.complete_after_binding_fate(event, resulting_debt, pending)
2041 }
2042
2043 pub(crate) fn complete_after_binding_fate(
2050 self,
2051 event: Event,
2052 resulting_debt: Option<ClosureDebt>,
2053 pending: PendingRecoveredCursorRelease,
2054 ) -> Result<ClosureState, PendingRecoveredCursorRelease> {
2055 if pending.predecessor != RecoveredStorageEdge::PhysicalCompaction(self)
2056 || physical_completion_floor(self, event).is_none()
2057 {
2058 return Err(pending);
2059 }
2060 Ok(preserve_or_clear(
2061 resulting_debt,
2062 StoredEdge::DetachedCursorRelease(pending.release),
2063 ))
2064 }
2065
2066 #[must_use]
2068 pub const fn clear_after_completion(
2069 &self,
2070 event: &Event,
2071 ) -> Option<ProjectionCompactionSuccessor> {
2072 if physical_completion_floor(*self, *event).is_some() {
2073 Some(ProjectionCompactionSuccessor {
2074 predecessor: StoredEdge::PhysicalCompaction(*self),
2075 event: *event,
2076 use_kind: SuccessorUse::PhysicalCompletion,
2077 state: ClosureState::Clear,
2078 })
2079 } else {
2080 None
2081 }
2082 }
2083
2084 #[must_use]
2086 pub const fn strict_after_completion(
2087 &self,
2088 event: &Event,
2089 debt: ClosureDebt,
2090 edge: StoredEdge,
2091 successor_boundary: DeliverySeq,
2092 ) -> Option<ProjectionCompactionSuccessor> {
2093 let Some(resulting_floor) = physical_completion_floor(*self, *event) else {
2094 return None;
2095 };
2096 if successor_boundary < resulting_floor
2097 || !strict_edge_matches_boundary(edge, successor_boundary)
2098 {
2099 return None;
2100 }
2101 Some(ProjectionCompactionSuccessor {
2102 predecessor: StoredEdge::PhysicalCompaction(*self),
2103 event: *event,
2104 use_kind: SuccessorUse::PhysicalCompletion,
2105 state: ClosureState::Owed { debt, edge },
2106 })
2107 }
2108
2109 pub fn complete(
2116 self,
2117 debt: ClosureDebt,
2118 event: Event,
2119 successor: ProjectionCompactionSuccessor,
2120 ) -> Result<ClosureState, ClosureState> {
2121 let original = owed(debt, StoredEdge::PhysicalCompaction(self));
2122 if successor.predecessor == StoredEdge::PhysicalCompaction(self)
2123 && successor.event == event
2124 && successor.use_kind == SuccessorUse::PhysicalCompletion
2125 && physical_completion_floor(self, event).is_some()
2126 {
2127 Ok(successor.state)
2128 } else {
2129 Err(original)
2130 }
2131 }
2132
2133 pub const fn marker_appended(
2141 self,
2142 debt: ClosureDebt,
2143 event: Event,
2144 ) -> Result<ClosureState, ClosureState> {
2145 let original = owed(debt, StoredEdge::PhysicalCompaction(self));
2146 let EventKind::MarkerAppended {
2147 marker_delivery_seq,
2148 resulting_projection_through,
2149 } = event.0
2150 else {
2151 return Err(original);
2152 };
2153 if marker_delivery_seq <= self.through_seq
2154 || resulting_projection_through < marker_delivery_seq
2155 {
2156 return Err(original);
2157 }
2158 Ok(original)
2159 }
2160
2161 pub const fn preserve_progress(
2169 self,
2170 debt: ClosureDebt,
2171 event: Event,
2172 resulting_debt: ClosureDebt,
2173 ) -> Result<ClosureState, ClosureState> {
2174 let original = owed(debt, StoredEdge::PhysicalCompaction(self));
2175 let Some(resulting_floor) = progress_event_floor(event) else {
2176 return Err(original);
2177 };
2178 if resulting_floor > self.through_seq {
2179 return Err(original);
2180 }
2181 Ok(owed(resulting_debt, StoredEdge::PhysicalCompaction(self)))
2182 }
2183
2184 #[must_use]
2186 pub const fn clear_after_progress(
2187 &self,
2188 event: &Event,
2189 ) -> Option<ProjectionCompactionSuccessor> {
2190 let Some(resulting_floor) = progress_event_floor(*event) else {
2191 return None;
2192 };
2193 if resulting_floor <= self.through_seq {
2194 return None;
2195 }
2196 Some(ProjectionCompactionSuccessor {
2197 predecessor: StoredEdge::PhysicalCompaction(*self),
2198 event: *event,
2199 use_kind: SuccessorUse::PhysicalCover,
2200 state: ClosureState::Clear,
2201 })
2202 }
2203
2204 #[must_use]
2206 pub const fn strict_after_progress(
2207 &self,
2208 event: &Event,
2209 debt: ClosureDebt,
2210 edge: StoredEdge,
2211 successor_boundary: DeliverySeq,
2212 ) -> Option<ProjectionCompactionSuccessor> {
2213 let Some(resulting_floor) = progress_event_floor(*event) else {
2214 return None;
2215 };
2216 if resulting_floor <= self.through_seq
2217 || successor_boundary < resulting_floor
2218 || !strict_edge_matches_boundary(edge, successor_boundary)
2219 {
2220 return None;
2221 }
2222 Some(ProjectionCompactionSuccessor {
2223 predecessor: StoredEdge::PhysicalCompaction(*self),
2224 event: *event,
2225 use_kind: SuccessorUse::PhysicalCover,
2226 state: ClosureState::Owed { debt, edge },
2227 })
2228 }
2229
2230 pub fn covered_by_progress(
2237 self,
2238 debt: ClosureDebt,
2239 event: Event,
2240 successor: ProjectionCompactionSuccessor,
2241 ) -> Result<ClosureState, ClosureState> {
2242 let original = owed(debt, StoredEdge::PhysicalCompaction(self));
2243 if successor.predecessor == StoredEdge::PhysicalCompaction(self)
2244 && successor.event == event
2245 && successor.use_kind == SuccessorUse::PhysicalCover
2246 {
2247 Ok(successor.state)
2248 } else {
2249 Err(original)
2250 }
2251 }
2252
2253 #[must_use]
2255 pub const fn unchanged(self, debt: ClosureDebt) -> ClosureState {
2256 owed(debt, StoredEdge::PhysicalCompaction(self))
2257 }
2258
2259 pub const fn charged_binding_change_preserving(
2266 self,
2267 debt: ClosureDebt,
2268 episode_churn_used: u64,
2269 delta_cycles: u64,
2270 episode_churn_limit: u64,
2271 resulting_floor: DeliverySeq,
2272 resulting_debt: ClosureDebt,
2273 ) -> Result<ClosureState, (ClosureState, DetachedAttachRefusal)> {
2274 let original = owed(debt, StoredEdge::PhysicalCompaction(self));
2275 if !charged_churn_fits(episode_churn_used, delta_cycles, episode_churn_limit) {
2276 return Err((original, DetachedAttachRefusal::EpisodeChurnLimit));
2277 }
2278 if resulting_floor > self.through_seq {
2279 return Err((original, DetachedAttachRefusal::StaleAuthority));
2280 }
2281 Ok(owed(resulting_debt, StoredEdge::PhysicalCompaction(self)))
2282 }
2283
2284 #[allow(clippy::too_many_arguments)]
2291 pub const fn charged_binding_change_covering(
2292 self,
2293 debt: ClosureDebt,
2294 episode_churn_used: u64,
2295 delta_cycles: u64,
2296 episode_churn_limit: u64,
2297 resulting_floor: DeliverySeq,
2298 resulting_debt: ClosureDebt,
2299 edge: StoredEdge,
2300 successor_boundary: DeliverySeq,
2301 ) -> Result<ClosureState, (ClosureState, DetachedAttachRefusal)> {
2302 let original = owed(debt, StoredEdge::PhysicalCompaction(self));
2303 if !charged_churn_fits(episode_churn_used, delta_cycles, episode_churn_limit) {
2304 return Err((original, DetachedAttachRefusal::EpisodeChurnLimit));
2305 }
2306 if resulting_floor <= self.through_seq
2307 || successor_boundary < resulting_floor
2308 || !strict_edge_matches_boundary(edge, successor_boundary)
2309 {
2310 return Err((original, DetachedAttachRefusal::StaleAuthority));
2311 }
2312 Ok(owed(resulting_debt, edge))
2313 }
2314}
2315
2316impl MarkerDelivery {
2317 pub fn delivered_progress(self, event: Event) -> Result<ParticipantCursorProgress, Self> {
2330 let EventKind::MarkerDelivered {
2331 participant_id,
2332 binding_epoch,
2333 marker_delivery_seq,
2334 } = event.0
2335 else {
2336 return Err(self);
2337 };
2338 if participant_id != self.participant_id
2339 || binding_epoch != self.binding_epoch
2340 || marker_delivery_seq != self.marker_delivery_seq
2341 {
2342 return Err(self);
2343 }
2344 Ok(ParticipantCursorProgress::Marker(CursorProgressMarker {
2345 conversation_id: self.conversation_id,
2346 participant_id,
2347 binding_epoch,
2348 through_seq: marker_delivery_seq,
2349 marker_delivery_seq,
2350 }))
2351 }
2352
2353 pub fn delivered(self, debt: ClosureDebt, event: Event) -> Result<ClosureState, ClosureState> {
2363 let original = owed(debt, StoredEdge::MarkerDelivery(self));
2364 let Ok(progress) = self.delivered_progress(event) else {
2365 return Err(original);
2366 };
2367 Ok(owed(debt, StoredEdge::ParticipantCursorProgress(progress)))
2368 }
2369
2370 pub const fn lower_progress(
2378 self,
2379 debt: ClosureDebt,
2380 event: Event,
2381 resulting_debt: Option<ClosureDebt>,
2382 ) -> Result<ClosureState, ClosureState> {
2383 let original = owed(debt, StoredEdge::MarkerDelivery(self));
2384 let is_lower = match event.0 {
2385 EventKind::CursorProgressed {
2386 progress: CursorProgressEvent::Normal { through_seq, .. },
2387 ..
2388 }
2389 | EventKind::ProjectionCompleted { through_seq } => {
2390 through_seq < self.marker_delivery_seq
2391 }
2392 EventKind::CompactionCompleted {
2393 through_seq,
2394 resulting_floor,
2395 ..
2396 } => {
2397 through_seq < self.marker_delivery_seq
2398 && resulting_floor <= self.marker_delivery_seq
2399 }
2400 _ => false,
2401 };
2402 if !is_lower {
2403 return Err(original);
2404 }
2405 Ok(preserve_or_clear(
2406 resulting_debt,
2407 StoredEdge::MarkerDelivery(self),
2408 ))
2409 }
2410
2411 pub fn binding_fate(
2418 self,
2419 debt: ClosureDebt,
2420 event: Event,
2421 ) -> Result<ClosureState, ClosureState> {
2422 let original = owed(debt, StoredEdge::MarkerDelivery(self));
2423 let EventKind::BindingFateObserved {
2424 participant_id,
2425 binding_epoch,
2426 ..
2427 } = event.0
2428 else {
2429 return Err(original);
2430 };
2431 if participant_id != self.participant_id || binding_epoch != self.binding_epoch {
2432 return Err(original);
2433 }
2434 Ok(owed(
2435 debt,
2436 StoredEdge::DetachedMarkerRelease(DetachedMarkerRelease {
2437 participant_id,
2438 marker_delivery_seq: self.marker_delivery_seq,
2439 last_dead_binding_epoch: binding_epoch,
2440 }),
2441 ))
2442 }
2443
2444 pub const fn retarget(
2451 self,
2452 new_binding_epoch: BindingEpoch,
2453 episode_churn_used: u64,
2454 delta_cycles: u64,
2455 episode_churn_limit: u64,
2456 ) -> Result<Self, (Self, DetachedAttachRefusal)> {
2457 if !charged_churn_fits(episode_churn_used, delta_cycles, episode_churn_limit) {
2458 return Err((self, DetachedAttachRefusal::EpisodeChurnLimit));
2459 }
2460 if !is_next_generation(self.binding_epoch, new_binding_epoch) {
2461 return Err((self, DetachedAttachRefusal::StaleAuthority));
2462 }
2463 Ok(Self {
2464 binding_epoch: new_binding_epoch,
2465 ..self
2466 })
2467 }
2468
2469 pub fn leave(
2476 self,
2477 debt: ClosureDebt,
2478 event: Event,
2479 successor: DebtCompletion,
2480 ) -> Result<ClosureState, ClosureState> {
2481 let original = owed(debt, StoredEdge::MarkerDelivery(self));
2482 let EventKind::LeaveCommitted {
2483 participant_id,
2484 authority: LeaveAuthority::Live(binding_epoch),
2485 ..
2486 } = event.0
2487 else {
2488 return Err(original);
2489 };
2490 if participant_id != self.participant_id || binding_epoch != self.binding_epoch {
2491 return Err(original);
2492 }
2493 Ok(successor.into_state())
2494 }
2495}
2496
2497impl ParticipantCursorProgress {
2498 pub fn complete_ack(
2506 self,
2507 debt: ClosureDebt,
2508 event: Event,
2509 successor: DebtCompletion,
2510 ) -> Result<ClosureState, ClosureState> {
2511 let original = owed(debt, StoredEdge::ParticipantCursorProgress(self));
2512 let exact = match (self, event.0) {
2513 (
2514 Self::Continuous(value),
2515 EventKind::CursorProgressed {
2516 participant_id,
2517 binding_epoch,
2518 progress: CursorProgressEvent::Normal { through_seq, .. },
2519 ..
2520 },
2521 ) => {
2522 participant_id == value.participant_id
2523 && binding_epoch == value.binding_epoch
2524 && through_seq == value.through_seq
2525 }
2526 (
2527 Self::Marker(value),
2528 EventKind::CursorProgressed {
2529 participant_id,
2530 binding_epoch,
2531 progress: CursorProgressEvent::Normal { through_seq, .. },
2532 ..
2533 },
2534 ) => {
2535 participant_id == value.participant_id
2536 && binding_epoch == value.binding_epoch
2537 && through_seq == value.through_seq
2538 }
2539 (
2540 Self::Marker(value),
2541 EventKind::CursorProgressed {
2542 participant_id,
2543 binding_epoch,
2544 progress:
2545 CursorProgressEvent::Marker {
2546 marker_delivery_seq,
2547 },
2548 ..
2549 },
2550 ) => {
2551 participant_id == value.participant_id
2552 && binding_epoch == value.binding_epoch
2553 && marker_delivery_seq == value.marker_delivery_seq
2554 }
2555 _ => false,
2556 };
2557 if exact {
2558 Ok(successor.into_state())
2559 } else {
2560 Err(original)
2561 }
2562 }
2563
2564 pub fn lesser_ack(
2571 self,
2572 debt: ClosureDebt,
2573 event: Event,
2574 resulting_debt: ClosureDebt,
2575 ) -> Result<ClosureState, ClosureState> {
2576 let original = owed(debt, StoredEdge::ParticipantCursorProgress(self));
2577 let EventKind::CursorProgressed {
2578 participant_id,
2579 binding_epoch,
2580 progress: CursorProgressEvent::Normal { through_seq, .. },
2581 ..
2582 } = event.0
2583 else {
2584 return Err(original);
2585 };
2586 if participant_id != self.participant_id()
2587 || binding_epoch != self.binding_epoch()
2588 || through_seq >= self.through_seq()
2589 {
2590 return Err(original);
2591 }
2592 Ok(owed(
2593 resulting_debt,
2594 StoredEdge::ParticipantCursorProgress(self),
2595 ))
2596 }
2597
2598 #[must_use]
2600 pub fn clear_after_greater_ack(&self, event: &Event) -> Option<ProjectionCompactionSuccessor> {
2601 if !greater_ack_matches(*self, *event) {
2602 return None;
2603 }
2604 Some(ProjectionCompactionSuccessor {
2605 predecessor: StoredEdge::ParticipantCursorProgress(*self),
2606 event: *event,
2607 use_kind: SuccessorUse::CursorGreaterAck,
2608 state: ClosureState::Clear,
2609 })
2610 }
2611
2612 #[must_use]
2614 pub fn strict_after_greater_ack(
2615 &self,
2616 event: &Event,
2617 debt: ClosureDebt,
2618 edge: StoredEdge,
2619 successor_boundary: DeliverySeq,
2620 ) -> Option<ProjectionCompactionSuccessor> {
2621 if !greater_ack_matches(*self, *event)
2622 || successor_boundary <= cursor_event_boundary(*event)
2623 || !strict_edge_matches_boundary(edge, successor_boundary)
2624 {
2625 return None;
2626 }
2627 Some(ProjectionCompactionSuccessor {
2628 predecessor: StoredEdge::ParticipantCursorProgress(*self),
2629 event: *event,
2630 use_kind: SuccessorUse::CursorGreaterAck,
2631 state: ClosureState::Owed { debt, edge },
2632 })
2633 }
2634
2635 pub fn greater_ack(
2642 self,
2643 debt: ClosureDebt,
2644 event: Event,
2645 successor: ProjectionCompactionSuccessor,
2646 ) -> Result<ClosureState, ClosureState> {
2647 let original = owed(debt, StoredEdge::ParticipantCursorProgress(self));
2648 if successor.predecessor == StoredEdge::ParticipantCursorProgress(self)
2649 && successor.event == event
2650 && successor.use_kind == SuccessorUse::CursorGreaterAck
2651 && greater_ack_matches(self, event)
2652 {
2653 Ok(successor.state)
2654 } else {
2655 Err(original)
2656 }
2657 }
2658
2659 #[must_use]
2661 pub const fn unchanged(self, debt: ClosureDebt) -> ClosureState {
2662 owed(debt, StoredEdge::ParticipantCursorProgress(self))
2663 }
2664
2665 pub const fn storage_progress(
2674 self,
2675 debt: ClosureDebt,
2676 event: Event,
2677 resulting_debt: Option<ClosureDebt>,
2678 ) -> Result<ClosureState, ClosureState> {
2679 let original = owed(debt, StoredEdge::ParticipantCursorProgress(self));
2680 let valid = match event.0 {
2681 EventKind::ProjectionCompleted { .. } => true,
2682 EventKind::CompactionCompleted {
2683 through_seq,
2684 resulting_floor,
2685 ..
2686 } => match self.marker_delivery_seq() {
2687 None => true,
2688 Some(marker) => through_seq < marker && resulting_floor <= marker,
2689 },
2690 _ => false,
2691 };
2692 if !valid {
2693 return Err(original);
2694 }
2695 Ok(preserve_or_clear(
2696 resulting_debt,
2697 StoredEdge::ParticipantCursorProgress(self),
2698 ))
2699 }
2700
2701 pub fn binding_fate(
2713 self,
2714 debt: ClosureDebt,
2715 event: Event,
2716 ) -> Result<CursorFateSuccessor, ClosureState> {
2717 let original = owed(debt, StoredEdge::ParticipantCursorProgress(self));
2718 let Self::Marker(value) = self else {
2719 return Err(original);
2720 };
2721 let EventKind::BindingFateObserved {
2722 participant_id,
2723 binding_epoch,
2724 ..
2725 } = event.0
2726 else {
2727 return Err(original);
2728 };
2729 if participant_id != value.participant_id || binding_epoch != value.binding_epoch {
2730 return Err(original);
2731 }
2732 Ok(CursorFateSuccessor::DetachedCredentialRecovery(
2733 DetachedCredentialRecovery {
2734 conversation_id: value.conversation_id,
2735 participant_id: value.participant_id,
2736 marker_delivery_seq: value.marker_delivery_seq,
2737 prior_binding_epoch: value.binding_epoch,
2738 },
2739 ))
2740 }
2741
2742 pub const fn retarget(
2749 self,
2750 new_binding_epoch: BindingEpoch,
2751 episode_churn_used: u64,
2752 delta_cycles: u64,
2753 episode_churn_limit: u64,
2754 ) -> Result<Self, (Self, DetachedAttachRefusal)> {
2755 let Self::Continuous(value) = self else {
2756 return Err((self, DetachedAttachRefusal::DeliveredMarkerAwaitingAck));
2757 };
2758 if !charged_churn_fits(episode_churn_used, delta_cycles, episode_churn_limit) {
2759 return Err((self, DetachedAttachRefusal::EpisodeChurnLimit));
2760 }
2761 if !is_next_generation(value.binding_epoch, new_binding_epoch) {
2762 return Err((self, DetachedAttachRefusal::StaleAuthority));
2763 }
2764 Ok(Self::Continuous(CursorProgressContinuous {
2765 binding_epoch: new_binding_epoch,
2766 ..value
2767 }))
2768 }
2769
2770 pub fn leave(
2777 self,
2778 debt: ClosureDebt,
2779 event: Event,
2780 successor: DebtCompletion,
2781 ) -> Result<ClosureState, ClosureState> {
2782 let original = owed(debt, StoredEdge::ParticipantCursorProgress(self));
2783 let EventKind::LeaveCommitted {
2784 participant_id,
2785 authority: LeaveAuthority::Live(binding_epoch),
2786 ..
2787 } = event.0
2788 else {
2789 return Err(original);
2790 };
2791 if participant_id != self.participant_id() || binding_epoch != self.binding_epoch() {
2792 return Err(original);
2793 }
2794 Ok(successor.into_state())
2795 }
2796}
2797
2798pub(super) struct FencedAttachRecordRefusal {
2801 record: ValidatedMarkerRecord,
2802}
2803
2804impl core::fmt::Debug for FencedAttachRecordRefusal {
2805 fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
2806 formatter.write_str("FencedAttachRecordRefusal { record: <linear> }")
2807 }
2808}
2809
2810impl FencedAttachRecordRefusal {
2811 pub(super) const fn into_record(self) -> ValidatedMarkerRecord {
2813 self.record
2814 }
2815}
2816
2817impl DetachedCredentialRecovery {
2818 #[must_use]
2820 pub const fn validate_leave_claim(
2821 &self,
2822 participant_id: ParticipantId,
2823 actual_charge: ResourceVector,
2824 remaining_k: ResourceVector,
2825 exit_claims: u64,
2826 ) -> Option<KClaimBackedDetachedLeave> {
2827 validate_detached_claim(
2828 self.participant_id,
2829 DetachedClaimTarget::CredentialRecovery {
2830 marker_delivery_seq: self.marker_delivery_seq,
2831 binding_epoch: self.prior_binding_epoch,
2832 },
2833 participant_id,
2834 actual_charge,
2835 remaining_k,
2836 exit_claims,
2837 )
2838 }
2839
2840 pub(super) fn fenced_attach(
2847 self,
2848 record_authority: ValidatedMarkerRecord,
2849 debt: ClosureDebt,
2850 event: Event,
2851 successor: DebtCompletion,
2852 ) -> Result<FencedAttachCommit, Box<FencedAttachRecordRefusal>> {
2853 let Some(proof) =
2854 FencedAttachCommit::restore_validated(self, &record_authority, debt, event, successor)
2855 else {
2856 return Err(Box::new(FencedAttachRecordRefusal {
2857 record: record_authority,
2858 }));
2859 };
2860 record_authority.consume();
2861 Ok(proof)
2862 }
2863
2864 pub fn detached_leave(
2871 self,
2872 debt: ClosureDebt,
2873 event: Event,
2874 evidence: KClaimBackedDetachedLeave,
2875 successor: DebtCompletion,
2876 ) -> Result<ClosureState, ClosureState> {
2877 let original = owed(debt, StoredEdge::DetachedCredentialRecovery(self));
2878 let EventKind::LeaveCommitted {
2879 participant_id,
2880 authority: LeaveAuthority::Detached,
2881 ..
2882 } = event.0
2883 else {
2884 return Err(original);
2885 };
2886 let target = DetachedClaimTarget::CredentialRecovery {
2887 marker_delivery_seq: self.marker_delivery_seq,
2888 binding_epoch: self.prior_binding_epoch,
2889 };
2890 if participant_id != self.participant_id
2891 || evidence.participant_id != self.participant_id
2892 || evidence.target != target
2893 {
2894 return Err(original);
2895 }
2896 Ok(successor.into_state())
2897 }
2898
2899 #[must_use]
2901 pub const fn ordinary_attach_refusal(self) -> DetachedAttachRefusal {
2902 let _ = self;
2903 DetachedAttachRefusal::RecoveryFence
2904 }
2905
2906 #[must_use]
2908 pub const fn marker_attach_refusal(
2909 self,
2910 presented_marker: DeliverySeq,
2911 ) -> Option<DetachedAttachRefusal> {
2912 if presented_marker == self.marker_delivery_seq {
2913 None
2914 } else {
2915 Some(DetachedAttachRefusal::MarkerMismatch)
2916 }
2917 }
2918
2919 #[must_use]
2921 pub const fn authority_superseded(self) -> (Self, DetachedAttachRefusal) {
2922 (self, DetachedAttachRefusal::StaleAuthority)
2923 }
2924
2925 pub const fn unrelated_event(
2933 self,
2934 debt: ClosureDebt,
2935 event: Event,
2936 resulting_debt: Option<ClosureDebt>,
2937 ) -> Result<ClosureState, ClosureState> {
2938 unrelated_detached_event(
2939 StoredEdge::DetachedCredentialRecovery(self),
2940 self.participant_id,
2941 debt,
2942 event,
2943 resulting_debt,
2944 )
2945 }
2946}
2947
2948mod sealed {
2949 pub trait Sealed {}
2950}
2951
2952pub trait LeaveOnlyEdge: sealed::Sealed + Sized + Copy {
2954 fn participant_id(self) -> ParticipantId;
2956
2957 fn validate_leave_claim(
2960 &self,
2961 participant_id: ParticipantId,
2962 actual_charge: ResourceVector,
2963 remaining_k: ResourceVector,
2964 exit_claims: u64,
2965 ) -> Option<KClaimBackedDetachedLeave>;
2966
2967 fn leave(
2974 self,
2975 debt: ClosureDebt,
2976 event: Event,
2977 evidence: KClaimBackedDetachedLeave,
2978 successor: DebtCompletion,
2979 ) -> Result<ClosureState, ClosureState>;
2980
2981 fn repeat_fate(self, event: Event) -> Result<Self, Self>;
2988
2989 fn authority_superseded(self) -> (Self, DetachedAttachRefusal) {
2991 (self, DetachedAttachRefusal::StaleAuthority)
2992 }
2993
2994 fn binding_required_refusal(self) -> DetachedAttachRefusal {
2996 let _ = self;
2997 DetachedAttachRefusal::NoBinding
2998 }
2999
3000 fn unrelated_event(
3008 self,
3009 debt: ClosureDebt,
3010 event: Event,
3011 resulting_debt: Option<ClosureDebt>,
3012 ) -> Result<ClosureState, ClosureState>;
3013}
3014
3015impl sealed::Sealed for DetachedMarkerRelease {}
3016impl sealed::Sealed for DetachedCursorRelease {}
3017
3018impl LeaveOnlyEdge for DetachedMarkerRelease {
3019 fn participant_id(self) -> ParticipantId {
3020 self.participant_id
3021 }
3022
3023 fn validate_leave_claim(
3024 &self,
3025 participant_id: ParticipantId,
3026 actual_charge: ResourceVector,
3027 remaining_k: ResourceVector,
3028 exit_claims: u64,
3029 ) -> Option<KClaimBackedDetachedLeave> {
3030 validate_detached_claim(
3031 self.participant_id,
3032 DetachedClaimTarget::MarkerRelease {
3033 marker_delivery_seq: self.marker_delivery_seq,
3034 binding_epoch: self.last_dead_binding_epoch,
3035 },
3036 participant_id,
3037 actual_charge,
3038 remaining_k,
3039 exit_claims,
3040 )
3041 }
3042
3043 fn leave(
3044 self,
3045 debt: ClosureDebt,
3046 event: Event,
3047 evidence: KClaimBackedDetachedLeave,
3048 successor: DebtCompletion,
3049 ) -> Result<ClosureState, ClosureState> {
3050 let original = owed(debt, StoredEdge::DetachedMarkerRelease(self));
3051 let EventKind::LeaveCommitted {
3052 participant_id,
3053 authority: LeaveAuthority::Detached,
3054 ..
3055 } = event.0
3056 else {
3057 return Err(original);
3058 };
3059 let target = DetachedClaimTarget::MarkerRelease {
3060 marker_delivery_seq: self.marker_delivery_seq,
3061 binding_epoch: self.last_dead_binding_epoch,
3062 };
3063 if participant_id != self.participant_id
3064 || evidence.participant_id != self.participant_id
3065 || evidence.target != target
3066 {
3067 return Err(original);
3068 }
3069 Ok(successor.into_state())
3070 }
3071
3072 fn repeat_fate(self, event: Event) -> Result<Self, Self> {
3073 let EventKind::BindingFateObserved {
3074 participant_id,
3075 binding_epoch,
3076 ..
3077 } = event.0
3078 else {
3079 return Err(self);
3080 };
3081 if participant_id == self.participant_id && binding_epoch == self.last_dead_binding_epoch {
3082 Ok(self)
3083 } else {
3084 Err(self)
3085 }
3086 }
3087
3088 fn unrelated_event(
3089 self,
3090 debt: ClosureDebt,
3091 event: Event,
3092 resulting_debt: Option<ClosureDebt>,
3093 ) -> Result<ClosureState, ClosureState> {
3094 unrelated_detached_event(
3095 StoredEdge::DetachedMarkerRelease(self),
3096 self.participant_id,
3097 debt,
3098 event,
3099 resulting_debt,
3100 )
3101 }
3102}
3103
3104impl DetachedMarkerRelease {
3105 #[must_use]
3107 pub const fn ordinary_attach_refusal(self) -> DetachedAttachRefusal {
3108 let _ = self;
3109 DetachedAttachRefusal::RecoveryFence
3110 }
3111
3112 #[must_use]
3115 pub const fn marker_attach_refusal(
3116 self,
3117 presented_marker: DeliverySeq,
3118 ) -> DetachedAttachRefusal {
3119 if presented_marker == self.marker_delivery_seq {
3120 DetachedAttachRefusal::MarkerNotDelivered
3121 } else {
3122 DetachedAttachRefusal::MarkerMismatch
3123 }
3124 }
3125}
3126
3127impl LeaveOnlyEdge for DetachedCursorRelease {
3128 fn participant_id(self) -> ParticipantId {
3129 self.participant_id
3130 }
3131
3132 fn validate_leave_claim(
3133 &self,
3134 participant_id: ParticipantId,
3135 actual_charge: ResourceVector,
3136 remaining_k: ResourceVector,
3137 exit_claims: u64,
3138 ) -> Option<KClaimBackedDetachedLeave> {
3139 validate_detached_claim(
3140 self.participant_id,
3141 DetachedClaimTarget::CursorRelease {
3142 binding_epoch: self.last_dead_binding_epoch,
3143 },
3144 participant_id,
3145 actual_charge,
3146 remaining_k,
3147 exit_claims,
3148 )
3149 }
3150
3151 fn leave(
3152 self,
3153 debt: ClosureDebt,
3154 event: Event,
3155 evidence: KClaimBackedDetachedLeave,
3156 successor: DebtCompletion,
3157 ) -> Result<ClosureState, ClosureState> {
3158 let original = owed(debt, StoredEdge::DetachedCursorRelease(self));
3159 let EventKind::LeaveCommitted {
3160 participant_id,
3161 authority: LeaveAuthority::Detached,
3162 ..
3163 } = event.0
3164 else {
3165 return Err(original);
3166 };
3167 let target = DetachedClaimTarget::CursorRelease {
3168 binding_epoch: self.last_dead_binding_epoch,
3169 };
3170 if participant_id != self.participant_id
3171 || evidence.participant_id != self.participant_id
3172 || evidence.target != target
3173 {
3174 return Err(original);
3175 }
3176 Ok(successor.into_state())
3177 }
3178
3179 fn repeat_fate(self, event: Event) -> Result<Self, Self> {
3180 let EventKind::BindingFateObserved {
3181 participant_id,
3182 binding_epoch,
3183 ..
3184 } = event.0
3185 else {
3186 return Err(self);
3187 };
3188 if participant_id == self.participant_id && binding_epoch == self.last_dead_binding_epoch {
3189 Ok(self)
3190 } else {
3191 Err(self)
3192 }
3193 }
3194
3195 fn unrelated_event(
3196 self,
3197 debt: ClosureDebt,
3198 event: Event,
3199 resulting_debt: Option<ClosureDebt>,
3200 ) -> Result<ClosureState, ClosureState> {
3201 unrelated_detached_event(
3202 StoredEdge::DetachedCursorRelease(self),
3203 self.participant_id,
3204 debt,
3205 event,
3206 resulting_debt,
3207 )
3208 }
3209}
3210
3211impl DetachedCursorRelease {
3212 #[must_use]
3214 pub const fn ordinary_attach_refusal(self) -> DetachedAttachRefusal {
3215 let _ = self;
3216 DetachedAttachRefusal::RecoveryFence
3217 }
3218
3219 #[must_use]
3221 pub const fn marker_attach_refusal(self) -> DetachedAttachRefusal {
3222 let _ = self;
3223 DetachedAttachRefusal::MarkerMismatch
3224 }
3225}
3226
3227const fn owed(debt: ClosureDebt, edge: StoredEdge) -> ClosureState {
3228 ClosureState::Owed { debt, edge }
3229}
3230
3231const fn preserve_or_clear(resulting_debt: Option<ClosureDebt>, edge: StoredEdge) -> ClosureState {
3232 match resulting_debt {
3233 Some(debt) => owed(debt, edge),
3234 None => ClosureState::Clear,
3235 }
3236}
3237
3238const fn projection_completion_boundary(
3239 edge: ObserverProjection,
3240 event: Event,
3241) -> Option<DeliverySeq> {
3242 let EventKind::ProjectionCompleted { through_seq } = event.0 else {
3243 return None;
3244 };
3245 if through_seq == edge.through_seq {
3246 Some(through_seq)
3247 } else {
3248 None
3249 }
3250}
3251
3252const fn physical_completion_floor(edge: PhysicalCompaction, event: Event) -> Option<DeliverySeq> {
3253 match event.0 {
3254 EventKind::CompactionCompleted {
3255 from_floor,
3256 through_seq,
3257 resulting_floor,
3258 } if from_floor == edge.from_floor
3259 && through_seq == edge.through_seq
3260 && resulting_floor > edge.through_seq =>
3261 {
3262 Some(resulting_floor)
3263 }
3264 _ => None,
3265 }
3266}
3267
3268const fn progress_event_floor(event: Event) -> Option<DeliverySeq> {
3269 match event.0 {
3270 EventKind::CursorProgressed {
3271 resulting_floor, ..
3272 }
3273 | EventKind::BindingFateObserved {
3274 resulting_floor, ..
3275 }
3276 | EventKind::LeaveCommitted {
3277 resulting_floor, ..
3278 } => Some(resulting_floor),
3279 _ => None,
3280 }
3281}
3282
3283const fn cursor_event_boundary(event: Event) -> DeliverySeq {
3284 match event.0 {
3285 EventKind::CursorProgressed {
3286 progress: CursorProgressEvent::Normal { through_seq, .. },
3287 ..
3288 } => through_seq,
3289 _ => 0,
3290 }
3291}
3292
3293fn greater_ack_matches(edge: ParticipantCursorProgress, event: Event) -> bool {
3294 let EventKind::CursorProgressed {
3295 participant_id,
3296 binding_epoch,
3297 progress:
3298 CursorProgressEvent::Normal {
3299 previous_cursor,
3300 through_seq,
3301 },
3302 ..
3303 } = event.0
3304 else {
3305 return false;
3306 };
3307 participant_id == edge.participant_id()
3308 && binding_epoch == edge.binding_epoch()
3309 && previous_cursor < edge.through_seq()
3310 && through_seq > edge.through_seq()
3311}
3312
3313const fn strict_edge_matches_boundary(edge: StoredEdge, boundary: DeliverySeq) -> bool {
3314 match edge {
3315 StoredEdge::ObserverProjection(value) => value.through_seq == boundary,
3316 StoredEdge::PhysicalCompaction(value) => value.through_seq == boundary,
3317 StoredEdge::MarkerDelivery(value) => value.marker_delivery_seq == boundary,
3318 StoredEdge::ParticipantCursorProgress(value) => value.through_seq() == boundary,
3319 StoredEdge::DetachedCredentialRecovery(_) => false,
3320 StoredEdge::DetachedMarkerRelease(value) => value.marker_delivery_seq == boundary,
3321 StoredEdge::DetachedCursorRelease(_) => true,
3325 }
3326}
3327
3328const fn charged_churn_fits(used: u64, delta: u64, limit: u64) -> bool {
3329 delta > 0 && widen_u64(used) + widen_u64(delta) <= widen_u64(limit)
3330}
3331
3332#[allow(clippy::cast_lossless)]
3333const fn widen_u64(value: u64) -> u128 {
3334 value as u128
3335}
3336
3337const fn is_next_generation(old: BindingEpoch, new: BindingEpoch) -> bool {
3338 match old.capability_generation.get().checked_add(1) {
3339 Some(expected) => new.capability_generation.get() == expected,
3340 None => false,
3341 }
3342}
3343
3344const fn validate_detached_claim(
3345 owner: ParticipantId,
3346 target: DetachedClaimTarget,
3347 participant_id: ParticipantId,
3348 actual_charge: ResourceVector,
3349 remaining_k: ResourceVector,
3350 exit_claims: u64,
3351) -> Option<KClaimBackedDetachedLeave> {
3352 if participant_id != owner
3353 || actual_charge.entries == 0
3354 || actual_charge.bytes == 0
3355 || actual_charge.entries > remaining_k.entries
3356 || actual_charge.bytes > remaining_k.bytes
3357 || exit_claims == 0
3358 {
3359 return None;
3360 }
3361 Some(KClaimBackedDetachedLeave {
3362 participant_id,
3363 target,
3364 actual_charge,
3365 })
3366}
3367
3368const fn event_participant(event: Event) -> Option<ParticipantId> {
3369 match event.0 {
3370 EventKind::MarkerDelivered { participant_id, .. }
3371 | EventKind::CursorProgressed { participant_id, .. }
3372 | EventKind::BindingFateObserved { participant_id, .. }
3373 | EventKind::LeaveCommitted { participant_id, .. }
3374 | EventKind::FencedRecoveryCommitted { participant_id, .. } => Some(participant_id),
3375 EventKind::ProjectionCompleted { .. }
3376 | EventKind::CompactionCompleted { .. }
3377 | EventKind::MarkerAppended { .. } => None,
3378 }
3379}
3380
3381const fn unrelated_detached_event(
3382 edge: StoredEdge,
3383 owner: ParticipantId,
3384 debt: ClosureDebt,
3385 event: Event,
3386 resulting_debt: Option<ClosureDebt>,
3387) -> Result<ClosureState, ClosureState> {
3388 let original = owed(debt, edge);
3389 let Some(participant_id) = event_participant(event) else {
3390 return Err(original);
3391 };
3392 if participant_id == owner {
3393 return Err(original);
3394 }
3395 Ok(preserve_or_clear(resulting_debt, edge))
3396}