laminar_core/time/
mod.rs

1//! Event time, watermarks, and timer management.
2mod cast;
3mod duration_str;
4mod event_time;
5mod filter;
6mod watermark;
7
8pub use cast::{cast_to_millis_array, CastError};
9pub use duration_str::parse_duration_str;
10pub use event_time::{EventTimeError, EventTimeExtractor, ExtractionMode, TimestampField};
11
12pub use filter::{filter_batch_by_timestamp, FilterError, ThresholdOp};
13
14pub use watermark::{
15    AscendingTimestampsGenerator, BoundedOutOfOrdernessGenerator, PeriodicGenerator,
16    ProcessingTimeGenerator, PunctuatedGenerator, SourceProvidedGenerator, WatermarkGenerator,
17    WatermarkRestoreError, WatermarkTracker, DEFAULT_MAX_FUTURE_SKEW_MS,
18};
19
20use smallvec::SmallVec;
21use std::cmp::Ordering;
22use std::collections::BinaryHeap;
23use std::time::{SystemTime, UNIX_EPOCH};
24
25/// Wall clock as epoch milliseconds; `0` if unreadable (callers fail open).
26#[must_use]
27pub fn now_unix_millis() -> i64 {
28    SystemTime::now()
29        .duration_since(UNIX_EPOCH)
30        .map_or(0, |d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX))
31}
32
33/// Timer key type optimized for window IDs (16 bytes).
34///
35/// Uses `SmallVec` to avoid heap allocation for keys up to 16 bytes,
36/// which covers the common case of `WindowId` keys.
37pub type TimerKey = SmallVec<[u8; 16]>;
38
39/// Collection type for fired timers.
40///
41/// Uses `SmallVec` to avoid heap allocation when few timers fire per poll.
42/// Size 8 covers most practical cases where timers fire in small batches.
43pub type FiredTimersVec = SmallVec<[TimerRegistration; 8]>;
44
45/// A monotonically-increasing assertion that no events with timestamps
46/// earlier than this will arrive. Drives window emission, late-event
47/// detection, and cross-operator time alignment.
48#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
49pub struct Watermark(pub i64);
50
51impl Watermark {
52    /// Creates a new watermark with the given timestamp.
53    #[inline]
54    #[must_use]
55    pub fn new(timestamp: i64) -> Self {
56        Self(timestamp)
57    }
58
59    /// Returns the watermark timestamp in milliseconds.
60    #[inline]
61    #[must_use]
62    pub fn timestamp(&self) -> i64 {
63        self.0
64    }
65
66    /// Checks if an event is late relative to this watermark.
67    ///
68    /// An event is considered late if its timestamp is strictly less than
69    /// the watermark timestamp.
70    #[inline]
71    #[must_use]
72    pub fn is_late(&self, event_time: i64) -> bool {
73        event_time < self.0
74    }
75
76    /// Returns the minimum (earlier) of two watermarks.
77    #[must_use]
78    pub fn min(self, other: Self) -> Self {
79        Self(self.0.min(other.0))
80    }
81
82    /// Returns the maximum (later) of two watermarks.
83    #[must_use]
84    pub fn max(self, other: Self) -> Self {
85        Self(self.0.max(other.0))
86    }
87}
88
89impl Default for Watermark {
90    fn default() -> Self {
91        Self(i64::MIN)
92    }
93}
94
95impl From<i64> for Watermark {
96    fn from(timestamp: i64) -> Self {
97        Self(timestamp)
98    }
99}
100
101impl From<Watermark> for i64 {
102    fn from(watermark: Watermark) -> Self {
103        watermark.0
104    }
105}
106
107/// A timer registration for delayed processing.
108///
109/// Timers are used by operators to schedule future actions, typically for
110/// window triggering or timeouts.
111#[derive(Debug, Clone, PartialEq, Eq)]
112pub struct TimerRegistration {
113    /// Unique timer ID
114    pub id: u64,
115    /// Scheduled timestamp (event time, in milliseconds)
116    pub timestamp: i64,
117    /// Timer key (for keyed operators).
118    /// Uses `TimerKey` (`SmallVec`) to avoid heap allocation for keys up to 16 bytes.
119    pub key: Option<TimerKey>,
120    /// The index of the operator that registered this timer
121    pub operator_index: Option<usize>,
122}
123
124impl Ord for TimerRegistration {
125    fn cmp(&self, other: &Self) -> Ordering {
126        // Reverse ordering for min-heap behavior (earliest first)
127        match other.timestamp.cmp(&self.timestamp) {
128            Ordering::Equal => other.id.cmp(&self.id),
129            ord => ord,
130        }
131    }
132}
133
134impl PartialOrd for TimerRegistration {
135    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
136        Some(self.cmp(other))
137    }
138}
139
140// Threshold at which the timer service logs a warning about accumulated timers.
141// This typically indicates a stalled watermark preventing timer firing.
142const TIMER_WARN_THRESHOLD: usize = 100_000;
143
144/// Timer service for scheduling and managing timers.
145///
146/// The timer service maintains a priority queue of timer registrations,
147/// ordered by timestamp. Operators can register timers to be fired at
148/// specific event times.
149pub struct TimerService {
150    timers: BinaryHeap<TimerRegistration>,
151    next_timer_id: u64,
152}
153
154impl TimerService {
155    /// Creates a new timer service.
156    #[must_use]
157    pub fn new() -> Self {
158        Self {
159            timers: BinaryHeap::new(),
160            next_timer_id: 0,
161        }
162    }
163
164    /// Creates a new timer service with pre-allocated capacity.
165    #[must_use]
166    pub fn with_capacity(capacity: usize) -> Self {
167        Self {
168            timers: BinaryHeap::with_capacity(capacity),
169            next_timer_id: 0,
170        }
171    }
172
173    /// Registers a new timer.
174    ///
175    /// Returns the unique timer ID that can be used to cancel the timer.
176    ///
177    /// # Arguments
178    ///
179    /// * `timestamp` - The event time at which the timer should fire
180    /// * `key` - Optional key for keyed operators
181    /// * `operator_index` - Optional index of the operator registering the timer(must match the index in the reactor)
182    pub fn register_timer(
183        &mut self,
184        timestamp: i64,
185        key: Option<TimerKey>,
186        operator_index: Option<usize>,
187    ) -> u64 {
188        let id = self.next_timer_id;
189        self.next_timer_id += 1;
190
191        self.timers.push(TimerRegistration {
192            id,
193            timestamp,
194            key,
195            operator_index,
196        });
197
198        if self.timers.len() == TIMER_WARN_THRESHOLD {
199            tracing::warn!(
200                pending = self.timers.len(),
201                "Timer heap reached {} pending timers — watermark may be stalled",
202                TIMER_WARN_THRESHOLD,
203            );
204        }
205
206        id
207    }
208
209    /// Polls for timers that should fire at or before the given timestamp.
210    ///
211    /// Returns all timers with timestamps <= `current_time`, in order.
212    /// Uses `FiredTimersVec` (`SmallVec`) to avoid heap allocation when few timers fire.
213    ///
214    /// # Panics
215    ///
216    /// This function should not panic under normal circumstances. The internal
217    /// `expect` is only called after verifying the heap is not empty via `peek`.
218    #[inline]
219    pub fn poll_timers(&mut self, current_time: i64) -> FiredTimersVec {
220        let mut fired = FiredTimersVec::new();
221
222        while let Some(timer) = self.timers.peek() {
223            if timer.timestamp <= current_time {
224                // SAFETY: We just peeked and confirmed the heap is not empty
225                fired.push(self.timers.pop().expect("heap should not be empty"));
226            } else {
227                break;
228            }
229        }
230
231        fired
232    }
233
234    /// Cancels a timer by ID.
235    ///
236    /// Returns `true` if the timer was found and cancelled.
237    pub fn cancel_timer(&mut self, id: u64) -> bool {
238        let count_before = self.timers.len();
239        self.timers.retain(|t| t.id != id);
240        self.timers.len() < count_before
241    }
242
243    /// Returns the number of pending timers.
244    #[must_use]
245    pub fn pending_count(&self) -> usize {
246        self.timers.len()
247    }
248
249    /// Returns the timestamp of the next timer to fire, if any.
250    #[must_use]
251    pub fn next_timer_timestamp(&self) -> Option<i64> {
252        self.timers.peek().map(|t| t.timestamp)
253    }
254
255    /// Clears all pending timers.
256    pub fn clear(&mut self) {
257        self.timers.clear();
258    }
259}
260
261impl Default for TimerService {
262    fn default() -> Self {
263        Self::new()
264    }
265}
266
267/// Errors that can occur in time operations.
268#[derive(Debug, thiserror::Error)]
269pub enum TimeError {
270    /// Invalid timestamp value
271    #[error("Invalid timestamp: {0}")]
272    InvalidTimestamp(i64),
273
274    /// Timer not found
275    #[error("Timer not found: {0}")]
276    TimerNotFound(u64),
277
278    /// Watermark regression (going backwards)
279    #[error("Watermark regression: current={current}, new={new}")]
280    WatermarkRegression {
281        /// Current watermark value
282        current: i64,
283        /// Attempted new watermark value
284        new: i64,
285    },
286}
287
288#[cfg(test)]
289mod tests;
laminar_core/time/mod.rs

laminar_core/time/
mod.rs
