active_rpc/transport/
mod.rs

1//! Transports.
2//!
3//! A transport moves [`Envelope`]s. The kernel is independent of it, so the same
4//! capability and promise semantics hold whether a call stays in the process,
5//! crosses a shared-memory segment, goes over QUIC, or is bridged to a
6//! gRPC/Connect client.
7//!
8//! | Transport | Module | Property |
9//! |-----------|--------|----------|
10//! | in-process | this module | no serialization: envelopes move by value |
11//! | TCP | this module | length-prefixed envelopes, cross-process |
12//! | shared memory | [`shm`] | zero-copy frames inside one segment |
13//! | QUIC | [`quic`] | real QUIC over loopback (feature `quic`) |
14//! | Connect/gRPC bridge | [`connect`] | HTTP/1.1 Connect + gRPC message framing |
15//!
16//! Every transport rejects a malformed frame with a stable error instead of
17//! panicking, and every transport carries the capability table in the same
18//! envelope shape.
19
20use std::collections::VecDeque;
21use std::io::{Read, Write};
22use std::net::{TcpListener, TcpStream};
23
24use crate::error::{Error, Result};
25use crate::wire::{Envelope, WireLimits};
26
27pub mod connect;
28#[cfg(feature = "quic")]
29pub mod quic;
30pub mod shm;
31
32/// Largest frame a byte-oriented transport accepts.
33pub const MAX_FRAME_BYTES: usize = 64 << 20;
34
35/// A bidirectional envelope pipe.
36pub trait Transport {
37    /// Stable transport name.
38    fn name(&self) -> &'static str;
39
40    /// Send one envelope.
41    fn send(&mut self, envelope: &Envelope) -> Result<()>;
42
43    /// Receive one envelope, if one is available.
44    fn recv(&mut self) -> Result<Option<Envelope>>;
45
46    /// Bytes written on the wire so far (zero for in-process transports).
47    fn bytes_written(&self) -> u64 {
48        0
49    }
50}
51
52// ---------------------------------------------------------------------------
53// In-process
54// ---------------------------------------------------------------------------
55
56#[derive(Default)]
57struct Mailboxes {
58    a_to_b: VecDeque<Envelope>,
59    b_to_a: VecDeque<Envelope>,
60}
61
62/// An in-process link: envelopes move by value, nothing is encoded.
63///
64/// The kernel sees the same `Envelope` type either way, so switching a peer from
65/// in-process to TCP changes the transport and nothing else.
66#[derive(Clone, Default)]
67pub struct InProcessTransport {
68    shared: Option<std::rc::Rc<std::cell::RefCell<Mailboxes>>>,
69    /// True on the endpoint that owns the `a` side.
70    is_a: bool,
71    /// Envelopes pushed into this endpoint directly (single-ended mode).
72    local: VecDeque<Envelope>,
73}
74
75impl std::fmt::Debug for InProcessTransport {
76    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
77        f.debug_struct("InProcessTransport")
78            .field("paired", &self.shared.is_some())
79            .finish()
80    }
81}
82
83impl InProcessTransport {
84    /// A pipe pair. Sending on one end is receiving on the other.
85    pub fn pair() -> (InProcessTransport, InProcessTransport) {
86        let shared = std::rc::Rc::new(std::cell::RefCell::new(Mailboxes::default()));
87        (
88            InProcessTransport {
89                shared: Some(std::rc::Rc::clone(&shared)),
90                is_a: true,
91                local: VecDeque::new(),
92            },
93            InProcessTransport {
94                shared: Some(shared),
95                is_a: false,
96                local: VecDeque::new(),
97            },
98        )
99    }
100
101    /// A single-ended queue: `send` feeds `recv` on the same endpoint.
102    pub fn queue() -> InProcessTransport {
103        InProcessTransport::default()
104    }
105}
106
107impl Transport for InProcessTransport {
108    fn name(&self) -> &'static str {
109        "in-process"
110    }
111
112    fn send(&mut self, envelope: &Envelope) -> Result<()> {
113        // No serialization: the envelope is cloned in memory. A real in-process
114        // link hands ownership to the peer task; the observable property is
115        // that no bytes are produced.
116        match &self.shared {
117            Some(shared) => {
118                let mut shared = shared.borrow_mut();
119                if self.is_a {
120                    shared.a_to_b.push_back(envelope.clone());
121                } else {
122                    shared.b_to_a.push_back(envelope.clone());
123                }
124            }
125            None => self.local.push_back(envelope.clone()),
126        }
127        Ok(())
128    }
129
130    fn recv(&mut self) -> Result<Option<Envelope>> {
131        match &self.shared {
132            Some(shared) => {
133                let mut shared = shared.borrow_mut();
134                Ok(if self.is_a {
135                    shared.b_to_a.pop_front()
136                } else {
137                    shared.a_to_b.pop_front()
138                })
139            }
140            None => Ok(self.local.pop_front()),
141        }
142    }
143
144    fn bytes_written(&self) -> u64 {
145        0
146    }
147}
148
149// ---------------------------------------------------------------------------
150// TCP (cross-process)
151// ---------------------------------------------------------------------------
152
153/// A length-prefixed envelope channel over a TCP connection.
154#[derive(Debug)]
155pub struct TcpTransport {
156    stream: TcpStream,
157    inbox: VecDeque<Envelope>,
158    written: u64,
159    limits: WireLimits,
160}
161
162impl TcpTransport {
163    /// Wrap an established stream.
164    pub fn new(stream: TcpStream) -> Result<Self> {
165        stream
166            .set_nodelay(true)
167            .map_err(|error| Error::Transport(error.to_string()))?;
168        Ok(Self {
169            stream,
170            inbox: VecDeque::new(),
171            written: 0,
172            limits: WireLimits::default(),
173        })
174    }
175
176    /// Use tighter wire limits (server side of an untrusted connection).
177    pub fn with_limits(mut self, limits: WireLimits) -> Self {
178        self.limits = limits;
179        self
180    }
181
182    /// Connect to a peer.
183    pub fn connect(address: &str) -> Result<Self> {
184        let stream = TcpStream::connect(address)
185            .map_err(|error| Error::Transport(format!("connect: {error}")))?;
186        Self::new(stream)
187    }
188}
189
190impl Transport for TcpTransport {
191    fn name(&self) -> &'static str {
192        "tcp"
193    }
194
195    fn send(&mut self, envelope: &Envelope) -> Result<()> {
196        let bytes = envelope.encode()?;
197        write_frame(&mut self.stream, &bytes)?;
198        self.written += (bytes.len() + 4) as u64;
199        Ok(())
200    }
201
202    fn recv(&mut self) -> Result<Option<Envelope>> {
203        if let Some(envelope) = self.inbox.pop_front() {
204            return Ok(Some(envelope));
205        }
206        let Some(body) = read_frame(&mut self.stream, &self.limits)? else {
207            return Ok(None);
208        };
209        Ok(Some(Envelope::decode_with(&body, self.limits)?))
210    }
211
212    fn bytes_written(&self) -> u64 {
213        self.written
214    }
215}
216
217/// Encode a frame length.
218pub fn frame(len: usize) -> Result<[u8; 4]> {
219    if len > MAX_FRAME_BYTES {
220        return Err(Error::Wire("frame exceeds the transport ceiling"));
221    }
222    Ok((len as u32).to_le_bytes())
223}
224
225/// Serve one connection: read envelopes and answer each with `handler`.
226///
227/// A malformed frame ends the connection with an error envelope instead of
228/// crashing the server.
229pub fn serve_connection<H>(mut transport: TcpTransport, mut handler: H) -> Result<u64>
230where
231    H: FnMut(Envelope) -> Envelope,
232{
233    let mut served = 0u64;
234    loop {
235        match transport.recv() {
236            Ok(Some(envelope)) => {
237                let response = handler(envelope);
238                transport.send(&response)?;
239                served += 1;
240            }
241            Ok(None) => return Ok(served),
242            Err(error) => {
243                let reply = Envelope::error(0, &error);
244                let _ = transport.send(&reply);
245                return Err(error);
246            }
247        }
248    }
249}
250
251/// Bind a loopback listener on an ephemeral port.
252pub fn bind_loopback() -> Result<(TcpListener, String)> {
253    let listener =
254        TcpListener::bind("127.0.0.1:0").map_err(|error| Error::Transport(error.to_string()))?;
255    let address = listener
256        .local_addr()
257        .map_err(|error| Error::Transport(error.to_string()))?
258        .to_string();
259    Ok((listener, address))
260}
261
262/// Read one exact frame from a generic reader.
263pub fn read_frame<R: Read>(reader: &mut R, limits: &WireLimits) -> Result<Option<Vec<u8>>> {
264    let mut header = [0u8; 4];
265    match reader.read_exact(&mut header) {
266        Ok(()) => {}
267        Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(None),
268        Err(error) => return Err(Error::Transport(error.to_string())),
269    }
270    let len = u32::from_le_bytes(header) as usize;
271    if len > limits.max_bytes {
272        return Err(Error::Wire("frame exceeds the size ceiling"));
273    }
274    let mut body = vec![0u8; len];
275    reader
276        .read_exact(&mut body)
277        .map_err(|error| Error::Transport(error.to_string()))?;
278    Ok(Some(body))
279}
280
281/// Write one exact frame to a generic writer.
282pub fn write_frame<W: Write>(writer: &mut W, body: &[u8]) -> Result<()> {
283    writer
284        .write_all(&frame(body.len())?)
285        .and_then(|_| writer.write_all(body))
286        .and_then(|_| writer.flush())
287        .map_err(|error| Error::Transport(error.to_string()))
288}
active_rpc/transport/mod.rs

active_rpc/transport/
mod.rs
