camel_api/
in_flight.rs

1//! Context-global in-flight claim (drainclaim), Exchange-carried
2//! extension (claimfamily).
3//!
4//! [`InFlightClaim`] is one accepted-not-completed unit on the
5//! context-global counter (`CamelContext::total_in_flight()`). The type
6//! lives in camel-api — not camel-component-api — so [`crate::Exchange`]
7//! can carry a claim directly (rc-hllkk, rc-qbigm): stash sites behind
8//! `dyn` traits and raw-`Exchange` channels (resequencer buffers,
9//! embedded aggregator buckets) receive resident exchanges as bare
10//! `Exchange` values with no envelope, and their residency stays counted
11//! only if the claim rides the exchange itself.
12
13use std::fmt;
14use std::sync::Arc;
15use std::sync::atomic::{AtomicU64, Ordering};
16
17/// One accepted-not-completed unit on the context-global in-flight
18/// counter.
19///
20/// Attaching a claim increments the counter; dropping it decrements the
21/// counter exactly once (RAII), covering every release path — normal
22/// pipeline completion, dispatch push failure, queued-envelope drop,
23/// pipeline task abort, panic, and readiness failure — with no manual
24/// rollback code. Fanout sites mint one sibling claim per subscriber copy
25/// via [`InFlightClaim::split`], so each copy counts and releases
26/// independently (drainclaim). The type is deliberately NOT `Clone`:
27/// duplicating a claim would double-release on drop; fanout copies mint
28/// siblings via [`InFlightClaim::split`] instead.
29///
30/// Claim lifecycle during a pipeline run (claimfamily): the pipeline
31/// drain site holds the envelope's claim in task scope AND splits a
32/// sibling onto the exchange, so residency inside pipeline-embedded
33/// stash sites (resequencer buffer, aggregator bucket) stays counted
34/// after the pipeline task itself completes. Out-of-band stash emissions
35/// escape with their carried claims; the in-band pipeline result has its
36/// sibling taken back before the reply, so release stays at task end for
37/// exchanges that complete inside the pipeline.
38pub struct InFlightClaim(Arc<AtomicU64>);
39
40impl InFlightClaim {
41    /// Mint a claim against `counter`, incrementing it by one.
42    pub fn attach(counter: &Arc<AtomicU64>) -> Self {
43        counter.fetch_add(1, Ordering::AcqRel);
44        Self(Arc::clone(counter))
45    }
46
47    /// Mint a sibling claim for a fanout copy, incrementing the same
48    /// counter by one. The original claim stays live.
49    pub fn split(&self) -> Self {
50        self.0.fetch_add(1, Ordering::AcqRel);
51        Self(Arc::clone(&self.0))
52    }
53}
54
55impl Drop for InFlightClaim {
56    fn drop(&mut self) {
57        self.0.fetch_sub(1, Ordering::AcqRel);
58    }
59}
60
61impl fmt::Debug for InFlightClaim {
62    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
63        // The live count is observable through the counter itself
64        // (`total_in_flight()`); the Debug shape deliberately avoids
65        // leaking the Arc pointer.
66        f.write_str("InFlightClaim")
67    }
68}
69
70#[cfg(test)]
71mod tests {
72    use super::*;
73
74    #[test]
75    fn claim_attach_increments_and_drop_decrements() {
76        let counter = Arc::new(AtomicU64::new(0));
77        let claim = InFlightClaim::attach(&counter);
78        assert_eq!(counter.load(Ordering::Acquire), 1);
79        drop(claim);
80        assert_eq!(counter.load(Ordering::Acquire), 0);
81    }
82
83    #[test]
84    fn claim_split_adds_one_sibling() {
85        let counter = Arc::new(AtomicU64::new(0));
86        let original = InFlightClaim::attach(&counter);
87        assert_eq!(counter.load(Ordering::Acquire), 1);
88        let sibling = original.split();
89        assert_eq!(counter.load(Ordering::Acquire), 2);
90        drop(sibling);
91        assert_eq!(counter.load(Ordering::Acquire), 1);
92        drop(original);
93        assert_eq!(counter.load(Ordering::Acquire), 0);
94    }
95
96    #[test]
97    fn claim_debug_does_not_leak_pointer() {
98        let counter = Arc::new(AtomicU64::new(0));
99        let claim = InFlightClaim::attach(&counter);
100        assert_eq!(format!("{claim:?}"), "InFlightClaim");
101    }
102}
camel_api/in_flight.rs

camel_api/
in_flight.rs
