node: Upgrade dependencies

Fixes issue with improperly closed connections.

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",

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 }

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<HostName> = 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::<G>(connection, self.signer.clone());
+                let session = accept::<G>(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(
+                self.inbound.insert(fd, Inbound { id: None, addr });
-                    fd,
-                    Inbound {
-                        id: None,
-                        addr: addr.into(),
-                    },
-                );
                 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(
+                    self.actions.push_back(reactor::Action::SetTimer(d.into()));
-                        // TODO: Remove this when `io-reactor` can handle `0` duration timeouts.
-                        d.max(localtime::LocalDuration::from_millis(1)).into(),
-                    ));
                 }
                 Io::Fetch {
                     rid,
@@ -924,16 +925,11 @@ pub fn dial<G: Signer + Ecdh<Pk = NodeId>>(
 
 /// Accept a new connection.
 pub fn accept<G: Signer + Ecdh<Pk = NodeId>>(
+    remote_addr: NetAddr<HostName>,
     connection: net::TcpStream,
     signer: G,
 ) -> WireSession<G> {
-    session::<G>(
+    session::<G>(remote_addr, None, connection, signer, false)
-        connection.remote_addr().into(),
-        None,
-        connection,
-        signer,
-        false,
-    )
 }
 
 /// Create a new [`WireSession`].