active_rpc/transport/
connect.rs

1//! gRPC / Connect bridge.
2//!
3//! The bridge lets an ordinary HTTP client — a gRPC toolchain, a Connect client,
4//! or `curl` — exchange Active RPC envelopes without knowing the kernel. It
5//! speaks the unary Connect protocol over HTTP/1.1 and carries messages in the
6//! gRPC length-prefixed framing, so the same codec sits behind an HTTP/2 gRPC
7//! server without changes:
8//!
9//! ```text
10//! POST /active.rpc.v1.Agent/Call HTTP/1.1
11//! content-type: application/connect+active-infinity
12//!
13//! [0x00][len u32 big-endian][envelope bytes]
14//! ```
15//!
16//! A malformed frame, an oversized body or a bad content type is answered with a
17//! `4xx` status and a stable code (`rpc/wire`), never a panic or a dropped
18//! connection.
19//!
20//! ADR-0004 records why this is Connect over HTTP/1.1 rather than a full HTTP/2
21//! stack: it is the interoperability surface, not the hot path, and it stays
22//! dependency-free so the crate still builds on a bare toolchain.
23
24use std::io::{BufRead, BufReader, Read, Write};
25use std::net::{TcpListener, TcpStream};
26
27use crate::error::{Error, Result};
28use crate::wire::{Envelope, WireLimits};
29
30/// Service path the bridge answers.
31pub const PATH: &str = "/active.rpc.v1.Agent/Call";
32/// Connect unary content type.
33pub const CONNECT_TYPE: &str = "application/connect+active-infinity";
34/// gRPC content type (accepted, same framing).
35pub const GRPC_TYPE: &str = "application/grpc+active-infinity";
36
37/// Largest request body the bridge accepts.
38pub const MAX_BODY: usize = 64 << 20;
39/// Largest request head (request line plus headers).
40pub const MAX_HEAD: usize = 16 << 10;
41
42/// Frame one message with the gRPC 5-byte prefix.
43pub fn grpc_frame(payload: &[u8]) -> Vec<u8> {
44    let mut out = Vec::with_capacity(5 + payload.len());
45    out.push(0);
46    out.extend_from_slice(&(payload.len() as u32).to_be_bytes());
47    out.extend_from_slice(payload);
48    out
49}
50
51/// Split one gRPC-framed message off `body`, returning it and the rest.
52pub fn grpc_unframe(body: &[u8]) -> Result<(&[u8], &[u8])> {
53    if body.len() < 5 {
54        return Err(Error::Wire("grpc frame shorter than its prefix"));
55    }
56    if body[0] != 0 {
57        return Err(Error::Wire("compressed grpc frames are not supported"));
58    }
59    let len = u32::from_be_bytes(body[1..5].try_into().expect("4 bytes")) as usize;
60    if len > MAX_BODY {
61        return Err(Error::Wire("grpc frame exceeds the ceiling"));
62    }
63    let end = 5 + len;
64    if end > body.len() {
65        return Err(Error::Wire("truncated grpc frame"));
66    }
67    Ok((&body[5..end], &body[end..]))
68}
69
70/// A bridge server bound to a loopback port.
71pub struct BridgeServer {
72    listener: TcpListener,
73    address: String,
74    limits: WireLimits,
75}
76
77impl BridgeServer {
78    /// Bind an ephemeral loopback port.
79    pub fn bind_loopback() -> Result<Self> {
80        let listener = TcpListener::bind("127.0.0.1:0")
81            .map_err(|error| Error::Transport(format!("bind: {error}")))?;
82        let address = listener
83            .local_addr()
84            .map_err(|error| Error::Transport(error.to_string()))?
85            .to_string();
86        Ok(Self {
87            listener,
88            address,
89            limits: WireLimits::default(),
90        })
91    }
92
93    /// The bound address.
94    pub fn address(&self) -> &str {
95        &self.address
96    }
97
98    /// Use tighter wire limits.
99    pub fn with_limits(mut self, limits: WireLimits) -> Self {
100        self.limits = limits;
101        self
102    }
103
104    /// Serve `requests` connections, then return the number of calls served.
105    pub fn serve<H>(&self, requests: usize, mut handler: H) -> Result<u64>
106    where
107        H: FnMut(Envelope) -> Envelope,
108    {
109        let mut served = 0u64;
110        for _ in 0..requests {
111            let (stream, _) = self
112                .listener
113                .accept()
114                .map_err(|error| Error::Transport(format!("accept: {error}")))?;
115            served += self.handle(stream, &mut handler)?;
116        }
117        Ok(served)
118    }
119
120    fn handle<H>(&self, mut stream: TcpStream, handler: &mut H) -> Result<u64>
121    where
122        H: FnMut(Envelope) -> Envelope,
123    {
124        stream
125            .set_nodelay(true)
126            .map_err(|error| Error::Transport(error.to_string()))?;
127        let mut reader = BufReader::new(
128            stream
129                .try_clone()
130                .map_err(|error| Error::Transport(error.to_string()))?,
131        );
132        let Some(head) = read_head(&mut reader, MAX_HEAD)? else {
133            return Ok(0);
134        };
135        let request_line = head.first().cloned().unwrap_or_default();
136        let mut parts = request_line.split_whitespace();
137        let method = parts.next().unwrap_or_default();
138        let target = parts.next().unwrap_or_default();
139        if method != "POST" || target != PATH {
140            write_http_response(&mut stream, 404, b"rpc/not-found")?;
141            return Ok(0);
142        }
143        let Some(content_length) = head.iter().find_map(|line| {
144            let (name, value) = line.split_once(':')?;
145            if name.eq_ignore_ascii_case("content-length") {
146                value.trim().parse::<usize>().ok()
147            } else {
148                None
149            }
150        }) else {
151            // A malformed request is a 400 for that connection, not a server
152            // failure: `serve` must keep accepting.
153            write_http_response(&mut stream, 400, b"rpc/wire")?;
154            return Ok(1);
155        };
156        if content_length > MAX_BODY {
157            write_http_response(&mut stream, 413, b"rpc/wire")?;
158            return Ok(0);
159        }
160        let content_type = head
161            .iter()
162            .find_map(|line| {
163                let (name, value) = line.split_once(':')?;
164                if name.eq_ignore_ascii_case("content-type") {
165                    Some(value.trim().to_string())
166                } else {
167                    None
168                }
169            })
170            .unwrap_or_default();
171        if content_type != CONNECT_TYPE && content_type != GRPC_TYPE {
172            write_http_response(&mut stream, 415, b"rpc/wire")?;
173            return Ok(0);
174        }
175        let mut body = vec![0u8; content_length];
176        if let Err(error) = reader.read_exact(&mut body) {
177            let _ = write_http_response(&mut stream, 400, b"rpc/wire");
178            return Err(Error::Transport(format!("body: {error}")));
179        }
180        let (message, rest) = match grpc_unframe(&body) {
181            Ok(parts) => parts,
182            Err(error) => {
183                write_http_response(&mut stream, 400, error.code().as_bytes())?;
184                return Ok(1);
185            }
186        };
187        if !rest.is_empty() {
188            write_http_response(&mut stream, 400, b"rpc/wire")?;
189            return Ok(1);
190        }
191        let envelope = match Envelope::decode_with(message, self.limits) {
192            Ok(envelope) => envelope,
193            Err(error) => {
194                write_http_response(&mut stream, 400, error.code().as_bytes())?;
195                return Ok(1);
196            }
197        };
198        let response = handler(envelope);
199        let bytes = match response.encode() {
200            Ok(bytes) => bytes,
201            Err(error) => {
202                write_http_response(&mut stream, 500, error.code().as_bytes())?;
203                return Ok(1);
204            }
205        };
206        let framed = grpc_frame(&bytes);
207        write_http_response(&mut stream, 200, &framed)?;
208        Ok(1)
209    }
210}
211
212fn read_head<R: BufRead>(reader: &mut R, max: usize) -> Result<Option<Vec<String>>> {
213    let mut head = Vec::new();
214    let mut total = 0usize;
215    loop {
216        let mut line = String::new();
217        let read = reader
218            .read_line(&mut line)
219            .map_err(|error| Error::Transport(format!("head: {error}")))?;
220        if read == 0 {
221            if head.is_empty() {
222                return Ok(None);
223            }
224            break;
225        }
226        total += read;
227        if total > max {
228            return Err(Error::Wire("request head exceeds the ceiling"));
229        }
230        let trimmed = line.trim_end_matches(['\r', '\n']).to_string();
231        if trimmed.is_empty() {
232            break;
233        }
234        head.push(trimmed);
235    }
236    Ok(Some(head))
237}
238
239fn write_http_response(stream: &mut TcpStream, status: u16, body: &[u8]) -> Result<()> {
240    let reason = match status {
241        200 => "OK",
242        400 => "Bad Request",
243        404 => "Not Found",
244        413 => "Payload Too Large",
245        415 => "Unsupported Media Type",
246        _ => "Internal Server Error",
247    };
248    let head = format!(
249        "HTTP/1.1 {status} {reason}\r\ncontent-type: {CONNECT_TYPE}\r\ncontent-length: {}\r\nconnection: close\r\n\r\n",
250        body.len()
251    );
252    stream
253        .write_all(head.as_bytes())
254        .and_then(|_| stream.write_all(body))
255        .and_then(|_| stream.flush())
256        .map_err(|error| Error::Transport(error.to_string()))
257}
258
259/// A Connect/gRPC bridge client.
260pub struct ConnectClient {
261    address: String,
262    limits: WireLimits,
263}
264
265impl ConnectClient {
266    /// Connect to a bridge.
267    pub fn connect(address: &str) -> Result<Self> {
268        // The bridge answers with `connection: close`, so the client dials per
269        // request. A pooled variant would negotiate keep-alive instead.
270        Ok(Self {
271            address: address.to_string(),
272            limits: WireLimits::default(),
273        })
274    }
275
276    /// Use tighter limits.
277    pub fn with_limits(mut self, limits: WireLimits) -> Self {
278        self.limits = limits;
279        self
280    }
281
282    /// Call the bridge with one envelope.
283    pub fn call(&mut self, envelope: &Envelope) -> Result<Envelope> {
284        let body = envelope.encode()?;
285        let framed = grpc_frame(&body);
286        let (status, response) = self.post(&framed, CONNECT_TYPE)?;
287        if status != 200 {
288            let code = String::from_utf8_lossy(&response).trim().to_string();
289            return Err(match code.as_str() {
290                "rpc/wire" => Error::Wire("bridge rejected the frame"),
291                other => Error::Provider {
292                    code: other.to_string(),
293                },
294            });
295        }
296        let (message, rest) = grpc_unframe(&response)?;
297        if !rest.is_empty() {
298            return Err(Error::Wire("trailing bytes in bridge response"));
299        }
300        Envelope::decode_with(message, self.limits)
301    }
302
303    /// Send a raw body (used to prove malformed frames are rejected).
304    pub fn post_raw(&mut self, body: &[u8]) -> Result<(u16, Vec<u8>)> {
305        self.post(body, CONNECT_TYPE)
306    }
307
308    /// Send a body with an explicit content type.
309    pub fn post_typed(&mut self, body: &[u8], content_type: &str) -> Result<(u16, Vec<u8>)> {
310        self.post(body, content_type)
311    }
312
313    fn post(&mut self, body: &[u8], content_type: &str) -> Result<(u16, Vec<u8>)> {
314        let head = format!(
315            "POST {PATH} HTTP/1.1\r\nhost: active-rpc\r\ncontent-type: {content_type}\r\ncontent-length: {}\r\nconnection: close\r\n\r\n",
316            body.len()
317        );
318        let mut stream = TcpStream::connect(&self.address)
319            .map_err(|error| Error::Transport(format!("connect: {error}")))?;
320        stream
321            .write_all(head.as_bytes())
322            .and_then(|_| stream.write_all(body))
323            .and_then(|_| stream.flush())
324            .map_err(|error| Error::Transport(error.to_string()))?;
325        let mut reader = BufReader::new(stream);
326        let Some(head) = read_head(&mut reader, MAX_HEAD)? else {
327            return Err(Error::Transport(
328                "bridge closed without a response".to_string(),
329            ));
330        };
331        let status: u16 = head
332            .first()
333            .and_then(|line| line.split_whitespace().nth(1))
334            .and_then(|code| code.parse().ok())
335            .ok_or(Error::Wire("malformed status line"))?;
336        let length: usize = head
337            .iter()
338            .find_map(|line| {
339                let (name, value) = line.split_once(':')?;
340                if name.eq_ignore_ascii_case("content-length") {
341                    value.trim().parse().ok()
342                } else {
343                    None
344                }
345            })
346            .ok_or(Error::Wire("missing response length"))?;
347        if length > MAX_BODY {
348            return Err(Error::Wire("response exceeds the ceiling"));
349        }
350        let mut response = vec![0u8; length];
351        reader
352            .read_exact(&mut response)
353            .map_err(|error| Error::Transport(format!("response: {error}")))?;
354        Ok((status, response))
355    }
356}
active_rpc/transport/connect.rs

active_rpc/transport/
connect.rs
