mab122
/
radicle-heartwood-lfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

node: Fix various small logic issues 

* Set peer's `last_active` even if they are being throttled
* Make sure not to break out of loop in wire module
* Add some more logging
* Use different variable for session filtering to make it clearer
This commit is contained in:
cloudhead 2023-11-10 14:44:52 +01:00
parent 2903b4f09a
commit 3aad82d158
No known key found for this signature in database
2 changed files with 13 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
radicle-node/src/service.rs
View File

@ -868,6 +868,8 @@ where
self.outbox.wakeup(delay); self.outbox.wakeup(delay);
} else { } else {
debug!(target: "service", "Dropping peer {remote}..");
self.sessions.remove(&remote); self.sessions.remove(&remote);
// Only re-attempt outbound connections, since we don't care if an inbound connection // Only re-attempt outbound connections, since we don't care if an inbound connection
// is dropped. // is dropped.
@ -1153,6 +1155,8 @@ where
warn!(target: "service", "Session not found for {remote}"); warn!(target: "service", "Session not found for {remote}");
return Ok(()); return Ok(());
}; };
peer.last_active = self.clock;
let limit = match peer.link { let limit = match peer.link {
Link::Outbound => &self.config.limits.rate.outbound, Link::Outbound => &self.config.limits.rate.outbound,
Link::Inbound => &self.config.limits.rate.inbound, Link::Inbound => &self.config.limits.rate.inbound,
@ -1164,7 +1168,6 @@ where
trace!(target: "service", "Rate limiting message from {remote} ({})", peer.addr); trace!(target: "service", "Rate limiting message from {remote} ({})", peer.addr);
return Ok(()); return Ok(());
} }
peer.last_active = self.clock;
message.log(log::Level::Debug, remote, Link::Inbound); message.log(log::Level::Debug, remote, Link::Inbound);
trace!(target: "service", "Received message {:?} from {}", &message, peer.id); trace!(target: "service", "Received message {:?} from {}", &message, peer.id);
@ -1184,7 +1187,7 @@ where
let relay_to = self let relay_to = self
.sessions .sessions
.connected() .connected()
.filter(|(id, _)| *id != remote && *id != &announcer) .filter(|(id, _)| *id != &relayer && *id != &announcer)
.map(|(_, p)| p); .map(|(_, p)| p);
self.outbox.relay(ann, relay_to); self.outbox.relay(ann, relay_to);
@ -1411,6 +1414,8 @@ where
} }
fn connect(&mut self, nid: NodeId, addr: Address) -> bool { fn connect(&mut self, nid: NodeId, addr: Address) -> bool {
debug!(target: "service", "Connecting to {nid} ({addr})..");
if self.sessions.contains_key(&nid) { if self.sessions.contains_key(&nid) {
warn!(target: "service", "Attempted connection to peer {nid} which already has a session"); warn!(target: "service", "Attempted connection to peer {nid} which already has a session");
return false; return false;
@ -1517,6 +1522,11 @@ where
.filter(|(_, session)| *now - session.last_active >= STALE_CONNECTION_TIMEOUT); .filter(|(_, session)| *now - session.last_active >= STALE_CONNECTION_TIMEOUT);
for (_, session) in stale { for (_, session) in stale {
debug!(target: "service", "Disconnecting unresponsive peer {}..", session.id);
// TODO: Should we switch the session state to "disconnected" even before receiving
// an official "disconnect"? Otherwise we keep pinging until we get the disconnection.
self.outbox.disconnect( self.outbox.disconnect(
session.id, session.id,
DisconnectReason::Session(session::Error::Timeout), DisconnectReason::Session(session::Error::Timeout),

4
radicle-node/src/wire/protocol.rs
View File

@ -780,7 +780,7 @@ where
target: "wire", target: "wire",
"Attempt to connect to already connected peer {node_id}" "Attempt to connect to already connected peer {node_id}"
); );
break; continue;
} }
match dial::<G>( match dial::<G>(
@ -801,7 +801,6 @@ where
transport.as_raw_fd(), transport.as_raw_fd(),
Peer::outbound(addr.to_inner(), node_id), Peer::outbound(addr.to_inner(), node_id),
); );
self.actions self.actions
.push_back(reactor::Action::RegisterTransport(transport)); .push_back(reactor::Action::RegisterTransport(transport));
} }
@ -810,7 +809,6 @@ where
self.service self.service
.disconnected(node_id, &DisconnectReason::Dial(Arc::new(err))); .disconnected(node_id, &DisconnectReason::Dial(Arc::new(err)));
break;
} }
} }
} }