Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions changelog.d/addrfetch-learns-reply.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
Fixed

- **An addr-fetch session keeps the addresses it asked for.** The `getaddr`
reply from an addr-fetch peer (the `--proxy` DNS seeds, `-seednode`) was
dropped before AddrMan and scored as misbehavior. As in Core, the
`getaddr` now buys `MAX_ADDR_TO_SEND` more tokens, the reply (`addrv2` or
v1 `addr`) is learned and not relayed, and more than one address ends the
session with a plain disconnect.
8 changes: 8 additions & 0 deletions changelog.d/proxy-seed-bootstrap-wait.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
Fixed

- **A fresh `--proxy` datadir waits for its seeds.** Catch-up read the dial
targets before any SOCKS seed addr-fetch had answered, so a node with an
empty book shut down about 60 ms after start. It now waits up to 180 s
for a seed's `getaddr` reply to be learned (a self-announce does not
count) and re-queues the seeds every 60 s while no addr-fetch session is
live. With no reply, the empty book takes the incomplete path as before.
56 changes: 50 additions & 6 deletions crates/rbitcoin-net/src/peer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -455,6 +455,10 @@ impl AddrTokenBucket {
self.tokens -= take as f64;
take
}

pub(crate) fn credit(&mut self, n: usize) {
self.tokens += n as f64;
}
}

