routers_realtime 0.5.0

//! Completion-frontier tracker.
//!
//! A partition worker reads its raw journal through a filtered JetStream view, so
//! stream sequences arrive in order but with gaps (other partitions' sequences are
//! skipped). The tracker records each sequence's observe and complete events and
//! reports the completion frontier `F`: the greatest position with no observed-but-
//! incomplete sequence at or below it. `F` never decreases and never leads commits;
//! recovery re-opens delivery at `F + 1`.

use alloc::collections::BTreeSet;
use core::time::Duration;

use tokio::time::Instant;
use tracing::warn;

use crate::store::checkpoint::PartitionFrontier;

/// The persistence cadence; a write is due on whichever trigger fires first.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct FrontierConfig {
    /// Persist once this many completions have landed since the last write.
    pub every: u32,
    /// Persist at least this often even when completions are sparse.
    pub at_least_every: Duration,
}

impl Default for FrontierConfig {
    /// The spec defaults: every 64 completions, and at least once per second.
    fn default() -> Self {
        Self {
            every: 64,
            at_least_every: Duration::from_secs(1),
        }
    }
}

/// The outcome of [`observe`](FrontierTracker::observe): what the delivered
/// sequence was relative to state the tracker already holds.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Observed {
    /// A sequence not seen before; it is now outstanding.
    New,
    /// A sequence already outstanding — a redelivery still in flight.
    Duplicate,
    /// A sequence at or below the frontier — already terminal; ack and skip.
    BehindFrontier,
}

/// How far [`complete`](FrontierTracker::complete) moved the frontier; `from`
/// and `to` are the values before and after, equal when nothing moved.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Advance {
    /// The frontier before this completion.
    pub from: u64,
    /// The frontier after this completion.
    pub to: u64,
}

impl Advance {
    /// Whether the frontier actually moved.
    pub fn advanced(&self) -> bool {
        self.to > self.from
    }
}

/// Tracks out-of-order completions for one partition and reports the contiguous
/// completed prefix of its filtered raw input as the frontier `F`.
#[derive(Clone, Debug)]
pub struct FrontierTracker {
    /// Stamped onto every [`PartitionFrontier`] it emits.
    partition: u16,
    /// The current frontier `F`, monotonic non-decreasing.
    frontier: u64,
    /// The greatest sequence ever completed.
    max_completed: u64,
    /// Observed-but-not-completed sequences; the oldest hole is `first()`.
    outstanding: BTreeSet<u64>,
    /// The frontier value most recently confirmed persisted by the caller.
    persisted: u64,
    /// Completions accumulated since the last persist, for the count trigger.
    completions_since_persist: u32,
    /// When the last persist was confirmed, for the time trigger.
    last_persist: Instant,
    cfg: FrontierConfig,
}

impl FrontierTracker {
    /// Build a tracker for `partition`, starting from the recovered `initial`
    /// frontier (`None` if never checkpointed).
    pub fn new(partition: u16, initial: Option<PartitionFrontier>, now: Instant) -> Self {
        Self::with_config(partition, initial, now, FrontierConfig::default())
    }

    /// [`new`](Self::new) with an explicit [`FrontierConfig`].
    pub fn with_config(
        partition: u16,
        initial: Option<PartitionFrontier>,
        now: Instant,
        cfg: FrontierConfig,
    ) -> Self {
        let start = initial.map_or(0, |f| {
            debug_assert_eq!(
                f.partition, partition,
                "recovered frontier for the wrong partition",
            );
            f.sequence
        });
        Self {
            partition,
            frontier: start,
            max_completed: start,
            outstanding: BTreeSet::new(),
            persisted: start,
            completions_since_persist: 0,
            last_persist: now,
            cfg,
        }
    }

    /// Record that `seq` has been delivered, before any processing.
    ///
    /// Reports [`Observed::BehindFrontier`] at or below the frontier,
    /// [`Observed::Duplicate`] if already outstanding, else [`Observed::New`].
    /// Observing may raise the frontier but never lowers it.
    pub fn observe(&mut self, seq: u64) -> Observed {
        if seq <= self.frontier {
            return Observed::BehindFrontier;
        }
        if !self.outstanding.insert(seq) {
            return Observed::Duplicate;
        }
        self.recompute();
        Observed::New
    }

