kevy_replicate/

slot.rs

//! Per-replica slot bookkeeping for the primary's streaming loop.
//!
//! Each connected (or recently-disconnected) replica owns one
//! [`ReplicaSlot`]. The [`SlotTable`] is the source of truth for "who
//! still has a chance to resume from the backlog vs whom we have
//! given up on". Slots are pure data; socket lifetime and the
//! streaming loop's wakeups live in the wiring layer.
//!
//! Clock policy: all timestamps are `u64` monotonic nanoseconds
//! supplied by the caller. The module never reads the clock itself —
//! tests pump synthetic timestamps, and the production hook uses a
//! single `Instant`-derived `u64`. Same pattern as the cached-clock
//! work in `kevy-store`.
//!
//! ```
//! use kevy_replicate::slot::SlotTable;
//!
//! const SEC: u64 = 1_000_000_000;
//! let mut slots = SlotTable::new();
//! slots.insert_or_touch("r1", 10, 0); // handshake / ack at t = 0 s
//! slots.insert_or_touch("r2", 4, 3 * SEC);
//! assert_eq!(slots.min_acked_offset(), Some(4)); // the backlog must keep offset 4 on
//!
//! // a 5 s reconnect window: r1 gives up at t = 5 s
//! assert_eq!(slots.expire(5 * SEC, 5 * SEC), ["r1"]);
//! ```

/// One connected-or-recently-disconnected replica.
///
/// ```
/// use kevy_replicate::slot::SlotTable;
///
/// let mut slots = SlotTable::new();
/// slots.insert_or_touch("r1", 42, 1_000);
/// let slot = slots.get("r1").expect("tracked");
/// assert_eq!((slot.id.as_str(), slot.acked_offset, slot.last_seen_ns), ("r1", 42, 1_000));
/// ```
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub struct ReplicaSlot {
    /// Operator-set replica identifier (opaque to the primary other
    /// than for slot bookkeeping).
    ///
    /// ```
    /// let mut slots = kevy_replicate::slot::SlotTable::new();
    /// slots.insert_or_touch("replica-a", 0, 0);
    /// assert_eq!(slots.iter().map(|s| s.id.as_str()).collect::<Vec<_>>(), ["replica-a"]);
    /// ```
    pub id: String,
    /// Monotonic ns timestamp of the most recent contact (handshake
    /// or ack). Drives expiry under `reconnect_window_ms`.
    ///
    /// ```
    /// let mut slots = kevy_replicate::slot::SlotTable::new();
    /// slots.insert_or_touch("r1", 0, 100);
    /// slots.touch_or_insert_unacked("r1", 250); // a later contact
    /// assert_eq!(slots.get("r1").map(|s| s.last_seen_ns), Some(250));
    /// ```
    pub last_seen_ns: u64,
    /// Highest offset the replica has acked. The streaming loop
    /// resumes sending from here on reconnect.
    ///
    /// ```
    /// let mut slots = kevy_replicate::slot::SlotTable::new();
    /// slots.insert_or_touch("r1", 9, 0);
    /// slots.insert_or_touch("r1", 7, 1); // a stale ack never moves it back
    /// assert_eq!(slots.get("r1").map(|s| s.acked_offset), Some(9));
    /// ```
    pub acked_offset: u64,
}

/// Mutable collection of [`ReplicaSlot`]s. Slots are addressed by id;
/// duplicate insertion upserts (newer state wins). Replica counts in
/// realistic deployments are small (< 16); a linear `Vec` is faster
/// than a `HashMap` at this size and avoids the cost of the hasher
/// the rest of the workspace uses.
///
/// ```
/// use kevy_replicate::slot::SlotTable;
///
/// let mut slots = SlotTable::new();
/// slots.insert_or_touch("r1", 3, 0);
/// slots.insert_or_touch("r1", 8, 1); // upsert, not a second row
/// assert_eq!(slots.len(), 1);
/// assert_eq!(slots.min_acked_offset(), Some(8));
/// ```
#[derive(Debug, Default)]
pub struct SlotTable {
    slots: Vec<ReplicaSlot>,
}

