clawft_kernel/
mesh.rs

1//! Mesh transport abstractions for WeftOS cluster networking (K6.1).
2//!
3//! Defines the transport-agnostic interface for node-to-node communication.
4//! Actual transport backends (QUIC, WebSocket, TCP) implement these traits.
5
6use std::net::SocketAddr;
7
8use async_trait::async_trait;
9use serde::{Deserialize, Serialize};
10
11/// Mesh networking errors.
12#[non_exhaustive]
13#[derive(Debug, thiserror::Error)]
14pub enum MeshError {
15    /// A transport-level failure (connection reset, DNS, etc.).
16    #[error("transport error: {0}")]
17    Transport(String),
18
19    /// Noise or WeftOS handshake failed.
20    #[error("handshake failed: {0}")]
21    Handshake(String),
22
23    /// Attempted to send to a peer that is not connected.
24    #[error("peer not connected: {0}")]
25    PeerNotConnected(String),
26
27    /// Message exceeds the maximum allowed size.
28    #[error("message too large: {size} bytes (max {max})")]
29    MessageTooLarge {
30        /// Actual size of the message.
31        size: usize,
32        /// Configured maximum.
33        max: usize,
34    },
35
36    /// The underlying connection was closed.
37    #[error("connection closed")]
38    ConnectionClosed,
39
40    /// An operation exceeded its deadline.
41    #[error("timeout: {0}")]
42    Timeout(String),
43
44    /// Encryption or decryption failure.
45    #[error("encryption error: {0}")]
46    Encryption(String),
47
48    /// Local and remote governance genesis hashes differ.
49    #[error("genesis mismatch: local vs remote cluster")]
50    GenesisMismatch,
51
52    /// Wraps a lower-level I/O error description.
53    #[error("io error: {0}")]
54    Io(String),
55}
56
57/// Maximum message size (16 MiB per D8).
58pub const MAX_MESSAGE_SIZE: usize = 16 * 1024 * 1024;
59
60/// A bidirectional byte stream over any transport.
61///
62/// # Cancel-safety (ADR-010 / WEFT-18)
63///
64/// `recv` **must** be cancel-safe: it is raced inside `tokio::select!` in
65/// the mesh peer loop (`boot.rs` bidirectional accept path). If a future is
66/// dropped mid-frame, the next `recv` on the same stream must resume cleanly
67/// without length-prefix desync. Implementations that stage multi-step reads
68/// (`read_exact` of length then body) **must** keep partial progress on `self`.
69///
70/// `send` is typically invoked only after a `select!` arm wins (not as a
71/// racing future). Prefer making multi-write sends resilient to task drop, but
72/// cancel-safety of `recv` is the load-bearing contract for mesh correctness.
73#[async_trait]
74pub trait MeshStream: Send + Sync + 'static {
75    /// Send raw bytes over the stream.
76    async fn send(&mut self, data: &[u8]) -> Result<(), MeshError>;
77
78    /// Receive raw bytes from the stream.
79    ///
80    /// Cancel-safe: may be dropped while pending inside `tokio::select!`.
81    async fn recv(&mut self) -> Result<Vec<u8>, MeshError>;
82
83    /// Close the stream gracefully.
84    async fn close(&mut self) -> Result<(), MeshError>;
85
86    /// Get the remote address if available.
87    fn remote_addr(&self) -> Option<SocketAddr>;
88}
89
90/// Listens for incoming mesh connections.
91#[async_trait]
92pub trait TransportListener: Send + Sync + 'static {
93    /// Accept the next incoming connection.
94    async fn accept(&mut self) -> Result<(Box<dyn MeshStream>, SocketAddr), MeshError>;
95
96    /// Get the local address this listener is bound to.
97    fn local_addr(&self) -> Result<SocketAddr, MeshError>;
98}
99
100/// Transport-agnostic mesh transport interface.
101///
102/// Implementations provide concrete connection semantics (QUIC, TCP,
103/// WebSocket, etc.) while the mesh layer stays transport-neutral.
104#[async_trait]
105pub trait MeshTransport: Send + Sync + 'static {
106    /// Human-readable transport name (e.g., "quic", "websocket", "tcp").
107    fn name(&self) -> &str;
108
109    /// Start listening for incoming connections on the given address.
110    async fn listen(&self, addr: &str) -> Result<Box<dyn TransportListener>, MeshError>;
111
112    /// Connect to a remote peer at the given address.
113    async fn connect(&self, addr: &str) -> Result<Box<dyn MeshStream>, MeshError>;
114
115    /// Check if this transport supports the given address scheme.
116    fn supports(&self, addr: &str) -> bool;
117}
118
119/// WeftOS mesh handshake payload (sent inside Noise XX step 3).
120///
121/// Both sides exchange this after the encrypted channel is established
122/// to verify governance compatibility and negotiate capabilities.
123#[derive(Debug, Clone, Serialize, Deserialize)]
124pub struct WeftHandshake {
125    /// Node identity (hex-encoded public key hash).
126    pub node_id: String,
127
128    /// Governance genesis hash for cluster trust root.
129    pub governance_genesis_hash: [u8; 32],
130
131    /// Governance engine version string.
132    pub governance_version: String,
133
134    /// Bitmap of supported capabilities.
135    pub capabilities: u32,
136
137    /// Whether this node supports post-quantum KEM.
138    pub kem_supported: bool,
139
140    /// Chain head sequence number.
141    pub chain_seq: u64,
142
143    /// Supported sync stream types (FrameType discriminants).
144    pub supported_sync_streams: Vec<u8>,
145}
146
147/// Information about a connected mesh peer.
148#[derive(Debug, Clone)]
149pub struct MeshPeer {
150    /// Peer's node identity.
151    pub node_id: String,
152
153    /// Peer's handshake data.
154    pub handshake: WeftHandshake,
155
156    /// Peer's network address.
157    pub address: SocketAddr,
158}
159
160#[cfg(test)]
161mod tests {
162    use super::*;
163
164    #[test]
165    fn mesh_error_display_transport() {
166        let err = MeshError::Transport("reset".into());
167        assert_eq!(err.to_string(), "transport error: reset");
168    }
169
170    #[test]
171    fn mesh_error_display_handshake() {
172        let err = MeshError::Handshake("bad key".into());
173        assert_eq!(err.to_string(), "handshake failed: bad key");
174    }
175
176    #[test]
177    fn mesh_error_display_peer_not_connected() {
178        let err = MeshError::PeerNotConnected("node-42".into());
179        assert_eq!(err.to_string(), "peer not connected: node-42");
180    }
181
182    #[test]
183    fn mesh_error_display_message_too_large() {
184        let err = MeshError::MessageTooLarge {
185            size: 20_000_000,
186            max: MAX_MESSAGE_SIZE,
187        };
188        let s = err.to_string();
189        assert!(s.contains("20000000"));
190        assert!(s.contains(&MAX_MESSAGE_SIZE.to_string()));
191    }
192
193    #[test]
194    fn mesh_error_display_connection_closed() {
195        let err = MeshError::ConnectionClosed;
196        assert_eq!(err.to_string(), "connection closed");
197    }
198
199    #[test]
200    fn mesh_error_display_timeout() {
201        let err = MeshError::Timeout("connect".into());
202        assert_eq!(err.to_string(), "timeout: connect");
203    }
204
205    #[test]
206    fn mesh_error_display_encryption() {
207        let err = MeshError::Encryption("decrypt failed".into());
208        assert_eq!(err.to_string(), "encryption error: decrypt failed");
209    }
210
211    #[test]
212    fn mesh_error_display_genesis_mismatch() {
213        let err = MeshError::GenesisMismatch;
214        assert!(err.to_string().contains("genesis mismatch"));
215    }
216
217    #[test]
218    fn mesh_error_display_io() {
219        let err = MeshError::Io("broken pipe".into());
220        assert_eq!(err.to_string(), "io error: broken pipe");
221    }
222
223    #[test]
224    fn max_message_size_is_16mib() {
225        assert_eq!(MAX_MESSAGE_SIZE, 16 * 1024 * 1024);
226    }
227
228    #[test]
229    fn weft_handshake_serde_roundtrip() {
230        let hs = WeftHandshake {
231            node_id: "abcdef1234567890".into(),
232            governance_genesis_hash: [0xAA; 32],
233            governance_version: "1.0.0".into(),
234            capabilities: 0b1011,
235            kem_supported: true,
236            chain_seq: 42,
237            supported_sync_streams: vec![0x01, 0x03, 0x07],
238        };
239        let json = serde_json::to_string(&hs).unwrap();
240        let restored: WeftHandshake = serde_json::from_str(&json).unwrap();
241        assert_eq!(restored.node_id, hs.node_id);
242        assert_eq!(restored.governance_genesis_hash, [0xAA; 32]);
243        assert_eq!(restored.governance_version, "1.0.0");
244        assert_eq!(restored.capabilities, 0b1011);
245        assert!(restored.kem_supported);
246        assert_eq!(restored.chain_seq, 42);
247        assert_eq!(restored.supported_sync_streams, vec![0x01, 0x03, 0x07]);
248    }
249
250    #[test]
251    fn mesh_peer_fields() {
252        let hs = WeftHandshake {
253            node_id: "peer-1".into(),
254            governance_genesis_hash: [0; 32],
255            governance_version: "0.1.0".into(),
256            capabilities: 0,
257            kem_supported: false,
258            chain_seq: 0,
259            supported_sync_streams: vec![],
260        };
261        let addr: SocketAddr = "127.0.0.1:9489".parse().unwrap();
262        let peer = MeshPeer {
263            node_id: "peer-1".into(),
264            handshake: hs,
265            address: addr,
266        };
267        assert_eq!(peer.node_id, "peer-1");
268        assert_eq!(peer.address.port(), 9489);
269    }
270}
clawft_kernel/mesh.rs

clawft_kernel/
mesh.rs
