reifydb_flow/window/kind/
rolling.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_value::value::{datetime::DateTime, duration::Duration};
5
6use crate::{
7	operator::state::seal::{
8		coord::Coord,
9		rule::{EvictionRule, SealRule},
10	},
11	window::coord::RowSpan,
12};
13
14pub struct RollingOverTime {
15	size: Duration,
16	lag: Duration,
17}
18
19impl RollingOverTime {
20	pub fn new(size: Duration, lag: Duration) -> Self {
21		Self {
22			size,
23			lag,
24		}
25	}
26
27	pub fn span(&self) -> Duration {
28		self.size.try_add(self.lag).unwrap_or(self.lag)
29	}
30
31	pub fn seal_rule(&self, lateness: Duration) -> SealRule {
32		SealRule::rolling(self.span(), lateness)
33	}
34
35	pub fn eviction_rule(&self) -> EvictionRule {
36		EvictionRule::rolling(self.span())
37	}
38
39	pub fn eviction_cutoff(&self, ledger: DateTime) -> Option<DateTime> {
40		ledger.checked_sub_span(self.span())
41	}
42
43	pub fn seal_horizon(&self, ledger: DateTime, lateness: Duration) -> DateTime {
44		ledger.saturating_sub_span(self.seal_rule(lateness).admissible().duration())
45	}
46}
47
48pub struct RollingOverRows {
49	capacity: RowSpan,
50}
51
52impl RollingOverRows {
53	pub fn new(capacity: RowSpan) -> Self {
54		Self {
55			capacity,
56		}
57	}
58
59	pub fn capacity(&self) -> usize {
60		self.capacity.rows() as usize
61	}
62}
63
64#[cfg(test)]
65mod tests {
66	use reifydb_value::factory::time::at_millis;
67
68	use super::*;
69
70	fn ms(millis: u64) -> Duration {
71		Duration::from_milliseconds_const(millis as i64)
72	}
73
74	#[test]
75	fn a_row_capacity_has_no_lag_no_lateness_and_no_horizon_to_ask_for() {
76		// A rolling window over ROWS always has a current value, so there is no lag, lateness or seal
77		// instant to ask for. Lag and lateness are milliseconds, and subtracting them from a row number
78		// would drop rows in proportion to the lag; the type carries no such method to answer wrongly.
79		let rows = RollingOverRows::new(RowSpan::of(64));
80
81		assert_eq!(rows.capacity(), 64);
82	}
83
84	#[test]
85	fn the_rolling_span_is_the_size_extended_by_the_lag() {
86		// Lag shifts the whole window back in time, so a lagged rolling window must retain size + lag
87		// or the lagged read falls off the end of the buffer. Eviction cutoff and seal policy are
88		// both built from this one span.
89		assert_eq!(RollingOverTime::new(ms(5_000), ms(0)).span(), ms(5_000));
90		assert_eq!(RollingOverTime::new(ms(5_000), ms(2_000)).span(), ms(7_000));
91	}
92
93	#[test]
94	fn eviction_uses_the_bare_span_and_sealing_adds_the_lateness() {
95		// Rolling admits a late row inside the lateness but evicts on the bare span. An eviction that
96		// also waited out the lateness keeps every window one lateness-period too wide, inflating every
97		// aggregate it publishes.
98		let rolling = RollingOverTime::new(ms(5_000), ms(0));
99
100		assert_eq!(rolling.eviction_cutoff(at_millis(8_000)), Some(at_millis(3_000)));
101		assert_eq!(rolling.seal_horizon(at_millis(8_000), ms(200)), at_millis(2_800));
102	}
103
104	#[test]
105	fn a_ledger_younger_than_the_span_evicts_nothing_rather_than_clamping_to_the_epoch() {
106		// At startup the ledger sits near the epoch while the span is minutes. Underflowing yields a
107		// cutoff near the maximum instant and evicts everything; clamping to the epoch is wrong too,
108		// since eviction is inclusive and a row at the epoch could then never be retained.
109		let rolling = RollingOverTime::new(ms(5_000), ms(0));
110
111		assert_eq!(rolling.eviction_cutoff(at_millis(0)), None);
112		assert_eq!(rolling.eviction_cutoff(at_millis(1_000)), None);
113		assert_eq!(
114			rolling.eviction_cutoff(at_millis(5_000)),
115			Some(at_millis(0)),
116			"a span that has exactly elapsed yields a real cutoff, not another None"
117		);
118		assert_eq!(rolling.seal_horizon(at_millis(1_000), ms(200)), at_millis(0));
119	}
120
121	#[test]
122	fn the_seal_horizon_never_sits_later_than_the_eviction_cutoff() {
123		// Sealing decides what may still be amended, eviction what is still stored. A horizon later
124		// than the cutoff declares a window sealed while its rows are still retained, refusing a late
125		// row the window could have accepted.
126		for lag in [ms(0), ms(1), ms(9_000)] {
127			for lateness in [ms(0), ms(1), ms(9_000)] {
128				let rolling = RollingOverTime::new(ms(5_000), lag);
129				let cutoff = rolling
130					.eviction_cutoff(at_millis(60_000))
131					.expect("a ledger well past the span must yield a cutoff");
132				assert!(
133					rolling.seal_horizon(at_millis(60_000), lateness) <= cutoff,
134					"horizon passed the cutoff at lag {lag:?} lateness {lateness:?}"
135				);
136			}
137		}
138	}
139}
reifydb_flow/window/kind/rolling.rs

reifydb_flow/window/kind/
rolling.rs