#[doc(hidden)]
Expand Down Expand Up @@ -2148,6 +2152,7 @@ pub fn maybe_queue_addrfetch_getaddr(
if session.conn_type != crate::peers::PeerConnType::AddrFetch {
return false;
}
session.credit_getaddr_reply();
let _ = queue_out(out, NetworkMessage::GetAddr);
true
}
Expand Down Expand Up @@ -2971,7 +2976,7 @@ pub fn handle_peer_inventory_msg(
return Ok(());
}
match payload {
NetworkMessage::Addr(list) => on_addr_list(follow, session, list.len())?,
NetworkMessage::Addr(list) => on_addr_list(follow, session, list)?,
NetworkMessage::AddrV2(list) => on_addrv2(follow, session, list)?,
NetworkMessage::GetHeaders(gh) => on_getheaders(hub, out_tx, follow, session, gh)?,
NetworkMessage::GetBlocks(gb) => on_getblocks(hub, out_tx, follow, session, gb)?,
Expand Down Expand Up @@ -3082,19 +3087,58 @@ pub fn on_sendaddrv2(follow: &mut PeerFollowState, session: Option<&crate::peers
punish_disconnect(&mut follow.ban_score, session);
}

/// v1 `addr` is learned only as an addr-fetch session's `getaddr` reply.
#[doc(hidden)]
pub fn on_addr_list(
follow: &mut PeerFollowState,
session: Option<&crate::peers::LivePeer>,
n: usize,
list: &[(u32, bitcoin::p2p::address::Address)],
) -> Result<(), NetError> {
if session.is_some_and(|s| s.conn_type == crate::peers::PeerConnType::AddrFetch && n > 1) {
punish_disconnect(&mut follow.ban_score, session);
let Some(s) = session.filter(|s| s.conn_type == crate::peers::PeerConnType::AddrFetch) else {
return Ok(());
};
if list.len() > MAX_ADDR_TO_SEND {
misbehaving(&mut follow.ban_score, session);
return Ok(());
}
let v2: Vec<_> = list
.iter()
.filter_map(|(time, a)| {
let sock = a.socket_addr().ok()?;
Some(bitcoin::p2p::address::AddrV2Message {
time: *time,
services: a.services,
addr: crate::NetAddr::from_socket(sock).to_addrv2(),
port: sock.port(),
})
})
.collect();
learn_addrfetch_reply(s, &v2, list.len());
Ok(())
}

/// Core `ProcessMessage` ADDR/ADDRV2 on an addr-fetch session: the reply to
/// the `getaddr` it sent is learned and not relayed (`m_getaddr_sent`).
/// More than one address ends the session, without misbehavior.
fn learn_addrfetch_reply(
s: &crate::peers::LivePeer,
list: &[bitcoin::p2p::address::AddrV2Message],
n: usize,
) {
if let Some(ph) = s.peer_hub() {
// Node time, as the bucket's start (`connected_at`) is.
let now_ms = s.clock_now().saturating_mul(1000);
let allow = s.take_addr_tokens(list.len(), now_ms);
ph.learn_addrv2(&list[..allow]);
if n > 1 {
ph.note_addrfetch_reply();
}
}
if n > 1 {
s.request_disconnect();
}
}

#[doc(hidden)]
pub fn on_addrv2(
follow: &mut PeerFollowState,
Expand All @@ -3111,8 +3155,8 @@ pub fn on_addrv2(
misbehaving(&mut follow.ban_score, session);
return Ok(());
}
if session.is_some_and(|s| s.conn_type == crate::peers::PeerConnType::AddrFetch && n > 1) {
punish_disconnect(&mut follow.ban_score, session);
if let Some(s) = session.filter(|s| s.conn_type == crate::peers::PeerConnType::AddrFetch) {
learn_addrfetch_reply(s, list, n);
return Ok(());
}
if let Some(s) = session {
Expand Down
22 changes: 22 additions & 0 deletions crates/rbitcoin-net/src/peers.rs
Original file line number Diff line number Diff line change
Expand Up @@ -983,6 +983,15 @@ impl LivePeer {
*self.out_tx.lock().unwrap_or_else(|e| e.into_inner()) = Some(tx);
}

/// Core `m_addr_token_bucket += MAX_ADDR_TO_SEND` on `getaddr`: the
/// reply may exceed the bucket cap.
pub(crate) fn credit_getaddr_reply(&self) {
self.addr_bucket
.lock()
.unwrap_or_else(|e| e.into_inner())
.credit(crate::peer::MAX_ADDR_TO_SEND);
}

/// How many of the `want` addresses in one message this peer may have
/// processed (learned and relayed). The rest count as rate limited. A
/// peer with the `addr` permission spends tokens but is never limited.
Expand Down Expand Up @@ -1457,6 +1466,9 @@ pub struct PeerHub {
/// Outbound dials still before session registration, by target. The
/// redial pass skips these, so a slow connect is not stacked every 2 s.
pub dials_in_flight: Arc<Mutex<HashMap<String, usize>>>,
/// Addr-fetch `getaddr` replies (more than one address) learned into the
/// book. A self-announce does not count.
addrfetch_replies: AtomicU64,
/// Network P2P port used when a remembered host omits `:port`. `0` = unset.
pub connect_default_port: AtomicU16,
pub dial_tx: Mutex<Option<mpsc::UnboundedSender<DialRequest>>>,
Expand Down Expand Up @@ -1579,6 +1591,7 @@ impl PeerHub {
connect_hosts: Mutex::new(Vec::new()),
connect_addrs: Mutex::new(Vec::new()),
dials_in_flight: Arc::new(Mutex::new(HashMap::new())),
addrfetch_replies: AtomicU64::new(0),
connect_default_port: AtomicU16::new(0),
dial_tx: Mutex::new(None),
hb_selected: Mutex::new(Vec::new()),
Expand Down Expand Up @@ -1807,6 +1820,15 @@ impl PeerHub {
///
/// Copies the I2P reachability bit onto that book. `learn_addrv2` stores
/// only rows the book will dial.
/// How many addr-fetch `getaddr` replies have been learned so far.
pub fn addrfetch_replies(&self) -> u64 {
self.addrfetch_replies.load(Ordering::Acquire)
}

pub(crate) fn note_addrfetch_reply(&self) {
self.addrfetch_replies.fetch_add(1, Ordering::Release);
}

pub fn set_addrman(&self, am: std::sync::Arc<Mutex<crate::seeds::AddrMan>>) {
let on = self.i2p_reachable.load(Ordering::Relaxed);
am.lock()
Expand Down
127 changes: 127 additions & 0 deletions crates/rbitcoin-net/tests/peer/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5075,3 +5075,130 @@ fn compact_filter_ranges_past_core_limits_disconnect() {
}

mod log_text_tests;

/// Core `ProcessMessage` ADDR/ADDRV2: the `getaddr` an addr-fetch session
/// sends buys `MAX_ADDR_TO_SEND` more addresses. The reply, v1 or v2, is
/// learned and not relayed (`m_getaddr_sent`). More than one address ends the
/// session without misbehavior; a one-address self-announce does not.
#[tokio::test]
async fn addrfetch_learns_the_getaddr_reply_then_disconnects() {
use bitcoin::p2p::address::{AddrV2, AddrV2Message, Address};
use bitcoin::p2p::message_network::VersionMessage;
use bitcoin::p2p::ServiceFlags;
use std::sync::atomic::Ordering;

let (dir, hub) = rbitcoin_net::chain::tiny_regtest_hub_labeled("addrfetch-reply");
let peers = rbitcoin_net::peers::PeerHub::new();
let book = std::sync::Arc::new(std::sync::Mutex::new(rbitcoin_net::seeds::AddrMan::new()));
peers.set_addrman(std::sync::Arc::clone(&book));
let register = |port: u16, typ: rbitcoin_net::peers::PeerConnType| {
let addr = std::net::SocketAddr::from(([127, 0, 0, 1], port));
let ver = VersionMessage {
version: bitcoin::p2p::PROTOCOL_VERSION,
services: ServiceFlags::NETWORK | ServiceFlags::WITNESS,
timestamp: 0,
receiver: Address::new(&addr, ServiceFlags::NONE),
sender: Address::new(&addr, ServiceFlags::NONE),
nonce: 0,
user_agent: "/rbitcoin:test/".into(),
start_height: 0,
relay: true,
};
peers.register(addr, addr, &ver, false, typ)
};
let neighbor = register(18459, rbitcoin_net::peers::PeerConnType::OutboundFullRelay);
neighbor.set_wants_addrv2();
let (relay_tx, mut relayed) = mpsc::unbounded_channel();
neighbor.attach_out(relay_tx);
let book_len = || book.lock().unwrap().len();
let routable = |i: u32| AddrV2::Ipv4(std::net::Ipv4Addr::from(0x0b00_0001 + (i << 16)));

let seed = register(18460, rbitcoin_net::peers::PeerConnType::AddrFetch);
let (out_tx, _out_rx) = mpsc::unbounded_channel();
assert!(maybe_queue_addrfetch_getaddr(&out_tx, &seed));
let time = seed.clock_now() as u32;
let mut follow = PeerFollowState::new();
push(
&hub,
&out_tx,
&mut follow,
Some(seed.as_ref()),
NetworkMessage::AddrV2(vec![AddrV2Message {
time,
services: ServiceFlags::NETWORK,
addr: AddrV2::Ipv4(std::net::Ipv4Addr::new(9, 9, 9, 9)),
port: 8333,
}]),
)
.await;
assert_eq!(book_len(), 1, "a self-announce is learned");
assert_eq!(
peers.addrfetch_replies(),
0,
"a self-announce is not the reply"
);
assert!(
!seed.stop.load(Ordering::SeqCst),
"one address is not the reply"
);
let reply = (0..MAX_ADDR_TO_SEND as u32)
.map(|i| AddrV2Message {
time,
services: ServiceFlags::NETWORK | ServiceFlags::WITNESS,
addr: routable(i),
port: 18444,
})
.collect();
push(
&hub,
&out_tx,
&mut follow,
Some(seed.as_ref()),
NetworkMessage::AddrV2(reply),
)
.await;
let learned = book_len() - 1;
assert!(
learned >= MAX_ADDR_TO_SEND * 9 / 10,
"the getaddr reply is learned past one token: {learned}"
);
assert!(
seed.stop.load(Ordering::SeqCst),
"the reply ends addr-fetch"
);
assert_eq!(follow.ban_score, 0, "a getaddr reply is not misbehavior");
assert_eq!(peers.addrfetch_replies(), 1);
assert!(
relayed.try_recv().is_err(),
"an addr-fetch reply is not relayed"
);

let v1_seed = register(18461, rbitcoin_net::peers::PeerConnType::AddrFetch);
assert!(maybe_queue_addrfetch_getaddr(&out_tx, &v1_seed));
let before = book_len();
let v1 = |last: u8| {
let addr = std::net::SocketAddr::from(([12, 0, 0, last], 18444));
(time, Address::new(&addr, ServiceFlags::NETWORK))
};
let mut follow = PeerFollowState::new();
push(
&hub,
&out_tx,
&mut follow,
Some(v1_seed.as_ref()),
NetworkMessage::Addr(vec![v1(1), v1(2)]),
)
.await;
assert!(book_len() > before, "a v1 getaddr reply is learned");
assert!(
v1_seed.stop.load(Ordering::SeqCst),
"a v1 reply ends addr-fetch"
);
assert_eq!(follow.ban_score, 0, "a v1 getaddr reply is not misbehavior");
assert_eq!(peers.addrfetch_replies(), 2);
assert!(
relayed.try_recv().is_err(),
"a v1 addr-fetch reply is not relayed"
);
let _ = std::fs::remove_dir_all(dir.path());
}
80 changes: 80 additions & 0 deletions crates/rbitcoin-node/src/run.rs
Original file line number Diff line number Diff line change
Expand Up @@ -685,6 +685,26 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> {
node.user_agent().to_string(),
);
}
if proxy_seeds_queued && addrman.is_empty() {
info!("ibd: waiting for a SOCKS seed addrfetch answer…");
if let Some(book) = await_proxy_seed_book(
&node.peers,
config.network,
&shared_peers,
&shutdown,
|| node.peers.addrfetch_replies(),
PROXY_SEED_RETRY,
PROXY_SEED_WAIT,
)
.await
{
addrman = book;
info!(
"ibd: seed addrfetch filled the book (book={})",
addrman.len()
);
}
}

let max_out = config.listen.max_outbound.max(1) as usize;
let candidate_n = max_out.saturating_mul(2).clamp(16, 48);
Expand Down Expand Up @@ -1589,6 +1609,66 @@ pub fn queue_proxy_seed_addrfetch(peers: &Arc<rbitcoin_net::PeerHub>, network: N
n
}

/// A SOCKS seed answer that lost its circuit is not retried by the dial
/// queue, so the wait re-queues the seeds on this period.
const PROXY_SEED_RETRY: Duration = Duration::from_secs(60);

/// How long catch-up waits for a `--proxy` seed answer. Past it the empty
/// book falls through to the `awaiting_dialable` path.
const PROXY_SEED_WAIT: Duration = Duration::from_secs(180);

/// The shared book once an addr-fetch `getaddr` reply has been learned
/// since the wait began (`replies` counts them; a self-announce does not),
/// or `None` on shutdown or after `wait`. Re-queues the seeds every `retry`
/// while no addr-fetch session is live.
#[doc(hidden)]
pub async fn await_proxy_seed_book(
peers: &Arc<rbitcoin_net::PeerHub>,
network: Network,
shared_peers: &Arc<std::sync::Mutex<AddrMan>>,
shutdown: &Shutdown,
replies: impl Fn() -> u64,
retry: Duration,
wait: Duration,
) -> Option<AddrMan> {
let start = replies();
let deadline = tokio::time::sleep(wait);
tokio::pin!(deadline);
let mut poll = tokio::time::interval(Duration::from_millis(250));
let mut requeue = tokio::time::interval_at(tokio::time::Instant::now() + retry, retry);
loop {
tokio::select! {
biased;
_ = shutdown.cancelled() => return None,
_ = &mut deadline => {
warn!(
"ibd: no SOCKS seed addrfetch answer in {}s — catch-up stays open",
wait.as_secs()
);
return None;
}
_ = requeue.tick() => {
let answering = peers
.snapshot()
.iter()
.any(|p| p.conn_type == PeerConnType::AddrFetch);
if !answering {
let n = queue_proxy_seed_addrfetch(peers, network);
info!("ibd: no seed answer yet — re-queued {n} seed hostnames via SOCKS addrfetch");
}
}
_ = poll.tick() => {
if replies() > start {
let book = shared_peers.lock().unwrap_or_else(|e| e.into_inner());
if !book.is_empty() {
return Some(book.clone());
}
}
}
}
}
}

/// IBD horizon after `sync_cancellable` (or no peers to dial).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CatchUp {
Expand Down
Loading
Loading