impl SlotTable {
    /// A fresh empty table.
    ///
    /// ```
    /// let slots = kevy_replicate::slot::SlotTable::new();
    /// assert!(slots.is_empty());
    /// ```
    pub fn new() -> Self {
        Self::default()
    }

    /// Number of slots currently tracked.
    ///
    /// ```
    /// let mut slots = kevy_replicate::slot::SlotTable::new();
    /// slots.insert_or_touch("r1", 0, 0);
    /// slots.insert_or_touch("r2", 0, 0);
    /// assert_eq!(slots.len(), 2);
    /// ```
    pub fn len(&self) -> usize {
        self.slots.len()
    }

    /// Whether the table has no slots.
    ///
    /// ```
    /// let mut slots = kevy_replicate::slot::SlotTable::new();
    /// assert!(slots.is_empty());
    /// slots.insert_or_touch("r1", 0, 0);
    /// assert!(!slots.is_empty());
    /// ```
    pub fn is_empty(&self) -> bool {
        self.slots.is_empty()
    }

    /// Look up a slot by id.
    ///
    /// ```
    /// let mut slots = kevy_replicate::slot::SlotTable::new();
    /// slots.insert_or_touch("r1", 5, 0);
    /// assert_eq!(slots.get("r1").map(|s| s.acked_offset), Some(5));
    /// assert!(slots.get("r2").is_none());
    /// ```
    pub fn get(&self, id: &str) -> Option<&ReplicaSlot> {
        self.slots.iter().find(|s| s.id == id)
    }

    /// Refresh a slot's `last_seen_ns` WITHOUT advancing its acked
    /// offset — the disconnect-time variant of
    /// [`Self::insert_or_touch`]. A closing conn knows what it SENT,
    /// not what the peer acked; recording sent-as-acked would let
    /// `WAIT` / min-replicas count bytes the replica never confirmed
    /// and let the backlog trim frames the peer may still need. A
    /// previously unseen id is inserted at `acked_offset = 0`
    /// (never acked — the truthful floor).
    ///
    /// ```
    /// let mut slots = kevy_replicate::slot::SlotTable::new();
    /// slots.insert_or_touch("r1", 12, 100);
    /// slots.touch_or_insert_unacked("r1", 200); // the link closed at t = 200
    /// slots.touch_or_insert_unacked("r2", 200); // never acked anything
    /// assert_eq!(slots.get("r1").map(|s| (s.acked_offset, s.last_seen_ns)), Some((12, 200)));
    /// assert_eq!(slots.get("r2").map(|s| s.acked_offset), Some(0));
    /// ```
    pub fn touch_or_insert_unacked(&mut self, id: &str, now_ns: u64) {
        if let Some(s) = self.slots.iter_mut().find(|s| s.id == id) {
            s.last_seen_ns = now_ns;
            return;
        }
        self.slots.push(ReplicaSlot { id: id.to_string(), last_seen_ns: now_ns, acked_offset: 0 });
    }

    /// Iterate over all slots.
    ///
    /// ```
    /// let mut slots = kevy_replicate::slot::SlotTable::new();
    /// slots.insert_or_touch("r1", 4, 0);
    /// slots.insert_or_touch("r2", 6, 0);
    /// let acked: u64 = slots.iter().map(|s| s.acked_offset).sum();
    /// assert_eq!(acked, 10);
    /// ```
    pub fn iter(&self) -> impl Iterator<Item = &ReplicaSlot> {
        self.slots.iter()
    }

    /// Insert a new slot or update an existing one. Touching always
    /// refreshes `last_seen_ns` and advances `acked_offset` if the
    /// new value is higher (a slot's acked offset is monotonic — a
    /// peer reporting a lower offset than we already recorded is
    /// almost always a bug; the silent max() here defends the
    /// invariant).
    ///
    /// ```
    /// let mut slots = kevy_replicate::slot::SlotTable::new();
    /// slots.insert_or_touch("r1", 5, 100); // first contact inserts
    /// slots.insert_or_touch("r1", 9, 200); // later acks advance
    /// assert_eq!(slots.get("r1").map(|s| (s.acked_offset, s.last_seen_ns)), Some((9, 200)));
    /// ```
    pub fn insert_or_touch(&mut self, id: &str, acked_offset: u64, now_ns: u64) {
        if let Some(s) = self.slots.iter_mut().find(|s| s.id == id) {
            s.last_seen_ns = now_ns;
            if acked_offset > s.acked_offset {
                s.acked_offset = acked_offset;
            }
            return;
        }
        self.slots.push(ReplicaSlot { id: id.to_string(), last_seen_ns: now_ns, acked_offset });
    }

