kevy_replicate/
feed.rs

1//! CDC feed layer over [`crate::source::ReplicationSource`] —
2//! the `(generation, offset)` cursor semantics the public FEED.* /
3//! `changes_since` surfaces speak.
4//!
5//! Cursor contract:
6//! - `generation` identifies one unbroken offset history. A given
7//!   `(gen, offset)` pair refers to the same stream prefix forever.
8//!   The value is an OPAQUE random identity, not a counter: two nodes
9//!   counting in lockstep (a startup election on one, a failover
10//!   promotion on the other) land on the same number for different
11//!   histories, and a stale cursor then slips through the fence into
12//!   offset aliasing (the availgate failover wedge). Randomness is
13//!   what makes cross-node fencing sound; there is no ordering.
14//! - Clean shutdown + restart preserves both (continuity); FLUSHALL,
15//!   restore-from-snapshot, or an unclean shutdown draws a fresh
16//!   `gen` and resets `offset` to 0.
17//! - A cursor from any OTHER generation, or one whose offsets were
18//!   evicted, answers `FeedRead::Resync` carrying the current tail —
19//!   the consumer rebuilds (SCAN) and resumes from there.
20//!
21//! ```
22//! use kevy_replicate::feed::{FeedPosition, FeedRead, FeedSource};
23//! use kevy_replicate::source::ReplicationSource;
24//!
25//! let mut feed = FeedSource::new(1, ReplicationSource::new(1 << 20));
26//! let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
27//! feed.source_mut().push_mutation(&set);
28//! let cursor = FeedPosition::new(1, 0);
29//! assert_eq!(feed.read(cursor, 10).map(|f| f.len()), Ok(1));
30//!
31//! feed.bump_generation(); // FLUSHALL: a new history begins
32//! assert!(matches!(feed.read(cursor, 10), Err(FeedRead::Resync { .. })));
33//! ```
34
35use crate::source::{FromOffset, ReplicationSource};
36
37/// A point in a feed: an offset within one generation's offset history.
38///
39/// The pair is the whole identity — an offset means nothing outside its
40/// generation, so the two travel together: a feed cursor, the tail a
41/// consumer resumes at, the position a replica claims at handshake, the
42/// cursor a snapshot records. `FeedPosition::default()` is generation 0,
43/// offset 0: the "no continuity claim" a fresh replica presents (no real
44/// feed ever runs at generation 0).
45///
46/// ```
47/// use kevy_replicate::feed::FeedPosition;
48///
49/// let at = FeedPosition::new(7, 42);
50/// assert_eq!((at.generation, at.offset), (7, 42));
51/// assert_eq!(FeedPosition::default(), FeedPosition::new(0, 0));
52/// ```
53#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
54#[non_exhaustive]
55pub struct FeedPosition {
56    /// The offset history this position belongs to: an opaque identity,
57    /// see [`fresh_generation`]. `0` = unknown.
58    ///
59    /// ```
60    /// use kevy_replicate::feed::FeedSource;
61    /// use kevy_replicate::source::ReplicationSource;
62    ///
63    /// let feed = FeedSource::new(9, ReplicationSource::new(1 << 20));
64    /// assert_eq!(feed.tail().generation, 9);
65    /// ```
66    pub generation: u64,
67    /// The offset within that history.
68    ///
69    /// ```
70    /// use kevy_replicate::feed::FeedSource;
71    /// use kevy_replicate::source::ReplicationSource;
72    ///
73    /// let mut feed = FeedSource::new(1, ReplicationSource::new(1 << 20));
74    /// let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
75    /// feed.source_mut().push_mutation(&set);
76    /// assert_eq!(feed.tail().offset, 1); // the next write lands at offset 1
77    /// ```
78    pub offset: u64,
79}
80
81impl FeedPosition {
82    /// The position at `offset` in `generation`.
83    ///
84    /// ```
85    /// let at = kevy_replicate::feed::FeedPosition::new(3, 0);
86    /// assert_eq!(at.generation, 3);
87    /// ```
88    #[inline]
89    #[must_use]
90    pub const fn new(generation: u64, offset: u64) -> Self {
91        Self { generation, offset }
92    }
93}
94
95/// Why a feed read could not be served from the backlog.
96///
97/// ```
98/// use kevy_replicate::feed::{FeedPosition, FeedRead, FeedSource};
99/// use kevy_replicate::source::ReplicationSource;
100///
101/// let feed = FeedSource::new(1, ReplicationSource::new(1 << 20));
102/// match feed.read(FeedPosition::new(2, 0), 10) {
103///     Err(FeedRead::Resync { tail }) => assert_eq!(tail, FeedPosition::new(1, 0)),
104///     other => panic!("unexpected {other:?}"),
105/// }
106/// ```
107#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
108pub enum FeedRead {
109    /// Cursor unservable (stale generation or evicted offset): rebuild
110    /// from a scan, then resume from the carried tail cursor.
111    ///
112    /// ```
113    /// use kevy_replicate::feed::{FeedPosition, FeedRead, FeedSource};
114    /// use kevy_replicate::source::ReplicationSource;
115    ///
116    /// // a 64-byte backlog keeps only the newest frame
117    /// let mut feed = FeedSource::new(1, ReplicationSource::new(64));
118    /// let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
119    /// for _ in 0..3 {
120    ///     feed.source_mut().push_mutation(&set);
121    /// }
122    /// let err = feed.read(FeedPosition::new(1, 0), 10).unwrap_err();
123    /// assert_eq!(err, FeedRead::Resync { tail: FeedPosition::new(1, 3) });
124    /// ```
125    Resync {
126        /// The current generation and the next offset the source will
127        /// assign: where to resume.
128        ///
129        /// ```
130        /// use kevy_replicate::feed::{FeedPosition, FeedRead, FeedSource};
131        /// use kevy_replicate::source::ReplicationSource;
132        ///
133        /// let feed = FeedSource::new(4, ReplicationSource::new(1 << 20));
134        /// let Err(FeedRead::Resync { tail }) = feed.read(FeedPosition::new(3, 0), 10) else { panic!() };
135        /// assert!(feed.read(tail, 10)?.is_empty()); // resuming at the tail is served
136        /// # Ok::<(), FeedRead>(())
137        /// ```
138        tail: FeedPosition,
139    },
140    /// Cursor is ahead of the stream (`offset > next`) in the CURRENT
141    /// generation — caller bug or epoch confusion; reject the read.
142    /// (A mismatched generation is `Resync`, never `Future`:
143    /// generations carry no order to be "ahead" in.)
144    ///
145    /// ```
146    /// use kevy_replicate::feed::{FeedPosition, FeedRead, FeedSource};
147    /// use kevy_replicate::source::ReplicationSource;
148    ///
149    /// let feed = FeedSource::new(1, ReplicationSource::new(1 << 20));
150    /// assert_eq!(feed.read(FeedPosition::new(1, 5), 10), Err(FeedRead::Future));
151    /// ```
152    Future,
153}
154
155/// One decoded feed entry: the offset plus the frame's wire bytes
156/// (envelope + offset + RESP argv — same encoding replicas consume;
157/// [`crate::wire::decode_frame`] parses it).
158///
159/// ```
160/// use kevy_replicate::feed::{FeedPosition, FeedSource};
161/// use kevy_replicate::source::ReplicationSource;
162///
163/// let mut feed = FeedSource::new(1, ReplicationSource::new(1 << 20));
164/// let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
165/// feed.source_mut().push_mutation(&set);
166/// let frames = feed.read(FeedPosition::new(1, 0), 10).expect("offset 0 is buffered");
167/// let (decoded, _) = kevy_replicate::wire::decode_frame(frames[0].bytes)?;
168/// assert_eq!((frames[0].offset, decoded.argv), (0, set));
169/// # Ok::<(), kevy_replicate::wire::WireError>(())
170/// ```
171#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
172#[non_exhaustive]
173pub struct FeedFrame<'a> {
174    /// Offset the source assigned at push time.
175    ///
176    /// ```
177    /// use kevy_replicate::feed::{FeedPosition, FeedSource};
178    /// use kevy_replicate::source::ReplicationSource;
179    ///
180    /// let mut feed = FeedSource::new(1, ReplicationSource::new(1 << 20));
181    /// let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
182    /// for _ in 0..3 {
183    ///     feed.source_mut().push_mutation(&set);
184    /// }
185    /// let frames = feed.read(FeedPosition::new(1, 1), 10)?;
186    /// assert_eq!(frames.iter().map(|f| f.offset).collect::<Vec<_>>(), [1, 2]);
187    /// # Ok::<(), kevy_replicate::feed::FeedRead>(())
188    /// ```
189    pub offset: u64,
190    /// Wire-encoded frame bytes.
191    ///
192    /// ```
193    /// use kevy_replicate::feed::{FeedPosition, FeedSource};
194    /// use kevy_replicate::source::ReplicationSource;
195    ///
196    /// let mut feed = FeedSource::new(1, ReplicationSource::new(1 << 20));
197    /// let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
198    /// feed.source_mut().push_mutation(&set);
199    /// let frames = feed.read(FeedPosition::new(1, 0), 1)?;
200    /// assert_eq!(frames[0].bytes, kevy_replicate::wire::encode_frame(0, &set));
201    /// # Ok::<(), kevy_replicate::feed::FeedRead>(())
202    /// ```
203    pub bytes: &'a [u8],
204}
205
206/// Generation-aware wrapper: owns the generation number alongside the
207/// backlog. The runtime persists `generation` via `kevy-persist`'s
208/// feed sidecars; this type only holds the in-memory value.
209///
210/// ```
211/// use kevy_replicate::feed::{FeedPosition, FeedSource};
212/// use kevy_replicate::source::ReplicationSource;
213///
214/// let mut feed = FeedSource::new(1, ReplicationSource::new(1 << 20));
215/// let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
216/// feed.source_mut().push_mutation(&set);
217/// let mut cursor = FeedPosition::new(1, 0);
218/// for frame in feed.read(cursor, 100)? {
219///     cursor.offset = frame.offset + 1; // a consumer advances past what it applied
220/// }
221/// assert_eq!(cursor, feed.tail());
222/// # Ok::<(), kevy_replicate::feed::FeedRead>(())
223/// ```
224#[derive(Debug)]
225pub struct FeedSource {
226    generation: u64,
227    source: ReplicationSource,
228}
229
230impl FeedSource {
231    /// Wrap a backlog at an explicit generation (loaded from the feed
232    /// sidecar at boot, or 1 for a fresh data dir).
233    ///
234    /// ```
235    /// use kevy_replicate::feed::{FeedPosition, FeedSource};
236    /// use kevy_replicate::source::ReplicationSource;
237    ///
238    /// let feed = FeedSource::new(1, ReplicationSource::new(1 << 20));
239    /// assert_eq!(feed.tail(), FeedPosition::new(1, 0));
240    /// ```
241    pub fn new(generation: u64, source: ReplicationSource) -> Self {
242        Self { generation, source }
243    }
244
245    /// Current generation.
246    ///
247    /// ```
248    /// use kevy_replicate::feed::FeedSource;
249    /// use kevy_replicate::source::ReplicationSource;
250    ///
251    /// let mut feed = FeedSource::new(1, ReplicationSource::new(1 << 20));
252    /// assert_eq!(feed.generation(), 1);
253    /// feed.bump_generation();
254    /// assert_ne!(feed.generation(), 1);
255    /// ```
256    pub fn generation(&self) -> u64 {
257        self.generation
258    }
259
260    /// Access the wrapped backlog (push side + replica streaming keep
261    /// their existing [`ReplicationSource`] API).
262    ///
263    /// ```
264    /// use kevy_replicate::feed::FeedSource;
265    /// use kevy_replicate::source::ReplicationSource;
266    ///
267    /// let feed = FeedSource::new(1, ReplicationSource::new(4096));
268    /// assert_eq!(feed.source().max_bytes(), 4096);
269    /// assert!(feed.source().is_empty());
270    /// ```
271    pub fn source(&self) -> &ReplicationSource {
272        &self.source
273    }
274
275    /// Mutable access for `push_mutation` / `drop_up_to`.
276    ///
277    /// ```
278    /// use kevy_replicate::feed::FeedSource;
279    /// use kevy_replicate::source::ReplicationSource;
280    ///
281    /// let mut feed = FeedSource::new(1, ReplicationSource::new(1 << 20));
282    /// let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
283    /// assert_eq!(feed.source_mut().push_mutation(&set), 0);
284    /// feed.source_mut().drop_up_to(1); // every consumer has passed offset 0
285    /// assert!(feed.source().is_empty());
286    /// ```
287    pub fn source_mut(&mut self) -> &mut ReplicationSource {
288        &mut self.source
289    }
290
291    /// Break continuity: draw a fresh generation and restart offsets
292    /// at 0 (FLUSHALL / restore / promotion / unclean-boot policy).
293    /// The backlog empties — frames from the old generation must
294    /// never be served under the new one.
295    ///
296    /// ```
297    /// use kevy_replicate::feed::{FeedPosition, FeedSource};
298    /// use kevy_replicate::source::ReplicationSource;
299    ///
300    /// let mut feed = FeedSource::new(1, ReplicationSource::new(1 << 20));
301    /// let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
302    /// feed.source_mut().push_mutation(&set);
303    /// feed.bump_generation();
304    /// assert_eq!(feed.tail(), FeedPosition::new(feed.generation(), 0));
305    /// assert!(feed.source().is_empty());
306    /// ```
307    pub fn bump_generation(&mut self) {
308        self.generation = fresh_generation(self.generation);
309        let budget = self.source.max_bytes();
310        self.source = ReplicationSource::new(budget);
311    }
312
313    /// The tail cursor: the current generation and the next offset —
314    /// where a consumer resuming after a rebuild starts.
315    ///
316    /// ```
317    /// use kevy_replicate::feed::{FeedPosition, FeedSource};
318    /// use kevy_replicate::source::ReplicationSource;
319    ///
320    /// let feed = FeedSource::new(5, ReplicationSource::new(1 << 20));
321    /// assert_eq!(feed.tail(), FeedPosition::new(5, 0));
322    /// ```
323    #[inline]
324    pub fn tail(&self) -> FeedPosition {
325        FeedPosition::new(self.generation, self.source.next_offset())
326    }
327
328    /// Serve up to `max` frames at cursor `at`.
329    ///
330    /// - Any OTHER generation → `Resync` (the cursor belongs to a
331    ///   different offset history — uniqueness of `(gen, offset)`
332    ///   forbids serving it; generations are identities, not ordered).
333    /// - Offset ahead of this generation's stream → `Future`.
334    /// - Evicted offset → `Resync` with the current tail.
335    ///
336    /// ```
337    /// use kevy_replicate::feed::{FeedPosition, FeedSource};
338    /// use kevy_replicate::source::ReplicationSource;
339    ///
340    /// let mut feed = FeedSource::new(1, ReplicationSource::new(1 << 20));
341    /// let set = kevy_resp::Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), b"v".to_vec()]);
342    /// for _ in 0..5 {
343    ///     feed.source_mut().push_mutation(&set);
344    /// }
345    /// let page = feed.read(FeedPosition::new(1, 0), 2)?; // at most 2 frames
346    /// assert_eq!(page.iter().map(|f| f.offset).collect::<Vec<_>>(), [0, 1]);
347    /// # Ok::<(), kevy_replicate::feed::FeedRead>(())
348    /// ```
349    pub fn read(&self, at: FeedPosition, max: usize) -> Result<Vec<FeedFrame<'_>>, FeedRead> {
350        if at.generation != self.generation {
351            return Err(FeedRead::Resync { tail: self.tail() });
352        }
353        match self.source.frames_from(at.offset) {
354            Ok(iter) => Ok(iter
355                .take(max)
356                .map(|f| FeedFrame { offset: f.offset, bytes: &f.bytes })
357                .collect()),
358            Err(FromOffset::TooOld) => Err(FeedRead::Resync { tail: self.tail() }),
359            Err(FromOffset::Future) => Err(FeedRead::Future),
360        }
361    }
362}
363
364/// Draw a fresh generation: a random nonzero u64 distinct from `old`.
365/// A generation is a HISTORY IDENTITY. Counters are unsound here: two
366/// nodes bumping in lockstep (startup election vs failover promotion)
367/// collide on the same number for different histories, and a replica's
368/// stale cursor then passes the generation fence into offset aliasing
369/// — served as "caught up" at a foreign offset, it wedges forever
370/// (the availgate failover-convergence flake). `RandomState`
371/// carries the process's OS-seeded entropy; identity, not crypto.
372///
373/// Values fit in 53 bits, so they survive a round trip through a
374/// JavaScript number.
375///
376/// ```
377/// let g = kevy_replicate::feed::fresh_generation(1);
378/// assert!(g != 0 && g != 1 && g < 1 << 53);
379/// ```
380#[must_use]
381pub fn fresh_generation(old: u64) -> u64 {
382    use std::hash::{BuildHasher, Hasher};
383    loop {
384        let mut h = std::collections::hash_map::RandomState::new().build_hasher();
385        h.write_u64(old);
386        // Mask to 53 bits: generations ride RESP integers (REPL.TOKEN
387        // / REPL.WAIT / FEED.TAIL), and client bindings surface them
388        // as JS Numbers, whose integer precision ends at 2^53 — a
389        // wider value round-trips corrupted and self-resyncs forever.
390        // A 2^53 identity space still makes collisions negligible.
391        let g = h.finish() & ((1u64 << 53) - 1);
392        if g != 0 && g != old {
393            return g;
394        }
395    }
396}
397
398#[cfg(test)]
399#[path = "feed_tests.rs"]
400mod tests;
kevy_replicate/feed.rs

kevy_replicate/
feed.rs
