Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268use tokio::sync::watch;
#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]pub struct HydrantCursor(u64);
impl HydrantCursor { pub const fn new(id: u64) -> Self { Self(id) }
pub const fn raw(self) -> u64 { self.0 }}
impl From<u64> for HydrantCursor { fn from(id: u64) -> Self { Self(id) }}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]pub enum Coverage { Warming { events_processed: u64, last_cursor: HydrantCursor, }, Ready { events_processed: u64, last_cursor: HydrantCursor, },}
impl Default for Coverage { fn default() -> Self { Self::Warming { events_processed: 0, last_cursor: HydrantCursor::default(), } }}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]pub struct PromotionSignal { pub rev_micros: Option<u64>, pub now_micros: u64, pub skew_micros: u64,}
impl PromotionSignal { pub fn caught_up(self) -> bool { let Some(rev) = self.rev_micros else { return false; }; self.now_micros.abs_diff(rev) <= self.skew_micros }}
impl Coverage { pub const fn is_ready(self) -> bool { matches!(self, Self::Ready { .. }) }
pub const fn events_processed(self) -> u64 { match self { Self::Warming { events_processed, .. } | Self::Ready { events_processed, .. } => events_processed, } }
pub const fn last_cursor(self) -> HydrantCursor { match self { Self::Warming { last_cursor, .. } | Self::Ready { last_cursor, .. } => last_cursor, } }
pub fn advance(self, cursor: HydrantCursor) -> Self { debug_assert!( cursor.raw() >= self.last_cursor().raw(), "hydrant cursor regressed: {} -> {}", self.last_cursor().raw(), cursor.raw(), ); let processed = self.events_processed().saturating_add(1); match self { Self::Warming { .. } => Self::Warming { events_processed: processed, last_cursor: cursor, }, Self::Ready { .. } => Self::Ready { events_processed: processed, last_cursor: cursor, }, } }
pub fn maybe_promote(self, signal: PromotionSignal) -> Self { match self { Self::Ready { .. } => self, Self::Warming { events_processed, last_cursor, } if signal.caught_up() => Self::Ready { events_processed, last_cursor, }, warming => warming, } }
pub const fn force_ready(self) -> Self { match self { Self::Ready { .. } => self, Self::Warming { events_processed, last_cursor, } => Self::Ready { events_processed, last_cursor, }, } }}
#[derive(Debug)]pub struct CoverageWatch { tx: watch::Sender<Coverage>,}
impl Default for CoverageWatch { fn default() -> Self { Self::new() }}
impl CoverageWatch { pub fn new() -> Self { let (tx, _) = watch::channel(Coverage::default()); Self { tx } }
pub fn snapshot(&self) -> Coverage { *self.tx.borrow() }
pub fn subscribe(&self) -> watch::Receiver<Coverage> { self.tx.subscribe() }
pub fn update<F>(&self, transform: F) where F: FnOnce(Coverage) -> Coverage, { self.tx.send_if_modified(|c| { let next = transform(*c); if *c == next { false } else { *c = next; true } }); }}
#[cfg(test)]mod tests { use super::*;
const SKEW: u64 = 60_000_000;
fn signal(rev: u64, now: u64) -> PromotionSignal { PromotionSignal { rev_micros: Some(rev), now_micros: now, skew_micros: SKEW, } }
#[test] fn defaults_to_warming_at_zero() { let c = Coverage::default(); assert!(!c.is_ready()); assert_eq!(c.events_processed(), 0); assert_eq!(c.last_cursor(), HydrantCursor::default()); }
#[test] fn advance_increments_and_tracks_cursor() { let c = Coverage::default() .advance(HydrantCursor::new(7)) .advance(HydrantCursor::new(9)); assert_eq!(c.events_processed(), 2); assert_eq!(c.last_cursor(), HydrantCursor::new(9)); assert!(!c.is_ready()); }
#[test] fn promotion_requires_recent_rev() { let now = 1_000_000_000; let recent = now - SKEW / 2; let stale = now - SKEW * 10;
let c = Coverage::default().advance(HydrantCursor::new(1)); assert!(!c.maybe_promote(signal(stale, now)).is_ready()); assert!(c.maybe_promote(signal(recent, now)).is_ready()); }
#[test] fn promotion_preserves_counters() { let now = 1_000_000_000; let c = Coverage::default() .advance(HydrantCursor::new(3)) .maybe_promote(signal(now, now)) .advance(HydrantCursor::new(4)); assert!(c.is_ready()); assert_eq!(c.events_processed(), 2); assert_eq!(c.last_cursor(), HydrantCursor::new(4)); }
#[test] fn ready_is_sticky() { let now = 1_000_000_000; let stale = now - SKEW * 10; let c = Coverage::default() .advance(HydrantCursor::new(1)) .maybe_promote(signal(now, now)) .advance(HydrantCursor::new(2)) .maybe_promote(signal(stale, now)); assert!(c.is_ready()); }
#[test] fn future_rev_within_skew_promotes() { let now = 1_000_000_000; let near_future = now + SKEW / 2; let c = Coverage::default() .advance(HydrantCursor::new(1)) .maybe_promote(signal(near_future, now)); assert!(c.is_ready()); }
#[test] fn future_rev_beyond_skew_does_not_promote() { let now = 1_000_000_000; let far_future = now + SKEW * 10; let c = Coverage::default() .advance(HydrantCursor::new(1)) .maybe_promote(signal(far_future, now)); assert!(!c.is_ready()); }
#[test] fn missing_rev_does_not_promote() { let c = Coverage::default().advance(HydrantCursor::new(1)); let s = PromotionSignal { rev_micros: None, now_micros: 1_000_000_000, skew_micros: SKEW, }; assert!(!c.maybe_promote(s).is_ready()); }}