deepstrike_core/scheduler/
mailbox.rs

1//! spc_006: Task-to-task IPC — point-to-point [`Mailbox`] (this card + spc_006-02/03) and
2//! many-to-one/one-to-many [`Channel`] (spc_006-04). Additive-only in this card: the message
3//! shape only, no send/receive, no wiring onto [`super::tcb::Tcb`].
4//!
5//! Naming note: spc_006 §3 names this struct `Message`, but `crate::types::message::Message`
6//! (the LLM conversation message) already owns that name and is glob-imported (`use super::*`)
7//! into `scheduler::state_machine::tests` — a second unqualified `Message` there would force
8//! every reference in this module's own integration tests to be fully qualified. `MailboxMessage`
9//! avoids the collision while staying unambiguous about what it is.
10
11use std::collections::{BTreeMap, BTreeSet, VecDeque};
12
13use compact_str::CompactString;
14use serde::{Deserialize, Serialize};
15
16use super::tcb::TaskId;
17use crate::mm::handle::HandleId;
18use crate::types::signal::Urgency;
19
20/// Opaque message id — mirrors [`TaskId`]'s convention of a plain `CompactString` alias rather
21/// than a validated newtype (no producer needs anything richer yet).
22pub type MessageId = CompactString;
23
24const IPC_DEDUPE_WINDOW: usize = 256;
25
26/// spc_006: no existing "logical time" counter type exists to reuse (`LoopStateMachine::turn` is
27/// a private `u32` field, not a public type) — a minimal placeholder newtype, following the same
28/// convention [`super::tcb::LogicalDeadline`] established in spc_003-02 for concepts with no
29/// producer yet.
30#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
31pub struct LogicalTime(pub u32);
32
33/// A point-to-point message between two tasks. Large payloads never live inline — `payload_handle`
34/// points into the sender's [`crate::mm::handle::HandleTable`] (spc_006 §5: pass handles, not
35/// prompts).
36#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
37pub struct MailboxMessage {
38    pub id: MessageId,
39    pub from: TaskId,
40    pub to: TaskId,
41    pub kind: CompactString,
42    pub payload_handle: HandleId,
43    pub priority: Urgency,
44    pub timestamp: LogicalTime,
45    /// Exclusive logical-turn expiry. `None` means the message does not expire.
46    #[serde(default, skip_serializing_if = "Option::is_none")]
47    pub expires_at: Option<LogicalTime>,
48}
49
50#[derive(Debug, Clone, Copy, PartialEq, Eq)]
51pub enum IpcEnqueueOutcome {
52    Accepted,
53    Duplicate,
54    Full,
55    Expired,
56}
57
58/// spc_006-02: a task's inbox — point-to-point only (no fan-out; that's [`Channel`]'s job,
59/// spc_006-04). Pure data structure: no `TaskTable`/`Tcb` reference, no send/receive wiring onto
60/// a real task yet (spc_006-03).
61#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
62pub struct Mailbox {
63    queue: VecDeque<MailboxMessage>,
64    #[serde(default)]
65    seen: BTreeSet<MessageId>,
66    #[serde(default)]
67    seen_order: VecDeque<MessageId>,
68    capacity: usize,
69}
70
71impl Default for Mailbox {
72    fn default() -> Self {
73        Self {
74            queue: VecDeque::new(),
75            seen: BTreeSet::new(),
76            seen_order: VecDeque::new(),
77            capacity: 64,
78        }
79    }
80}
81
82impl Mailbox {
83    pub fn new() -> Self {
84        Self::default()
85    }
86
87    pub fn send(&mut self, msg: MailboxMessage) {
88        let _ = self.try_send(msg, LogicalTime(0));
89    }
90
91    pub fn try_send(&mut self, msg: MailboxMessage, now: LogicalTime) -> IpcEnqueueOutcome {
92        normalize_seen_order(&mut self.seen, &mut self.seen_order);
93        self.drop_expired(now);
94        if msg.expires_at.is_some_and(|deadline| now >= deadline) {
95            return IpcEnqueueOutcome::Expired;
96        }
97        if self.seen.contains(&msg.id) {
98            return IpcEnqueueOutcome::Duplicate;
99        }
100        if self.queue.len() >= self.capacity {
101            return IpcEnqueueOutcome::Full;
102        }
103        self.seen.insert(msg.id.clone());
104        remember_seen(&mut self.seen, &mut self.seen_order, msg.id.clone());
105        self.queue.push_back(msg);
106        IpcEnqueueOutcome::Accepted
107    }
108
109    /// FIFO — oldest message first.
110    pub fn receive(&mut self) -> Option<MailboxMessage> {
111        self.queue.pop_front()
112    }
113
114    pub fn receive_at(&mut self, now: LogicalTime) -> Option<MailboxMessage> {
115        self.drop_expired(now);
116        self.receive()
117    }
118
119    pub fn snapshot(&self) -> Vec<MailboxMessage> {
120        self.queue.iter().cloned().collect()
121    }
122
123    pub fn is_empty(&self) -> bool {
124        self.queue.is_empty() && self.seen.is_empty() && self.seen_order.is_empty()
125    }
126
127    fn drop_expired(&mut self, now: LogicalTime) {
128        self.queue
129            .retain(|message| message.expires_at.is_none_or(|deadline| now < deadline));
130    }
131}
132
133/// spc_006-04: many-to-one / one-to-many fan-in. Unlike [`Mailbox`], a `publish`ed message is
134/// never removed from the shared `buffer` — each subscriber reads independently via its own
135/// cursor into that buffer (the doc's "共享 buffer + per-consumer 已读游标" design; the `cursors`
136/// field is not shown in spc_006 §3's abbreviated struct sketch but is mechanically required to
137/// realize that description — without it a second consumer draining after a first would see
138/// nothing).
139#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
140pub struct Channel {
141    pub subscribers: Vec<TaskId>,
142    buffer: VecDeque<MailboxMessage>,
143    #[serde(default)]
144    cursors: BTreeMap<TaskId, usize>,
145    #[serde(default)]
146    seen: BTreeSet<MessageId>,
147    #[serde(default)]
148    seen_order: VecDeque<MessageId>,
149    capacity: usize,
150}
151
152impl Default for Channel {
153    fn default() -> Self {
154        Self {
155            subscribers: Vec::new(),
156            buffer: VecDeque::new(),
157            cursors: BTreeMap::new(),
158            seen: BTreeSet::new(),
159            seen_order: VecDeque::new(),
160            capacity: 64,
161        }
162    }
163}
164
165impl Channel {
166    pub fn new(subscribers: Vec<TaskId>) -> Self {
167        Self {
168            subscribers,
169            buffer: VecDeque::new(),
170            cursors: BTreeMap::new(),
171            seen: BTreeSet::new(),
172            seen_order: VecDeque::new(),
173            capacity: 64,
174        }
175    }
176
177    pub fn publish(&mut self, msg: MailboxMessage) {
178        let _ = self.publish_at(msg, LogicalTime(0));
179    }
180
181    pub fn publish_at(&mut self, msg: MailboxMessage, now: LogicalTime) -> IpcEnqueueOutcome {
182        normalize_seen_order(&mut self.seen, &mut self.seen_order);
183        self.drop_expired(now);
184        if msg.expires_at.is_some_and(|deadline| now >= deadline) {
185            return IpcEnqueueOutcome::Expired;
186        }
187        if self.seen.contains(&msg.id) {
188            return IpcEnqueueOutcome::Duplicate;
189        }
190        if self.buffer.len() >= self.capacity {
191            return IpcEnqueueOutcome::Full;
192        }
193        self.seen.insert(msg.id.clone());
194        remember_seen(&mut self.seen, &mut self.seen_order, msg.id.clone());
195        self.buffer.push_back(msg);
196        IpcEnqueueOutcome::Accepted
197    }
198
199    /// Every message published since `consumer`'s last `drain_for`, oldest first; advances that
200    /// consumer's cursor to the current end of the buffer. Independent of every other consumer's
201    /// cursor — draining does not consume the buffer.
202    pub fn drain_for(&mut self, consumer: TaskId) -> Vec<MailboxMessage> {
203        let cursor = self.cursors.entry(consumer).or_insert(0);
204        let unread: Vec<MailboxMessage> = self.buffer.iter().skip(*cursor).cloned().collect();
205        *cursor = self.buffer.len();
206        self.compact_consumed();
207        unread
208    }
209
210    pub fn drain_for_at(&mut self, consumer: TaskId, now: LogicalTime) -> Vec<MailboxMessage> {
211        if !self.subscribers.contains(&consumer) {
212            return Vec::new();
213        }
214        self.drain_for(consumer)
215            .into_iter()
216            .filter(|message| message.expires_at.is_none_or(|deadline| now < deadline))
217            .collect()
218    }
219
220    fn compact_consumed(&mut self) {
221        let consumed = self
222            .subscribers
223            .iter()
224            .map(|subscriber| self.cursors.get(subscriber).copied().unwrap_or(0))
225            .min()
226            .unwrap_or(0);
227        for _ in 0..consumed {
228            self.buffer.pop_front();
229        }
230        if consumed > 0 {
231            for cursor in self.cursors.values_mut() {
232                *cursor = cursor.saturating_sub(consumed);
233            }
234        }
235    }
236
237    fn drop_expired(&mut self, now: LogicalTime) {
238        if self.buffer.is_empty() {
239            return;
240        }
241        let previous: Vec<_> = self.buffer.drain(..).collect();
242        for cursor in self.cursors.values_mut() {
243            *cursor = previous
244                .iter()
245                .take(*cursor)
246                .filter(|message| message.expires_at.is_none_or(|deadline| now < deadline))
247                .count();
248        }
249        self.buffer = previous
250            .into_iter()
251            .filter(|message| message.expires_at.is_none_or(|deadline| now < deadline))
252            .collect();
253    }
254}
255
256fn normalize_seen_order(seen: &mut BTreeSet<MessageId>, order: &mut VecDeque<MessageId>) {
257    if order.len() < seen.len() {
258        for id in seen.iter() {
259            if !order.contains(id) {
260                order.push_back(id.clone());
261            }
262        }
263    }
264    while order.len() > IPC_DEDUPE_WINDOW {
265        if let Some(expired) = order.pop_front() {
266            seen.remove(&expired);
267        }
268    }
269}
270
271fn remember_seen(seen: &mut BTreeSet<MessageId>, order: &mut VecDeque<MessageId>, id: MessageId) {
272    order.push_back(id);
273    while order.len() > IPC_DEDUPE_WINDOW {
274        if let Some(expired) = order.pop_front() {
275            seen.remove(&expired);
276        }
277    }
278}
279
280#[cfg(test)]
281mod tests {
282    use super::*;
283
284    #[test]
285    fn spc_006_01_mailbox_message_fields_are_readable() {
286        let msg = MailboxMessage {
287            id: MessageId::from("msg-1"),
288            from: TaskId::from("a"),
289            to: TaskId::from("b"),
290            kind: CompactString::from("research_result"),
291            payload_handle: 42,
292            priority: Urgency::High,
293            timestamp: LogicalTime(7),
294            expires_at: None,
295        };
296
297        assert_eq!(msg.id, MessageId::from("msg-1"));
298        assert_eq!(msg.from, TaskId::from("a"));
299        assert_eq!(msg.to, TaskId::from("b"));
300        assert_eq!(msg.kind, CompactString::from("research_result"));
301        assert_eq!(msg.payload_handle, 42);
302        assert_eq!(msg.priority, Urgency::High);
303        assert_eq!(msg.timestamp, LogicalTime(7));
304    }
305
306    fn msg(id: &str) -> MailboxMessage {
307        MailboxMessage {
308            id: MessageId::from(id),
309            from: TaskId::from("a"),
310            to: TaskId::from("b"),
311            kind: CompactString::from("kind"),
312            payload_handle: 1,
313            priority: Urgency::Normal,
314            timestamp: LogicalTime(0),
315            expires_at: None,
316        }
317    }
318
319    #[test]
320    fn spc_006_02_receive_on_an_empty_mailbox_returns_none() {
321        let mut mailbox = Mailbox::new();
322        assert_eq!(mailbox.receive(), None);
323    }
324
325    #[test]
326    fn spc_006_02_receive_returns_sent_messages_in_fifo_order() {
327        let mut mailbox = Mailbox::new();
328        mailbox.send(msg("first"));
329        mailbox.send(msg("second"));
330
331        assert_eq!(
332            mailbox.receive().map(|m| m.id),
333            Some(MessageId::from("first"))
334        );
335        assert_eq!(
336            mailbox.receive().map(|m| m.id),
337            Some(MessageId::from("second"))
338        );
339    }
340
341    #[test]
342    fn mailbox_dedupe_history_is_bounded() {
343        let mut mailbox = Mailbox::new();
344        for index in 0..=IPC_DEDUPE_WINDOW {
345            assert_eq!(
346                mailbox.try_send(msg(&format!("message-{index}")), LogicalTime(0)),
347                IpcEnqueueOutcome::Accepted
348            );
349            mailbox.receive();
350        }
351        assert_eq!(mailbox.seen.len(), IPC_DEDUPE_WINDOW);
352        assert_eq!(mailbox.seen_order.len(), IPC_DEDUPE_WINDOW);
353        assert_eq!(
354            mailbox.try_send(msg("message-0"), LogicalTime(0)),
355            IpcEnqueueOutcome::Accepted
356        );
357    }
358
359    #[test]
360    fn spc_006_02_receive_returns_none_once_drained() {
361        let mut mailbox = Mailbox::new();
362        mailbox.send(msg("only"));
363        assert!(mailbox.receive().is_some());
364        assert_eq!(mailbox.receive(), None);
365    }
366
367    fn msg_from(id: &str, from: &str) -> MailboxMessage {
368        MailboxMessage {
369            id: MessageId::from(id),
370            from: TaskId::from(from),
371            to: TaskId::from("coordinator"),
372            kind: CompactString::from("kind"),
373            payload_handle: 1,
374            priority: Urgency::Normal,
375            timestamp: LogicalTime(0),
376            expires_at: None,
377        }
378    }
379
380    #[test]
381    fn spc_006_04_drain_for_gathers_every_producers_message_in_arrival_order() {
382        let mut channel = Channel::new(vec![TaskId::from("coordinator")]);
383        channel.publish(msg_from("m1", "worker-1"));
384        channel.publish(msg_from("m2", "worker-2"));
385        channel.publish(msg_from("m3", "worker-3"));
386
387        let drained = channel.drain_for(TaskId::from("coordinator"));
388        let ids: Vec<_> = drained.iter().map(|m| m.id.clone()).collect();
389        assert_eq!(
390            ids,
391            vec![
392                MessageId::from("m1"),
393                MessageId::from("m2"),
394                MessageId::from("m3"),
395            ]
396        );
397    }
398
399    #[test]
400    fn spc_006_04_drain_for_only_returns_messages_published_since_the_last_drain() {
401        let mut channel = Channel::new(vec![TaskId::from("coordinator")]);
402        channel.publish(msg_from("m1", "worker-1"));
403        assert_eq!(channel.drain_for(TaskId::from("coordinator")).len(), 1);
404        assert_eq!(
405            channel.drain_for(TaskId::from("coordinator")),
406            Vec::new(),
407            "a second drain with nothing new published must come back empty"
408        );
409
410        channel.publish(msg_from("m2", "worker-2"));
411        let second_batch = channel.drain_for(TaskId::from("coordinator"));
412        assert_eq!(second_batch.len(), 1);
413        assert_eq!(second_batch[0].id, MessageId::from("m2"));
414    }
415
416    #[test]
417    fn spc_006_04_two_consumers_drain_independently_from_the_same_buffer() {
418        let mut channel = Channel::new(vec![TaskId::from("c1"), TaskId::from("c2")]);
419        channel.publish(msg_from("m1", "worker-1"));
420
421        let c1_drained = channel.drain_for(TaskId::from("c1"));
422        assert_eq!(c1_drained.len(), 1);
423
424        // c2 has never drained yet — it must still see m1, not have it "stolen" by c1's read.
425        let c2_drained = channel.drain_for(TaskId::from("c2"));
426        assert_eq!(c2_drained.len(), 1);
427        assert_eq!(c2_drained[0].id, MessageId::from("m1"));
428    }
429
430    #[test]
431    fn spc_019_08_mailbox_dedupes_bounds_and_expires_on_logical_time() {
432        let mut mailbox = Mailbox::new();
433        let mut first = msg("first");
434        first.expires_at = Some(LogicalTime(2));
435        assert_eq!(
436            mailbox.try_send(first.clone(), LogicalTime(0)),
437            IpcEnqueueOutcome::Accepted
438        );
439        assert_eq!(
440            mailbox.try_send(first, LogicalTime(0)),
441            IpcEnqueueOutcome::Duplicate
442        );
443        assert!(mailbox.receive_at(LogicalTime(2)).is_none());
444
445        for index in 0..64 {
446            assert_eq!(
447                mailbox.try_send(msg(&format!("m-{index}")), LogicalTime(2)),
448                IpcEnqueueOutcome::Accepted
449            );
450        }
451        assert_eq!(
452            mailbox.try_send(msg("overflow"), LogicalTime(2)),
453            IpcEnqueueOutcome::Full
454        );
455    }
456
457    #[test]
458    fn spc_019_08_channel_dedupes_and_filters_expired_messages_per_subscriber() {
459        let mut channel = Channel::new(vec![TaskId::from("b")]);
460        let mut expiring = msg_from("m1", "a");
461        expiring.expires_at = Some(LogicalTime(3));
462        assert_eq!(
463            channel.publish_at(expiring.clone(), LogicalTime(1)),
464            IpcEnqueueOutcome::Accepted
465        );
466        assert_eq!(
467            channel.publish_at(expiring, LogicalTime(1)),
468            IpcEnqueueOutcome::Duplicate
469        );
470        assert!(
471            channel
472                .drain_for_at(TaskId::from("not-subscribed"), LogicalTime(1))
473                .is_empty()
474        );
475        assert!(
476            channel
477                .drain_for_at(TaskId::from("b"), LogicalTime(3))
478                .is_empty()
479        );
480    }
481}
deepstrike_core/scheduler/mailbox.rs

deepstrike_core/scheduler/
mailbox.rs