    /// Remove the slot with the given id. Returns `true` if a slot
    /// was actually removed.
    ///
    /// ```
    /// let mut slots = kevy_replicate::slot::SlotTable::new();
    /// slots.insert_or_touch("r1", 0, 0);
    /// assert!(slots.remove("r1"));
    /// assert!(!slots.remove("r1")); // already gone
    /// ```
    pub fn remove(&mut self, id: &str) -> bool {
        if let Some(pos) = self.slots.iter().position(|s| s.id == id) {
            self.slots.swap_remove(pos);
            true
        } else {
            false
        }
    }

    /// Drop slots whose `last_seen_ns + window_ns ≤ now_ns`. Returns
    /// the ids of the dropped slots so callers can fire metrics or
    /// log lines. Order is unspecified (swap-remove internally).
    ///
    /// ```
    /// let mut slots = kevy_replicate::slot::SlotTable::new();
    /// slots.insert_or_touch("stale", 0, 100);
    /// slots.insert_or_touch("live", 0, 900);
    /// assert_eq!(slots.expire(1_000, 500), ["stale"]); // 100 + 500 <= 1000
    /// assert_eq!(slots.len(), 1);
    /// ```
    pub fn expire(&mut self, now_ns: u64, window_ns: u64) -> Vec<String> {
        let mut dropped = Vec::new();
        // Walk backward so swap_remove doesn't shift indices we still
        // need to visit.
        let mut i = self.slots.len();
        while i > 0 {
            i -= 1;
            let cutoff = self.slots[i].last_seen_ns.saturating_add(window_ns);
            if cutoff <= now_ns {
                let s = self.slots.swap_remove(i);
                dropped.push(s.id);
            }
        }
        dropped
    }

