orbit_stream/exchange/
protocol.rs

1use std::fmt;
2
3use crate::StreamId;
4
5pub(crate) const CONTROL_FRAME_BYTES: usize = 64;
6
7const MAGIC: [u8; 4] = *b"OEXC";
8const VERSION: u8 = 1;
9const EVENT_START: u8 = 1;
10const EVENT_DATA: u8 = 2;
11const EVENT_FIN: u8 = 3;
12const EVENT_RESET: u8 = 4;
13const FLOW_A_TO_B: u8 = 1;
14const FLOW_B_TO_A: u8 = 2;
15const HAS_PAYLOAD: u8 = 1;
16
17/// The identity shared by both directions of one exchange.
18#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
19pub struct ExchangeId(StreamId);
20
21impl ExchangeId {
22    pub const fn from_stream_id(id: StreamId) -> Self {
23        Self(id)
24    }
25
26    pub const fn stream_id(self) -> StreamId {
27        self.0
28    }
29}
30
31impl fmt::Display for ExchangeId {
32    fn fmt(
33        &self,
34        f: &mut fmt::Formatter<'_>,
35    ) -> fmt::Result {
36        self.0.fmt(f)
37    }
38}
39
40/// The two independent byte flows paired by an [`ExchangeId`].
41#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
42pub enum Flow {
43    AtoB,
44    BtoA,
45}
46
47impl Flow {
48    const fn wire(self) -> u8 {
49        match self {
50            Self::AtoB => FLOW_A_TO_B,
51            Self::BtoA => FLOW_B_TO_A,
52        }
53    }
54
55    fn from_wire(value: u8) -> Option<Self> {
56        match value {
57            FLOW_A_TO_B => Some(Self::AtoB),
58            FLOW_B_TO_A => Some(Self::BtoA),
59            _ => None,
60        }
61    }
62}
63
64/// Application-defined reason attached to a reset.
65#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
66pub struct ResetCode(u32);
67
68impl ResetCode {
69    pub const fn new(value: u32) -> Self {
70        Self(value)
71    }
72
73    pub const fn get(self) -> u32 {
74        self.0
75    }
76}
77
78/// One immutable payload publication. Slots are physical allocation units;
79/// this descriptor is the variable-sized chunk decision that spans them.
80#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
81pub struct ChunkDescriptor {
82    exchange: ExchangeId,
83    flow: Flow,
84    chunk_id: u64,
85    allocation_generation: u64,
86    first_slot: u32,
87    slot_count: u32,
88    payload_len: u32,
89    arena_kind: u8,
90    owner_node: u16,
91}
92
93impl ChunkDescriptor {
94    #[allow(clippy::too_many_arguments)]
95    pub(crate) const fn new(
96        exchange: ExchangeId,
97        flow: Flow,
98        chunk_id: u64,
99        allocation_generation: u64,
100        first_slot: u32,
101        slot_count: u32,
102        payload_len: u32,
103        arena_kind: u8,
104        owner_node: u16,
105    ) -> Self {
106        Self {
107            exchange,
108            flow,
109            chunk_id,
110            allocation_generation,
111            first_slot,
112            slot_count,
113            payload_len,
114            arena_kind,
115            owner_node,
116        }
117    }
118
119    pub const fn exchange(self) -> ExchangeId {
120        self.exchange
121    }
122
123    pub const fn flow(self) -> Flow {
124        self.flow
125    }
126
127    pub const fn chunk_id(self) -> u64 {
128        self.chunk_id
129    }
130
131    pub const fn allocation_generation(self) -> u64 {
132        self.allocation_generation
133    }
134
135    pub const fn first_slot(self) -> u32 {
136        self.first_slot
137    }
138
139    pub const fn slot_count(self) -> u32 {
140        self.slot_count
141    }
142
143    pub const fn payload_len(self) -> u32 {
144        self.payload_len
145    }
146
147    pub const fn arena_kind(self) -> u8 {
148        self.arena_kind
149    }
150
151    pub const fn owner_node(self) -> u16 {
152        self.owner_node
153    }
154}
155
156/// A lossless control-plane event. Payload bytes live in the arena named by
157/// the descriptor and never in this frame.
158#[derive(Clone, Copy, Debug, Eq, PartialEq)]
159pub enum ControlEvent {
160    Start {
161        exchange: ExchangeId,
162        flow: Flow,
163        metadata: Option<ChunkDescriptor>,
164    },
165    Data(ChunkDescriptor),
166    Fin {
167        exchange: ExchangeId,
168        flow: Flow,
169    },
170    Reset {
171        exchange: ExchangeId,
172        flow: Flow,
173        code: ResetCode,
174    },
175}
176
177impl ControlEvent {
178    pub const fn exchange(self) -> ExchangeId {
179        match self {
180            Self::Start { exchange, .. }
181            | Self::Fin { exchange, .. }
182            | Self::Reset { exchange, .. } => exchange,
183            Self::Data(descriptor) => descriptor.exchange(),
184        }
185    }
186
187    pub const fn flow(self) -> Flow {
188        match self {
189            Self::Start { flow, .. }
190            | Self::Fin { flow, .. }
191            | Self::Reset { flow, .. } => flow,
192            Self::Data(descriptor) => descriptor.flow(),
193        }
194    }
195
196    pub(crate) fn encode(self) -> [u8; CONTROL_FRAME_BYTES] {
197        let mut frame = [0_u8; CONTROL_FRAME_BYTES];
198        frame[..4].copy_from_slice(&MAGIC);
199        frame[4] = VERSION;
200        frame[6] = self.flow().wire();
201        put_u64(&mut frame, 8, self.exchange().stream_id().net_id().raw());
202
203        match self {
204            Self::Start { metadata, .. } => {
205                frame[5] = EVENT_START;
206                if let Some(descriptor) = metadata {
207                    frame[7] = HAS_PAYLOAD;
208                    encode_descriptor(&mut frame, descriptor);
209                }
210            }
211            Self::Data(descriptor) => {
212                frame[5] = EVENT_DATA;
213                frame[7] = HAS_PAYLOAD;
214                encode_descriptor(&mut frame, descriptor);
215            }
216            Self::Fin { .. } => frame[5] = EVENT_FIN,
217            Self::Reset { code, .. } => {
218                frame[5] = EVENT_RESET;
219                put_u32(&mut frame, 50, code.get());
220            }
221        }
222        frame
223    }
224
225    pub(crate) fn decode(
226        frame: &[u8; CONTROL_FRAME_BYTES]
227    ) -> Result<Self, &'static str> {
228        if frame[..4] != MAGIC {
229            return Err("wrong exchange control magic");
230        }
231        if frame[4] != VERSION {
232            return Err("unsupported exchange control version");
233        }
234        let flow = Flow::from_wire(frame[6]).ok_or("unknown exchange flow")?;
235        let exchange = ExchangeId::from_stream_id(StreamId::from_net_id(
236            orbit_core::NetId64::from_raw(get_u64(frame, 8)),
237        ));
238        let has_payload = frame[7] & HAS_PAYLOAD != 0;
239        match frame[5] {
240            EVENT_START => Ok(Self::Start {
241                exchange,
242                flow,
243                metadata: has_payload
244                    .then(|| decode_descriptor(frame, exchange, flow)),
245            }),
246            EVENT_DATA if has_payload => {
247                Ok(Self::Data(decode_descriptor(frame, exchange, flow)))
248            }
249            EVENT_DATA => Err("data event has no payload descriptor"),
250            EVENT_FIN if !has_payload => Ok(Self::Fin { exchange, flow }),
251            EVENT_RESET if !has_payload => Ok(Self::Reset {
252                exchange,
253                flow,
254                code: ResetCode::new(get_u32(frame, 50)),
255            }),
256            EVENT_FIN | EVENT_RESET => {
257                Err("terminal event carries a payload descriptor")
258            }
259            _ => Err("unknown exchange event"),
260        }
261    }
262}
263
264fn encode_descriptor(
265    frame: &mut [u8; CONTROL_FRAME_BYTES],
266    descriptor: ChunkDescriptor,
267) {
268    put_u64(frame, 16, descriptor.chunk_id());
269    put_u64(frame, 24, descriptor.allocation_generation());
270    put_u32(frame, 32, descriptor.first_slot());
271    put_u32(frame, 36, descriptor.slot_count());
272    put_u32(frame, 40, descriptor.payload_len());
273    frame[44] = descriptor.arena_kind();
274    frame[48..50].copy_from_slice(&descriptor.owner_node().to_le_bytes());
275}
276
277fn decode_descriptor(
278    frame: &[u8; CONTROL_FRAME_BYTES],
279    exchange: ExchangeId,
280    flow: Flow,
281) -> ChunkDescriptor {
282    ChunkDescriptor::new(
283        exchange,
284        flow,
285        get_u64(frame, 16),
286        get_u64(frame, 24),
287        get_u32(frame, 32),
288        get_u32(frame, 36),
289        get_u32(frame, 40),
290        frame[44],
291        u16::from_le_bytes(frame[48..50].try_into().expect("two bytes")),
292    )
293}
294
295fn put_u32(
296    frame: &mut [u8; CONTROL_FRAME_BYTES],
297    at: usize,
298    value: u32,
299) {
300    frame[at..at + 4].copy_from_slice(&value.to_le_bytes());
301}
302
303fn put_u64(
304    frame: &mut [u8; CONTROL_FRAME_BYTES],
305    at: usize,
306    value: u64,
307) {
308    frame[at..at + 8].copy_from_slice(&value.to_le_bytes());
309}
310
311fn get_u32(
312    frame: &[u8; CONTROL_FRAME_BYTES],
313    at: usize,
314) -> u32 {
315    u32::from_le_bytes(frame[at..at + 4].try_into().expect("four bytes"))
316}
317
318fn get_u64(
319    frame: &[u8; CONTROL_FRAME_BYTES],
320    at: usize,
321) -> u64 {
322    u64::from_le_bytes(frame[at..at + 8].try_into().expect("eight bytes"))
323}
324
325#[cfg(test)]
326mod tests {
327    use super::*;
328
329    fn exchange() -> ExchangeId {
330        ExchangeId::from_stream_id(StreamId::from_net_id(
331            orbit_core::NetId64::make(240, 3, 91),
332        ))
333    }
334
335    fn descriptor(flow: Flow) -> ChunkDescriptor {
336        ChunkDescriptor::new(exchange(), flow, 7, 19, 31, 4, 777, 241, 5)
337    }
338
339    #[test]
340    fn every_control_event_survives_its_fixed_frame() {
341        let events = [
342            ControlEvent::Start {
343                exchange: exchange(),
344                flow: Flow::AtoB,
345                metadata: Some(descriptor(Flow::AtoB)),
346            },
347            ControlEvent::Data(descriptor(Flow::BtoA)),
348            ControlEvent::Fin { exchange: exchange(), flow: Flow::AtoB },
349            ControlEvent::Reset {
350                exchange: exchange(),
351                flow: Flow::BtoA,
352                code: ResetCode::new(503),
353            },
354        ];
355        for event in events {
356            assert_eq!(ControlEvent::decode(&event.encode()), Ok(event));
357        }
358    }
359
360    #[test]
361    fn control_frames_refuse_unknown_or_incomplete_events() {
362        let mut frame = ControlEvent::Data(descriptor(Flow::AtoB)).encode();
363        frame[7] = 0;
364        assert_eq!(
365            ControlEvent::decode(&frame),
366            Err("data event has no payload descriptor")
367        );
368        frame = ControlEvent::Fin { exchange: exchange(), flow: Flow::AtoB }
369            .encode();
370        frame[6] = 99;
371        assert_eq!(ControlEvent::decode(&frame), Err("unknown exchange flow"));
372    }
373}
orbit_stream/exchange/protocol.rs

orbit_stream/exchange/
protocol.rs
