Struct WatermarkAlignmentGroup 

pub struct WatermarkAlignmentGroup { /* private fields */ }

Manages watermark alignment across sources in a group.

Tracks watermarks from multiple sources and enforces that no source gets too far ahead of others. When a source exceeds the maximum drift, it is paused (or warned/dropped depending on enforcement mode) until slower sources catch up.

Example

use laminar_core::time::{
    WatermarkAlignmentGroup, AlignmentGroupConfig, AlignmentGroupId,
    EnforcementMode, AlignmentAction,
};
use std::time::Duration;

let config = AlignmentGroupConfig {
    group_id: AlignmentGroupId("join-group".to_string()),
    max_drift: Duration::from_secs(60), // 1 minute
    update_interval: Duration::from_millis(100),
    enforcement_mode: EnforcementMode::Pause,
};

let mut group = WatermarkAlignmentGroup::new(config);
group.register_source(0);
group.register_source(1);

// Both sources start at 0
group.report_watermark(0, 0);
group.report_watermark(1, 0);

// Source 0 advances within limit
let action = group.report_watermark(0, 50_000); // 50 seconds
assert_eq!(action, AlignmentAction::Continue);

// Source 0 advances beyond limit (>60s drift)
let action = group.report_watermark(0, 70_000); // 70 seconds
assert_eq!(action, AlignmentAction::Pause);

// Source 1 catches up
group.report_watermark(1, 30_000); // 30 seconds
assert!(group.should_resume(0)); // Drift now 40s < 60s

Implementations

impl WatermarkAlignmentGroup

pub fn new(config: AlignmentGroupConfig) -> Self

Creates a new alignment group.

pub fn group_id(&self) -> &AlignmentGroupId

Returns the group ID.

pub fn config(&self) -> &AlignmentGroupConfig

Returns the configuration.

pub fn register_source(&mut self, source_id: usize)

Registers a source with this alignment group.

pub fn unregister_source(&mut self, source_id: usize)

Removes a source from this alignment group.

pub fn report_watermark( &mut self, source_id: usize, watermark: i64, ) -> AlignmentAction

Reports a watermark update from a source.

Returns the action the source should take.

pub fn should_resume(&self, source_id: usize) -> bool

Checks if a paused source should resume.

pub fn current_drift(&self) -> Duration

Returns the current drift (max - min watermark).

pub fn is_paused(&self, source_id: usize) -> bool

Returns whether a specific source is paused.

pub fn min_watermark(&self) -> i64

Returns the minimum watermark in the group.

pub fn max_watermark(&self) -> i64

Returns the maximum watermark in the group.

pub fn metrics(&self) -> &AlignmentGroupMetrics

Returns metrics for this group.

pub fn source_count(&self) -> usize

Returns the number of registered sources.

pub fn paused_source_count(&self) -> usize

Returns the number of paused sources.

pub fn active_source_count(&self) -> usize

Returns the number of active (non-paused) sources.

pub fn check_alignment(&mut self) -> Vec<(usize, AlignmentAction)>

Performs periodic alignment check.

Should be called from Ring 1 at the configured update_interval. Returns list of (source_id, action) pairs.

pub fn source_state(&self, source_id: usize) -> Option<&AlignmentSourceState>

Returns the state of a specific source.

Trait Implementations

impl Debug for WatermarkAlignmentGroup

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

Formats the value using the given formatter.