    /// Lowest acked offset across all tracked slots. Useful for the
    /// streaming loop to know how far back the backlog must still
    /// retain frames; `None` when the table is empty.
    ///
    /// ```
    /// use kevy_replicate::slot::SlotTable;
    ///
    /// let mut slots = SlotTable::new();
    /// assert_eq!(slots.min_acked_offset(), None);
    /// slots.insert_or_touch("fast", 90, 0);
    /// slots.insert_or_touch("slow", 40, 0);
    /// assert_eq!(slots.min_acked_offset(), Some(40)); // trim the backlog below 40 only
    /// ```
    pub fn min_acked_offset(&self) -> Option<u64> {
        self.slots.iter().map(|s| s.acked_offset).min()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn fresh_table_is_empty() {
        let t = SlotTable::new();
        assert!(t.is_empty());
        assert_eq!(t.len(), 0);
        assert_eq!(t.min_acked_offset(), None);
    }

    #[test]
    fn insert_then_get_returns_the_slot() {
        let mut t = SlotTable::new();
        t.insert_or_touch("a", 5, 100);
        let s = t.get("a").unwrap();
        assert_eq!(s.id, "a");
        assert_eq!(s.acked_offset, 5);
        assert_eq!(s.last_seen_ns, 100);
        assert_eq!(t.len(), 1);
    }

    #[test]
    fn touch_advances_last_seen_and_acked_offset() {
        let mut t = SlotTable::new();
        t.insert_or_touch("a", 5, 100);
        t.insert_or_touch("a", 9, 200);
        let s = t.get("a").unwrap();
        assert_eq!(s.acked_offset, 9);
        assert_eq!(s.last_seen_ns, 200);
        // No duplicate row.
        assert_eq!(t.len(), 1);
    }

    #[test]
    fn touch_with_lower_acked_offset_keeps_the_higher_one() {
        let mut t = SlotTable::new();
        t.insert_or_touch("a", 10, 100);
        // Peer reports a lower offset (bug / stale ack); slot keeps 10.
        t.insert_or_touch("a", 7, 200);
        let s = t.get("a").unwrap();
        assert_eq!(s.acked_offset, 10);
        assert_eq!(s.last_seen_ns, 200, "last_seen still advances");
    }

    #[test]
    fn remove_existing_returns_true_and_drops_slot() {
        let mut t = SlotTable::new();
        t.insert_or_touch("a", 1, 100);
        assert!(t.remove("a"));
        assert!(t.is_empty());
        assert_eq!(t.get("a"), None);
    }

    #[test]
    fn remove_missing_returns_false() {
        let mut t = SlotTable::new();
        assert!(!t.remove("missing"));
    }

    #[test]
    fn expire_drops_slots_past_window() {
        let mut t = SlotTable::new();
        t.insert_or_touch("old", 1, 100);
        t.insert_or_touch("fresh", 1, 500);
        // Window 200 ns; "now" = 350. "old" expires (100+200 ≤ 350),
        // "fresh" survives (500+200 > 350).
        let dropped = t.expire(350, 200);
        assert_eq!(dropped, vec!["old".to_string()]);
        assert_eq!(t.len(), 1);
        assert!(t.get("fresh").is_some());
    }

    #[test]
    fn expire_when_nothing_expires_returns_empty() {
        let mut t = SlotTable::new();
        t.insert_or_touch("a", 1, 1000);
        let dropped = t.expire(1100, 500); // 1000 + 500 = 1500 > 1100
        assert!(dropped.is_empty());
        assert_eq!(t.len(), 1);
    }

    #[test]
    fn expire_with_overflow_window_saturates_does_not_panic() {
        // `last_seen + window` must saturate at u64::MAX rather than
        // wrap. With now=u64::MAX-1 the saturated cutoff stays above
        // now, so the slot survives. (At now=u64::MAX a saturated
        // cutoff would equal now and the slot does expire — but the
        // safety invariant being asserted is "no panic".)
        let mut t = SlotTable::new();
        t.insert_or_touch("a", 1, u64::MAX - 10);
        let dropped = t.expire(u64::MAX - 1, u64::MAX);
        assert!(dropped.is_empty());
        assert_eq!(t.len(), 1);
    }

    #[test]
    fn min_acked_offset_returns_the_floor() {
        let mut t = SlotTable::new();
        t.insert_or_touch("a", 7, 100);
        t.insert_or_touch("b", 3, 100);
        t.insert_or_touch("c", 12, 100);
        assert_eq!(t.min_acked_offset(), Some(3));
    }

    #[test]
    fn touch_or_insert_unacked_never_advances_acked() {
        let mut t = SlotTable::new();
        t.insert_or_touch("a", 7, 100);
        // Disconnect-time touch: last_seen refreshes, acked stays.
        t.touch_or_insert_unacked("a", 200);
        let s = t.get("a").unwrap();
        assert_eq!(s.last_seen_ns, 200);
        assert_eq!(s.acked_offset, 7, "close must not write sent-as-acked");
        // Unknown id inserts at the truthful floor: never acked.
        t.touch_or_insert_unacked("b", 300);
        assert_eq!(t.get("b").unwrap().acked_offset, 0);
    }

    #[test]
    fn iter_visits_every_slot() {
        let mut t = SlotTable::new();
        t.insert_or_touch("a", 1, 100);
        t.insert_or_touch("b", 2, 100);
        let mut ids: Vec<_> = t.iter().map(|s| s.id.clone()).collect();
        ids.sort();
        assert_eq!(ids, vec!["a".to_string(), "b".to_string()]);
    }

    #[test]
    fn expire_all_when_window_zero() {
        let mut t = SlotTable::new();
        t.insert_or_touch("a", 1, 100);
        t.insert_or_touch("b", 1, 100);
        let dropped = t.expire(100, 0);
        assert_eq!(dropped.len(), 2);
        assert!(t.is_empty());
    }
}