diff --git a/Cargo.lock b/Cargo.lock index e9fb2abc..59b195b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1370,7 +1370,7 @@ dependencies = [ [[package]] name = "io-reactor" version = "0.1.0" -source = "git+https://github.com/cyphernet-wg/rust-netservices#2a09c1714bea9baf8a96b9bd09d2dac4181a5019" +source = "git+https://github.com/cyphernet-wg/rust-netservices#32b3d308566228ebeb04b6aea4d6f17a9c4dfe6c" dependencies = [ "amplify", "crossbeam-channel", @@ -1618,7 +1618,7 @@ dependencies = [ [[package]] name = "netservices" version = "0.1.0" -source = "git+https://github.com/cyphernet-wg/rust-netservices#2a09c1714bea9baf8a96b9bd09d2dac4181a5019" +source = "git+https://github.com/cyphernet-wg/rust-netservices#32b3d308566228ebeb04b6aea4d6f17a9c4dfe6c" dependencies = [ "amplify", "cyphernet", diff --git a/radicle-node/src/wire/transport.rs b/radicle-node/src/wire/transport.rs index 8dd8819d..a989e23e 100644 --- a/radicle-node/src/wire/transport.rs +++ b/radicle-node/src/wire/transport.rs @@ -24,11 +24,12 @@ use radicle::storage::WriteStorage; use crate::crypto::Signer; use crate::service::reactor::{Fetch, Io}; use crate::service::{routing, session, Message, Service}; +use crate::wire::{Decode, Encode}; use crate::worker::{WorkerReq, WorkerResp}; use crate::{address, service}; /// Reactor action. -type Action = reactor::Action>, NetTransport, Message>>; +type Action = reactor::Action>, NetTransport>>; /// Peer connection state machine. #[derive(Debug)] @@ -113,12 +114,12 @@ impl Peer { } /// Switch back from upgraded to connected state. - fn downgrade(&mut self) -> Link { + fn downgrade(&mut self) { if let Self::Upgraded { id, link, .. } = self { - let link = *link; - *self = Self::Connected { id: *id, link }; - - link + *self = Self::Connected { + id: *id, + link: *link, + }; } else { panic!("Peer::downgrade: can't downgrade if not in upgraded state"); } @@ -139,6 +140,8 @@ pub struct Transport { peers: HashMap>, /// SOCKS5 proxy address. proxy: net::SocketAddr, + /// Buffer for incoming peer data. + read_queue: VecDeque, } impl Transport @@ -164,6 +167,7 @@ where proxy, actions: VecDeque::new(), peers: HashMap::default(), + read_queue: VecDeque::new(), } } @@ -214,28 +218,21 @@ where self.actions.push_back(Action::UnregisterTransport(fd)); } - fn upgraded(&mut self, transport: NetTransport, Message>) { - let fd = transport.as_raw_fd(); + fn upgraded(&mut self, session: NetTransport>) { + let fd = session.as_raw_fd(); let Some(peer) = self.peers.get_mut(&fd) else { log::error!(target: "transport", "Peer with fd {fd} was not found"); return; }; let (send, recv) = chan::bounded::>(1); let fetch = peer.upgraded(recv); - // Downgrade the transport to a simple session and the buffer of incoming data that is - // unprocessed. This buffer is provided as initial input to the worker. - let Ok((session, drain)) = transport.downgrade() else { - // This can happen in case the service attempts to send data to a peer after it has - // initiated an upgrade protocol. - panic!("Transport::upgraded: outgoing messages buffer is not empty"); - }; if self .worker .send(WorkerReq { fetch, session, - drain, + drain: self.read_queue.drain(..).collect(), channel: send, }) .is_err() @@ -255,11 +252,9 @@ where log::error!(target: "transport", "Peer with fd {fd} is already disconnected"); return; }; - let link = peer.downgrade(); - let transport = NetTransport::upgrade(session, link == Link::Inbound) - .expect("unable to set socket into non-blocking mode"); + peer.downgrade(); - self.actions.push_back(Action::RegisterTransport(transport)); + self.actions.push_back(Action::RegisterTransport(session)); self.service.fetch_complete(resp.result); } } @@ -272,7 +267,7 @@ where G: Signer + Negotiator + Send, { type Listener = NetAccept>; - type Transport = NetTransport, Message>; + type Transport = NetTransport>; type Command = service::Command; fn tick(&mut self, time: Instant) { @@ -313,7 +308,7 @@ where self.peers .insert(session.as_raw_fd(), Peer::connecting(Link::Inbound)); - let transport = match NetTransport::, Message>::upgrade(session, true) { + let transport = match NetTransport::>::accept(session) { Ok(transport) => transport, Err(err) => { log::error!(target: "transport", "Failed to upgrade accepted peer socket: {err}"); @@ -324,20 +319,15 @@ where self.actions .push_back(reactor::Action::RegisterTransport(transport)) } - ListenerEvent::Error(err) => { + ListenerEvent::Failure(err) => { log::error!(target: "transport", "Error listening for inbound connections: {err}"); } } } - fn handle_transport_event( - &mut self, - fd: RawFd, - event: SessionEvent, Message>, - _: Instant, - ) { + fn handle_transport_event(&mut self, fd: RawFd, event: SessionEvent>, _: Instant) { match event { - SessionEvent::SessionEstablished(node_id) => { + SessionEvent::Established(node_id) => { log::debug!(target: "transport", "Session established with {node_id}"); let conflicting = self @@ -374,34 +364,42 @@ where peer.connected(node_id); self.service.connected(node_id, link); } - SessionEvent::Message(msg) => { + SessionEvent::Data(data) => { if let Some(Peer::Connected { link, id }) = self.peers.get(&fd) { - log::debug!( - target: "transport", "Received message {:?} from {} ({:?})", msg, id, link - ); - self.service.received_message(*id, msg); + self.read_queue.extend(data); + + loop { + match Message::decode(&mut self.read_queue) { + Ok(msg) => { + log::debug!( + target: "transport", "Received message {:?} from {} ({:?})", msg, id, link + ); + self.service.received_message(*id, msg) + } + Err(err) if err.is_eof() => { + // Buffer is empty, or message isn't complete. + break; + } + Err(err) => { + // TODO(cloudhead): Include error in reason. + log::error!(target: "transport", "Invalid message from {}: {err}", id); + self.disconnect( + fd, + DisconnectReason::Protocol(service::DisconnectReason::Error( + session::Error::Misbehavior, + )), + ); + break; + } + } + } } else { log::warn!(target: "transport", "Dropping message from unconnected peer with fd {fd}"); } } - SessionEvent::FrameFailure(_err) => { - // TODO(cloudhead): Include error in reason. - self.disconnect( - fd, - DisconnectReason::Protocol(service::DisconnectReason::Error( - session::Error::Misbehavior, - )), - ); - } - SessionEvent::ConnectionFailure(err) => { + SessionEvent::Terminated(err) => { self.disconnect(fd, DisconnectReason::ConnectionError(Arc::new(err))); } - SessionEvent::Disconnected => { - self.disconnect( - fd, - DisconnectReason::Protocol(service::DisconnectReason::Peer), - ); - } } } @@ -462,7 +460,7 @@ where W: WriteStorage + 'static, G: Signer + Negotiator, { - type Item = reactor::Action>, NetTransport, Message>>; + type Item = reactor::Action>, NetTransport>>; fn next(&mut self) -> Option { if let Some(event) = self.actions.pop_front() { @@ -476,8 +474,11 @@ where target: "transport", "Sending {} message(s) to {}", msgs.len(), node_id ); let fd = self.by_id(&node_id); - - return Some(reactor::Action::Send(fd, msgs)); + let mut data = Vec::new(); + for msg in msgs { + msg.encode(&mut data).expect("in-memory writes never fail"); + } + return Some(reactor::Action::Send(fd, data)); } Io::Event(_e) => { log::warn!( @@ -499,7 +500,7 @@ where break; } - match NetTransport::, Message>::connect( + match NetTransport::>::connect( PeerAddr::new(node_id, socket_addr), &self.keypair, ) { diff --git a/radicle-node/src/worker.rs b/radicle-node/src/worker.rs index 280bc425..e90b223a 100644 --- a/radicle-node/src/worker.rs +++ b/radicle-node/src/worker.rs @@ -1,5 +1,6 @@ use crossbeam_channel as chan; use netservices::noise::NoiseXk; +use netservices::wire::NetTransport; use std::thread; use std::thread::JoinHandle; @@ -13,7 +14,7 @@ use crate::service::FetchResult; /// Worker request. pub struct WorkerReq { pub fetch: Fetch, - pub session: NoiseXk, + pub session: NetTransport>, pub drain: Vec, pub channel: chan::Sender>, } @@ -21,7 +22,7 @@ pub struct WorkerReq { /// Worker response. pub struct WorkerResp { pub result: FetchResult, - pub session: NoiseXk, + pub session: NetTransport>, } pub struct Worker {