From bc040527e54e687bf1448e5d0f687eb48b3549a4 Mon Sep 17 00:00:00 2001 From: Kinflou Date: Wed, 2 Sep 2026 02:57:28 +0800 Subject: [PATCH] feat: connection handshake -- Client::connect / Server::serve_handshaked Before any calls, each end declares { ir_hash, wire_format, framing, capabilities } and refuses on a mismatch -- catches "one end msgpack, one end JSON" and two ends built from different schema versions. - contract::Handshake -- lifetime-free, Copy. wire_format / framing are carried as name_hash (FNV-1a) of a *name*, not a numeric id: a user's add-on WireFormat picks a namespaced name and its hash won't collide with a built-in, no central registry. Fixed 31-byte frame ([magic][ver][ir_hash u64][wire_format u64][framing u64][caps u32]). Handshake::new(ir_hash, wire_format_name, framing_name, caps) hashes. - WireFormat::name() -> &'static str (was going to be id() -> u16; dropped). MsgPack -> "msgpack". contract::FRAMING_DATAGRAM. - RuntimeError::Handshake (non_exhaustive, so additive). - Client::connect(transport, format, local) -> Result and Server::serve_handshaked(transport, local): send ours, read + check the peer's, then proceed. Client::new / Server::serve are unchanged and skip it entirely -- "misaligned mode", documented, for legacy peers. - BufMut::put_u32_le. tests/handshake_roundtrip.rs: matching handshakes connect + a call goes through; a schema-hash mismatch is refused before any call (both ends); misaligned mode skips it. Plus handshake.rs unit tests. All feature configs green. Generated code supplies IR_HASH and calls Client::connect -- a follow-up codegen PR. --- runtime/core/src/client.rs | 27 +++- runtime/core/src/contract/buf.rs | 3 + runtime/core/src/contract/error.rs | 5 + runtime/core/src/contract/handshake.rs | 154 ++++++++++++++++++++++ runtime/core/src/contract/mod.rs | 2 + runtime/core/src/contract/wire.rs | 7 + runtime/core/src/format/msgpack.rs | 4 + runtime/core/src/serve.rs | 25 +++- runtime/core/tests/handshake_roundtrip.rs | 134 +++++++++++++++++++ 9 files changed, 358 insertions(+), 3 deletions(-) create mode 100644 runtime/core/src/contract/handshake.rs create mode 100644 runtime/core/tests/handshake_roundtrip.rs diff --git a/runtime/core/src/client.rs b/runtime/core/src/client.rs index c8c6b58..2f6ab8e 100644 --- a/runtime/core/src/client.rs +++ b/runtime/core/src/client.rs @@ -19,7 +19,7 @@ use core::time::Duration; use serde::Serialize; -use crate::contract::{Envelope, RuntimeError, WireFormat}; +use crate::contract::{Envelope, Handshake, RuntimeError, WireFormat}; use crate::transport::Transport; use crate::wire; @@ -33,6 +33,11 @@ pub struct Client { } impl Client { + /// Bind a client to a transport **without a handshake** — "misaligned + /// mode". Use when the peer can't take part (a legacy server); a + /// wire-format or schema mismatch then surfaces later as a decode / framing + /// error instead of up front. [`connect`](Self::connect) is the checked + /// path. pub fn new(transport: T, format: W) -> Self { Self { transport, @@ -43,6 +48,26 @@ impl Client { } } + /// Bind a client and run the connection [`Handshake`]: send `local`, read + /// the peer's, and refuse (`RuntimeError::Handshake`) if they disagree on + /// schema hash, wire format, or framing. + pub fn connect(transport: T, format: W, local: Handshake) -> Result { + let mut client = Self::new(transport, format); + client.exchange_handshake(local)?; + Ok(client) + } + + fn exchange_handshake(&mut self, local: Handshake) -> Result<(), RuntimeError> { + self.request.clear(); + local.encode(&mut self.request); + self.transport.send(&self.request)?; + + self.response.clear(); + self.transport.recv(&mut self.response)?; + let peer = Handshake::decode(&self.response).ok_or(RuntimeError::Handshake)?; + local.check(&peer) + } + /// 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 diff --git a/runtime/core/src/contract/buf.rs b/runtime/core/src/contract/buf.rs index 9d35a92..9df4a36 100644 --- a/runtime/core/src/contract/buf.rs +++ b/runtime/core/src/contract/buf.rs @@ -13,6 +13,9 @@ pub trait BufMut { fn put_u16_le(&mut self, value: u16) { self.put_slice(&value.to_le_bytes()); } + fn put_u32_le(&mut self, value: u32) { + self.put_slice(&value.to_le_bytes()); + } fn put_u64_le(&mut self, value: u64) { self.put_slice(&value.to_le_bytes()); } diff --git a/runtime/core/src/contract/error.rs b/runtime/core/src/contract/error.rs index 397a6d2..38ea0a3 100644 --- a/runtime/core/src/contract/error.rs +++ b/runtime/core/src/contract/error.rs @@ -26,6 +26,10 @@ pub enum RuntimeError { /// knows — a newer schema. `id` is the schema-global error ordinal; the /// still-encoded body is not retained. Remote { id: u16 }, + /// The connection handshake disagreed on schema hash, wire format, or + /// framing — or none arrived. Never raised when the handshake is skipped + /// (`Client::new` / `Server::serve` — "misaligned mode"). + Handshake, } impl fmt::Display for RuntimeError { @@ -37,6 +41,7 @@ impl fmt::Display for RuntimeError { RuntimeError::Timeout => f.write_str("call timed out"), RuntimeError::UnknownCall => f.write_str("unknown call"), RuntimeError::Remote { id } => write!(f, "unrecognised remote error (ordinal {id})"), + RuntimeError::Handshake => f.write_str("connection handshake mismatch"), } } } diff --git a/runtime/core/src/contract/handshake.rs b/runtime/core/src/contract/handshake.rs new file mode 100644 index 0000000..d4d07e3 --- /dev/null +++ b/runtime/core/src/contract/handshake.rs @@ -0,0 +1,154 @@ +use crate::contract::{BufMut, RuntimeError}; + +const MAGIC: [u8; 2] = *b"CO"; +const VERSION: u8 = 1; +/// `[MAGIC:2][VERSION:1][ir_hash:u64 LE][wire_format:u64 LE][framing:u64 LE][capabilities:u32 LE]` +const LEN: usize = 2 + 1 + 8 + 8 + 8 + 4; + +/// Name of the Comline datagram framing (`wire::encode_request` / +/// `encode_response`). Pass it to [`Handshake::new`]. +pub const FRAMING_DATAGRAM: &str = "comline.datagram"; + +/// 64-bit FNV-1a. Folds a wire-format / framing **name** into the fixed-size +/// handshake without a central id registry — a user's add-on format picks a +/// namespaced name (`"com.acme.myformat"`) and its hash won't collide with a +/// built-in. Stable across platforms. +pub fn name_hash(name: &str) -> u64 { + let mut h: u64 = 0xcbf2_9ce4_8422_2325; + for b in name.as_bytes() { + h ^= u64::from(*b); + h = h.wrapping_mul(0x0000_0100_0000_01b3); + } + h +} + +/// What each end declares when a connection opens: the schema fingerprint the +/// generated code was built from, the serialization + framing it speaks, and +/// its transport capability bits. +/// +/// [`Client::connect`](crate::client::Client::connect) / +/// [`Server::serve_handshaked`](crate::serve::Server::serve_handshaked) exchange +/// these and refuse on a mismatch — catching "one end MessagePack, one end +/// JSON" or two ends built from different schema versions. `Client::new` / +/// `Server::serve` skip it ("misaligned mode" — for legacy peers that can't +/// take part). +/// +/// Lifetime-free and `Copy`: the format / framing are carried as +/// [`name_hash`]es of their names, not the strings. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Handshake { + /// Fingerprint of the frozen IR the two ends generated from. A generator + /// emits this as a constant. + pub ir_hash: u64, + /// [`name_hash`] of the [`WireFormat::name`](crate::contract::WireFormat::name). + pub wire_format: u64, + /// [`name_hash`] of the framing name ([`FRAMING_DATAGRAM`], …). + pub framing: u64, + /// Transport capability bits (reliable / ordered / duplex / …). Advisory: + /// a difference here is not a mismatch. + pub capabilities: u32, +} + +impl Handshake { + /// Build one from the format / framing **names** (hashed in). + pub fn new(ir_hash: u64, wire_format: &str, framing: &str, capabilities: u32) -> Self { + Self { + ir_hash, + wire_format: name_hash(wire_format), + framing: name_hash(framing), + capabilities, + } + } + + /// Write the fixed-size handshake frame. + pub fn encode(&self, out: &mut dyn BufMut) { + out.put_slice(&MAGIC); + out.put_u8(VERSION); + out.put_u64_le(self.ir_hash); + out.put_u64_le(self.wire_format); + out.put_u64_le(self.framing); + out.put_u32_le(self.capabilities); + } + + /// Parse one. `None` if it is truncated, or the magic / version is wrong + /// (a peer that never sent a handshake, or a different protocol). + pub fn decode(frame: &[u8]) -> Option { + let frame: [u8; LEN] = frame.get(..LEN)?.try_into().ok()?; + if frame[0] != MAGIC[0] || frame[1] != MAGIC[1] || frame[2] != VERSION { + return None; + } + Some(Self { + ir_hash: u64::from_le_bytes(frame[3..11].try_into().unwrap()), + wire_format: u64::from_le_bytes(frame[11..19].try_into().unwrap()), + framing: u64::from_le_bytes(frame[19..27].try_into().unwrap()), + capabilities: u32::from_le_bytes(frame[27..31].try_into().unwrap()), + }) + } + + /// Check a peer's declaration against ours. `Err(RuntimeError::Handshake)` + /// if `ir_hash`, `wire_format`, or `framing` disagree; capability bits are + /// allowed to differ. + pub fn check(&self, peer: &Handshake) -> Result<(), RuntimeError> { + let agree = self.ir_hash == peer.ir_hash + && self.wire_format == peer.wire_format + && self.framing == peer.framing; + if agree { + Ok(()) + } else { + Err(RuntimeError::Handshake) + } + } +} + +#[cfg(all(test, feature = "alloc"))] +mod tests { + use super::*; + use alloc::vec::Vec; + + fn sample() -> Handshake { + Handshake::new(0xdead_beef_0102_0304, "msgpack", FRAMING_DATAGRAM, 0b101) + } + + #[test] + fn round_trips() { + let mut frame = Vec::new(); + sample().encode(&mut frame); + assert_eq!(frame.len(), LEN); + assert_eq!(Handshake::decode(&frame), Some(sample())); + } + + #[test] + fn name_hash_is_deterministic_and_name_specific() { + assert_eq!(name_hash("msgpack"), name_hash("msgpack")); + assert_ne!(name_hash("msgpack"), name_hash("json")); + assert_ne!(name_hash("msgpack"), name_hash("com.acme.msgpack")); + } + + #[test] + fn rejects_truncated_or_foreign() { + assert_eq!(Handshake::decode(&[]), None); + assert_eq!(Handshake::decode(&[0u8; LEN]), None); // bad magic + let mut frame = Vec::new(); + sample().encode(&mut frame); + frame.truncate(LEN - 1); + assert_eq!(Handshake::decode(&frame), None); + } + + #[test] + fn check_agrees_and_disagrees() { + let ours = sample(); + assert!(ours.check(&sample()).is_ok()); + + let mut caps_differ = sample(); + caps_differ.capabilities = 0; + assert!(ours.check(&caps_differ).is_ok(), "capability bits may differ"); + + let fmt_differ = + Handshake::new(ours.ir_hash, "json", FRAMING_DATAGRAM, ours.capabilities); + assert_eq!(ours.check(&fmt_differ), Err(RuntimeError::Handshake)); + + let mut schema_differ = sample(); + schema_differ.ir_hash = 0; + assert_eq!(ours.check(&schema_differ), Err(RuntimeError::Handshake)); + } +} diff --git a/runtime/core/src/contract/mod.rs b/runtime/core/src/contract/mod.rs index 9c6e48b..5192eb5 100644 --- a/runtime/core/src/contract/mod.rs +++ b/runtime/core/src/contract/mod.rs @@ -11,6 +11,7 @@ mod call; mod dispatch; mod envelope; mod error; +mod handshake; mod wire; pub use buf::{BufMut, SliceBuf}; @@ -18,4 +19,5 @@ pub use call::CallError; pub use dispatch::{Dispatch, Kind}; pub use envelope::Envelope; pub use error::RuntimeError; +pub use handshake::{name_hash, Handshake, FRAMING_DATAGRAM}; pub use wire::WireFormat; diff --git a/runtime/core/src/contract/wire.rs b/runtime/core/src/contract/wire.rs index 3b9885e..a792f5b 100644 --- a/runtime/core/src/contract/wire.rs +++ b/runtime/core/src/contract/wire.rs @@ -9,6 +9,13 @@ use crate::contract::{BufMut, RuntimeError}; /// `decode` borrows from the input, so a generated `Msg<'de>` points straight /// into the receive buffer. pub trait WireFormat { + /// A stable name for this format — the connection + /// [`Handshake`](crate::contract::Handshake) folds it in (hashed) so the + /// two ends can catch "one speaks MessagePack, one speaks JSON". Built-ins + /// are bare (`"msgpack"`); a third-party format should namespace it + /// (`"com.acme.myformat"`) to avoid a clash. + fn name(&self) -> &'static str; + fn encode( &self, value: &T, diff --git a/runtime/core/src/format/msgpack.rs b/runtime/core/src/format/msgpack.rs index 9c747c1..7f612f3 100644 --- a/runtime/core/src/format/msgpack.rs +++ b/runtime/core/src/format/msgpack.rs @@ -26,6 +26,10 @@ impl Write for BufWriter<'_> { } impl WireFormat for MsgPack { + fn name(&self) -> &'static str { + "msgpack" + } + fn encode( &self, value: &T, diff --git a/runtime/core/src/serve.rs b/runtime/core/src/serve.rs index f4a03d6..0d2e1e1 100644 --- a/runtime/core/src/serve.rs +++ b/runtime/core/src/serve.rs @@ -3,7 +3,7 @@ use alloc::vec::Vec; -use crate::contract::{Dispatch, Kind, RuntimeError, WireFormat}; +use crate::contract::{Dispatch, Handshake, Kind, RuntimeError, WireFormat}; use crate::transport::Transport; use crate::wire; @@ -56,9 +56,30 @@ impl Server { Ok(true) } - /// Serve calls until the transport closes. + /// Serve calls until the transport closes. **No handshake** — pair with a + /// `Client::new` peer, or an aligned setup where a mismatch can't happen. pub fn serve(&mut self, transport: &mut T) -> Result<(), RuntimeError> { while self.serve_one(transport)? {} Ok(()) } + + /// Run the connection [`Handshake`] against the connecting peer — send + /// `local`, read theirs, refuse (`RuntimeError::Handshake`) on a schema / + /// wire-format / framing mismatch — then [`serve`](Self::serve). + pub fn serve_handshaked( + &mut self, + transport: &mut T, + local: Handshake, + ) -> Result<(), RuntimeError> { + self.response.clear(); + local.encode(&mut self.response); + transport.send(&self.response)?; + + self.recv.clear(); + transport.recv(&mut self.recv)?; + let peer = Handshake::decode(&self.recv).ok_or(RuntimeError::Handshake)?; + local.check(&peer)?; + + self.serve(transport) + } } diff --git a/runtime/core/tests/handshake_roundtrip.rs b/runtime/core/tests/handshake_roundtrip.rs new file mode 100644 index 0000000..0fc5df6 --- /dev/null +++ b/runtime/core/tests/handshake_roundtrip.rs @@ -0,0 +1,134 @@ +//! The connection handshake: `Client::connect` ⇆ `Server::serve_handshaked` +//! agree and a call goes through; a schema-hash mismatch is refused before any +//! call; `Client::new` / `Server::serve` skip it ("misaligned mode"). +#![cfg(feature = "std")] + +use std::thread; + +use comline_runtime::client::Client; +use comline_runtime::contract::{ + BufMut, Dispatch, Envelope, Handshake, Kind, RuntimeError, WireFormat, FRAMING_DATAGRAM, +}; +use comline_runtime::format::MsgPack; +use comline_runtime::serve::Server; +use comline_runtime::transport::duplex; +use serde::{Deserialize, Serialize}; + +// protocol Echo { function bump(n: u32) -> u32; } + +#[derive(Serialize, Deserialize)] +struct BumpParams { + n: u32, +} + +const CALLS: &[&str] = &["bump"]; + +trait Echo { + fn bump(&self, n: u32) -> u32; +} + +struct EchoDispatcher(T); + +impl Dispatch for EchoDispatcher { + 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: BumpParams = fmt.decode(params)?; + let mut body = Vec::new(); + fmt.encode(&self.0.bump(p.n), &mut body)?; + Envelope::encode_ok(&body, out); + Ok(()) + } + _ => Err(RuntimeError::UnknownCall), + } + } +} + +struct Inc; +impl Echo for Inc { + fn bump(&self, n: u32) -> u32 { + n + 1 + } +} + +fn hs(ir_hash: u64) -> Handshake { + Handshake::new(ir_hash, MsgPack.name(), FRAMING_DATAGRAM, 0) +} + +const SCHEMA: u64 = 0x0011_2233_4455_6677; + +#[test] +fn matching_handshakes_connect_and_a_call_round_trips() { + let (client_side, provider_side) = duplex(); + + let provider = thread::spawn(move || { + let mut provider_side = provider_side; + Server::new(EchoDispatcher(Inc), MsgPack) + .serve_handshaked(&mut provider_side, hs(SCHEMA)) + .unwrap(); + }); + + let mut client = Client::connect(client_side, MsgPack, hs(SCHEMA)).expect("handshake"); + + let (reply, fmt) = client.call(0, &BumpParams { n: 41 }).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(); +} + +#[test] +fn a_schema_hash_mismatch_is_refused_before_any_call() { + let (client_side, provider_side) = duplex(); + + // The provider speaks a different schema version. + let provider = thread::spawn(move || { + let mut provider_side = provider_side; + Server::new(EchoDispatcher(Inc), MsgPack) + .serve_handshaked(&mut provider_side, hs(0xdead)) + .unwrap_err() + }); + + let err = match Client::connect(client_side, MsgPack, hs(SCHEMA)) { + Err(e) => e, + Ok(_) => panic!("connect should have refused the mismatched schema"), + }; + assert_eq!(err, RuntimeError::Handshake); + + assert_eq!(provider.join().unwrap(), RuntimeError::Handshake); +} + +#[test] +fn misaligned_mode_skips_the_handshake_entirely() { + // `Client::new` + `Server::serve` — no handshake frame at all, the call + // is the first thing on the wire. + 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(0, &BumpParams { n: 7 }).unwrap(); + let n: u32 = match reply { + Envelope::Ok(payload) => fmt.decode(payload).unwrap(), + Envelope::Err { .. } => panic!("unexpected error frame"), + }; + assert_eq!(n, 8); + + drop(client); + provider.join().unwrap(); +}