diff --git a/Cargo.lock b/Cargo.lock index 4411c804..a180cf89 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1604,9 +1604,9 @@ dependencies = [ [[package]] name = "io-reactor" -version = "0.3.0" +version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2457e8fb1b1f298809fcd93cd15d485f52ef565f7ad47970583c7a80ae6c7e78" +checksum = "bd2eb3f6ef3b908c8144cd1d1a3069ecfdb60cbc795a303fd2d17b96c1c67c77" dependencies = [ "amplify", "crossbeam-channel", @@ -1838,9 +1838,9 @@ dependencies = [ [[package]] name = "netservices" -version = "0.5.0" +version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e227ff39744a414d32b9b473266c1739a950b89cd03694a58b624f4bde926c10" +checksum = "dcd7e1d8a6763072326ece5a4329c6a924d7e8889a301334169e46613c5ef7a4" dependencies = [ "amplify", "cyphernet", diff --git a/radicle-node/Cargo.toml b/radicle-node/Cargo.toml index 95091516..123e4217 100644 --- a/radicle-node/Cargo.toml +++ b/radicle-node/Cargo.toml @@ -19,12 +19,12 @@ colored = { version = "2.1.0" } crossbeam-channel = { version = "0.5.6" } cyphernet = { version = "0.4.1", features = ["tor", "dns", "ed25519", "p2p-ed25519"] } fastrand = { version = "2.0.0" } -io-reactor = { version = "0.3.0", features = ["popol"] } +io-reactor = { version = "0.5.0", features = ["popol"] } lexopt = { version = "0.3.0" } libc = { version = "0.2.137" } log = { version = "0.4.17", features = ["std"] } localtime = { version = "1.2.0" } -netservices = { version = "0.5.0", features = ["io-reactor", "socket2"] } +netservices = { version = "0.7.0", features = ["io-reactor", "socket2"] } nonempty = { version = "0.9.0", features = ["serialize"] } once_cell = { version = "1.13" } qcheck = { version = "1", default-features = false, optional = true } diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs index 1cc02a3a..9904d114 100644 --- a/radicle-node/src/wire/protocol.rs +++ b/radicle-node/src/wire/protocol.rs @@ -18,7 +18,7 @@ use localtime::LocalTime; use netservices::resource::{ListenerEvent, NetAccept, NetTransport, SessionEvent}; use netservices::session::{ProtocolArtifact, Socks5Session}; use netservices::{NetConnection, NetProtocol, NetReader, NetWriter}; -use reactor::{ResourceId, Timestamp}; +use reactor::{ResourceId, ResourceType, Timestamp}; use radicle::collections::RandomMap; use radicle::node::NodeId; @@ -456,20 +456,26 @@ where ) { match event { ListenerEvent::Accepted(connection) => { - let addr = connection.remote_addr(); + let Ok(addr) = connection.remote_addr() else { + log::warn!(target: "wire", "Accepted connection doesn't have remote address; dropping.."); + drop(connection); + + return; + }; + let addr: NetAddr = addr.into(); let fd = connection.as_raw_fd(); log::debug!(target: "wire", "Accepting inbound connection from {addr} (fd={fd}).."); // If the service doesn't want to accept this connection, // we drop the connection here, which disconnects the socket. - if !self.service.accepted(NetAddr::from(addr.clone()).into()) { + if !self.service.accepted(addr.clone().into()) { log::debug!(target: "wire", "Rejecting inbound connection from {addr} (fd={fd}).."); drop(connection); return; } - let session = accept::(connection, self.signer.clone()); + let session = accept::(addr.clone(), connection, self.signer.clone()); let transport = match NetTransport::with_session(session, Link::Inbound) { Ok(transport) => transport, Err(err) => { @@ -478,13 +484,7 @@ where } }; - self.inbound.insert( - fd, - Inbound { - id: None, - addr: addr.into(), - }, - ); + self.inbound.insert(fd, Inbound { id: None, addr }); self.actions .push_back(reactor::Action::RegisterTransport(transport)) } @@ -494,7 +494,11 @@ where } } - fn handle_registered(&mut self, fd: RawFd, id: ResourceId) { + fn handle_registered(&mut self, fd: RawFd, id: ResourceId, typ: ResourceType) { + if typ == ResourceType::Listener { + // Not interested in listener resource registration. + return; + } if let Some(outbound) = self.outbound.get_mut(&fd) { log::debug!(target: "wire", "Outbound peer resource registered for {} with id={id} (fd={fd})", outbound.nid); outbound.id = Some(id); @@ -837,10 +841,7 @@ where } } Io::Wakeup(d) => { - self.actions.push_back(reactor::Action::SetTimer( - // TODO: Remove this when `io-reactor` can handle `0` duration timeouts. - d.max(localtime::LocalDuration::from_millis(1)).into(), - )); + self.actions.push_back(reactor::Action::SetTimer(d.into())); } Io::Fetch { rid, @@ -924,16 +925,11 @@ pub fn dial>( /// Accept a new connection. pub fn accept>( + remote_addr: NetAddr, connection: net::TcpStream, signer: G, ) -> WireSession { - session::( - connection.remote_addr().into(), - None, - connection, - signer, - false, - ) + session::(remote_addr, None, connection, signer, false) } /// Create a new [`WireSession`].