Struct KeyedWatermarkTracker 

pub struct KeyedWatermarkTracker<K: Hash + Eq + Clone> { /* private fields */ }

Tracks watermarks per logical key.

Provides fine-grained watermark tracking for multi-tenant workloads and scenarios with significant event-time skew between keys.

Research Background

Based on research (March 2025), keyed watermarks achieve 99%+ accuracy compared to 63-67% with global watermarks.

Example

use laminar_core::time::{KeyedWatermarkTracker, KeyedWatermarkConfig, Watermark};
use std::time::Duration;

let config = KeyedWatermarkConfig::with_bounded_delay(Duration::from_secs(5));
let mut tracker: KeyedWatermarkTracker<String> = KeyedWatermarkTracker::new(config);

// Fast tenant advances quickly
tracker.update("tenant_a".to_string(), 10_000);
tracker.update("tenant_a".to_string(), 15_000);

// Slow tenant at earlier time
tracker.update("tenant_b".to_string(), 5_000);

// Per-key watermarks differ
assert_eq!(tracker.watermark_for_key(&"tenant_a".to_string()), Some(10_000));
assert_eq!(tracker.watermark_for_key(&"tenant_b".to_string()), Some(0));

// Global watermark is minimum
assert_eq!(tracker.global_watermark(), Some(Watermark::new(0)));

Implementations

impl<K: Hash + Eq + Clone> KeyedWatermarkTracker<K>

pub fn new(config: KeyedWatermarkConfig) -> Self

Creates a new keyed watermark tracker with the given configuration.

pub fn with_defaults() -> Self

Creates a tracker with default configuration.

pub fn update( &mut self, key: K, event_time: i64, ) -> Result<Option<Watermark>, KeyedWatermarkError>

Updates the watermark for a specific key.

Returns

• Ok(Some(Watermark)) if the global watermark changes
• Ok(None) if no global change

Errors

Returns KeyedWatermarkError::MaxKeysReached if max_keys is reached and the RejectNew eviction policy is configured.

Example

use laminar_core::time::{KeyedWatermarkTracker, KeyedWatermarkConfig};
use std::time::Duration;

let config = KeyedWatermarkConfig::with_bounded_delay(Duration::from_millis(100));
let mut tracker: KeyedWatermarkTracker<String> = KeyedWatermarkTracker::new(config);

// First update creates the key
let wm = tracker.update("key1".to_string(), 1000).unwrap();
assert!(wm.is_some()); // Global watermark advances

pub fn update_batch( &mut self, events: &[(K, i64)], ) -> Result<Option<Watermark>, KeyedWatermarkError>

Batch update for multiple events (more efficient).

Returns the new global watermark if it changed.

Errors

Returns KeyedWatermarkError::MaxKeysReached if max_keys is reached and the RejectNew eviction policy is configured.

pub fn watermark_for_key(&self, key: &K) -> Option<i64>

Returns the watermark for a specific key.

pub fn global_watermark(&self) -> Option<Watermark>

Returns the global watermark (minimum across active keys).

pub fn is_late(&self, key: &K, event_time: i64) -> bool

Checks if an event is late for its key.

Uses the key’s individual watermark, not the global watermark. If the key doesn’t exist, returns false (not late).

pub fn is_late_global(&self, event_time: i64) -> bool

Checks if an event is late using the global watermark.

Use this for cross-key ordering guarantees.

pub fn mark_idle(&mut self, key: &K) -> Option<Watermark>

Marks a key as idle, excluding it from global watermark calculation.

Returns Some(Watermark) if the global watermark advances.

pub fn mark_active(&mut self, key: &K)

Marks a key as active again.

pub fn check_idle_keys(&mut self) -> Option<Watermark>

Checks for keys that have been idle longer than the timeout.

Should be called periodically from Ring 1.

Returns Some(Watermark) if marking idle keys causes the global watermark to advance.

pub fn active_key_count(&self) -> usize

Returns the number of active (non-idle) keys.

pub fn total_key_count(&self) -> usize

Returns the total number of tracked keys.

pub fn metrics(&self) -> &KeyedWatermarkMetrics

Returns metrics.

pub fn recalculate_global(&mut self) -> Option<Watermark>

Forces recalculation of global watermark.

Useful after bulk operations or recovery.

pub fn remove_key(&mut self, key: &K) -> Option<KeyWatermarkState>

Removes a key from tracking.

Returns the key’s watermark state if it existed.

pub fn clear(&mut self)

Clears all tracked keys.

pub fn key_state(&self, key: &K) -> Option<&KeyWatermarkState>

Returns the state for a specific key.

pub fn config(&self) -> &KeyedWatermarkConfig

Returns the configuration.

pub fn contains_key(&self, key: &K) -> bool

Checks if a key exists in the tracker.

pub fn keys(&self) -> impl Iterator<Item = &K>

Returns an iterator over all keys.

pub fn iter(&self) -> impl Iterator<Item = (&K, &KeyWatermarkState)>

Returns an iterator over all key-state pairs.

pub fn bounded_delay_ms(&self) -> i64

Returns the bounded delay in milliseconds.

Trait Implementations

impl<K: Debug + Hash + Eq + Clone> Debug for KeyedWatermarkTracker<K>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.