From 83d1058498b46e16506645b5e96c1c3e7baad47f Mon Sep 17 00:00:00 2001 From: Kinflou Date: Tue, 1 Sep 2026 22:51:36 +0800 Subject: [PATCH] feat: consumer-side Client + a TCP transport (step 7e) The mirror of 7d's Server, and a real stream transport under it. - client Client -- owns a Transport + a WireFormat, hands out request ids, reuses its send / recv buffers. call(call_id, ¶ms) encodes straight into the frame header (no params scratch), sends, blocks for the response, and returns (Envelope, &W) -- both views out of one &mut self borrow, so the generated stub can decode Ok as R or map an Err ordinal without reaching back into the client. One outstanding call at a time; pipelining is a later, additive layer. - transport Tcp -- a Transport over a TCP byte stream. No message boundaries on a stream, so each frame gets a u32 LE length prefix; recv reads exactly one and rejects a prefix over MAX_FRAME (16 MiB) before allocating. new() wraps an accepted stream, connect() dials. - wire encode_request_header split out of encode_request, so a client frames the header then serializes params into the same buffer. tests/client_roundtrip.rs: a Greet protocol (with a raised schema error) driven through a Client-backed stub against a Server -- once over InMemory, once over loopback Tcp -- plus a check that back-to-back Tcp frames keep their boundaries. Client is alloc-gated (Vec buffers), like serve / transport; Tcp is std-gated, like InMemory. All feature configs green: cargo test (13 lib + 3 client + 3 dispatch + 1 serve), --no-default-features (2), --features alloc (9). Warnings unchanged (1, the pre-existing abi_stable macro lint). --- runtime/core/src/client.rs | 91 +++++++++++++ runtime/core/src/lib.rs | 4 +- runtime/core/src/transport.rs | 76 +++++++++++ runtime/core/src/wire.rs | 23 +++- runtime/core/tests/client_roundtrip.rs | 182 +++++++++++++++++++++++++ 5 files changed, 373 insertions(+), 3 deletions(-) create mode 100644 runtime/core/src/client.rs create mode 100644 runtime/core/tests/client_roundtrip.rs diff --git a/runtime/core/src/client.rs b/runtime/core/src/client.rs new file mode 100644 index 0000000..c15ecfc --- /dev/null +++ b/runtime/core/src/client.rs @@ -0,0 +1,91 @@ +//! Consumer-side calling: the mirror of [`Server`](crate::serve::Server). +//! +//! A [`Client`] owns a [`Transport`] and a [`WireFormat`], hands out request +//! ids, and reuses its send / receive buffers across calls (§4.6 — no per-call +//! allocation on the frame path). One call is: encode the params straight into +//! the frame, send, block for the response, hand back the [`Envelope`] +//! borrowing the receive buffer. +//! +//! The generated `Client` stub wraps this: it knows each function's +//! ordinal and its params / result / error types, so it calls [`Client::call`] +//! with a `u16` and turns the returned `(Envelope, &W)` into `Result>` — `Ok` payload decoded as `R`, an `Err` ordinal mapped through +//! its generated error table. This type stays protocol-agnostic; it does no +//! decoding, which is why `call` also hands back the format. + +use alloc::vec::Vec; + +use serde::Serialize; + +use crate::contract::{Envelope, RuntimeError, WireFormat}; +use crate::transport::Transport; +use crate::wire; + +/// A calling endpoint bound to one transport. +pub struct Client { + transport: T, + format: W, + next_id: u64, + request: Vec, + response: Vec, +} + +impl Client { + pub fn new(transport: T, format: W) -> Self { + Self { + transport, + format, + next_id: 0, + request: Vec::new(), + response: Vec::new(), + } + } + + /// Make call `call_id` with `params`, block for the response, and return + /// its [`Envelope`] (borrowing this client's receive buffer) together with + /// the format to decode it: `Ok(payload)` for the stub to read as `R`, or + /// `Err { id, body }` to map through the generated error table. + /// + /// The format rides along because both views come out of the same + /// `&mut self` borrow — the stub can't reach back into the client for it + /// while holding the envelope. + /// + /// Borrows `self` mutably for as long as the result is held: the previous + /// response must be decoded (or dropped) before the next call. A pipelined + /// / multiplexed client is a later, additive layer. + pub fn call

