sectorsync_core/
event.rs

1//! Cross-station event envelopes and bounded priority queues.
2
3use std::collections::VecDeque;
4
5use crate::ids::{EntityId, EventId, OwnerEpoch, StationId, Tick};
6
7/// Event priority class.
8#[derive(Clone, Copy, Debug, PartialEq, Eq)]
9pub enum EventPriority {
10    /// Must be delivered or cause backpressure.
11    Critical,
12    /// Should be delivered with bounded retry/ack policy.
13    Important,
14    /// Can be dropped, merged, or downgraded under pressure.
15    BestEffort,
16}
17
18/// Core event kind understood by the runtime.
19#[derive(Clone, Debug, PartialEq, Eq)]
20pub enum EventKind {
21    /// User-defined event kind id.
22    Custom(u32),
23    /// Prepare a two-phase entity handoff.
24    HandoffPrepare {
25        /// Entity being handed off.
26        entity_id: EntityId,
27    },
28    /// Commit a two-phase entity handoff.
29    HandoffCommit {
30        /// Entity being handed off.
31        entity_id: EntityId,
32        /// New owner epoch.
33        owner_epoch: OwnerEpoch,
34    },
35}
36
37/// Event envelope routed between stations.
38#[derive(Clone, Debug, PartialEq, Eq)]
39pub struct StationEvent {
40    /// Idempotency key.
41    pub id: EventId,
42    /// Source station.
43    pub source: StationId,
44    /// Target station.
45    pub target: StationId,
46    /// Tick observed at source.
47    pub source_tick: Tick,
48    /// Tick at which target should apply the event.
49    pub target_tick: Tick,
50    /// Priority class.
51    pub priority: EventPriority,
52    /// Event payload kind.
53    pub kind: EventKind,
54}
55
56/// Bounded queue limits by priority.
57#[derive(Clone, Copy, Debug, PartialEq, Eq)]
58pub struct EventQueueLimits {
59    /// Critical queue capacity.
60    pub critical: usize,
61    /// Important queue capacity.
62    pub important: usize,
63    /// Best-effort queue capacity.
64    pub best_effort: usize,
65}
66
67impl Default for EventQueueLimits {
68    fn default() -> Self {
69        Self {
70            critical: 1024,
71            important: 4096,
72            best_effort: 8192,
73        }
74    }
75}
76
77/// Outcome of a queue push.
78#[derive(Clone, Copy, Debug, PartialEq, Eq)]
79pub enum PushOutcome {
80    /// Event was accepted without dropping another event.
81    Accepted,
82    /// A best-effort event was dropped to admit the new one.
83    DroppedOldestBestEffort,
84}
85
86/// Event queue error.
87#[derive(Clone, Copy, Debug, PartialEq, Eq)]
88pub enum EventQueueError {
89    /// Reliable queue is full and caller must apply backpressure.
90    QueueFull(EventPriority),
91}
92
93impl core::fmt::Display for EventQueueError {
94    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
95        match self {
96            Self::QueueFull(priority) => write!(f, "{priority:?} event queue is full"),
97        }
98    }
99}
100
101impl std::error::Error for EventQueueError {}
102
103/// Bounded priority queues for station events.
104#[derive(Clone, Debug)]
105pub struct EventQueues {
106    limits: EventQueueLimits,
107    critical: VecDeque<StationEvent>,
108    important: VecDeque<StationEvent>,
109    best_effort: VecDeque<StationEvent>,
110}
111
112impl EventQueues {
113    /// Creates empty event queues.
114    pub fn new(limits: EventQueueLimits) -> Self {
115        Self {
116            limits,
117            critical: VecDeque::with_capacity(limits.critical),
118            important: VecDeque::with_capacity(limits.important),
119            best_effort: VecDeque::with_capacity(limits.best_effort),
120        }
121    }
122
123    /// Pushes an event according to priority and queue semantics.
124    pub fn push(&mut self, event: StationEvent) -> Result<PushOutcome, EventQueueError> {
125        match event.priority {
126            EventPriority::Critical => {
127                if self.critical.len() >= self.limits.critical {
128                    Err(EventQueueError::QueueFull(EventPriority::Critical))
129                } else {
130                    self.critical.push_back(event);
131                    Ok(PushOutcome::Accepted)
132                }
133            }
134            EventPriority::Important => {
135                if self.important.len() >= self.limits.important {
136                    Err(EventQueueError::QueueFull(EventPriority::Important))
137                } else {
138                    self.important.push_back(event);
139                    Ok(PushOutcome::Accepted)
140                }
141            }
142            EventPriority::BestEffort => {
143                let outcome = if self.best_effort.len() >= self.limits.best_effort {
144                    self.best_effort.pop_front();
145                    PushOutcome::DroppedOldestBestEffort
146                } else {
147                    PushOutcome::Accepted
148                };
149                self.best_effort.push_back(event);
150                Ok(outcome)
151            }
152        }
153    }
154
155    /// Pops the next event, preferring higher priority.
156    pub fn pop_next(&mut self) -> Option<StationEvent> {
157        self.critical
158            .pop_front()
159            .or_else(|| self.important.pop_front())
160            .or_else(|| self.best_effort.pop_front())
161    }
162
163    /// Returns total queued events.
164    pub fn len(&self) -> usize {
165        self.critical.len() + self.important.len() + self.best_effort.len()
166    }
167
168    /// Returns whether all queues are empty.
169    pub fn is_empty(&self) -> bool {
170        self.len() == 0
171    }
172}
sectorsync_core/event.rs

sectorsync_core/
event.rs
