1use alloc::boxed::Box;
2
3use crate::algebra::{ResourceDimension, ResourceVector, WideResourceVector};
4use crate::wire::{
5 ClosureCapacityReason, ClosureCheckedEnvelope, ClosureRefusalReason, ClosureSnapshot,
6 MarkerClosureCapacityExceeded, ParticipantCursorProgressEdge, RepaymentEdge,
7};
8
9use super::{ClosureState, StoredEdge};
10
11#[derive(Clone, Copy, Debug, PartialEq, Eq)]
13pub enum ClosureAccountingError {
14 ZeroChurnLimit,
16 ChurnUsedExceedsLimit {
18 used: u64,
20 limit: u64,
22 },
23 MarkerAnchorsExceedCredits {
25 anchors: u64,
27 credits: u64,
29 },
30 BaselineExceedsCapacity {
32 dimension: ResourceDimension,
34 },
35 ClearStateOwnsEdgeResources,
37 DebtOutsideWireDomain {
39 dimension: ResourceDimension,
41 },
42}
43
44#[derive(Clone, Copy, Debug, PartialEq, Eq)]
50pub struct ClosureAccounting {
51 state: ClosureState,
52 marker_capacity_credits: u64,
53 marker_anchors: u64,
54 edge_sequence_claims: u64,
55 edge_order_position_claims: u64,
56 edge_k_remaining: ResourceVector,
57 baseline: WideResourceVector,
58 configured_cap: ResourceVector,
59 episode_churn_used: u64,
60 episode_churn_limit: u64,
61}
62
63impl ClosureAccounting {
64 #[allow(clippy::too_many_arguments)]
76 pub fn try_new(
77 state: ClosureState,
78 marker_capacity_credits: u64,
79 marker_anchors: u64,
80 edge_sequence_claims: u64,
81 edge_order_position_claims: u64,
82 edge_k_remaining: ResourceVector,
83 baseline: WideResourceVector,
84 configured_cap: ResourceVector,
85 episode_churn_used: u64,
86 episode_churn_limit: u64,
87 ) -> Result<Self, ClosureAccountingError> {
88 if episode_churn_limit == 0 {
89 return Err(ClosureAccountingError::ZeroChurnLimit);
90 }
91 if episode_churn_used > episode_churn_limit {
92 return Err(ClosureAccountingError::ChurnUsedExceedsLimit {
93 used: episode_churn_used,
94 limit: episode_churn_limit,
95 });
96 }
97 if marker_anchors > marker_capacity_credits {
98 return Err(ClosureAccountingError::MarkerAnchorsExceedCredits {
99 anchors: marker_anchors,
100 credits: marker_capacity_credits,
101 });
102 }
103 if baseline.entries > u128::from(configured_cap.entries) {
104 return Err(ClosureAccountingError::BaselineExceedsCapacity {
105 dimension: ResourceDimension::Entries,
106 });
107 }
108 if baseline.bytes > u128::from(configured_cap.bytes) {
109 return Err(ClosureAccountingError::BaselineExceedsCapacity {
110 dimension: ResourceDimension::Bytes,
111 });
112 }
113 match state {
114 ClosureState::Clear => {
115 if edge_sequence_claims != 0
116 || edge_order_position_claims != 0
117 || edge_k_remaining != ResourceVector::default()
118 {
119 return Err(ClosureAccountingError::ClearStateOwnsEdgeResources);
120 }
121 }
122 ClosureState::Owed { debt, .. } => {
123 let value = debt.value();
124 if u64::try_from(value.entries).is_err() {
125 return Err(ClosureAccountingError::DebtOutsideWireDomain {
126 dimension: ResourceDimension::Entries,
127 });
128 }
129 if u64::try_from(value.bytes).is_err() {
130 return Err(ClosureAccountingError::DebtOutsideWireDomain {
131 dimension: ResourceDimension::Bytes,
132 });
133 }
134 }
135 }
136 Ok(Self {
137 state,
138 marker_capacity_credits,
139 marker_anchors,
140 edge_sequence_claims,
141 edge_order_position_claims,
142 edge_k_remaining,
143 baseline,
144 configured_cap,
145 episode_churn_used,
146 episode_churn_limit,
147 })
148 }
149
150 #[must_use]
152 pub const fn state(self) -> ClosureState {
153 self.state
154 }
155
156 #[must_use]
158 pub const fn marker_capacity_credits(self) -> u64 {
159 self.marker_capacity_credits
160 }
161
162 #[must_use]
164 pub const fn marker_anchors(self) -> u64 {
165 self.marker_anchors
166 }
167
168 #[must_use]
170 pub const fn edge_sequence_claims(self) -> u64 {
171 self.edge_sequence_claims
172 }
173
174 #[must_use]
176 pub const fn edge_order_position_claims(self) -> u64 {
177 self.edge_order_position_claims
178 }
179
180 #[must_use]
182 pub const fn edge_k_remaining(self) -> ResourceVector {
183 self.edge_k_remaining
184 }
185
186 #[must_use]
188 pub const fn baseline(self) -> WideResourceVector {
189 self.baseline
190 }
191
192 #[must_use]
194 pub const fn configured_cap(self) -> ResourceVector {
195 self.configured_cap
196 }
197
198 #[must_use]
200 pub const fn episode_churn_used(self) -> u64 {
201 self.episode_churn_used
202 }
203
204 #[must_use]
206 pub const fn episode_churn_limit(self) -> u64 {
207 self.episode_churn_limit
208 }
209
210 fn snapshot(self, delta_cycles: u64) -> ClosureSnapshot {
211 let (debt, repayment_edge) = closure_state_wire(self.state);
212 ClosureSnapshot {
213 marker_capacity_credits: self.marker_capacity_credits,
214 marker_anchors: self.marker_anchors,
215 entry_debt: debt.entries,
216 byte_debt: debt.bytes,
217 repayment_edge,
218 edge_sequence_claims: self.edge_sequence_claims,
219 edge_order_position_claims: self.edge_order_position_claims,
220 edge_k_remaining: self.edge_k_remaining,
221 k_headroom: WideResourceVector::new(
222 u128::from(self.configured_cap.entries) - self.baseline.entries,
223 u128::from(self.configured_cap.bytes) - self.baseline.bytes,
224 ),
225 episode_churn_used: self.episode_churn_used,
226 delta_cycles,
227 episode_churn_limit: self.episode_churn_limit,
228 }
229 }
230}
231
232#[derive(Clone, Copy, Debug, PartialEq, Eq)]
234pub enum RequiredCapacityPlanError {
235 EmptySuccessorSet,
237 ArithmeticOverflow {
239 dimension: ResourceDimension,
241 },
242}
243
244#[derive(Clone, Copy, Debug, PartialEq, Eq)]
246pub struct RequiredCapacityPlan {
247 maximum: WideResourceVector,
248}
249
250impl RequiredCapacityPlan {
251 pub fn from_successors(
258 successors: &[WideResourceVector],
259 ) -> Result<Self, RequiredCapacityPlanError> {
260 let Some(first) = successors.first().copied() else {
261 return Err(RequiredCapacityPlanError::EmptySuccessorSet);
262 };
263 let maximum = successors.iter().skip(1).fold(first, |maximum, state| {
264 WideResourceVector::new(
265 maximum.entries.max(state.entries),
266 maximum.bytes.max(state.bytes),
267 )
268 });
269 Ok(Self { maximum })
270 }
271
272 pub fn ordinary(
279 resulting_baseline: WideResourceVector,
280 mandatory_bound: ResourceVector,
281 full_recovery_claim: ResourceVector,
282 ) -> Result<Self, RequiredCapacityPlanError> {
283 let Some(entries_with_q) = resulting_baseline
284 .entries
285 .checked_add(u128::from(mandatory_bound.entries))
286 else {
287 return Err(RequiredCapacityPlanError::ArithmeticOverflow {
288 dimension: ResourceDimension::Entries,
289 });
290 };
291 let Some(entries) = entries_with_q.checked_add(u128::from(full_recovery_claim.entries))
292 else {
293 return Err(RequiredCapacityPlanError::ArithmeticOverflow {
294 dimension: ResourceDimension::Entries,
295 });
296 };
297 let Some(bytes_with_q) = resulting_baseline
298 .bytes
299 .checked_add(u128::from(mandatory_bound.bytes))
300 else {
301 return Err(RequiredCapacityPlanError::ArithmeticOverflow {
302 dimension: ResourceDimension::Bytes,
303 });
304 };
305 let Some(bytes) = bytes_with_q.checked_add(u128::from(full_recovery_claim.bytes)) else {
306 return Err(RequiredCapacityPlanError::ArithmeticOverflow {
307 dimension: ResourceDimension::Bytes,
308 });
309 };
310 Ok(Self {
311 maximum: WideResourceVector::new(entries, bytes),
312 })
313 }
314
315 #[must_use]
317 pub const fn maximum(self) -> WideResourceVector {
318 self.maximum
319 }
320}
321
322#[derive(Clone, Copy, Debug, PartialEq, Eq)]
324pub struct RecoveryFencePermit {
325 accounting: ClosureAccounting,
326}
327
328impl RecoveryFencePermit {
329 #[must_use]
331 pub const fn accounting(self) -> ClosureAccounting {
332 self.accounting
333 }
334}
335
336#[derive(Clone, Debug, PartialEq, Eq)]
338pub enum RecoveryFenceDecision {
339 Eligible(RecoveryFencePermit),
341 Respond(Box<MarkerClosureCapacityExceeded>),
343}
344
345#[must_use]
347pub fn check_recovery_fence(
348 request: &ClosureCheckedEnvelope,
349 accounting: ClosureAccounting,
350 recovery_fence: bool,
351) -> RecoveryFenceDecision {
352 if recovery_fence {
353 return RecoveryFenceDecision::Respond(Box::new(closure_refusal(
354 request.clone(),
355 accounting,
356 0,
357 ClosureRefusalReason::RecoveryFence,
358 )));
359 }
360 RecoveryFenceDecision::Eligible(RecoveryFencePermit { accounting })
361}
362
363#[derive(Clone, Copy, Debug, PartialEq, Eq)]
365pub struct RemainingClosurePermit {
366 accounting: ClosureAccounting,
367 required_capacity: RequiredCapacityPlan,
368 delta_cycles: u64,
369}
370
371impl RemainingClosurePermit {
372 #[must_use]
374 pub const fn accounting(self) -> ClosureAccounting {
375 self.accounting
376 }
377
378 #[must_use]
380 pub const fn required_capacity(self) -> RequiredCapacityPlan {
381 self.required_capacity
382 }
383
384 #[must_use]
386 pub const fn delta_cycles(self) -> u64 {
387 self.delta_cycles
388 }
389}
390
391#[derive(Clone, Debug, PartialEq, Eq)]
393pub enum RemainingClosureDecision {
394 Eligible(Box<RemainingClosurePermit>),
396 Respond(Box<MarkerClosureCapacityExceeded>),
398}
399
400#[must_use]
405pub fn check_remaining_closure(
406 request: &ClosureCheckedEnvelope,
407 accounting: ClosureAccounting,
408 delivered_marker_awaiting_ack: bool,
409 delta_cycles: u64,
410 required_capacity: RequiredCapacityPlan,
411) -> RemainingClosureDecision {
412 if delivered_marker_awaiting_ack {
413 return RemainingClosureDecision::Respond(Box::new(closure_refusal(
414 request.clone(),
415 accounting,
416 0,
417 ClosureRefusalReason::DeliveredMarkerAwaitingAck,
418 )));
419 }
420 let resulting_cycles = u128::from(accounting.episode_churn_used) + u128::from(delta_cycles);
421 if resulting_cycles > u128::from(accounting.episode_churn_limit) {
422 return RemainingClosureDecision::Respond(Box::new(closure_refusal(
423 request.clone(),
424 accounting,
425 delta_cycles,
426 ClosureRefusalReason::EpisodeChurnLimit,
427 )));
428 }
429 let maximum = required_capacity.maximum;
430 if maximum.entries > u128::from(accounting.configured_cap.entries) {
431 return RemainingClosureDecision::Respond(Box::new(closure_refusal(
432 request.clone(),
433 accounting,
434 delta_cycles,
435 ClosureRefusalReason::Capacity(ClosureCapacityReason {
436 dimension: ResourceDimension::Entries,
437 required: maximum.entries,
438 limit: u128::from(accounting.configured_cap.entries),
439 }),
440 )));
441 }
442 if maximum.bytes > u128::from(accounting.configured_cap.bytes) {
443 return RemainingClosureDecision::Respond(Box::new(closure_refusal(
444 request.clone(),
445 accounting,
446 delta_cycles,
447 ClosureRefusalReason::Capacity(ClosureCapacityReason {
448 dimension: ResourceDimension::Bytes,
449 required: maximum.bytes,
450 limit: u128::from(accounting.configured_cap.bytes),
451 }),
452 )));
453 }
454 RemainingClosureDecision::Eligible(Box::new(RemainingClosurePermit {
455 accounting,
456 required_capacity,
457 delta_cycles,
458 }))
459}
460
461fn closure_refusal(
462 request: ClosureCheckedEnvelope,
463 accounting: ClosureAccounting,
464 delta_cycles: u64,
465 reason: ClosureRefusalReason,
466) -> MarkerClosureCapacityExceeded {
467 MarkerClosureCapacityExceeded {
468 request,
469 snapshot: accounting.snapshot(delta_cycles),
470 reason,
471 }
472}
473
474fn closure_state_wire(state: ClosureState) -> (ResourceVector, RepaymentEdge) {
475 match state {
476 ClosureState::Clear => (ResourceVector::default(), RepaymentEdge::None),
477 ClosureState::Owed { debt, edge } => {
478 let debt = debt.value();
479 let wire_debt = ResourceVector::new(
480 u64::try_from(debt.entries).map_or(u64::MAX, core::convert::identity),
481 u64::try_from(debt.bytes).map_or(u64::MAX, core::convert::identity),
482 );
483 (wire_debt, stored_edge_wire(edge))
484 }
485 }
486}
487
488const fn stored_edge_wire(edge: StoredEdge) -> RepaymentEdge {
489 match edge {
490 StoredEdge::ObserverProjection(edge) => RepaymentEdge::ObserverProjection {
491 through_seq: edge.through_seq(),
492 },
493 StoredEdge::PhysicalCompaction(edge) => RepaymentEdge::PhysicalCompaction {
494 from_floor: edge.from_floor(),
495 through_seq: edge.through_seq(),
496 },
497 StoredEdge::MarkerDelivery(edge) => RepaymentEdge::MarkerDelivery {
498 participant_id: edge.participant_id(),
499 binding_epoch: edge.binding_epoch(),
500 marker_delivery_seq: edge.marker_delivery_seq(),
501 },
502 StoredEdge::ParticipantCursorProgress(edge) => {
503 RepaymentEdge::ParticipantCursorProgress(ParticipantCursorProgressEdge {
504 participant_id: edge.participant_id(),
505 binding_epoch: edge.binding_epoch(),
506 through_seq: edge.through_seq(),
507 marker_delivery_seq: edge.marker_delivery_seq(),
508 })
509 }
510 StoredEdge::DetachedCredentialRecovery(edge) => RepaymentEdge::DetachedCredentialRecovery {
511 participant_id: edge.participant_id(),
512 marker_delivery_seq: edge.marker_delivery_seq(),
513 prior_binding_epoch: edge.prior_binding_epoch(),
514 },
515 StoredEdge::DetachedMarkerRelease(edge) => RepaymentEdge::DetachedMarkerRelease {
516 participant_id: edge.participant_id(),
517 marker_delivery_seq: edge.marker_delivery_seq(),
518 last_dead_binding_epoch: edge.last_dead_binding_epoch(),
519 },
520 StoredEdge::DetachedCursorRelease(edge) => RepaymentEdge::DetachedCursorRelease {
521 participant_id: edge.participant_id(),
522 last_dead_binding_epoch: edge.last_dead_binding_epoch(),
523 },
524 }
525}