1use serde::{Deserialize, Serialize};
2use std::{
3 collections::{BTreeMap, BTreeSet},
4 sync::{
5 Arc, Mutex,
6 atomic::{AtomicU64, Ordering},
7 },
8 time::{SystemTime, UNIX_EPOCH},
9};
10
11#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
12#[serde(rename_all = "camelCase", deny_unknown_fields)]
13pub struct CallPolicyDeclaration {
14 pub max_attempts: u32,
15 #[serde(default, skip_serializing_if = "Option::is_none")]
16 pub circuit_breaker: Option<CallPolicyCircuitBreaker>,
17 #[serde(default, skip_serializing_if = "Option::is_none")]
18 pub concurrency: Option<CallPolicyConcurrency>,
19 #[serde(default, skip_serializing_if = "Option::is_none")]
20 pub overload: Option<CallPolicyOverload>,
21 #[serde(default, skip_serializing_if = "Option::is_none")]
22 pub fallback: Option<CallPolicyFallback>,
23}
24
25impl Default for CallPolicyDeclaration {
26 fn default() -> Self {
27 Self {
28 max_attempts: 2,
29 circuit_breaker: None,
30 concurrency: None,
31 overload: None,
32 fallback: None,
33 }
34 }
35}
36
37impl CallPolicyDeclaration {
38 #[must_use]
39 pub fn validate(&self, retry_safe: bool) -> Vec<CallPolicyValidationIssue> {
40 let mut issues = Vec::new();
41 if self.max_attempts == 0 {
42 issues.push(CallPolicyValidationIssue::new(
43 "maxAttempts",
44 "max_attempts_invalid",
45 ));
46 } else if self.max_attempts > 1 && !retry_safe {
47 issues.push(CallPolicyValidationIssue::new(
48 "maxAttempts",
49 "unsafe_retry_policy",
50 ));
51 }
52 if let Some(circuit) = &self.circuit_breaker {
53 if circuit.failure_threshold == 0 {
54 issues.push(CallPolicyValidationIssue::new(
55 "circuitBreaker.failureThreshold",
56 "circuit_failure_threshold_invalid",
57 ));
58 }
59 if circuit.open_for_ms == 0 {
60 issues.push(CallPolicyValidationIssue::new(
61 "circuitBreaker.openForMs",
62 "circuit_open_duration_invalid",
63 ));
64 }
65 if circuit.half_open_max_calls == 0 {
66 issues.push(CallPolicyValidationIssue::new(
67 "circuitBreaker.halfOpenMaxCalls",
68 "circuit_half_open_limit_invalid",
69 ));
70 }
71 }
72 if self
73 .concurrency
74 .as_ref()
75 .is_some_and(|value| value.max_in_flight == 0)
76 {
77 issues.push(CallPolicyValidationIssue::new(
78 "concurrency.maxInFlight",
79 "concurrency_limit_invalid",
80 ));
81 }
82 if self
83 .overload
84 .as_ref()
85 .is_some_and(|value| value.max_in_flight == 0)
86 {
87 issues.push(CallPolicyValidationIssue::new(
88 "overload.maxInFlight",
89 "overload_limit_invalid",
90 ));
91 }
92 if let Some(fallback) = &self.fallback {
93 if fallback.handler.trim().is_empty() {
94 issues.push(CallPolicyValidationIssue::new(
95 "fallback.handler",
96 "fallback_handler_required",
97 ));
98 }
99 let unique = fallback.on.iter().copied().collect::<BTreeSet<_>>();
100 if unique.len() != fallback.on.len() {
101 issues.push(CallPolicyValidationIssue::new(
102 "fallback.on",
103 "fallback_trigger_duplicate",
104 ));
105 }
106 if fallback.on.is_empty() {
107 issues.push(CallPolicyValidationIssue::new(
108 "fallback.on",
109 "fallback_trigger_required",
110 ));
111 }
112 }
113 issues
114 }
115
116 #[must_use]
117 pub fn fallback_for(&self, failure: CallPolicyFailure) -> Option<&CallPolicyFallback> {
118 self.fallback
119 .as_ref()
120 .filter(|fallback| fallback.on.contains(&failure))
121 }
122}
123
124#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
125#[serde(rename_all = "camelCase", deny_unknown_fields)]
126pub struct CallPolicyCircuitBreaker {
127 pub failure_threshold: u32,
128 pub open_for_ms: u64,
129 pub half_open_max_calls: u32,
130}
131
132#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
133#[serde(rename_all = "camelCase", deny_unknown_fields)]
134pub struct CallPolicyConcurrency {
135 pub max_in_flight: u32,
136}
137
138#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
139#[serde(rename_all = "camelCase", deny_unknown_fields)]
140pub struct CallPolicyOverload {
141 pub max_in_flight: u32,
142}
143
144#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
145#[serde(rename_all = "camelCase", deny_unknown_fields)]
146pub struct CallPolicyFallback {
147 pub handler: String,
148 pub on: Vec<CallPolicyFailure>,
149}
150
151#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
152#[serde(rename_all = "snake_case")]
153pub enum CallPolicyFailure {
154 CircuitOpen,
155 BulkheadSaturated,
156 OverloadRejected,
157 DeadlineExpired,
158 RetryableFailure,
159 NonRetryableFailure,
160 TransportFailure,
161}
162
163#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
164#[serde(rename_all = "snake_case")]
165pub enum CallPolicyEvent {
166 CircuitOpen,
167 CircuitHalfOpen,
168 CircuitOpened,
169 CircuitRecovered,
170 BulkheadSaturated,
171 OverloadRejected,
172 DeadlineExpired,
173 FallbackApplied,
174 RetryScheduled,
175 CallCompleted,
176 CallFailed,
177}
178
179#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
180#[serde(rename_all = "snake_case")]
181pub enum CallPolicyTerminalOutcome {
182 Completed,
183 Failed,
184 Rejected,
185 Fallback,
186}
187
188#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
189#[serde(rename_all = "camelCase")]
190pub struct CallPolicyEvidence {
191 pub events: Vec<CallPolicyEvent>,
192 pub attempts: u32,
193 pub terminal_outcome: CallPolicyTerminalOutcome,
194 #[serde(default, skip_serializing_if = "Option::is_none")]
195 pub fallback_handler: Option<String>,
196}
197
198impl CallPolicyEvent {
199 #[must_use]
200 pub const fn as_str(self) -> &'static str {
201 match self {
202 Self::CircuitOpen => "circuit_open",
203 Self::CircuitHalfOpen => "circuit_half_open",
204 Self::CircuitOpened => "circuit_opened",
205 Self::CircuitRecovered => "circuit_recovered",
206 Self::BulkheadSaturated => "bulkhead_saturated",
207 Self::OverloadRejected => "overload_rejected",
208 Self::DeadlineExpired => "deadline_expired",
209 Self::FallbackApplied => "fallback_applied",
210 Self::RetryScheduled => "retry_scheduled",
211 Self::CallCompleted => "call_completed",
212 Self::CallFailed => "call_failed",
213 }
214 }
215}
216
217#[derive(Debug, Clone, PartialEq, Eq)]
218pub struct CallPolicyValidationIssue {
219 pub path: String,
220 pub code: &'static str,
221}
222
223impl CallPolicyValidationIssue {
224 fn new(path: impl Into<String>, code: &'static str) -> Self {
225 Self {
226 path: path.into(),
227 code,
228 }
229 }
230}
231
232pub trait CallPolicyClock: Send + Sync {
233 fn now_ms(&self) -> u64;
234}
235
236#[derive(Debug, Default)]
237pub struct SystemCallPolicyClock;
238
239impl CallPolicyClock for SystemCallPolicyClock {
240 fn now_ms(&self) -> u64 {
241 SystemTime::now()
242 .duration_since(UNIX_EPOCH)
243 .unwrap_or_default()
244 .as_millis() as u64
245 }
246}
247
248#[derive(Debug)]
249pub struct ManualCallPolicyClock {
250 now_ms: AtomicU64,
251}
252
253impl ManualCallPolicyClock {
254 #[must_use]
255 pub const fn new(now_ms: u64) -> Self {
256 Self {
257 now_ms: AtomicU64::new(now_ms),
258 }
259 }
260
261 pub fn advance_ms(&self, duration_ms: u64) {
262 self.now_ms.fetch_add(duration_ms, Ordering::SeqCst);
263 }
264}
265
266impl CallPolicyClock for ManualCallPolicyClock {
267 fn now_ms(&self) -> u64 {
268 self.now_ms.load(Ordering::SeqCst)
269 }
270}
271
272#[derive(Clone)]
273pub struct CallPolicyRuntime {
274 inner: Arc<RuntimeInner>,
275}
276
277struct RuntimeInner {
278 clock: Arc<dyn CallPolicyClock>,
279 states: Mutex<BTreeMap<String, OperationState>>,
280}
281
282#[derive(Debug, Default)]
283struct OperationState {
284 consecutive_failures: u32,
285 circuit_open_until_ms: Option<u64>,
286 half_open_in_flight: u32,
287 caller_in_flight: u32,
288 receiver_in_flight: u32,
289}
290
291impl std::fmt::Debug for CallPolicyRuntime {
292 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
293 formatter
294 .debug_struct("CallPolicyRuntime")
295 .finish_non_exhaustive()
296 }
297}
298
299impl Default for CallPolicyRuntime {
300 fn default() -> Self {
301 Self::new(Arc::new(SystemCallPolicyClock))
302 }
303}
304
305impl CallPolicyRuntime {
306 #[must_use]
307 pub fn new(clock: Arc<dyn CallPolicyClock>) -> Self {
308 Self {
309 inner: Arc::new(RuntimeInner {
310 clock,
311 states: Mutex::new(BTreeMap::new()),
312 }),
313 }
314 }
315
316 #[must_use]
317 pub fn now_ms(&self) -> u64 {
318 self.inner.clock.now_ms()
319 }
320
321 pub fn begin_call(
322 &self,
323 operation_key: impl Into<String>,
324 policy: &CallPolicyDeclaration,
325 ) -> Result<CallPolicyPermit, CallPolicyEvent> {
326 let operation_key = operation_key.into();
327 let mut states = self
328 .inner
329 .states
330 .lock()
331 .expect("call policy state poisoned");
332 let state = states.entry(operation_key.clone()).or_default();
333 let now_ms = self.inner.clock.now_ms();
334 let mut half_open = false;
335 let mut events = Vec::new();
336 if let Some(circuit) = &policy.circuit_breaker {
337 if let Some(open_until) = state.circuit_open_until_ms {
338 if now_ms < open_until || state.half_open_in_flight >= circuit.half_open_max_calls {
339 return Err(CallPolicyEvent::CircuitOpen);
340 }
341 half_open = true;
342 state.half_open_in_flight += 1;
343 events.push(CallPolicyEvent::CircuitHalfOpen);
344 }
345 }
346 if policy
347 .concurrency
348 .as_ref()
349 .is_some_and(|limit| state.caller_in_flight >= limit.max_in_flight)
350 {
351 if half_open {
352 state.half_open_in_flight -= 1;
353 }
354 return Err(CallPolicyEvent::BulkheadSaturated);
355 }
356 state.caller_in_flight += 1;
357 drop(states);
358 Ok(CallPolicyPermit {
359 runtime: self.clone(),
360 operation_key,
361 policy: policy.clone(),
362 mode: PermitMode::Caller { half_open },
363 events,
364 released: false,
365 })
366 }
367
368 pub fn admit(
369 &self,
370 operation_key: impl Into<String>,
371 policy: &CallPolicyDeclaration,
372 ) -> Result<CallPolicyPermit, CallPolicyEvent> {
373 let operation_key = operation_key.into();
374 let mut states = self
375 .inner
376 .states
377 .lock()
378 .expect("call policy state poisoned");
379 let state = states.entry(operation_key.clone()).or_default();
380 if policy
381 .overload
382 .as_ref()
383 .is_some_and(|limit| state.receiver_in_flight >= limit.max_in_flight)
384 {
385 return Err(CallPolicyEvent::OverloadRejected);
386 }
387 state.receiver_in_flight += 1;
388 drop(states);
389 Ok(CallPolicyPermit {
390 runtime: self.clone(),
391 operation_key,
392 policy: policy.clone(),
393 mode: PermitMode::Receiver,
394 events: Vec::new(),
395 released: false,
396 })
397 }
398}
399
400#[derive(Debug, Clone, Copy)]
401enum PermitMode {
402 Caller { half_open: bool },
403 Receiver,
404}
405
406#[derive(Debug)]
407pub struct CallPolicyPermit {
408 runtime: CallPolicyRuntime,
409 operation_key: String,
410 policy: CallPolicyDeclaration,
411 mode: PermitMode,
412 events: Vec<CallPolicyEvent>,
413 released: bool,
414}
415
416impl CallPolicyPermit {
417 #[must_use]
418 pub fn events(&self) -> &[CallPolicyEvent] {
419 &self.events
420 }
421
422 #[must_use]
423 pub fn success(mut self) -> Vec<CallPolicyEvent> {
424 if let PermitMode::Caller { half_open } = self.mode {
425 let mut states = self
426 .runtime
427 .inner
428 .states
429 .lock()
430 .expect("call policy state poisoned");
431 let state = states
432 .get_mut(&self.operation_key)
433 .expect("permit state exists");
434 state.consecutive_failures = 0;
435 if half_open {
436 state.circuit_open_until_ms = None;
437 self.events.push(CallPolicyEvent::CircuitRecovered);
438 }
439 }
440 self.release();
441 self.events.clone()
442 }
443
444 #[must_use]
445 pub fn success_after(self, earlier_events: Vec<CallPolicyEvent>) -> Vec<CallPolicyEvent> {
446 ordered_events(self.success(), earlier_events)
447 }
448
449 #[must_use]
450 pub fn failure(mut self) -> Vec<CallPolicyEvent> {
451 if let (PermitMode::Caller { half_open }, Some(circuit)) =
452 (self.mode, self.policy.circuit_breaker.as_ref())
453 {
454 let mut states = self
455 .runtime
456 .inner
457 .states
458 .lock()
459 .expect("call policy state poisoned");
460 let state = states
461 .get_mut(&self.operation_key)
462 .expect("permit state exists");
463 state.consecutive_failures = state.consecutive_failures.saturating_add(1);
464 if half_open || state.consecutive_failures >= circuit.failure_threshold {
465 state.circuit_open_until_ms = Some(
466 self.runtime
467 .inner
468 .clock
469 .now_ms()
470 .saturating_add(circuit.open_for_ms),
471 );
472 self.events.push(CallPolicyEvent::CircuitOpened);
473 }
474 }
475 self.release();
476 self.events.clone()
477 }
478
479 #[must_use]
480 pub fn failure_after(self, earlier_events: Vec<CallPolicyEvent>) -> Vec<CallPolicyEvent> {
481 ordered_events(self.failure(), earlier_events)
482 }
483
484 fn release(&mut self) {
485 if self.released {
486 return;
487 }
488 let mut states = self
489 .runtime
490 .inner
491 .states
492 .lock()
493 .expect("call policy state poisoned");
494 let state = states
495 .get_mut(&self.operation_key)
496 .expect("permit state exists");
497 match self.mode {
498 PermitMode::Caller { half_open } => {
499 state.caller_in_flight = state.caller_in_flight.saturating_sub(1);
500 if half_open {
501 state.half_open_in_flight = state.half_open_in_flight.saturating_sub(1);
502 }
503 }
504 PermitMode::Receiver => {
505 state.receiver_in_flight = state.receiver_in_flight.saturating_sub(1);
506 }
507 }
508 self.released = true;
509 }
510}
511
512fn ordered_events(
513 mut state_events: Vec<CallPolicyEvent>,
514 mut earlier_events: Vec<CallPolicyEvent>,
515) -> Vec<CallPolicyEvent> {
516 let terminal_transition = state_events.last().copied().filter(|event| {
517 matches!(
518 event,
519 CallPolicyEvent::CircuitOpened | CallPolicyEvent::CircuitRecovered
520 )
521 });
522 if terminal_transition.is_some() {
523 state_events.pop();
524 }
525 state_events.append(&mut earlier_events);
526 if let Some(event) = terminal_transition {
527 state_events.push(event);
528 }
529 state_events
530}
531
532impl Drop for CallPolicyPermit {
533 fn drop(&mut self) {
534 self.release();
535 }
536}