    /// Record that `seq` was made terminal and advance the frontier over any
    /// prefix this unblocks; a sequence already behind the frontier is a no-op.
    pub fn complete(&mut self, seq: u64) -> Advance {
        let from = self.frontier;
        if seq <= self.frontier {
            return Advance { from, to: from };
        }
        if !self.outstanding.remove(&seq) {
            debug_assert!(
                false,
                "completing unobserved sequence {seq} in partition {}",
                self.partition,
            );
            warn!(
                partition = self.partition,
                sequence = seq,
                "completing a sequence that was never observed",
            );
        }
        self.max_completed = self.max_completed.max(seq);
        self.completions_since_persist = self.completions_since_persist.saturating_add(1);
        let to = self.recompute();
        Advance { from, to }
    }

    /// Recompute the frontier from `outstanding` and `max_completed`, clamped so
    /// it never decreases, and return the new value.
    fn recompute(&mut self) -> u64 {
        let candidate = match self.outstanding.iter().next() {
            Some(&oldest) => oldest - 1,
            None => self.max_completed,
        };
        self.frontier = self.frontier.max(candidate);
        self.frontier
    }

    /// The current frontier `F`.
    pub fn frontier(&self) -> u64 {
        self.frontier
    }

    /// How many sequences are observed but not yet completed.
    pub fn outstanding(&self) -> usize {
        self.outstanding.len()
    }

    /// The oldest still-incomplete sequence, or `None` if all have completed.
    pub fn oldest_outstanding(&self) -> Option<u64> {
        self.outstanding.iter().next().copied()
    }

    /// The record to persist now, or `None` when no write is due.
    ///
    /// Due when `frontier > persisted` and either trigger has fired. The caller
    /// writes it, then confirms via [`persisted`](Self::persisted).
    pub fn due(&self, now: Instant) -> Option<PartitionFrontier> {
        if self.frontier <= self.persisted {
            return None;
        }
        let by_count = self.completions_since_persist >= self.cfg.every;
        let by_time = now.saturating_duration_since(self.last_persist) >= self.cfg.at_least_every;
        (by_count || by_time).then_some(PartitionFrontier {
            partition: self.partition,
            sequence: self.frontier,
        })
    }

