subetha_cxc/
replay_positions.rs

1//! `SubscriberPosition`: Aeron-inspired MMF-resident position
2//! counter for resumable cross-process subscribers.
3//!
4//! A subscriber that consumes from a ring needs a way to checkpoint
5//! "I have consumed up to position N" so that if the subscriber
6//! crashes / restarts, it can resume from N rather than from the
7//! ring's tail (which may have advanced past lost items).
8//! `SubscriberPosition` is the standalone primitive that holds N
9//! in an MMF file, surviving process restarts.
10//!
11//! # Why standalone
12//!
13//! The substrate's ring primitives (SpscRingCore, SharedRing,
14//! AdaptiveRing) maintain head/tail counters internally; those
15//! counters track ring slot positions, not subscriber positions.
16//! `SubscriberPosition` is the caller-managed counter that bridges
17//! "ring is at slot K" with "this subscriber has acknowledged up
18//! to absolute position P". Callers compute the relationship
19//! between K and P themselves (typically by walking the ring's
20//! head pointer at startup + remembering the offset).
21//!
22//! # MMF residency
23//!
24//! The position lives in a `SharedAtomicU64` file. Two processes
25//! that open the same path see the same position counter
26//! atomically. This matches the substrate's MMF-resident-control
27//! pattern (locale_tag, pin_generation, etc.).
28//!
29//! # Replay semantics
30//!
31//! After a restart, the subscriber:
32//! 1. Reopens the SubscriberPosition file by path.
33//! 2. Reads the persisted position via [`get`](SubscriberPosition::get).
34//! 3. Reopens the source ring and re-attaches.
35//! 4. Resumes consumption from the recorded position (caller's
36//!    responsibility to map absolute position to ring slot index).
37//!
38//! Wraparound caveat: a regular ring's slot array is bounded; if
39//! the producer outraces the subscriber's checkpoint by more than
40//! ring capacity, the lost items are gone. Callers needing
41//! guaranteed-no-loss replay back the source ring with a large
42//! enough capacity OR snapshot positions frequently enough that
43//! checkpoint-position never lags producer-position by more than
44//! one ring sweep.
45
46use std::path::Path;
47use std::sync::Arc;
48use std::sync::atomic::Ordering;
49
50use crate::shared_atomic::SharedAtomicU64;
51
52/// MMF-backed monotonically-increasing consumer position counter.
53pub struct SubscriberPosition {
54    counter: Arc<SharedAtomicU64>,
55}
56
57impl SubscriberPosition {
58    /// Create a new position counter at `path` initialised to
59    /// `initial`.
60    pub fn create(
61        path: impl AsRef<Path>,
62        initial: u64,
63    ) -> Result<Self, std::io::Error> {
64        let counter = SharedAtomicU64::create(path, initial)
65            .map_err(|e| std::io::Error::other(format!("{e:?}")))?;
66        Ok(Self { counter: Arc::new(counter) })
67    }
68
69    /// Open an existing position counter at `path` for read/write.
70    /// Used by a subscriber restart path.
71    pub fn open(path: impl AsRef<Path>) -> Result<Self, std::io::Error> {
72        let counter = SharedAtomicU64::open(path)
73            .map_err(|e| std::io::Error::other(format!("{e:?}")))?;
74        Ok(Self { counter: Arc::new(counter) })
75    }
76
77    /// Current position (Acquire load).
78    pub fn get(&self) -> u64 { self.counter.load(Ordering::Acquire) }
79
80    /// Advance the position by `by`. Returns the NEW position.
81    /// Atomic; safe for one subscriber to call concurrently with
82    /// another holder reading via `get`.
83    pub fn advance(&self, by: u64) -> u64 {
84        let prior = self.counter.fetch_add(by, Ordering::AcqRel);
85        prior + by
86    }
87
88    /// Set the position to `new` unconditionally. Used by restart
89    /// paths that want to reset rather than advance.
90    pub fn set(&self, new: u64) {
91        self.counter.store(new, Ordering::Release);
92    }
93
94    /// Compare-and-set semantics. Returns `Ok(new)` if the previous
95    /// value matched `expected`; `Err(actual)` otherwise.
96    pub fn compare_and_set(
97        &self,
98        expected: u64,
99        new: u64,
100    ) -> Result<u64, u64> {
101        match self.counter.compare_exchange(
102            expected, new, Ordering::AcqRel, Ordering::Acquire,
103        ) {
104            Ok(_) => Ok(new),
105            Err(actual) => Err(actual),
106        }
107    }
108
109    /// Clone the underlying `Arc<SharedAtomicU64>` so a second
110    /// in-process holder can read the same counter cheaply.
111    pub fn counter_handle(&self) -> Arc<SharedAtomicU64> {
112        self.counter.clone()
113    }
114}
115
116#[cfg(test)]
117mod tests {
118    use super::*;
119
120    fn tmp(name: &str) -> std::path::PathBuf {
121        let mut p = std::env::temp_dir();
122        let pid = std::process::id();
123        let nonce = std::time::SystemTime::now()
124            .duration_since(std::time::UNIX_EPOCH)
125            .map(|d| d.as_nanos())
126            .unwrap_or(0);
127        p.push(format!("subpos_{pid}_{nonce}_{name}.bin"));
128        p
129    }
130
131    #[test]
132    fn create_then_get_returns_initial() {
133        let path = tmp("init");
134        let pos = SubscriberPosition::create(&path, 42).expect("create");
135        assert_eq!(pos.get(), 42);
136        std::fs::remove_file(&path).ok();
137    }
138
139    #[test]
140    fn advance_returns_new_position() {
141        let path = tmp("advance");
142        let pos = SubscriberPosition::create(&path, 0).expect("create");
143        assert_eq!(pos.advance(5), 5);
144        assert_eq!(pos.advance(7), 12);
145        assert_eq!(pos.get(), 12);
146        std::fs::remove_file(&path).ok();
147    }
148
149    #[test]
150    fn set_overrides_unconditionally() {
151        let path = tmp("set");
152        let pos = SubscriberPosition::create(&path, 100).expect("create");
153        pos.set(9999);
154        assert_eq!(pos.get(), 9999);
155        std::fs::remove_file(&path).ok();
156    }
157
158    #[test]
159    fn compare_and_set_succeeds_on_expected() {
160        let path = tmp("cas_ok");
161        let pos = SubscriberPosition::create(&path, 0).expect("create");
162        assert_eq!(pos.compare_and_set(0, 10), Ok(10));
163        assert_eq!(pos.get(), 10);
164        std::fs::remove_file(&path).ok();
165    }
166
167    #[test]
168    fn compare_and_set_fails_on_mismatch() {
169        let path = tmp("cas_fail");
170        let pos = SubscriberPosition::create(&path, 5).expect("create");
171        assert_eq!(pos.compare_and_set(0, 999), Err(5));
172        assert_eq!(pos.get(), 5);
173        std::fs::remove_file(&path).ok();
174    }
175
176    #[test]
177    fn open_after_create_sees_same_position() {
178        let path = tmp("reopen");
179        let a = SubscriberPosition::create(&path, 100).expect("create");
180        a.advance(50);
181        let b = SubscriberPosition::open(&path).expect("open");
182        assert_eq!(b.get(), 150);
183        b.advance(25);
184        assert_eq!(a.get(), 175);
185        std::fs::remove_file(&path).ok();
186    }
187
188    #[test]
189    fn position_survives_drop_then_reopen() {
190        let path = tmp("survive_drop");
191        {
192            let a = SubscriberPosition::create(&path, 0).expect("create");
193            a.advance(42);
194            // a drops here; underlying MMF file persists.
195        }
196        let b = SubscriberPosition::open(&path).expect("reopen after drop");
197        assert_eq!(b.get(), 42);
198        std::fs::remove_file(&path).ok();
199    }
200}
subetha_cxc/replay_positions.rs

subetha_cxc/
replay_positions.rs