(&mut self, call_id: u16, params: &P) -> Result<(Envelope<'_>, &W), RuntimeError> + where + P: Serialize + ?Sized, + { + let request_id = self.next_id; + self.next_id = self.next_id.wrapping_add(1); + + self.request.clear(); + wire::encode_request_header(call_id, request_id, &mut self.request); + self.format.encode(params, &mut self.request)?; + self.transport.send(&self.request)?; + + self.response.clear(); + self.transport.recv(&mut self.response)?; + + let (echoed, envelope) = + wire::decode_response(&self.response).ok_or(RuntimeError::Framing)?; + if echoed != request_id { + // One outstanding call at a time, so a mismatched id is a stale or + // corrupt frame, not reordering. + return Err(RuntimeError::Framing); + } + let envelope = Envelope::decode(envelope).ok_or(RuntimeError::Framing)?; + Ok((envelope, &self.format)) + } + + /// The underlying transport, e.g. to close it or read its peer address. + pub fn transport_mut(&mut self) -> &mut T { + &mut self.transport + } + + /// Consume the client, returning its transport. + pub fn into_transport(self) -> T { + self.transport + } +} diff --git a/runtime/core/src/lib.rs b/runtime/core/src/lib.rs index 68f44dd..95a0100 100644 --- a/runtime/core/src/lib.rs +++ b/runtime/core/src/lib.rs @@ -15,7 +15,9 @@ pub mod wire; #[cfg(feature = "std")] pub mod format; -// The frame transport and the provider serve loop. +// The frame transport, the provider serve loop, and the consumer call side. +#[cfg(feature = "alloc")] +pub mod client; #[cfg(feature = "alloc")] pub mod serve; #[cfg(feature = "alloc")] diff --git a/runtime/core/src/transport.rs b/runtime/core/src/transport.rs index 3b59cf8..46d36eb 100644 --- a/runtime/core/src/transport.rs +++ b/runtime/core/src/transport.rs @@ -1,6 +1,10 @@ //! The byte-frame transport. Message-oriented — `send` / `recv` move whole //! request / response frames (see [`wire`](crate::wire)). Sync, to pair with //! the sync [`Dispatch`](crate::contract::Dispatch) and [`Server`](crate::serve::Server). +//! +//! A datagram medium (`InMemory`, and later UDP) carries one frame per message +//! natively. A byte stream ([`Tcp`]) has no message boundaries, so it adds a +//! `u32` length prefix per frame. use alloc::vec::Vec; @@ -55,3 +59,75 @@ mod in_memory { #[cfg(feature = "std")] pub use in_memory::{duplex, InMemory}; + +#[cfg(feature = "std")] +mod tcp { + use std::io::{Read, Write}; + use std::net::{TcpStream, ToSocketAddrs}; + + use super::{Transport, Vec}; + use crate::contract::RuntimeError; + + /// Reject a length prefix larger than this before allocating for it — a + /// peer claiming a 4 GiB frame should not cost 4 GiB. Frames are datagrams + /// (one call), so the ceiling is generous. + const MAX_FRAME: usize = 16 * 1024 * 1024; + + /// A [`Transport`] over a TCP byte stream. Each frame is `[len: u32 LE] + /// [frame bytes]`; `recv` reads exactly one. + pub struct Tcp { + stream: TcpStream, + len: [u8; 4], + } + + impl Tcp { + /// Wrap an established stream (e.g. from `TcpListener::accept`). + pub fn new(stream: TcpStream) -> Self { + Self { + stream, + len: [0; 4], + } + } + + /// Connect to `addr`. + pub fn connect(addr: impl ToSocketAddrs) -> Result { + TcpStream::connect(addr) + .map(Self::new) + .map_err(|_| RuntimeError::Transport) + } + + /// The wrapped stream. + pub fn stream(&self) -> &TcpStream { + &self.stream + } + } + + impl Transport for Tcp { + fn send(&mut self, frame: &[u8]) -> Result<(), RuntimeError> { + let len = u32::try_from(frame.len()).map_err(|_| RuntimeError::Framing)?; + self.stream + .write_all(&len.to_le_bytes()) + .and_then(|()| self.stream.write_all(frame)) + .and_then(|()| self.stream.flush()) + .map_err(|_| RuntimeError::Transport) + } + + fn recv(&mut self, buf: &mut Vec) -> Result<(), RuntimeError> { + self.stream + .read_exact(&mut self.len) + .map_err(|_| RuntimeError::Transport)?; + let len = u32::from_le_bytes(self.len) as usize; + if len > MAX_FRAME { + return Err(RuntimeError::Framing); + } + buf.clear(); + buf.resize(len, 0); + self.stream + .read_exact(buf) + .map_err(|_| RuntimeError::Transport) + } + } +} + +#[cfg(feature = "std")] +pub use tcp::Tcp; diff --git a/runtime/core/src/wire.rs b/runtime/core/src/wire.rs index 9a5f09f..334923e 100644 --- a/runtime/core/src/wire.rs +++ b/runtime/core/src/wire.rs @@ -12,11 +12,18 @@ use crate::contract::BufMut; const REQUEST_HEADER: usize = 2 + 8; const RESPONSE_HEADER: usize = 8; -/// Write a request frame. +/// Write a request frame around already-encoded params. pub fn encode_request(call_id: u16, request_id: u64, params: &[u8], out: &mut dyn BufMut) { + encode_request_header(call_id, request_id, out); + out.put_slice(params); +} + +/// Write just the request header. A client follows it by serializing the params +/// straight into the same buffer — no params scratch buffer, one contiguous +/// frame. +pub fn encode_request_header(call_id: u16, request_id: u64, out: &mut dyn BufMut) { out.put_u16_le(call_id); out.put_u64_le(request_id); - out.put_slice(params); } /// `(call_id, request_id, params)` — `params` borrows `frame`. `None` if the @@ -66,4 +73,16 @@ mod tests { assert_eq!(decode_request(&[0, 0, 0]), None); assert_eq!(decode_response(&[0, 0, 0]), None); } + + #[test] + fn header_then_params_equals_whole_frame() { + let mut split = Vec::new(); + encode_request_header(7, 99, &mut split); + split.put_slice(b"body"); + + let mut whole = Vec::new(); + encode_request(7, 99, b"body", &mut whole); + + assert_eq!(split, whole); + } } diff --git a/runtime/core/tests/client_roundtrip.rs b/runtime/core/tests/client_roundtrip.rs new file mode 100644 index 0000000..ac8b121 --- /dev/null +++ b/runtime/core/tests/client_roundtrip.rs @@ -0,0 +1,182 @@ +//! The consumer side end to end: a hand-written stand-in for what +//! `comline-rust` will generate — a `Client`-backed stub — talking to a +//! `Server` over a real transport. Once over `InMemory`, once over loopback +//! `Tcp` (exercising the length-prefixed stream framing). +#![cfg(feature = "std")] + +use std::net::{TcpListener, TcpStream}; +use std::thread; + +use comline_runtime::client::Client; +use comline_runtime::contract::{ + BufMut, CallError, Dispatch, Envelope, Kind, RuntimeError, WireFormat, +}; +use comline_runtime::format::MsgPack; +use comline_runtime::serve::Server; +use comline_runtime::transport::{duplex, Tcp, Transport}; +use serde::{Deserialize, Serialize}; + +// protocol Greet { +// function hello(name: str) -> str ! Rude; +// } + +#[derive(Serialize, Deserialize)] +struct HelloParams<'a> { + #[serde(borrow)] + name: &'a str, +} + +#[derive(Serialize, Deserialize, PartialEq, Eq, Debug)] +struct Rude { + reason: String, +} + +const CALLS: &[&str] = &["hello"]; +const ERR_RUDE: u16 = 0; // schema-global error ordinal + +/// The generated per-function error enum for `hello`. +#[derive(Debug, PartialEq, Eq)] +enum GreetHelloError { + Rude(Rude), +} + +// ── provider ────────────────────────────────────────────────────────────── + +trait Greet { + fn hello(&self, name: &str) -> Result; +} + +struct GreetDispatcher(T); + +impl Dispatch for GreetDispatcher { + fn dispatch( + &self, + call: Kind, + params: &[u8], + fmt: &W, + out: &mut dyn BufMut, + ) -> Result<(), RuntimeError> { + match call.resolve(CALLS).ok_or(RuntimeError::UnknownCall)? { + 0 => { + let p: HelloParams = fmt.decode(params)?; + match self.0.hello(p.name) { + Ok(reply) => { + let mut body = Vec::new(); + fmt.encode(&reply, &mut body)?; + Envelope::encode_ok(&body, out); + } + Err(GreetHelloError::Rude(e)) => { + let mut body = Vec::new(); + fmt.encode(&e, &mut body)?; + Envelope::encode_err(ERR_RUDE, &body, out); + } + } + Ok(()) + } + _ => Err(RuntimeError::UnknownCall), + } + } +} + +struct Polite; + +impl Greet for Polite { + fn hello(&self, name: &str) -> Result { + if name.is_empty() { + return Err(GreetHelloError::Rude(Rude { + reason: "no name given".into(), + })); + } + Ok(format!("hi, {name}")) + } +} + +// ── consumer: stand-in for the generated `GreetClient` ──────────────────── + +struct GreetClient(Client); + +impl GreetClient { + fn hello(&mut self, name: &str) -> Result> { + let (reply, fmt) = self.0.call(0, &HelloParams { name })?; + match reply { + Envelope::Ok(payload) => fmt.decode(payload).map_err(CallError::Runtime), + Envelope::Err { id: ERR_RUDE, body } => { + let e: Rude = fmt.decode(body)?; + Err(CallError::App(GreetHelloError::Rude(e))) + } + Envelope::Err { id, .. } => Err(CallError::Runtime(RuntimeError::Remote { id })), + } + } +} + +// ── the tests ──────────────────────────────────────────────────────────── + +#[test] +fn a_call_and_a_raised_error_round_trip_over_in_memory() { + let (client_side, provider_side) = duplex(); + + let provider = thread::spawn(move || { + let mut provider_side = provider_side; + Server::new(GreetDispatcher(Polite), MsgPack) + .serve(&mut provider_side) + .unwrap(); + }); + + let mut client = GreetClient(Client::new(client_side, MsgPack)); + + assert_eq!(client.hello("world").unwrap(), "hi, world"); + assert_eq!( + client.hello("").unwrap_err(), + CallError::App(GreetHelloError::Rude(Rude { + reason: "no name given".into(), + })), + ); + + drop(client); // closes the transport -> `serve` returns + provider.join().unwrap(); +} + +#[test] +fn a_call_round_trips_over_loopback_tcp() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + + let provider = thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut transport = Tcp::new(stream); + Server::new(GreetDispatcher(Polite), MsgPack) + .serve(&mut transport) + .unwrap(); + }); + + let mut client = GreetClient(Client::new(Tcp::connect(addr).unwrap(), MsgPack)); + + assert_eq!(client.hello("over tcp").unwrap(), "hi, over tcp"); + assert_eq!(client.hello("again").unwrap(), "hi, again"); // second call, same framing + + drop(client); // half-closes -> provider's `read_exact` hits EOF -> `serve` returns + provider.join().unwrap(); +} + +/// A stream `Transport` must not confuse two frames that arrive back to back. +#[test] +fn tcp_framing_keeps_frame_boundaries() { + let (a, b) = { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let client = TcpStream::connect(addr).unwrap(); + let (server, _) = listener.accept().unwrap(); + (Tcp::new(client), Tcp::new(server)) + }; + let mut a = a; + let mut b = b; + + a.send(b"first").unwrap(); + a.send(b"second and longer").unwrap(); + + let mut buf = Vec::new(); + b.recv(&mut buf).unwrap(); + assert_eq!(buf, b"first"); + b.recv(&mut buf).unwrap(); + assert_eq!(buf, b"second and longer"); +}