    /// Confirm the frontier was durably persisted up to `seq`, resetting both
    /// triggers; the persisted mark only ever advances.
    pub fn persisted(&mut self, seq: u64, now: Instant) {
        self.persisted = self.persisted.max(seq);
        self.completions_since_persist = 0;
        self.last_persist = now;
    }
}

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

    fn fresh(now: Instant) -> FrontierTracker {
        FrontierTracker::new(0, None, now)
    }

    #[test]
    fn fresh_partition_starts_at_zero() {
        let t = fresh(Instant::now());
        assert_eq!(t.frontier(), 0);
        assert_eq!(t.outstanding(), 0);
        assert_eq!(t.oldest_outstanding(), None);
    }

    #[test]
    fn in_order_completions_advance_one_by_one() {
        let t0 = Instant::now();
        let mut t = fresh(t0);
        for seq in 1..=3 {
            assert_eq!(t.observe(seq), Observed::New);
        }
        assert_eq!(t.frontier(), 0);

        assert_eq!(t.complete(1), Advance { from: 0, to: 1 });
        assert_eq!(t.complete(2), Advance { from: 1, to: 2 });
        assert_eq!(t.complete(3), Advance { from: 2, to: 3 });
        assert_eq!(t.outstanding(), 0);
    }

    #[test]
    fn out_of_order_holds_then_jumps() {
        let mut t = fresh(Instant::now());
        for seq in 3..=5 {
            t.observe(seq);
        }
        assert_eq!(t.frontier(), 2);

        let held = t.complete(5);
        assert_eq!(held, Advance { from: 2, to: 2 });
        assert!(!held.advanced());
        assert_eq!(t.oldest_outstanding(), Some(3));

        assert_eq!(t.complete(3), Advance { from: 2, to: 3 });
        assert_eq!(t.complete(4), Advance { from: 3, to: 5 });
        assert_eq!(t.frontier(), 5);
        assert_eq!(t.outstanding(), 0);
    }

    #[test]
    fn gaps_in_delivered_sequences_are_not_holes() {
        let mut t = fresh(Instant::now());
        t.observe(10);
        t.observe(20);
        t.observe(30);
        assert_eq!(t.frontier(), 9);

        assert_eq!(t.complete(10), Advance { from: 9, to: 19 });
        assert_eq!(t.complete(20), Advance { from: 19, to: 29 });
        assert_eq!(t.complete(30), Advance { from: 29, to: 30 });
    }

    #[test]
    fn frontier_never_regresses() {
        let mut t = fresh(Instant::now());
        t.observe(5);
        t.complete(5);
        assert_eq!(t.frontier(), 5);

        assert_eq!(t.observe(5), Observed::BehindFrontier);
        assert_eq!(t.observe(3), Observed::BehindFrontier);
        assert_eq!(t.complete(4), Advance { from: 5, to: 5 });
        assert_eq!(t.frontier(), 5);
        assert_eq!(t.outstanding(), 0);
    }

    #[test]
    fn redelivery_classification() {
        let mut t = fresh(Instant::now());
        assert_eq!(t.observe(7), Observed::New);
        assert_eq!(t.observe(7), Observed::Duplicate);
        t.complete(7);
        assert_eq!(t.observe(7), Observed::BehindFrontier);
    }

    #[test]
    fn recovery_honours_initial_frontier() {
        let initial = PartitionFrontier {
            partition: 4,
            sequence: 100,
        };
        let mut t = FrontierTracker::new(4, Some(initial), Instant::now());
        assert_eq!(t.frontier(), 100);

        assert_eq!(t.observe(100), Observed::BehindFrontier);
        assert_eq!(t.observe(50), Observed::BehindFrontier);

        assert_eq!(t.observe(101), Observed::New);
        assert_eq!(t.frontier(), 100);
        assert_eq!(t.complete(101), Advance { from: 100, to: 101 });
    }

    #[test]
    fn due_obeys_the_count_trigger() {
        let t0 = Instant::now();
        let cfg = FrontierConfig {
            every: 3,
            at_least_every: Duration::from_secs(3600),
        };
        let mut t = FrontierTracker::with_config(0, None, t0, cfg);
        for seq in 1..=3 {
            t.observe(seq);
        }

        t.complete(1);
        assert_eq!(t.due(t0), None);

        t.complete(2);
        t.complete(3);
        let due = t.due(t0).expect("count trigger should fire");
        assert_eq!(
            due,
            PartitionFrontier {
                partition: 0,
                sequence: 3,
            }
        );

        t.persisted(due.sequence, t0);
        assert_eq!(t.due(t0), None);
    }

    #[test]
    fn due_obeys_the_time_trigger() {
        let t0 = Instant::now();
        let cfg = FrontierConfig {
            every: 1_000,
            at_least_every: Duration::from_secs(1),
        };
        let mut t = FrontierTracker::with_config(0, None, t0, cfg);
        t.observe(1);
        t.complete(1);

        assert_eq!(t.due(t0), None);
        assert_eq!(t.due(t0 + Duration::from_millis(999)), None);

        let now = t0 + Duration::from_millis(1_000);
        let due = t.due(now).expect("time trigger should fire");
        assert_eq!(due.sequence, 1);

        t.persisted(due.sequence, now);
        assert_eq!(t.due(now + Duration::from_millis(500)), None);
    }

    #[test]
    fn due_requires_the_frontier_to_have_moved() {
        let t0 = Instant::now();
        let cfg = FrontierConfig {
            every: 1,
            at_least_every: Duration::from_millis(0),
        };
        // Recover at 5, so frontier and persisted mark start equal.
        let initial = PartitionFrontier {
            partition: 0,
            sequence: 5,
        };
        let mut t = FrontierTracker::with_config(0, Some(initial), t0, cfg);

        assert_eq!(t.observe(6), Observed::New);
        assert_eq!(t.frontier(), 5);
        assert_eq!(t.due(t0 + Duration::from_secs(10)), None);

        assert_eq!(t.complete(6), Advance { from: 5, to: 6 });
        assert_eq!(
            t.due(t0),
            Some(PartitionFrontier {
                partition: 0,
                sequence: 6,
            })
        );
    }
}