active_rpc/
channel.rs

1//! Channels: named streams of messages and capability transfers.
2//!
3//! A channel is a typed, named stream between lanes of one agent. Direction is
4//! declared up front; a receive-only channel refuses sends, and a send-only
5//! channel refuses receives.
6//!
7//! Capabilities named in a message are **shared (borrowed)**, not copied or
8//! moved: the ids travel, the sender keeps its handles, and the receiver's
9//! authority is whatever its own kernel registry already holds. Across agents
10//! the same ids travel in an envelope's capability table and each receiver looks
11//! them up in its own registry — authority never transfers implicitly, which is
12//! why a channel message cannot inflate a reference count.
13
14use std::collections::{BTreeSet, VecDeque};
15
16use crate::cap::CapId;
17use crate::error::{Error, Result};
18use crate::lane::LaneId;
19use crate::value::Value;
20
21/// Maximum messages queued on one channel.
22///
23/// A channel is a mailbox, not an unbounded log. The ceiling is high enough for
24/// a burst and finite so one sender with a valid capability cannot grow kernel
25/// memory without bound.
26pub const MAX_QUEUED_PER_CHANNEL: usize = 4_096;
27
28/// Channel identity.
29#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
30pub struct ChannelId(pub u64);
31
32impl std::fmt::Display for ChannelId {
33    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34        write!(f, "ch{}", self.0)
35    }
36}
37
38/// Allowed traffic direction.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub enum Direction {
41    /// The lane may only send.
42    Send,
43    /// The lane may only receive.
44    Receive,
45    /// Both directions.
46    Both,
47}
48
49impl Direction {
50    /// Whether this direction permits sending.
51    pub fn can_send(self) -> bool {
52        matches!(self, Direction::Send | Direction::Both)
53    }
54
55    /// Whether this direction permits receiving.
56    pub fn can_receive(self) -> bool {
57        matches!(self, Direction::Receive | Direction::Both)
58    }
59}
60
61/// One message on a channel.
62#[derive(Debug, Clone, PartialEq)]
63pub struct Message {
64    /// Sending lane.
65    pub from: LaneId,
66    /// Payload value (never secret material).
67    pub payload: Value,
68    /// Capabilities transferred with this message.
69    pub caps: Vec<CapId>,
70    /// Monotonic sequence within the channel.
71    pub sequence: u64,
72}
73
74/// A named channel.
75#[derive(Debug, Clone)]
76pub struct Channel {
77    /// Kernel-assigned identifier.
78    pub id: ChannelId,
79    /// The name a peer subscribes by.
80    pub name: String,
81    /// Which ways messages may flow.
82    pub direction: Direction,
83    queue: VecDeque<Message>,
84    next_sequence: u64,
85    closed: bool,
86    /// Lanes subscribed to this channel.
87    pub subscribers: BTreeSet<u64>,
88    /// Total messages sent (including those already received).
89    pub sent: u64,
90    /// Total messages received.
91    pub received: u64,
92}
93
94impl Channel {
95    /// Create an open channel with an empty queue.
96    pub fn new(id: ChannelId, name: impl Into<String>, direction: Direction) -> Self {
97        Self {
98            id,
99            name: name.into(),
100            direction,
101            queue: VecDeque::new(),
102            next_sequence: 0,
103            closed: false,
104            subscribers: BTreeSet::new(),
105            sent: 0,
106            received: 0,
107        }
108    }
109
110    /// Whether the channel accepts sends.
111    pub fn can_send(&self) -> bool {
112        self.direction.can_send() && !self.closed
113    }
114
115    /// Whether the channel has messages for a receiver.
116    pub fn can_receive(&self) -> bool {
117        self.direction.can_receive() && !self.queue.is_empty()
118    }
119
120    /// Queue a message.
121    pub fn send(&mut self, from: LaneId, payload: Value, caps: Vec<CapId>) -> Result<Message> {
122        if self.closed {
123            return Err(Error::InvalidArgument("channel is closed"));
124        }
125        if !self.direction.can_send() {
126            return Err(Error::InvalidArgument("channel is receive-only"));
127        }
128        if self.queue.len() >= MAX_QUEUED_PER_CHANNEL {
129            return Err(Error::BudgetExceeded("channel queue"));
130        }
131        let message = Message {
132            from,
133            payload,
134            caps,
135            sequence: self.next_sequence,
136        };
137        self.next_sequence += 1;
138        self.sent += 1;
139        self.queue.push_back(message.clone());
140        Ok(message)
141    }
142
143    /// Take the oldest message.
144    pub fn recv(&mut self) -> Result<Option<Message>> {
145        if !self.direction.can_receive() {
146            return Err(Error::InvalidArgument("channel is send-only"));
147        }
148        match self.queue.pop_front() {
149            Some(message) => {
150                self.received += 1;
151                Ok(Some(message))
152            }
153            None => Ok(None),
154        }
155    }
156
157    /// Peek at the oldest message without consuming it.
158    pub fn peek(&self) -> Option<&Message> {
159        self.queue.front()
160    }
161
162    /// Number of queued messages.
163    pub fn len(&self) -> usize {
164        self.queue.len()
165    }
166
167    /// Whether the queue holds no messages.
168    pub fn is_empty(&self) -> bool {
169        self.queue.is_empty()
170    }
171
172    /// Close the channel: further sends fail; messages already queued remain
173    /// receivable.
174    pub fn close(&mut self) {
175        self.closed = true;
176    }
177
178    /// Whether the channel has been closed.
179    pub fn is_closed(&self) -> bool {
180        self.closed
181    }
182}
active_rpc/channel.rs

active_rpc/
channel.rs
