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 { exchange: ExchangeId, flow: Flow, metadata: Option<ChunkDescriptor> },
161    Data(ChunkDescriptor),
162    Fin { exchange: ExchangeId, flow: Flow },
163    Reset { exchange: ExchangeId, flow: Flow, code: ResetCode }
164}
165
166impl ControlEvent {
167    pub const fn exchange(self) -> ExchangeId {
168        match self {
169            Self::Start { exchange, .. }
170            | Self::Fin { exchange, .. }
171            | Self::Reset { exchange, .. } => exchange,
172            Self::Data(descriptor) => descriptor.exchange()
173        }
174    }
175
176    pub const fn flow(self) -> Flow {
177        match self {
178            Self::Start { flow, .. } | Self::Fin { flow, .. } | Self::Reset { flow, .. } => flow,
179            Self::Data(descriptor) => descriptor.flow()
180        }
181    }
182
183    pub(crate) fn encode(self) -> [u8; CONTROL_FRAME_BYTES] {
184        let mut frame = [0_u8; CONTROL_FRAME_BYTES];
185        frame[..4].copy_from_slice(&MAGIC);
186        frame[4] = VERSION;
187        frame[6] = self.flow().wire();
188        put_u64(&mut frame, 8, self.exchange().stream_id().net_id().raw());
189
190        match self {
191            Self::Start { metadata, .. } => {
192                frame[5] = EVENT_START;
193                if let Some(descriptor) = metadata {
194                    frame[7] = HAS_PAYLOAD;
195                    encode_descriptor(&mut frame, descriptor);
196                }
197            }
198            Self::Data(descriptor) => {
199                frame[5] = EVENT_DATA;
200                frame[7] = HAS_PAYLOAD;
201                encode_descriptor(&mut frame, descriptor);
202            }
203            Self::Fin { .. } => frame[5] = EVENT_FIN,
204            Self::Reset { code, .. } => {
205                frame[5] = EVENT_RESET;
206                put_u32(&mut frame, 50, code.get());
207            }
208        }
209        frame
210    }
211
212    pub(crate) fn decode(frame: &[u8; CONTROL_FRAME_BYTES]) -> Result<Self, &'static str> {
213        if frame[..4] != MAGIC {
214            return Err("wrong exchange control magic");
215        }
216        if frame[4] != VERSION {
217            return Err("unsupported exchange control version");
218        }
219        let flow = Flow::from_wire(frame[6]).ok_or("unknown exchange flow")?;
220        let exchange = ExchangeId::from_stream_id(StreamId::from_net_id(
221            orbit_core::NetId64::from_raw(get_u64(frame, 8))
222        ));
223        let has_payload = frame[7] & HAS_PAYLOAD != 0;
224        match frame[5] {
225            EVENT_START => Ok(Self::Start {
226                exchange,
227                flow,
228                metadata: has_payload.then(|| decode_descriptor(frame, exchange, flow))
229            }),
230            EVENT_DATA if has_payload => Ok(Self::Data(decode_descriptor(frame, exchange, flow))),
231            EVENT_DATA => Err("data event has no payload descriptor"),
232            EVENT_FIN if !has_payload => Ok(Self::Fin { exchange, flow }),
233            EVENT_RESET if !has_payload => {
234                Ok(Self::Reset { exchange, flow, code: ResetCode::new(get_u32(frame, 50)) })
235            }
236            EVENT_FIN | EVENT_RESET => Err("terminal event carries a payload descriptor"),
237            _ => Err("unknown exchange event")
238        }
239    }
240}
241
242fn encode_descriptor(
243    frame: &mut [u8; CONTROL_FRAME_BYTES],
244    descriptor: ChunkDescriptor
245) {
246    put_u64(frame, 16, descriptor.chunk_id());
247    put_u64(frame, 24, descriptor.allocation_generation());
248    put_u32(frame, 32, descriptor.first_slot());
249    put_u32(frame, 36, descriptor.slot_count());
250    put_u32(frame, 40, descriptor.payload_len());
251    frame[44] = descriptor.arena_kind();
252    frame[48..50].copy_from_slice(&descriptor.owner_node().to_le_bytes());
253}
254
255fn decode_descriptor(
256    frame: &[u8; CONTROL_FRAME_BYTES],
257    exchange: ExchangeId,
258    flow: Flow
259) -> ChunkDescriptor {
260    ChunkDescriptor::new(
261        exchange,
262        flow,
263        get_u64(frame, 16),
264        get_u64(frame, 24),
265        get_u32(frame, 32),
266        get_u32(frame, 36),
267        get_u32(frame, 40),
268        frame[44],
269        u16::from_le_bytes(frame[48..50].try_into().expect("two bytes"))
270    )
271}
272
273fn put_u32(
274    frame: &mut [u8; CONTROL_FRAME_BYTES],
275    at: usize,
276    value: u32
277) {
278    frame[at..at + 4].copy_from_slice(&value.to_le_bytes());
279}
280
281fn put_u64(
282    frame: &mut [u8; CONTROL_FRAME_BYTES],
283    at: usize,
284    value: u64
285) {
286    frame[at..at + 8].copy_from_slice(&value.to_le_bytes());
287}
288
289fn get_u32(
290    frame: &[u8; CONTROL_FRAME_BYTES],
291    at: usize
292) -> u32 {
293    u32::from_le_bytes(frame[at..at + 4].try_into().expect("four bytes"))
294}
295
296fn get_u64(
297    frame: &[u8; CONTROL_FRAME_BYTES],
298    at: usize
299) -> u64 {
300    u64::from_le_bytes(frame[at..at + 8].try_into().expect("eight bytes"))
301}
302
303#[cfg(test)]
304mod tests {
305    use super::*;
306
307    fn exchange() -> ExchangeId {
308        ExchangeId::from_stream_id(StreamId::from_net_id(orbit_core::NetId64::make(240, 3, 91)))
309    }
310
311    fn descriptor(flow: Flow) -> ChunkDescriptor {
312        ChunkDescriptor::new(exchange(), flow, 7, 19, 31, 4, 777, 241, 5)
313    }
314
315    #[test]
316    fn every_control_event_survives_its_fixed_frame() {
317        let events = [
318            ControlEvent::Start {
319                exchange: exchange(),
320                flow: Flow::AtoB,
321                metadata: Some(descriptor(Flow::AtoB))
322            },
323            ControlEvent::Data(descriptor(Flow::BtoA)),
324            ControlEvent::Fin { exchange: exchange(), flow: Flow::AtoB },
325            ControlEvent::Reset {
326                exchange: exchange(),
327                flow: Flow::BtoA,
328                code: ResetCode::new(503)
329            }
330        ];
331        for event in events {
332            assert_eq!(ControlEvent::decode(&event.encode()), Ok(event));
333        }
334    }
335
336    #[test]
337    fn control_frames_refuse_unknown_or_incomplete_events() {
338        let mut frame = ControlEvent::Data(descriptor(Flow::AtoB)).encode();
339        frame[7] = 0;
340        assert_eq!(ControlEvent::decode(&frame), Err("data event has no payload descriptor"));
341        frame = ControlEvent::Fin { exchange: exchange(), flow: Flow::AtoB }.encode();
342        frame[6] = 99;
343        assert_eq!(ControlEvent::decode(&frame), Err("unknown exchange flow"));
344    }
345}
orbit_stream/exchange/protocol.rs

orbit_stream/exchange/
protocol.rs
