From ce2b3fbfb8e721c21359a4d652dfb5b43e727ccd Mon Sep 17 00:00:00 2001 From: Mike Young Date: Mon, 4 May 2026 22:14:23 +0100 Subject: [PATCH 1/3] Add builds in core to gitignore --- .gitignore | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index c41cc9e..a03c5d3 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ -/target \ No newline at end of file +/target +/core/* \ No newline at end of file From 8ab4cf8e6d194f361e9de3f3d28dd3e2e2966a86 Mon Sep 17 00:00:00 2001 From: Mike Young Date: Mon, 4 May 2026 22:29:50 +0100 Subject: [PATCH 2/3] Update Packet variant to reflect change of `Message` to `WireMessage` --- crates/net/src/packet.rs | 2 +- crates/net/src/transport/error.rs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/net/src/packet.rs b/crates/net/src/packet.rs index d650f38..a56ecd4 100644 --- a/crates/net/src/packet.rs +++ b/crates/net/src/packet.rs @@ -31,5 +31,5 @@ pub type SessionId = Id; pub enum Packet { Handshake(Handshake), HandshakeAck(HandshakeAck), - Message(WireMessage), + WireMessage(WireMessage), } \ No newline at end of file diff --git a/crates/net/src/transport/error.rs b/crates/net/src/transport/error.rs index 613c796..db6e1d0 100644 --- a/crates/net/src/transport/error.rs +++ b/crates/net/src/transport/error.rs @@ -2,7 +2,7 @@ use crypto::{error::CryptoError, id::Id}; use thiserror::Error; use tokio::{net::tcp::OwnedReadHalf, sync::mpsc::error::SendError, task::JoinError}; -use crate::{error::NetError, packet::Message}; +use crate::{error::NetError, packet::WireMessage}; // use crate::{net::SessionManagerDispatcherError, utils::ChannelError}; @@ -56,7 +56,7 @@ pub enum TransportError { Serialization(#[from] postcard::Error), #[error(transparent)] - Reading(#[from] SendError::<(Id,Message)>), + Reading(#[from] SendError::<(Id,WireMessage)>), #[error(transparent)] IO(#[from] std::io::Error), From 33ea946320bc4fa471eac0895b824df669cf2b6b Mon Sep 17 00:00:00 2001 From: Mike Young Date: Fri, 8 May 2026 06:21:52 +0100 Subject: [PATCH 3/3] Merge IncomingMessage and OutgoingMessage --- crates/net/src/message.rs | 114 +++++++++++++++++++-------- crates/net/src/payload/mod.rs | 6 +- crates/net/src/payload/test.rs | 4 +- crates/net/src/service.rs | 5 +- crates/net/src/transport/controls.rs | 18 ++--- crates/net/src/transport/mock.rs | 7 +- crates/net/src/transport/tcp.rs | 15 ++-- 7 files changed, 111 insertions(+), 58 deletions(-) diff --git a/crates/net/src/message.rs b/crates/net/src/message.rs index 2be2058..f2f250f 100644 --- a/crates/net/src/message.rs +++ b/crates/net/src/message.rs @@ -7,50 +7,100 @@ use crate::{payload::{Payload, Query, Reply}}; pub type MsgId = u64; #[derive(Serialize, Deserialize, Debug, PartialEq)] -pub struct IncomingMessage { +pub struct Message { + pub id: MsgId, pub from: Id, pub payload: Payload, - pub id: MsgId, + pub to: Id, } -impl IncomingMessage { - pub fn receive(from: Id, msg: OutgoingMessage) -> Self { - Self { from, payload: msg.payload, id: msg.id } +impl Message { + + pub fn new(from: &Id, to: &Id, payload: Payload) -> Self { + Message{ + from: from.clone(), + id: OsRng.next_u64(), + payload, + to: to.clone(), + } } - pub fn reply(self, reply: Reply) -> OutgoingMessage { - OutgoingMessage { to: self.from, payload: Payload::Reply(reply), id: self.id } + pub fn query(from: &Id, to: &Id, q: impl Into) -> Self { + Self::new(from, to, Payload::Query(q.into())) } -} -#[derive(Serialize, Deserialize, Debug, PartialEq)] -pub struct OutgoingMessage { - pub to: Id, - pub payload: Payload, - pub id: MsgId, + pub fn reply(&self, reply: impl Into) -> Self { + // At the moment I think it only makes sense to reply to an existing message + // I don't think we need to consume the original message - but that's up for debate + Self { from: self.to.clone(), to: self.from.clone(), payload: Payload::Reply(reply.into()), id: self.id } + } } -impl OutgoingMessage { - pub fn new(to: &Id, payload: Payload) -> Self { - Self { to: to.clone(), payload, id: OsRng.next_u64() } - } - pub fn query(to: &Id, q: impl Into) -> Self { - Self::new(to, Payload::Query(q.into())) - } - pub fn reply(to: &Id, r: impl Into) -> Self { - Self::new(to, Payload::Reply(r.into())) +#[cfg(test)] +mod test{ + use super::*; + use crate::payload::{TestQuery,TestReply}; + + #[test] + fn new_message() { + let alice = Id::new([0u8; 32]); + let bob = Id::new([1u8; 32]); + let payload = Payload::Query(Query::Mock(TestQuery::Ping)); + let expect = Message{ + from: alice.clone(), + id: 1234567890, + payload: payload.clone(), + to: bob.clone(), + }; + + let got = Message::new(&alice, &bob, payload); + + // ID is assigned randomly so can't compare structs directly + assert_eq!(expect.from,got.from,"From should match"); + assert_eq!(expect.to,got.to,"To should match"); + assert_eq!(expect.payload,got.payload,"Payload should match"); } -} -pub struct OutgoingMessageBuilder { - to: Id, - id: MsgId, -} + #[test] + fn test_query() { + let alice = Id::new([0u8; 32]); + let bob = Id::new([1u8; 32]); + let payload = Payload::Query(Query::Mock(TestQuery::Ping)); + let expect = Message{ + from: alice.clone(), + id: 1234567890, + payload: payload.clone(), + to: bob.clone(), + }; + + let got = Message::new(&alice, &bob, payload); -impl OutgoingMessageBuilder { - pub fn new(incoming: &IncomingMessage) -> Self { - Self { to: incoming.from.clone(), id: incoming.id } + // ID is assigned randomly so can't compare structs directly + assert_eq!(expect.from,got.from,"From should match"); + assert_eq!(expect.to,got.to,"To should match"); + assert_eq!(expect.payload,got.payload,"Payload should match"); } - pub fn reply(self, reply: Reply) -> OutgoingMessage { - OutgoingMessage { to: self.to, payload: Payload::Reply(reply), id: self.id } + + #[test] + fn test_reply() { + let alice = Id::new([0u8; 32]); + let bob = Id::new([1u8; 32]); + let payload = Payload::Query(Query::Mock(TestQuery::Ping)); + let message = Message{ + from: alice.clone(), + id: 1234567890, + payload, + to: bob.clone(), + }; + + let expect = Message { + from: bob.clone(), + id: 1234567890, + payload: Payload::Reply(Reply::Mock(TestReply::Pong)), + to: alice.clone(), + }; + + // ID should match so we can compare directly + let got = message.reply(Reply::Mock(TestReply::Pong)); + assert_eq!(expect,got,"Reply should match"); } } \ No newline at end of file diff --git a/crates/net/src/payload/mod.rs b/crates/net/src/payload/mod.rs index 3e7885b..a3e694d 100644 --- a/crates/net/src/payload/mod.rs +++ b/crates/net/src/payload/mod.rs @@ -10,13 +10,13 @@ pub use test::*; use serde::{Deserialize, Serialize}; -#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)] +#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)] pub enum Payload { Query(Query), Reply(Reply), } -#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)] +#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)] pub enum Query { // Pow(PowQuery), // Tag(TagQuery), @@ -24,7 +24,7 @@ pub enum Query { Mock(TestQuery), } -#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)] +#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)] pub enum Reply { // Empty, // Ok, diff --git a/crates/net/src/payload/test.rs b/crates/net/src/payload/test.rs index 85db3f2..536bd21 100644 --- a/crates/net/src/payload/test.rs +++ b/crates/net/src/payload/test.rs @@ -2,12 +2,12 @@ use serde::{Deserialize, Serialize}; use crate::payload::{Query, Reply}; -#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)] +#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)] pub enum TestQuery { Ping, } -#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)] +#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)] pub enum TestReply { Pong, } diff --git a/crates/net/src/service.rs b/crates/net/src/service.rs index 7f3209e..ed435a2 100644 --- a/crates/net/src/service.rs +++ b/crates/net/src/service.rs @@ -2,7 +2,7 @@ use crypto::{error::CryptoError, id::Id}; use tokio_util::sync::CancellationToken; use tokio::task::JoinError; use thiserror::Error; -use crate::{/*net::session::ActiveSession, */error::NetError,message::IncomingMessage,payload::Payload,peer::Peer,/*utils::{SerdeError,ChannelError}, */transport::error::TransportError}; +use crate::{/*net::session::ActiveSession, */error::NetError,message::Message,payload::Payload,peer::Peer,/*utils::{SerdeError,ChannelError}, */transport::error::TransportError}; #[derive(Debug,Error)] pub enum NetServError { @@ -56,13 +56,12 @@ pub trait NetService{ fn drop_session(&mut self, peer: &Id) -> impl Future> + Send; // Listen for incoming messages from all peers - fn listen(&mut self, token: CancellationToken) -> impl Future> + Send; + fn listen(&mut self, token: CancellationToken) -> impl Future> + Send; // Broadcast messages to all sessions // Responsible for encrypting for each peer fn broadcast(&mut self, msg: Payload, token: CancellationToken) -> impl Future> + Send; // Transmit messages to a specific session - // OutgoingMessage has its own PeerID fn transmit(&mut self, msg: Payload,target: Id, token: CancellationToken) -> impl Future> + Send; } \ No newline at end of file diff --git a/crates/net/src/transport/controls.rs b/crates/net/src/transport/controls.rs index 6f7db50..1c0c5fc 100644 --- a/crates/net/src/transport/controls.rs +++ b/crates/net/src/transport/controls.rs @@ -1,24 +1,24 @@ use tokio::sync::mpsc; -use crate::{message::{IncomingMessage, OutgoingMessage}, utils::ChannelError}; +use crate::{message::{Message}, utils::ChannelError}; const CHAN_SIZE: usize = 128; pub struct TransportDispatcher { - pub(super) rx: mpsc::Receiver, - pub(super) tx: mpsc::Sender, + pub(super) rx: mpsc::Receiver, + pub(super) tx: mpsc::Sender, } pub struct TransportHandler { - pub(super) rx: mpsc::Receiver, - pub(super) tx: mpsc::Sender, + pub(super) rx: mpsc::Receiver, + pub(super) tx: mpsc::Sender, } impl TransportDispatcher { - pub async fn send(&self, msg: OutgoingMessage) -> Result<(), ChannelError> { + pub async fn send(&self, msg: Message) -> Result<(), ChannelError> { self.tx.send(msg).await.map_err(|_| ChannelError::Closed) } - pub async fn recv(&mut self) -> Option { + pub async fn recv(&mut self) -> Option { self.rx.recv().await } } @@ -29,10 +29,10 @@ impl TransportHandler { let (recv_tx, recv_rx) = mpsc::channel(CHAN_SIZE); (Self { rx: send_rx, tx: recv_tx }, TransportDispatcher { rx: recv_rx, tx: send_tx }) } - pub async fn send(&self, msg: IncomingMessage) -> Result<(), ChannelError> { + pub async fn send(&self, msg: Message) -> Result<(), ChannelError> { self.tx.send(msg).await.map_err(|_| ChannelError::Closed) } - pub async fn recv(&mut self) -> Option { + pub async fn recv(&mut self) -> Option { self.rx.recv().await } } \ No newline at end of file diff --git a/crates/net/src/transport/mock.rs b/crates/net/src/transport/mock.rs index 7c02224..e2d7ab2 100644 --- a/crates/net/src/transport/mock.rs +++ b/crates/net/src/transport/mock.rs @@ -92,7 +92,7 @@ mod tests { use tokio::time::timeout; use tokio_util::sync::CancellationToken; - use crate::{message::OutgoingMessage, net::{NetClient, SessionManager, SessionManagerDispatcher}, payload::{Payload, Reply, TagQuery}, peer::{PeerTable, PeerTableDispatcher}, service::Service, transport::{MockTransport, TransportHandler}, utils::random_bytes}; + use crate::{message::Message, net::{NetClient, SessionManager, SessionManagerDispatcher}, payload::{Payload, Reply, TagQuery}, peer::{PeerTable, PeerTableDispatcher}, service::Service, transport::{MockTransport, TransportHandler}, utils::random_bytes}; fn setup_peer_table() -> PeerTableDispatcher { let (service, dispatcher) = PeerTable::new(); @@ -126,12 +126,15 @@ mod tests { let (alice_handler, mut alice) = TransportHandler::new(); let (bob_handler, mut bob) = TransportHandler::new(); + // Alice needs an ID for Bob to be able to reply back + // TODO - NetIdentity has been removed. In favour of crypto::id? + // let alice_id = Id::new([0u8;32]); transport.add_participant(alice_sessions, None, alice_handler).await.expect("add participant failed"); transport.add_participant(bob_sessions, Some(bob_identity), bob_handler).await.expect("add participant failed"); tokio::spawn(async { transport.run(CancellationToken::new()).await.unwrap() }); - alice.send(OutgoingMessage::query(&bob_id, TagQuery::Get)).await.expect("alice query failed"); + alice.send(Message::query(&alice_id,&bob_id, TagQuery::Get)).await.expect("alice query failed"); let incoming = timeout(TIMEOUT, bob.recv()).await.expect("timeout").expect("channel should not be closed"); diff --git a/crates/net/src/transport/tcp.rs b/crates/net/src/transport/tcp.rs index c357e2f..0be6206 100644 --- a/crates/net/src/transport/tcp.rs +++ b/crates/net/src/transport/tcp.rs @@ -9,7 +9,7 @@ use bytes::BytesMut; use postcard::{to_stdvec,take_from_bytes,from_bytes}; use crypto::id::Id; -use crate::{/*net::{ActiveSession, PendingSession, NetClient},*/ message::{IncomingMessage,OutgoingMessage}, packet::Message, payload::Payload, peer::Peer, service::NetService, transport::error::TransportError}; +use crate::{/*net::{ActiveSession, PendingSession, NetClient},*/ message::{Message}, packet::Message, payload::Payload, peer::Peer, service::NetService, transport::error::TransportError}; const CHANNELSIZE: usize = 128; // Max message len current 226 @@ -141,7 +141,7 @@ impl NetService for TcpTransport { Ok(()) } - async fn listen(&mut self, token: CancellationToken) -> Result { + async fn listen(&mut self, token: CancellationToken) -> Result { loop { tokio::select!{ _ = token.cancelled() => { return Err(TransportError::Cancelled); }, @@ -176,8 +176,7 @@ impl NetService for TcpTransport { if let Some((id,encrypted)) = msg { let session = self.sessions.get_mut(&id).ok_or(TransportError::SessionNotFound(Some(id)))?; let decrypted = session.receive(encrypted)?; - let outgoing = from_bytes::(&decrypted)?; - return Ok(IncomingMessage::receive(id, outgoing)) + return Ok(from_bytes::(&decrypted)?); } else { // Channel has been closed return Err(TransportError::MessageChannelClosed) @@ -232,7 +231,7 @@ impl NetService for TcpTransport { #[cfg(test)] mod test { use super::*; - use crate::{message::OutgoingMessage,payload::{Action, Query, Reply, TagQuery}, pow::Pow, tag::{Tag,TagPayload}}; + use crate::{message::Message,payload::{Action, Query, Reply, TagQuery}, pow::Pow, tag::{Tag,TagPayload}}; use std::{io::Write, sync::mpsc::{Sender, channel}}; use tokio::time::timeout; use std::{time::{Duration}}; @@ -420,15 +419,17 @@ mod test { let peer = Peer::new(client.identity().expect("Expect static ID"), addr.to_string()); expect_messages.push( - IncomingMessage{ + Message{ + to: transport.client.identity().expect("Expected an ID").peer_id(), from: peer.id.clone(), payload: msg.clone(), id: 0, } ); - let send_message = OutgoingMessage{ + let send_message = Message{ to: transport.client.identity().expect("Expected an ID").peer_id(), + from: peer.id.clone(), payload: msg.clone(), id: 0, };