orbit_core/fleet/
cursor.rs

1use std::marker::PhantomData;
2
3use crate::OrbitTyped;
4use crate::fleet::{Fleet, NodeId};
5use crate::ring::cursor::{
6    RingCursor, RingFrameSource, RingLoss, RingPoll, RingRead, poll_ring,
7};
8use crate::ring::{Frame, RingTopology};
9
10struct FleetRingSource<'a, T: OrbitTyped> {
11    fleet: &'a Fleet,
12    _t: PhantomData<T>,
13}
14
15struct FleetLaneSource<'a, T: OrbitTyped> {
16    fleet: &'a Fleet,
17    node_id: NodeId,
18    _t: PhantomData<T>,
19}
20
21impl<'a, T: OrbitTyped> FleetLaneSource<'a, T> {
22    fn new(
23        fleet: &'a Fleet,
24        node_id: NodeId,
25    ) -> Self {
26        Self { fleet, node_id, _t: PhantomData }
27    }
28}
29
30impl<T: OrbitTyped> RingFrameSource for FleetLaneSource<'_, T> {
31    fn kind(&self) -> u8 {
32        T::KIND
33    }
34
35    fn head(&self) -> u64 {
36        self.fleet.lane_head::<T>(self.node_id)
37    }
38
39    fn capacity(&self) -> usize {
40        self.fleet.ring_capacity::<T>()
41    }
42
43    fn read_at(
44        &self,
45        counter: u64,
46    ) -> Option<Frame> {
47        self.fleet.read_lane_at::<T>(self.node_id, counter)
48    }
49
50    fn read_state_at(
51        &self,
52        counter: u64,
53    ) -> RingRead {
54        self.fleet.read_lane_state_at::<T>(self.node_id, counter)
55    }
56}
57
58/// Caller-owned positions for every node lane of one ring type.
59#[derive(Clone, Debug, PartialEq, Eq)]
60pub struct FleetLaneCursor {
61    lanes: Vec<RingCursor>,
62    initial_counter: u64,
63}
64
65impl FleetLaneCursor {
66    /// Start every physical lane at the same lane-local counter.
67    ///
68    /// This does not restore a multi-lane checkpoint: independently advanced
69    /// lane positions must be preserved by retaining or cloning the cursor.
70    pub const fn from_counter(initial_counter: u64) -> Self {
71        Self { lanes: Vec::new(), initial_counter }
72    }
73
74    /// Numerical minimum of the lane-local next counters, including idle lanes.
75    ///
76    /// Counters belong to independent physical lanes. This value is not a global
77    /// cursor, event order, completion watermark, or multi-lane checkpoint. An
78    /// idle lane at zero keeps the minimum at zero even after all available
79    /// frames in other lanes have been read. Retain or clone this cursor to
80    /// preserve each lane's position; passing this minimum to
81    /// [`Self::from_counter`] can replay already consumed frames.
82    ///
83    /// Before lanes are initialized, returns the initial counter. Computing the
84    /// minimum scans only caller-owned memory and does not poll the rings.
85    pub fn minimum_lane_next_counter(&self) -> u64 {
86        self.lanes
87            .iter()
88            .map(|cursor| cursor.next_counter())
89            .min()
90            .unwrap_or(self.initial_counter)
91    }
92
93    /// Numerical minimum of independent lane-local counters, not a global cursor.
94    #[deprecated(
95        note = "use minimum_lane_next_counter; this is not a global cursor or checkpoint"
96    )]
97    pub fn minimum_next_counter(&self) -> u64 {
98        self.minimum_lane_next_counter()
99    }
100}
101
102/// Combined result of walking every node lane once.
103#[derive(Clone, Debug, Default, PartialEq, Eq)]
104pub struct FleetLanePoll {
105    pub frames: Vec<Frame>,
106    pub loss: RingLoss,
107}
108
109impl<'a, T: OrbitTyped> FleetRingSource<'a, T> {
110    fn new(fleet: &'a Fleet) -> Self {
111        Self { fleet, _t: PhantomData }
112    }
113}
114
115impl<T: OrbitTyped> RingFrameSource for FleetRingSource<'_, T> {
116    fn kind(&self) -> u8 {
117        T::KIND
118    }
119
120    fn head(&self) -> u64 {
121        self.fleet.head::<T>()
122    }
123
124    fn capacity(&self) -> usize {
125        self.fleet.ring_capacity::<T>()
126    }
127
128    fn read_at(
129        &self,
130        counter: u64,
131    ) -> Option<Frame> {
132        self.fleet.read_at::<T>(counter)
133    }
134
135    fn read_state_at(
136        &self,
137        counter: u64,
138    ) -> RingRead {
139        self.fleet.read_state_at::<T>(counter)
140    }
141}
142
143impl Fleet {
144    /// Cursor that starts after every counter currently claimed for `T`.
145    /// Useful for subscribers that only want future writes.
146    pub fn cursor_at_head<T: OrbitTyped>(&self) -> RingCursor {
147        RingCursor::from_counter(self.head::<T>())
148    }
149
150    /// Cursor that starts at counter 0 for `T`.
151    pub const fn cursor_from_start<T: OrbitTyped>(&self) -> RingCursor {
152        let _ = self;
153        let _ = PhantomData::<T>;
154        RingCursor::from_start()
155    }
156
157    /// Walk `cursor` toward the current claim head for `T`, stopping at
158    /// an in-flight counter and reporting definitive losses as
159    /// [`crate::ring::cursor::RingLoss`].
160    pub fn poll_ring<T: OrbitTyped>(
161        &self,
162        cursor: &mut RingCursor,
163    ) -> RingPoll {
164        poll_ring(&FleetRingSource::<T>::new(self), cursor)
165    }
166
167    /// Create one caller-owned cursor per physical node lane, starting at each
168    /// lane's current head.
169    pub fn lane_cursor_at_head<T: OrbitTyped>(&self) -> FleetLaneCursor {
170        self.assert_per_node::<T>();
171        let lanes = (0..self.fleet_capacity())
172            .map(|node| {
173                RingCursor::from_counter(self.lane_head::<T>(NodeId::new(node)))
174            })
175            .collect();
176        FleetLaneCursor { lanes, initial_counter: 0 }
177    }
178
179    /// Create one caller-owned cursor per physical node lane, starting at
180    /// counter zero.
181    pub fn lane_cursor_from_start<T: OrbitTyped>(&self) -> FleetLaneCursor {
182        self.assert_per_node::<T>();
183        FleetLaneCursor {
184            lanes: vec![
185                RingCursor::from_start();
186                usize::from(self.fleet_capacity())
187            ],
188            initial_counter: 0,
189        }
190    }
191
192    /// Poll every physical node lane and combine the retained frames and loss
193    /// counters into one result. Frames retain their writer node in `id`;
194    /// callers that need semantic ordering across lanes must provide it.
195    pub fn poll_lanes<T: OrbitTyped>(
196        &self,
197        cursor: &mut FleetLaneCursor,
198    ) -> FleetLanePoll {
199        self.assert_per_node::<T>();
200        if cursor.lanes.is_empty() {
201            cursor.lanes = vec![
202                RingCursor::from_counter(cursor.initial_counter);
203                usize::from(self.fleet_capacity())
204            ];
205        }
206        assert_eq!(
207            cursor.lanes.len(),
208            usize::from(self.fleet_capacity()),
209            "lane cursor belongs to a different fleet capacity"
210        );
211
212        let mut combined = FleetLanePoll::default();
213        for (node, lane_cursor) in cursor.lanes.iter_mut().enumerate() {
214            let source =
215                FleetLaneSource::<T>::new(self, NodeId::new(node as u16));
216            let poll = poll_ring(&source, lane_cursor);
217            combined.frames.extend(poll.frames);
218            combined.loss.overwritten =
219                combined.loss.overwritten.saturating_add(poll.loss.overwritten);
220            combined.loss.unavailable =
221                combined.loss.unavailable.saturating_add(poll.loss.unavailable);
222        }
223        combined
224    }
225
226    fn assert_per_node<T: OrbitTyped>(&self) {
227        assert_eq!(
228            T::RING_SPEC.topology,
229            RingTopology::PerNode,
230            "lane cursor requires a per-node ring"
231        );
232    }
233}
orbit_core/fleet/cursor.rs

orbit_core/fleet/
cursor.rs
