feat: per-call timeout — Client::call_with_timeout + Transport::recv_timeout

runtime/core/src/client.rs

@@ -15,6 +15,8 @@
 
 use alloc::vec::Vec;
 
+use core::time::Duration;
+
 use serde::Serialize;
 
 use crate::contract::{Envelope, RuntimeError, WireFormat};
@@ -54,6 +56,34 @@ impl<T: Transport, W: WireFormat> Client<T, W> {
     /// 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,
+    {
+        self.request_response(call_id, params, None)
+    }
+
+    /// [`call`](Self::call), but give up after `timeout` waiting for the
+    /// response — `Err(RuntimeError::Timeout)`. What a generated stub emits for
+    /// a `@timeout_ms` function annotation. Honoured only by transports that
+    /// override [`Transport::recv_timeout`] (the `std` ones do); others block.
+    pub fn call_with_timeout<P>(
+        &mut self,
+        call_id: u16,
+        params: &P,
+        timeout: Duration,
+    ) -> Result<(Envelope<'_>, &W), RuntimeError>
+    where
+        P: Serialize + ?Sized,
+    {
+        self.request_response(call_id, params, Some(timeout))
+    }
+
+    fn request_response<P>(
+        &mut self,
+        call_id: u16,
+        params: &P,
+        timeout: Option<Duration>,
+    ) -> Result<(Envelope<'_>, &W), RuntimeError>
     where
         P: Serialize + ?Sized,
     {
@@ -66,7 +96,14 @@ impl<T: Transport, W: WireFormat> Client<T, W> {
         self.transport.send(&self.request)?;
 
         self.response.clear();
-        self.transport.recv(&mut self.response)?;
+        match timeout {
+            None => self.transport.recv(&mut self.response)?,
+            Some(d) => {
+                if !self.transport.recv_timeout(&mut self.response, d)? {
+                    return Err(RuntimeError::Timeout);
+                }
+            }
+        }
 
         let (echoed, envelope) =
             wire::decode_response(&self.response).ok_or(RuntimeError::Framing)?;

runtime/core/src/transport.rs

@@ -6,6 +6,8 @@
 //! natively. A byte stream ([`Tcp`]) has no message boundaries, so it adds a
 //! `u32` length prefix per frame.
 
+use core::time::Duration;
+
 use alloc::vec::Vec;
 
 use crate::contract::RuntimeError;
@@ -18,6 +20,19 @@ pub trait Transport {
     /// Receive the next frame into `buf` (the caller clears and reuses it).
     /// `Err(RuntimeError::Transport)` once the peer is gone.
     fn recv(&mut self, buf: &mut Vec<u8>) -> Result<(), RuntimeError>;
+
+    /// Receive the next frame, waiting at most `timeout`. `Ok(true)` — a
+    /// frame was read into `buf`; `Ok(false)` — the timeout elapsed first.
+    ///
+    /// The default **blocks**, ignoring `timeout` — a transport with no clock
+    /// still works, it just can't honour a per-call deadline. `std` transports
+    /// override this. [`Client::call_with_timeout`](crate::client::Client::call_with_timeout)
+    /// is the caller.
+    fn recv_timeout(&mut self, buf: &mut Vec<u8>, timeout: Duration) -> Result<bool, RuntimeError> {
+        let _ = timeout;
+        self.recv(buf)?;
+        Ok(true)
+    }
 }
 
 #[cfg(feature = "std")]
@@ -72,6 +87,23 @@ mod in_memory {
             buf.extend_from_slice(&frame);
             Ok(())
         }
+
+        fn recv_timeout(
+            &mut self,
+            buf: &mut Vec<u8>,
+            timeout: super::Duration,
+        ) -> Result<bool, RuntimeError> {
+            use std::sync::mpsc::RecvTimeoutError;
+            match self.rx.recv_timeout(timeout) {
+                Ok(frame) => {
+                    buf.clear();
+                    buf.extend_from_slice(&frame);
+                    Ok(true)
+                }
+                Err(RecvTimeoutError::Timeout) => Ok(false),
+                Err(RecvTimeoutError::Disconnected) => Err(RuntimeError::Transport),
+            }
+        }
     }
 }
 
@@ -80,10 +112,10 @@ pub use in_memory::{duplex, InMemory};
 
 #[cfg(feature = "std")]
 mod tcp {
-    use std::io::{Read, Write};
+    use std::io::{ErrorKind, Read, Write};
     use std::net::{TcpStream, ToSocketAddrs};
 
-    use super::{Transport, Vec};
+    use super::{Duration, Transport, Vec};
     use crate::contract::RuntimeError;
 
     /// Reject a length prefix larger than this before allocating for it — a
@@ -144,6 +176,49 @@ mod tcp {
                 .read_exact(buf)
                 .map_err(|_| RuntimeError::Transport)
         }
+
+        /// Best-effort: sets a read timeout for the length-prefix and body
+        /// reads, then clears it. A timeout part-way through a frame leaves the
+        /// stream desynced — treat a timed-out call as fatal to the connection
+        /// and drop the [`Tcp`].
+        fn recv_timeout(
+            &mut self,
+            buf: &mut Vec<u8>,
+            timeout: Duration,
+        ) -> Result<bool, RuntimeError> {
+            self.stream
+                .set_read_timeout(Some(timeout))
+                .map_err(|_| RuntimeError::Transport)?;
+            let outcome = self.recv_within(buf);
+            let _ = self.stream.set_read_timeout(None);
+            outcome
+        }
+    }
+
+    impl Tcp {
+        fn recv_within(&mut self, buf: &mut Vec<u8>) -> Result<bool, RuntimeError> {
+            let timed_out = |e: &std::io::Error| {
+                matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut)
+            };
+            if let Err(e) = self.stream.read_exact(&mut self.len) {
+                return if timed_out(&e) {
+                    Ok(false)
+                } else {
+                    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);
+            match self.stream.read_exact(buf) {
+                Ok(()) => Ok(true),
+                Err(e) if timed_out(&e) => Ok(false),
+                Err(_) => Err(RuntimeError::Transport),
+            }
+        }
     }
 }
 

runtime/core/tests/timeout_roundtrip.rs

@@ -0,0 +1,97 @@
+//! `Client::call_with_timeout` — what a generated stub emits for a
+//! `@timeout_ms` function annotation. Over `InMemory` (which overrides
+//! `Transport::recv_timeout`): the call gives up when no reply comes, and
+//! succeeds normally when one does before the deadline.
+#![cfg(feature = "std")]
+
+use std::thread;
+use std::time::Duration;
+
+use comline_runtime::client::Client;
+use comline_runtime::contract::{BufMut, Dispatch, Envelope, Kind, RuntimeError, WireFormat};
+use comline_runtime::format::MsgPack;
+use comline_runtime::serve::Server;
+use comline_runtime::transport::duplex;
+use serde::{Deserialize, Serialize};
+
+// protocol Echo { function ping(n: u32) -> u32; }
+
+#[derive(Serialize, Deserialize)]
+struct PingParams {
+    n: u32,
+}
+
+const CALLS: &[&str] = &["ping"];
+
+trait Echo {
+    fn ping(&self, n: u32) -> u32;
+}
+
+struct EchoDispatcher<T>(T);
+
+impl<T: Echo> Dispatch for EchoDispatcher<T> {
+    fn dispatch<W: WireFormat>(
+        &self,
+        call: Kind,
+        params: &[u8],
+        fmt: &W,
+        out: &mut dyn BufMut,
+    ) -> Result<(), RuntimeError> {
+        match call.resolve(CALLS).ok_or(RuntimeError::UnknownCall)? {
+            0 => {
+                let p: PingParams = fmt.decode(params)?;
+                let mut body = Vec::new();
+                fmt.encode(&self.0.ping(p.n), &mut body)?;
+                Envelope::encode_ok(&body, out);
+                Ok(())
+            }
+            _ => Err(RuntimeError::UnknownCall),
+        }
+    }
+}
+
+struct Inc;
+impl Echo for Inc {
+    fn ping(&self, n: u32) -> u32 {
+        n + 1
+    }
+}
+
+#[test]
+fn a_call_that_gets_no_reply_times_out() {
+    // `_provider` end stays bound (so the channel isn't disconnected) but
+    // nothing serves it.
+    let (client_side, _provider) = duplex();
+    let mut client = Client::new(client_side, MsgPack);
+
+    let err = client
+        .call_with_timeout(0, &PingParams { n: 1 }, Duration::from_millis(50))
+        .unwrap_err();
+    assert_eq!(err, RuntimeError::Timeout);
+}
+
+#[test]
+fn a_call_answered_before_the_deadline_succeeds() {
+    let (client_side, provider_side) = duplex();
+
+    let provider = thread::spawn(move || {
+        let mut provider_side = provider_side;
+        Server::new(EchoDispatcher(Inc), MsgPack)
+            .serve(&mut provider_side)
+            .unwrap();
+    });
+
+    let mut client = Client::new(client_side, MsgPack);
+
+    let (reply, fmt) = client
+        .call_with_timeout(0, &PingParams { n: 41 }, Duration::from_secs(5))
+        .unwrap();
+    let n: u32 = match reply {
+        Envelope::Ok(payload) => fmt.decode(payload).unwrap(),
+        Envelope::Err { .. } => panic!("unexpected error frame"),
+    };
+    assert_eq!(n, 42);
+
+    drop(client);
+    provider.join().unwrap();
+}