mnesis_store/
step.rs

1//! The catch-up→live phase marker for typed subscription consumption (#250).
2
3/// One item of a phase-aware subscription cursor: a decoded event, or the
4/// **caught-up** boundary marker emitted once when replay finishes and the
5/// cursor switches to live tailing.
6///
7/// Everything yielded before [`CaughtUp`](Step::CaughtUp) is replay (catch-up
8/// over the backlog); everything after is live. `T` is the item payload —
9/// `Decoded<E>` per-stream, `(AllPosition, StreamKey, Decoded<E>)` for `$all`.
10///
11/// Exhaustive (no `#[non_exhaustive]`, per project rule): the two variants are
12/// frozen at 1.0. A lag signal (`FellBehind`) is intentionally omitted — the
13/// live loop does not distinguish a lagging live consumer from a caught-up one,
14/// and a consumer can observe lag from positions.
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16pub enum Step<T> {
17    /// A delivered event (replay or live — the phase is told by the preceding
18    /// [`CaughtUp`](Step::CaughtUp)).
19    Event(T),
20    /// The backlog is drained; subsequent items are live. Emitted exactly once.
21    CaughtUp,
22}
23
24impl<T> Step<T> {
25    /// The event payload, or `None` for [`CaughtUp`](Step::CaughtUp).
26    #[must_use]
27    pub fn event(self) -> Option<T> {
28        match self {
29            Self::Event(t) => Some(t),
30            Self::CaughtUp => None,
31        }
32    }
33
34    /// `true` iff this is the [`CaughtUp`](Step::CaughtUp) marker.
35    #[must_use]
36    pub const fn is_caught_up(&self) -> bool {
37        matches!(self, Self::CaughtUp)
38    }
39
40    /// Map the carried payload, leaving [`CaughtUp`](Step::CaughtUp) untouched.
41    /// The phase marker flows through every transform (drop-the-tag in
42    /// `subscribe`, decode in `.decoded()`) so a boundary is never lost.
43    #[must_use]
44    pub fn map<U>(self, f: impl FnOnce(T) -> U) -> Step<U> {
45        match self {
46            Self::Event(t) => Step::Event(f(t)),
47            Self::CaughtUp => Step::CaughtUp,
48        }
49    }
50}
51
52#[cfg(test)]
53mod tests {
54    use super::*;
55
56    #[test]
57    fn event_and_caught_up_accessors() {
58        assert_eq!(Step::Event(7u8).event(), Some(7));
59        assert_eq!(Step::<u8>::CaughtUp.event(), None);
60        assert!(Step::<u8>::CaughtUp.is_caught_up());
61        assert!(!Step::Event(7u8).is_caught_up());
62    }
63
64    #[test]
65    fn map_transforms_event_and_passes_caught_up_through() {
66        assert_eq!(
67            Step::Event(3u8).map(|n| u32::from(n) * 2),
68            Step::Event(6u32)
69        );
70        assert_eq!(Step::<u8>::CaughtUp.map(u32::from), Step::<u32>::CaughtUp);
71    }
72}
mnesis_store/step.rs

mnesis_store/
step.rs
