feat: consumer-side Client + a TCP transport (rollout step 7e)

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 `<Proto>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<R,
+//! CallError<E>>` — `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<T, W> {
+    transport: T,
+    format: W,
+    next_id: u64,
+    request: Vec<u8>,
+    response: Vec<u8>,
+}
+
+impl<T: Transport, W: WireFormat> Client<T, W> {
+    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<P>(&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
+    }
+}

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")]

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<Self, RuntimeError> {
+            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<u8>) -> 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;

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);
+    }
 }