use serde::{Deserialize, Serialize};
use crate::ipc::{KernelMessage, MessageTarget};
const MAX_IPC_MESSAGE_SIZE: usize = 16 * 1024 * 1024;
const MAX_HOPS: u8 = 8;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[repr(u8)]
pub enum MeshIpcEncoding {
#[default]
Json = 0x01,
#[cfg(feature = "mesh-rvf")]
Rvf = 0x02,
}
pub const DEFAULT_MESH_IPC_ENCODING: MeshIpcEncoding = MeshIpcEncoding::Json;
impl MeshIpcEncoding {
pub const fn as_str(self) -> &'static str {
match self {
Self::Json => "json",
#[cfg(feature = "mesh-rvf")]
Self::Rvf => "rvf",
}
}
pub fn from_byte(b: u8) -> Option<Self> {
match b {
0x01 => Some(Self::Json),
#[cfg(feature = "mesh-rvf")]
0x02 => Some(Self::Rvf),
_ => None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Scope {
pub user_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub project_id: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MeshIpcEnvelope {
pub source_node: String,
pub dest_node: String,
pub message: KernelMessage,
pub hop_count: u8,
pub envelope_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub dest_scope: Option<Scope>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub src_scope: Option<Scope>,
}
impl MeshIpcEnvelope {
pub fn new(source_node: String, dest_node: String, message: KernelMessage) -> Self {
Self {
source_node,
dest_node,
message,
hop_count: 0,
envelope_id: uuid::Uuid::new_v4().to_string(),
dest_scope: None,
src_scope: None,
}
}
pub fn to_bytes(&self) -> Result<Vec<u8>, MeshIpcError> {
self.to_bytes_with_encoding(DEFAULT_MESH_IPC_ENCODING)
}
pub fn to_bytes_with_encoding(
&self,
encoding: MeshIpcEncoding,
) -> Result<Vec<u8>, MeshIpcError> {
match encoding {
MeshIpcEncoding::Json => {
let bytes = serde_json::to_vec(self)
.map_err(|e| MeshIpcError::Serialization(e.to_string()))?;
if bytes.len() > MAX_IPC_MESSAGE_SIZE {
return Err(MeshIpcError::MessageTooLarge {
size: bytes.len(),
max: MAX_IPC_MESSAGE_SIZE,
});
}
Ok(bytes)
}
#[cfg(feature = "mesh-rvf")]
MeshIpcEncoding::Rvf => Err(MeshIpcError::UnsupportedEncoding {
encoding: encoding.as_str(),
}),
}
}
pub fn from_bytes(data: &[u8]) -> Result<Self, MeshIpcError> {
Self::from_bytes_with_encoding(data, DEFAULT_MESH_IPC_ENCODING)
}
pub fn from_bytes_with_encoding(
data: &[u8],
encoding: MeshIpcEncoding,
) -> Result<Self, MeshIpcError> {
if data.len() > MAX_IPC_MESSAGE_SIZE {
return Err(MeshIpcError::MessageTooLarge {
size: data.len(),
max: MAX_IPC_MESSAGE_SIZE,
});
}
match encoding {
MeshIpcEncoding::Json => {
let envelope: Self = serde_json::from_slice(data)
.map_err(|e| MeshIpcError::Deserialization(e.to_string()))?;
Self::validate_boundary(&envelope)?;
Ok(envelope)
}
#[cfg(feature = "mesh-rvf")]
MeshIpcEncoding::Rvf => Err(MeshIpcError::UnsupportedEncoding {
encoding: encoding.as_str(),
}),
}
}
fn validate_boundary(envelope: &Self) -> Result<(), MeshIpcError> {
if envelope.envelope_id.is_empty() {
return Err(MeshIpcError::Deserialization(
"envelope_id must not be empty".into(),
));
}
if envelope.source_node.is_empty() {
return Err(MeshIpcError::Deserialization(
"source_node must not be empty".into(),
));
}
Ok(())
}
pub fn increment_hop(&mut self) -> Result<(), MeshIpcError> {
self.hop_count += 1;
if self.hop_count > MAX_HOPS {
return Err(MeshIpcError::MaxHopsExceeded {
hops: self.hop_count,
});
}
Ok(())
}
pub fn inner_target(&self) -> &MessageTarget {
match &self.message.target {
MessageTarget::RemoteNode { target, .. } => target.as_ref(),
other => other,
}
}
}
#[non_exhaustive]
#[derive(Debug, thiserror::Error)]
pub enum MeshIpcError {
#[error("serialization error: {0}")]
Serialization(String),
#[error("deserialization error: {0}")]
Deserialization(String),
#[error("message too large: {size} bytes (max {max})")]
MessageTooLarge { size: usize, max: usize },
#[error("max hops exceeded: {hops}")]
MaxHopsExceeded { hops: u8 },
#[error("duplicate message: {envelope_id}")]
DuplicateMessage { envelope_id: String },
#[error("unsupported mesh IPC encoding: {encoding} (ADR-031 / WEFT-683 — deferred)")]
UnsupportedEncoding { encoding: &'static str },
}
#[derive(Debug)]
pub struct MeshRequest {
pub request: MeshIpcEnvelope,
pub correlation_id: String,
pub sent_at: std::time::Instant,
pub timeout: std::time::Duration,
}
impl MeshRequest {
pub fn new(mut envelope: MeshIpcEnvelope, timeout: std::time::Duration) -> Self {
let correlation_id = uuid::Uuid::new_v4().to_string();
envelope.message.correlation_id = Some(correlation_id.clone());
Self {
request: envelope,
correlation_id,
sent_at: std::time::Instant::now(),
timeout,
}
}
pub fn is_timed_out(&self) -> bool {
self.sent_at.elapsed() >= self.timeout
}
pub fn matches_response(&self, response: &MeshIpcEnvelope) -> bool {
response.message.correlation_id.as_deref() == Some(&self.correlation_id)
}
}
pub struct PendingRequests {
requests: std::collections::HashMap<String, MeshRequest>,
#[cfg(feature = "exochain")]
chain_manager: Option<std::sync::Arc<crate::chain::ChainManager>>,
}
impl PendingRequests {
pub fn new() -> Self {
Self {
requests: std::collections::HashMap::new(),
#[cfg(feature = "exochain")]
chain_manager: None,
}
}
#[cfg(feature = "exochain")]
pub fn set_chain_manager(&mut self, cm: std::sync::Arc<crate::chain::ChainManager>) {
self.chain_manager = Some(cm);
}
pub fn register(&mut self, request: MeshRequest) {
#[cfg(feature = "exochain")]
if let Some(ref cm) = self.chain_manager {
cm.append(
"mesh_ipc",
crate::chain::EVENT_KIND_MESH_IPC_SEND,
Some(serde_json::json!({
"correlation_id": &request.correlation_id,
"source_node": &request.request.source_node,
"dest_node": &request.request.dest_node,
"envelope_id": &request.request.envelope_id,
})),
);
}
self.requests
.insert(request.correlation_id.clone(), request);
}
pub fn try_complete(&mut self, response: &MeshIpcEnvelope) -> Option<MeshRequest> {
if let Some(corr_id) = &response.message.correlation_id {
self.requests.remove(corr_id)
} else {
None
}
}
pub fn evict_timed_out(&mut self) -> Vec<String> {
let timed_out: Vec<String> = self
.requests
.iter()
.filter(|(_, r)| r.is_timed_out())
.map(|(id, _)| id.clone())
.collect();
for id in &timed_out {
self.requests.remove(id);
}
timed_out
}
pub fn len(&self) -> usize {
self.requests.len()
}
pub fn is_empty(&self) -> bool {
self.requests.is_empty()
}
}
impl Default for PendingRequests {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ipc::{KernelMessage, MessagePayload, MessageTarget};
fn make_test_message() -> KernelMessage {
KernelMessage::text(1, MessageTarget::Process(2), "hello mesh")
}
#[test]
fn envelope_new_sets_defaults() {
let msg = make_test_message();
let env = MeshIpcEnvelope::new("node-a".into(), "node-b".into(), msg);
assert_eq!(env.source_node, "node-a");
assert_eq!(env.dest_node, "node-b");
assert_eq!(env.hop_count, 0);
assert!(!env.envelope_id.is_empty());
}
#[test]
fn to_bytes_from_bytes_roundtrip() {
let msg = make_test_message();
let env = MeshIpcEnvelope::new("src".into(), "dst".into(), msg);
let bytes = env.to_bytes().unwrap();
let restored = MeshIpcEnvelope::from_bytes(&bytes).unwrap();
assert_eq!(restored.source_node, "src");
assert_eq!(restored.dest_node, "dst");
assert_eq!(restored.envelope_id, env.envelope_id);
assert_eq!(restored.message.id, env.message.id);
}
#[test]
fn from_bytes_rejects_oversized() {
let big = vec![0u8; MAX_IPC_MESSAGE_SIZE + 1];
let err = MeshIpcEnvelope::from_bytes(&big).unwrap_err();
assert!(matches!(err, MeshIpcError::MessageTooLarge { .. }));
}
#[test]
fn from_bytes_rejects_invalid_json() {
let err = MeshIpcEnvelope::from_bytes(b"not json").unwrap_err();
assert!(matches!(err, MeshIpcError::Deserialization(_)));
}
#[test]
fn increment_hop_within_limit() {
let msg = make_test_message();
let mut env = MeshIpcEnvelope::new("a".into(), "b".into(), msg);
for _ in 0..MAX_HOPS {
env.increment_hop().unwrap();
}
assert_eq!(env.hop_count, MAX_HOPS);
}
#[test]
fn increment_hop_exceeds_max() {
let msg = make_test_message();
let mut env = MeshIpcEnvelope::new("a".into(), "b".into(), msg);
env.hop_count = MAX_HOPS;
let err = env.increment_hop().unwrap_err();
assert!(matches!(err, MeshIpcError::MaxHopsExceeded { .. }));
}
#[test]
fn inner_target_unwraps_remote_node() {
let inner = MessageTarget::Process(7);
let remote_target = MessageTarget::RemoteNode {
node_id: "node-42".into(),
target: Box::new(inner),
};
let msg = KernelMessage::new(1, remote_target, MessagePayload::Text("hi".into()));
let env = MeshIpcEnvelope::new("a".into(), "node-42".into(), msg);
assert!(matches!(env.inner_target(), MessageTarget::Process(7)));
}
#[test]
fn inner_target_returns_non_remote_as_is() {
let msg = KernelMessage::text(1, MessageTarget::Broadcast, "hi");
let env = MeshIpcEnvelope::new("a".into(), "b".into(), msg);
assert!(matches!(env.inner_target(), MessageTarget::Broadcast));
}
#[test]
fn from_bytes_rejects_empty_source_node() {
let msg = make_test_message();
let mut env = MeshIpcEnvelope::new("".into(), "b".into(), msg);
env.source_node = "".into();
let bytes = serde_json::to_vec(&env).unwrap();
let err = MeshIpcEnvelope::from_bytes(&bytes).unwrap_err();
assert!(matches!(err, MeshIpcError::Deserialization(_)));
}
#[test]
fn from_bytes_rejects_empty_envelope_id() {
let msg = make_test_message();
let mut env = MeshIpcEnvelope::new("a".into(), "b".into(), msg);
env.envelope_id = "".into();
let bytes = serde_json::to_vec(&env).unwrap();
let err = MeshIpcEnvelope::from_bytes(&bytes).unwrap_err();
assert!(matches!(err, MeshIpcError::Deserialization(_)));
}
#[test]
fn k6_cross_node_ipc_via_did_addressing() {
use crate::ipc::GlobalPid;
let node_a_did = "a1b2c3d4e5f6a7b8a1b2c3d4e5f6a7b8";
let node_b_did = "b2c3d4e5f6a7b8a1b2c3d4e5f6a7b8a1";
let target_pid = GlobalPid::local(42, node_b_did);
assert_eq!(target_pid.node_id, node_b_did);
assert!(!target_pid.is_local(node_a_did));
assert!(target_pid.is_local(node_b_did));
let msg = KernelMessage::text(
1,
MessageTarget::RemoteNode {
node_id: node_b_did.to_string(),
target: Box::new(MessageTarget::Process(42)),
},
"hello via DID",
);
let envelope = MeshIpcEnvelope::new(node_a_did.to_string(), node_b_did.to_string(), msg);
let bytes = envelope.to_bytes().unwrap();
let restored = MeshIpcEnvelope::from_bytes(&bytes).unwrap();
assert_eq!(restored.dest_node, node_b_did);
assert_eq!(restored.source_node, node_a_did);
assert!(matches!(
restored.inner_target(),
MessageTarget::Process(42)
));
}
#[test]
fn mesh_request_correlation() {
let msg = make_test_message();
let env = MeshIpcEnvelope::new("a".into(), "b".into(), msg);
let req = MeshRequest::new(env, std::time::Duration::from_secs(5));
assert!(!req.correlation_id.is_empty());
assert_eq!(
req.request.message.correlation_id.as_deref(),
Some(req.correlation_id.as_str())
);
}
#[test]
fn mesh_request_timeout() {
let msg = make_test_message();
let env = MeshIpcEnvelope::new("a".into(), "b".into(), msg);
let req = MeshRequest::new(env, std::time::Duration::from_secs(0));
assert!(req.is_timed_out());
}
#[test]
fn pending_requests_complete() {
let msg = make_test_message();
let env = MeshIpcEnvelope::new("a".into(), "b".into(), msg);
let req = MeshRequest::new(env, std::time::Duration::from_secs(30));
let corr_id = req.correlation_id.clone();
let mut pending = PendingRequests::new();
pending.register(req);
assert_eq!(pending.len(), 1);
let resp_msg = KernelMessage::with_correlation(
2,
MessageTarget::Process(1),
MessagePayload::Text("ok".into()),
corr_id.clone(),
);
let resp_env = MeshIpcEnvelope::new("b".into(), "a".into(), resp_msg);
let completed = pending.try_complete(&resp_env);
assert!(completed.is_some());
assert_eq!(completed.unwrap().correlation_id, corr_id);
assert!(pending.is_empty());
}
#[test]
fn pending_requests_evict_timed_out() {
let msg = make_test_message();
let env = MeshIpcEnvelope::new("a".into(), "b".into(), msg);
let req = MeshRequest::new(env, std::time::Duration::from_secs(0));
let corr_id = req.correlation_id.clone();
let mut pending = PendingRequests::new();
pending.register(req);
assert_eq!(pending.len(), 1);
let evicted = pending.evict_timed_out();
assert_eq!(evicted.len(), 1);
assert_eq!(evicted[0], corr_id);
assert!(pending.is_empty());
}
#[test]
fn envelope_serde_with_correlation_id() {
let msg = KernelMessage::with_correlation(
1,
MessageTarget::Process(2),
MessagePayload::Text("req".into()),
"corr-99".into(),
);
let env = MeshIpcEnvelope::new("n1".into(), "n2".into(), msg);
let bytes = env.to_bytes().unwrap();
let restored = MeshIpcEnvelope::from_bytes(&bytes).unwrap();
assert_eq!(restored.message.correlation_id, Some("corr-99".into()));
}
#[test]
fn default_encoding_is_json() {
assert_eq!(DEFAULT_MESH_IPC_ENCODING, MeshIpcEncoding::Json);
assert_eq!(MeshIpcEncoding::default(), MeshIpcEncoding::Json);
assert_eq!(MeshIpcEncoding::Json.as_str(), "json");
assert_eq!(
MeshIpcEncoding::from_byte(0x01),
Some(MeshIpcEncoding::Json)
);
#[cfg(not(feature = "mesh-rvf"))]
assert_eq!(MeshIpcEncoding::from_byte(0x02), None);
#[cfg(feature = "mesh-rvf")]
assert_eq!(MeshIpcEncoding::from_byte(0x02), Some(MeshIpcEncoding::Rvf));
}
#[test]
fn to_bytes_with_encoding_json_matches_to_bytes() {
let msg = make_test_message();
let env = MeshIpcEnvelope::new("src".into(), "dst".into(), msg);
let a = env.to_bytes().unwrap();
let b = env.to_bytes_with_encoding(MeshIpcEncoding::Json).unwrap();
assert_eq!(a, b);
let restored =
MeshIpcEnvelope::from_bytes_with_encoding(&a, MeshIpcEncoding::Json).unwrap();
assert_eq!(restored.envelope_id, env.envelope_id);
}
#[cfg(feature = "mesh-rvf")]
#[test]
fn rvf_encoding_returns_unsupported() {
let msg = make_test_message();
let env = MeshIpcEnvelope::new("src".into(), "dst".into(), msg);
let err = env
.to_bytes_with_encoding(MeshIpcEncoding::Rvf)
.unwrap_err();
assert!(matches!(
err,
MeshIpcError::UnsupportedEncoding { encoding: "rvf" }
));
let err =
MeshIpcEnvelope::from_bytes_with_encoding(b"{}", MeshIpcEncoding::Rvf).unwrap_err();
assert!(matches!(
err,
MeshIpcError::UnsupportedEncoding { encoding: "rvf" }
));
}
const FROZEN_PRE_SCOPE_ENVELOPE: &str = concat!(
r#"{"source_node":"a","dest_node":"b","message":{"id":"m1","from":1,"#,
r#""target":{"Topic":"t"},"payload":{"Text":"x"},"#,
r#""timestamp":"2026-01-02T03:04:05Z"},"hop_count":0,"envelope_id":"fixed-id"}"#
);
#[test]
fn envelope_without_scopes_serializes_byte_identical_to_old_format() {
let env = MeshIpcEnvelope::from_bytes(FROZEN_PRE_SCOPE_ENVELOPE.as_bytes()).unwrap();
assert!(env.dest_scope.is_none() && env.src_scope.is_none());
assert_eq!(serde_json::to_string(&env).unwrap(), FROZEN_PRE_SCOPE_ENVELOPE);
assert!(!FROZEN_PRE_SCOPE_ENVELOPE.contains("scope"));
}
#[test]
fn old_json_without_scopes_deserializes() {
let msg = KernelMessage::text(1, MessageTarget::Topic("t".into()), "x");
let env = MeshIpcEnvelope::new("a".into(), "b".into(), msg);
let bytes = env.to_bytes().unwrap();
let back = MeshIpcEnvelope::from_bytes(&bytes).unwrap();
assert!(back.dest_scope.is_none() && back.src_scope.is_none());
}
#[test]
fn scoped_envelope_round_trips() {
let msg = KernelMessage::text(1, MessageTarget::Topic("t".into()), "x");
let mut env = MeshIpcEnvelope::new("a".into(), "b".into(), msg);
env.dest_scope = Some(Scope {
user_id: "u".repeat(32),
project_id: Some("P".repeat(26)),
});
env.src_scope = Some(Scope {
user_id: "v".repeat(32),
project_id: None,
});
let back = MeshIpcEnvelope::from_bytes(&env.to_bytes().unwrap()).unwrap();
assert_eq!(back.dest_scope, env.dest_scope);
assert_eq!(back.src_scope, env.src_scope);
}
}