harmos_sidecar/
transport.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12mod connection;
13mod consumption;
14mod credit;
15mod payload {
16    use serde::{Deserialize, Deserializer, Serializer};
17
18    pub fn serialize<S: Serializer>(bytes: &[u8], serializer: S) -> Result<S::Ok, S::Error> {
19        serializer.serialize_bytes(bytes)
20    }
21
22    pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<&'de [u8], D::Error> {
23        <&[u8]>::deserialize(deserializer)
24    }
25}
26
27/* === Re-Exports === */
28
29pub use connection::{Connection, Driver, Io, Received, Receiver, Sender};
30pub use consumption::{Consumption, ReceiveCredit};
31pub use credit::{Credit, Reservation};
32
33/* === Imports === */
34
35// Standard Library Imports
36use std::io::{self, IoSlice};
37use std::ops::Range;
38use std::sync::Arc;
39// Third Party Imports
40use serde::{Deserialize, Serialize};
41use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
42// Local Imports
43use crate::{Frame, FramePayload};
44
45/* === Constants === */
46
47const VERSION: u8 = 1;
48const CONTROL_LIMIT: usize = 64;
49const COALESCE_BYTES: usize = 64 * 1024;
50
51/* === Macros === */
52
53/* === Type Aliases === */
54
55/* === Structs === */
56
57/* === Enums === */
58
59#[derive(Clone, Copy, Debug, PartialEq, Eq)]
60pub enum Role {
61    Data,
62    Control,
63}
64
65#[derive(Debug, PartialEq, Eq)]
66pub enum Packet {
67    Frame(Frame),
68    Credit(u64),
69    Cancellation { id: u64, preceding: u64 },
70}
71
72// Variant Order Is Fixed By Version 1. No Owned Allocation Occurs While Decoding.
73#[derive(Serialize, Deserialize)]
74enum Body<'a> {
75    Call {
76        id: u64,
77        method: &'a str,
78
79        #[serde(borrow, with = "payload")]
80        params: &'a [u8],
81    },
82    Cancel {
83        id: u64,
84        preceding: u64,
85    },
86    Push {
87        method: &'a str,
88
89        #[serde(borrow, with = "payload")]
90        params: &'a [u8],
91    },
92    Answer {
93        id: u64,
94
95        #[serde(borrow, with = "payload")]
96        result: &'a [u8],
97    },
98    Refusal {
99        id: u64,
100        refusal: std::borrow::Cow<'a, harmos_domain::Refusal>,
101    },
102    Credit {
103        consumed: u64,
104    },
105}
106
107/* === Traits === */
108
109/* === Implementations === */
110
111impl Role {
112    fn limit(self) -> usize {
113        match self {
114            Self::Data => Frame::LIMIT,
115            Self::Control => CONTROL_LIMIT,
116        }
117    }
118}
119
120impl Packet {
121    pub fn channel(&self) -> Role {
122        match self {
123            Self::Frame(Frame::Cancel { .. }) | Self::Credit(_) | Self::Cancellation { .. } => Role::Control,
124            _ => Role::Data,
125        }
126    }
127
128    pub fn encoded_len(&self) -> io::Result<usize> {
129        let size = postcard::experimental::serialized_size(&(VERSION, self.body()?)).map_err(invalid)?;
130        if size > self.channel().limit() {
131            return Err(invalid("packet exceeds channel limit"));
132        }
133        Ok(size + 4)
134    }
135
136    pub fn encode(&self) -> io::Result<Vec<u8>> {
137        let mut output = Vec::new();
138        self.encode_into(&mut output)?;
139        Ok(output)
140    }
141
142    fn encode_into(&self, output: &mut Vec<u8>) -> io::Result<()> {
143        let size = self.encoded_len()?;
144        if size > output.capacity() {
145            output.try_reserve_exact(size - output.len()).map_err(io::Error::other)?;
146        }
147        output.resize(size, 0);
148        output[..4].copy_from_slice(&((size - 4) as u32).to_be_bytes());
149        postcard::to_slice(&(VERSION, self.body()?), &mut output[4..]).map_err(invalid)?;
150        Ok(())
151    }
152
153    #[tracing::instrument(target = "harmos::carrier", name = "sidecar.frame.decode", level = "trace", skip_all, fields(wire_bytes = bytes.len(), channel = ?channel))]
154    pub fn decode(channel: Role, bytes: Vec<u8>) -> io::Result<Self> {
155        if bytes.is_empty() || bytes.len() > channel.limit() {
156            return Err(invalid("packet exceeds channel limit or is empty"));
157        }
158        let (_version, rest) = postcard::take_from_bytes::<u8>(&bytes).map_err(invalid)?;
159        let (body, rest) = postcard::take_from_bytes::<Body<'_>>(rest).map_err(invalid)?;
160        if !rest.is_empty() {
161            return Err(invalid("trailing sidecar transport bytes"));
162        }
163        let control = matches!(body, Body::Cancel { .. } | Body::Credit { .. });
164        if control != (channel == Role::Control) {
165            return Err(invalid("packet arrived on the wrong physical channel"));
166        }
167        Ok(match body {
168            Body::Call { id, method, params } => {
169                let method = method.to_owned();
170                let range = payload_range(&bytes, params)?;
171                Self::Frame(Frame::Call {
172                    id,
173                    method,
174                    params: FramePayload::from_range(bytes, range),
175                })
176            }
177            Body::Cancel { id, preceding } => Self::Cancellation { id, preceding },
178            Body::Push { method, params } => {
179                let method = method.to_owned();
180                let range = payload_range(&bytes, params)?;
181                Self::Frame(Frame::Push {
182                    method,
183                    params: FramePayload::from_range(bytes, range),
184                })
185            }
186            Body::Answer { id, result } => {
187                let range = payload_range(&bytes, result)?;
188                Self::Frame(Frame::Answer {
189                    id,
190                    result: FramePayload::from_range(bytes, range),
191                })
192            }
193            Body::Refusal { id, refusal } => Self::Frame(Frame::Refusal { id, refusal: refusal.into_owned() }),
194            Body::Credit { consumed } => Self::Credit(consumed),
195        })
196    }
197
198    fn body(&self) -> io::Result<Body<'_>> {
199        Ok(match self {
200            Self::Frame(Frame::Call { id, method, params }) => Body::Call { id: *id, method, params },
201            Self::Frame(Frame::Cancel { .. }) => {
202                return Err(invalid("cancellation requires a data watermark"));
203            }
204            Self::Frame(Frame::Push { method, params }) => Body::Push { method, params },
205            Self::Frame(Frame::Answer { id, result }) => Body::Answer { id: *id, result },
206            Self::Frame(Frame::Refusal { id, refusal }) => Body::Refusal {
207                id: *id,
208                refusal: std::borrow::Cow::Borrowed(refusal),
209            },
210            Self::Credit(consumed) => Body::Credit { consumed: *consumed },
211            Self::Cancellation { id, preceding } => Body::Cancel { id: *id, preceding: *preceding },
212        })
213    }
214}
215
216/* === Public Functions === */
217
218#[tracing::instrument(target = "harmos::carrier", name = "sidecar.frame.read", level = "trace", skip_all, fields(channel = ?channel))]
219pub async fn read<R: AsyncRead + Unpin>(reader: &mut R, channel: Role) -> io::Result<Option<Packet>> {
220    let Some(length) = read_length(reader, channel).await? else {
221        return Ok(None);
222    };
223    let mut body = vec![0; length];
224    reader.read_exact(&mut body).await?;
225    Packet::decode(channel, body).map(Some)
226}
227
228#[tracing::instrument(target = "harmos::carrier", name = "sidecar.frame.read", level = "trace", skip_all, fields(channel = "Data"))]
229pub async fn read_data<R: AsyncRead + Unpin>(reader: &mut R, credit: &Arc<ReceiveCredit>) -> io::Result<Option<(Packet, Consumption)>> {
230    let Some(length) = read_length(reader, Role::Data).await? else {
231        return Ok(None);
232    };
233    let consumption = credit.admit(length + 4)?;
234    let mut body = vec![0; length];
235    reader.read_exact(&mut body).await?;
236    let packet = Packet::decode(Role::Data, body)?;
237    Ok(Some((packet, consumption)))
238}
239
240pub async fn write<W: AsyncWrite + Unpin>(writer: &mut W, packet: &Packet) -> io::Result<()> {
241    write_with(writer, packet, &mut Vec::new()).await
242}
243
244/* === Private Functions === */
245
246fn payload_range(bytes: &[u8], payload: &[u8]) -> io::Result<Range<usize>> {
247    if payload.is_empty() {
248        return Ok(0..0);
249    }
250    let start =
251        (payload.as_ptr() as usize).checked_sub(bytes.as_ptr() as usize).ok_or_else(|| invalid("payload lies outside its packet"))?;
252    let end = start.checked_add(payload.len()).filter(|end| *end <= bytes.len()).ok_or_else(|| invalid("payload exceeds its packet"))?;
253    Ok(start..end)
254}
255
256#[tracing::instrument(target = "harmos::carrier", name = "sidecar.frame.write", level = "trace", skip_all, fields(wire_bytes = tracing::field::Empty, coalesced = tracing::field::Empty))]
257async fn write_with<W: AsyncWrite + Unpin>(writer: &mut W, packet: &Packet, header: &mut Vec<u8>) -> io::Result<()> {
258    let payload = match packet {
259        Packet::Frame(Frame::Call { params, .. } | Frame::Push { params, .. }) => params.as_slice(),
260        Packet::Frame(Frame::Answer { result, .. }) => result.as_slice(),
261        _ => {
262            packet.encode_into(header)?;
263            tracing::Span::current().record("wire_bytes", header.len());
264            return writer.write_all(header).await;
265        }
266    };
267    let size = packet.encoded_len()?;
268    let prefix_size = size - payload.len();
269    let coalesce = !writer.is_write_vectored() && size <= COALESCE_BYTES;
270    tracing::Span::current().record("wire_bytes", size);
271    tracing::Span::current().record("coalesced", coalesce);
272    let capacity = if coalesce { size } else { prefix_size };
273    if capacity > header.capacity() {
274        header.try_reserve_exact(capacity - header.len()).map_err(io::Error::other)?;
275    }
276    header.resize(prefix_size, 0);
277    header[..4].copy_from_slice(&((size - 4) as u32).to_be_bytes());
278    // Version 1's Fixed Body Tags Match The Canonical Serializer Above.
279    match packet {
280        Packet::Frame(Frame::Call { id, method, .. }) => {
281            postcard::to_slice(&(VERSION, 0_u32, id, method.as_str(), payload.len()), &mut header[4..]).map_err(invalid)?;
282        }
283        Packet::Frame(Frame::Push { method, .. }) => {
284            postcard::to_slice(&(VERSION, 2_u32, method.as_str(), payload.len()), &mut header[4..]).map_err(invalid)?;
285        }
286        Packet::Frame(Frame::Answer { id, .. }) => {
287            postcard::to_slice(&(VERSION, 3_u32, id, payload.len()), &mut header[4..]).map_err(invalid)?;
288        }
289        _ => unreachable!("only payload frames use a split write"),
290    }
291    // Scalar Pipe Writers Otherwise Schedule Separate Header And Payload Writes.
292    // Large Payloads Keep Their Original Allocation And The Split Write Path.
293    if coalesce {
294        header.extend_from_slice(payload);
295        return writer.write_all(header).await;
296    }
297    let mut prefix = header.as_slice();
298    let mut remaining = payload;
299    while !prefix.is_empty() || !remaining.is_empty() {
300        let slices = [IoSlice::new(prefix), IoSlice::new(remaining)];
301        let written = writer.write_vectored(&slices).await?;
302        if written == 0 {
303            return Err(io::Error::new(io::ErrorKind::WriteZero, "sidecar packet write made no progress"));
304        }
305        let prefix_bytes = written.min(prefix.len());
306        prefix = &prefix[prefix_bytes..];
307        remaining = &remaining[written - prefix_bytes..];
308    }
309    Ok(())
310}
311
312async fn read_length<R: AsyncRead + Unpin>(reader: &mut R, channel: Role) -> io::Result<Option<usize>> {
313    let mut header = [0; 4];
314    let read = reader.read(&mut header).await?;
315    if read == 0 {
316        return Ok(None);
317    }
318    reader.read_exact(&mut header[read..]).await?;
319    let length = u32::from_be_bytes(header) as usize;
320    if length == 0 || length > channel.limit() {
321        return Err(invalid("invalid sidecar packet length"));
322    }
323    Ok(Some(length))
324}
325
326fn invalid(error: impl Into<Box<dyn std::error::Error + Send + Sync>>) -> io::Error {
327    io::Error::new(io::ErrorKind::InvalidData, error)
328}
329
330/* === Tests === */
331
332#[cfg(test)]
333mod tests {
334    use super::*;
335    use std::pin::Pin;
336    use std::task::{Context, Poll};
337
338    #[derive(Default)]
339    struct ScalarWriter {
340        bytes: Vec<u8>,
341        writes: usize,
342        maximum: Option<usize>,
343    }
344
345    struct PrefixReader {
346        bytes: Vec<u8>,
347        position: usize,
348        maximum: usize,
349        reads: usize,
350        interrupted: bool,
351    }
352
353    impl PrefixReader {
354        fn new(bytes: Vec<u8>, maximum: usize) -> Self {
355            Self {
356                bytes,
357                position: 0,
358                maximum,
359                reads: 0,
360                interrupted: false,
361            }
362        }
363    }
364
365    impl AsyncRead for PrefixReader {
366        fn poll_read(mut self: Pin<&mut Self>, _: &mut Context<'_>, output: &mut tokio::io::ReadBuf<'_>) -> Poll<io::Result<()>> {
367            self.reads += 1;
368            if std::mem::take(&mut self.interrupted) {
369                let error = io::ErrorKind::Interrupted.into();
370                return Poll::Ready(Err(error));
371            }
372            let remaining = self.bytes.len() - self.position;
373            let count = remaining.min(self.maximum).min(output.remaining());
374            let end = self.position + count;
375            output.put_slice(&self.bytes[self.position..end]);
376            self.position = end;
377            Poll::Ready(Ok(()))
378        }
379    }
380
381    impl AsyncWrite for ScalarWriter {
382        fn poll_write(mut self: Pin<&mut Self>, _: &mut Context<'_>, bytes: &[u8]) -> Poll<io::Result<usize>> {
383            let count = bytes.len().min(self.maximum.unwrap_or(usize::MAX));
384            self.bytes.extend_from_slice(&bytes[..count]);
385            self.writes += 1;
386            Poll::Ready(Ok(count))
387        }
388
389        fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
390            Poll::Ready(Ok(()))
391        }
392
393        fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
394            Poll::Ready(Ok(()))
395        }
396    }
397
398    #[tokio::test]
399    async fn prefix_reads_once() {
400        let mut reader = PrefixReader::new(128_u32.to_be_bytes().to_vec(), 4);
401        assert_eq!(read_length(&mut reader, Role::Data).await.unwrap(), Some(128));
402        assert_eq!(reader.reads, 1);
403
404        for maximum in 1..4 {
405            let mut reader = PrefixReader::new(128_u32.to_be_bytes().to_vec(), maximum);
406            assert_eq!(read_length(&mut reader, Role::Data).await.unwrap(), Some(128));
407            assert_eq!(reader.reads, 4_usize.div_ceil(maximum));
408        }
409    }
410
411    #[tokio::test]
412    async fn prefix_eof_errors() {
413        let mut reader = PrefixReader::new(Vec::new(), 4);
414        assert_eq!(read_length(&mut reader, Role::Data).await.unwrap(), None);
415        for length in 1..4 {
416            let bytes = 128_u32.to_be_bytes()[..length].to_vec();
417            let mut reader = PrefixReader::new(bytes, 4);
418            let error = read_length(&mut reader, Role::Data).await.unwrap_err();
419            assert_eq!(error.kind(), io::ErrorKind::UnexpectedEof);
420        }
421    }
422
423    #[tokio::test]
424    async fn prefix_invalid_interrupted() {
425        for (channel, length) in [(Role::Data, 0), (Role::Data, Frame::LIMIT + 1), (Role::Control, 0), (Role::Control, CONTROL_LIMIT + 1)] {
426            let bytes = u32::try_from(length).unwrap().to_be_bytes().to_vec();
427            let mut reader = PrefixReader::new(bytes, 4);
428            let error = read_length(&mut reader, channel).await.unwrap_err();
429            assert_eq!(error.kind(), io::ErrorKind::InvalidData);
430        }
431
432        let mut reader = PrefixReader::new(128_u32.to_be_bytes().to_vec(), 4);
433        reader.interrupted = true;
434        let error = read_length(&mut reader, Role::Data).await.unwrap_err();
435        assert_eq!(error.kind(), io::ErrorKind::Interrupted);
436        assert_eq!(reader.position, 0);
437        assert_eq!(read_length(&mut reader, Role::Data).await.unwrap(), Some(128));
438    }
439
440    #[tokio::test]
441    async fn scalar_frames_coalesce() {
442        let mut scratch = Vec::new();
443        for frame in [
444            Frame::Call {
445                id: 1,
446                method: "call".into(),
447                params: vec![23; 4096].into(),
448            },
449            Frame::Push {
450                method: "push".into(),
451                params: vec![23; 8192].into(),
452            },
453            Frame::Answer { id: 2, result: vec![23; 32_768].into() },
454        ] {
455            let packet = Packet::Frame(frame);
456            let expected = packet.encode().unwrap();
457            let mut writer = ScalarWriter::default();
458            write_with(&mut writer, &packet, &mut scratch).await.unwrap();
459            assert_eq!(writer.bytes, expected);
460            assert_eq!(writer.writes, 1);
461            assert_eq!(scratch.capacity(), expected.len());
462        }
463    }
464
465    #[tokio::test]
466    async fn scalar_partial_bulk() {
467        let packet = Packet::Frame(Frame::Answer { id: 1, result: vec![7; 4096].into() });
468        let mut writer = ScalarWriter {
469            maximum: Some(3),
470            ..ScalarWriter::default()
471        };
472        write_with(&mut writer, &packet, &mut Vec::new()).await.unwrap();
473        assert_eq!(writer.bytes, packet.encode().unwrap());
474
475        let packet = Packet::Frame(Frame::Answer {
476            id: 2,
477            result: vec![7; COALESCE_BYTES].into(),
478        });
479        let mut writer = ScalarWriter::default();
480        let mut scratch = Vec::new();
481        write_with(&mut writer, &packet, &mut scratch).await.unwrap();
482        assert_eq!(writer.bytes, packet.encode().unwrap());
483        assert_eq!(writer.writes, 2);
484        assert!(scratch.capacity() < 32);
485    }
486
487    #[test]
488    fn malformed_payloads() {
489        let bytes = vec![1, 2, 3];
490        let foreign = vec![4, 5, 6];
491        assert!(payload_range(&bytes, &foreign).is_err());
492        assert!(std::panic::catch_unwind(|| FramePayload::from_range(vec![1], 0..2)).is_err());
493
494        let packet = Packet::Frame(Frame::Answer { id: 1, result: vec![7; 128].into() });
495        let encoded = packet.encode().unwrap();
496        let mut body = encoded[4..].to_vec();
497        body.pop();
498        assert!(Packet::decode(Role::Data, body).is_err());
499        let mut body = encoded[4..].to_vec();
500        body.push(0);
501        assert!(Packet::decode(Role::Data, body).is_err());
502    }
503
504    #[test]
505    fn payload_keeps_backing() {
506        for frame in [
507            Frame::Call {
508                id: 1,
509                method: "call".into(),
510                params: vec![23; 4096].into(),
511            },
512            Frame::Push {
513                method: "push".into(),
514                params: vec![23; 4096].into(),
515            },
516            Frame::Answer { id: 2, result: vec![23; 4096].into() },
517        ] {
518            let encoded = Packet::Frame(frame).encode().unwrap();
519            let body = encoded[4..].to_vec();
520            let expected = body.as_ptr().wrapping_add(body.len() - 4096);
521            let Packet::Frame(frame) = Packet::decode(Role::Data, body).unwrap() else { panic!("payload frame") };
522            let payload = match frame {
523                Frame::Call { params, .. } | Frame::Push { params, .. } => params,
524                Frame::Answer { result, .. } => result,
525                _ => unreachable!(),
526            };
527            assert_eq!(payload.as_ptr(), expected);
528            let retained = payload.clone();
529            drop(payload);
530            assert_eq!(retained.as_ptr(), expected);
531            assert_eq!(retained.as_slice(), [23; 4096]);
532        }
533    }
534}
harmos_sidecar/transport.rs

harmos_sidecar/
transport.rs
