node: Don't connect to recently attempted peers
This commit is contained in:
parent
ca0abc4356
commit
0349699200
|
|
@ -12,6 +12,7 @@ use crate::address::types;
|
||||||
use crate::address::{KnownAddress, Source};
|
use crate::address::{KnownAddress, Source};
|
||||||
use crate::service::NodeId;
|
use crate::service::NodeId;
|
||||||
use crate::wire::AddressType;
|
use crate::wire::AddressType;
|
||||||
|
use crate::LocalTime;
|
||||||
|
|
||||||
#[derive(Error, Debug)]
|
#[derive(Error, Debug)]
|
||||||
pub enum Error {
|
pub enum Error {
|
||||||
|
|
@ -175,7 +176,7 @@ impl Store for Book {
|
||||||
fn entries(&self) -> Result<Box<dyn Iterator<Item = (NodeId, KnownAddress)>>, Error> {
|
fn entries(&self) -> Result<Box<dyn Iterator<Item = (NodeId, KnownAddress)>>, Error> {
|
||||||
let mut stmt = self
|
let mut stmt = self
|
||||||
.db
|
.db
|
||||||
.prepare("SELECT node, type, value, source FROM addresses ORDER BY node")?
|
.prepare("SELECT node, type, value, source, last_success, last_attempt FROM addresses ORDER BY node")?
|
||||||
.into_iter();
|
.into_iter();
|
||||||
let mut entries = Vec::new();
|
let mut entries = Vec::new();
|
||||||
|
|
||||||
|
|
@ -184,14 +185,18 @@ impl Store for Book {
|
||||||
let _typ = row.read::<AddressType, _>("type");
|
let _typ = row.read::<AddressType, _>("type");
|
||||||
let addr = row.read::<Address, _>("value");
|
let addr = row.read::<Address, _>("value");
|
||||||
let source = row.read::<Source, _>("source");
|
let source = row.read::<Source, _>("source");
|
||||||
|
let last_success = row.read::<Option<i64>, _>("last_success");
|
||||||
|
let last_attempt = row.read::<Option<i64>, _>("last_attempt");
|
||||||
|
let last_success = last_success.map(|t| LocalTime::from_millis(t as u128));
|
||||||
|
let last_attempt = last_attempt.map(|t| LocalTime::from_millis(t as u128));
|
||||||
|
|
||||||
entries.push((
|
entries.push((
|
||||||
node,
|
node,
|
||||||
KnownAddress {
|
KnownAddress {
|
||||||
addr,
|
addr,
|
||||||
source,
|
source,
|
||||||
last_success: None,
|
last_success,
|
||||||
last_attempt: None,
|
last_attempt,
|
||||||
},
|
},
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -20,7 +20,7 @@ use localtime::{LocalDuration, LocalTime};
|
||||||
use log::*;
|
use log::*;
|
||||||
|
|
||||||
use crate::address;
|
use crate::address;
|
||||||
use crate::address::AddressBook;
|
use crate::address::{AddressBook, KnownAddress};
|
||||||
use crate::crypto;
|
use crate::crypto;
|
||||||
use crate::crypto::{Signer, Verified};
|
use crate::crypto::{Signer, Verified};
|
||||||
use crate::identity::IdentityError;
|
use crate::identity::IdentityError;
|
||||||
|
|
@ -997,7 +997,7 @@ where
|
||||||
timestamp,
|
timestamp,
|
||||||
addresses
|
addresses
|
||||||
.iter()
|
.iter()
|
||||||
.map(|a| address::KnownAddress::new(a.clone(), address::Source::Peer)),
|
.map(|a| KnownAddress::new(a.clone(), address::Source::Peer)),
|
||||||
) {
|
) {
|
||||||
Ok(updated) => {
|
Ok(updated) => {
|
||||||
// Only relay if we received new information.
|
// Only relay if we received new information.
|
||||||
|
|
@ -1404,7 +1404,7 @@ where
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get a list of peers available to connect to.
|
/// Get a list of peers available to connect to.
|
||||||
fn available_peers(&mut self) -> Vec<(NodeId, Address)> {
|
fn available_peers(&mut self) -> Vec<(NodeId, KnownAddress)> {
|
||||||
let outbound = self
|
let outbound = self
|
||||||
.sessions
|
.sessions
|
||||||
.values()
|
.values()
|
||||||
|
|
@ -1425,7 +1425,6 @@ where
|
||||||
entries
|
entries
|
||||||
.filter(|(nid, _)| !self.sessions.contains_key(nid))
|
.filter(|(nid, _)| !self.sessions.contains_key(nid))
|
||||||
.take(wanted)
|
.take(wanted)
|
||||||
.map(|(n, s)| (n, s.addr))
|
|
||||||
.collect()
|
.collect()
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
|
|
@ -1473,8 +1472,17 @@ where
|
||||||
}
|
}
|
||||||
|
|
||||||
fn maintain_connections(&mut self) {
|
fn maintain_connections(&mut self) {
|
||||||
for (id, addr) in self.available_peers() {
|
let now = self.clock;
|
||||||
self.connect(id, addr.clone());
|
|
||||||
|
// Nb. We use the `MAX_RECONNECTION_DELTA` to know when it's ok to reconnect, because
|
||||||
|
// these aren't persistent peers. They could go offline for a long time and we don't want to
|
||||||
|
// be too persistent.
|
||||||
|
for (id, ka) in self
|
||||||
|
.available_peers()
|
||||||
|
.into_iter()
|
||||||
|
.filter(|(_, ka)| now - ka.last_attempt.unwrap_or_default() >= MAX_RECONNECTION_DELTA)
|
||||||
|
{
|
||||||
|
self.connect(id, ka.addr.clone());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -202,12 +202,9 @@ where
|
||||||
Address::from(net::SocketAddr::from((self.ip, 8776)))
|
Address::from(net::SocketAddr::from((self.ip, 8776)))
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn import_addresses<P>(&mut self, peers: P)
|
pub fn import_addresses<'a>(&mut self, peers: impl IntoIterator<Item = &'a Self>) {
|
||||||
where
|
|
||||||
P: AsRef<[Self]>,
|
|
||||||
{
|
|
||||||
let timestamp = self.timestamp();
|
let timestamp = self.timestamp();
|
||||||
for peer in peers.as_ref() {
|
for peer in peers.into_iter() {
|
||||||
let known_address = address::KnownAddress::new(peer.address(), address::Source::Peer);
|
let known_address = address::KnownAddress::new(peer.address(), address::Source::Peer);
|
||||||
self.service
|
self.service
|
||||||
.addresses_mut()
|
.addresses_mut()
|
||||||
|
|
|
||||||
|
|
@ -1018,6 +1018,41 @@ fn test_maintain_connections() {
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_maintain_connections_failed_attempt() {
|
||||||
|
let bob = Peer::new("bob", [8, 8, 8, 8]);
|
||||||
|
let eve = Peer::new("eve", [9, 9, 9, 9]);
|
||||||
|
let mut alice = Peer::new("alice", [7, 7, 7, 7]);
|
||||||
|
|
||||||
|
alice.connect_to(&bob);
|
||||||
|
// Make sure Alice knows about Eve.
|
||||||
|
alice.receive(bob.id, eve.node_announcement());
|
||||||
|
alice.disconnected(bob.id(), &DisconnectReason::Command);
|
||||||
|
alice
|
||||||
|
.outbox()
|
||||||
|
.find(|o| matches!(o, Io::Connect(id, _) if id == &eve.id))
|
||||||
|
.expect("Alice attempts Eve");
|
||||||
|
alice.attempted(eve.id, eve.addr());
|
||||||
|
|
||||||
|
// Disconnect Eve and make sure Alice doesn't try to re-connect immediately.
|
||||||
|
alice.disconnected(eve.id(), &DisconnectReason::Command);
|
||||||
|
assert!(!alice.outbox().any(|o| matches!(o, Io::Connect(_, _))));
|
||||||
|
|
||||||
|
// Now pass some time and try again.
|
||||||
|
alice.elapse(MAX_RECONNECTION_DELTA);
|
||||||
|
alice
|
||||||
|
.outbox()
|
||||||
|
.find(|o| matches!(o, Io::Connect(id, _) if id == &eve.id))
|
||||||
|
.expect("Alice attempts Eve again");
|
||||||
|
|
||||||
|
// Disconnect Eve and make sure Alice doesn't try to re-connect immediately.
|
||||||
|
alice.disconnected(eve.id(), &DisconnectReason::Command);
|
||||||
|
assert!(!alice.outbox().any(|o| matches!(o, Io::Connect(_, _))));
|
||||||
|
// Or even after some short time..
|
||||||
|
alice.elapse(MIN_RECONNECTION_DELTA);
|
||||||
|
assert!(!alice.outbox().any(|o| matches!(o, Io::Connect(_, _))));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_track_repo_subscribe() {
|
fn test_track_repo_subscribe() {
|
||||||
let mut alice = Peer::new("alice", [7, 7, 7, 7]);
|
let mut alice = Peer::new("alice", [7, 7, 7, 7]);
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue