From eb7f3e46941440ab8ceffb2e46a972932c0c23c0 Mon Sep 17 00:00:00 2001 From: Benjamen Keroack Date: Wed, 7 Oct 2026 16:18:18 -0600 Subject: [PATCH 1/3] p2p: apply Core's message size and count limits, and bound queued sends per peer - v1 frames are capped at Core's MAX_PROTOCOL_MESSAGE_LENGTH (4,000,000 bytes) instead of 32 MiB. The v1 and v2 receive buffers grow as payload bytes arrive, at most 256 KiB ahead of them (V1Transport::readData, V2Transport MAX_RESERVE_AHEAD), and a large v1 buffer is released after its message. - inv/getdata over MAX_INV_SZ (50,000), headers over MAX_HEADERS_RESULTS (2000) and addr/addrv2 over MAX_ADDR_TO_SEND (1000) are misbehaviour, as in Core; a notfound over 5016 entries is ignored. - A peer's queue counts its bytes. getdata is served as Core's ProcessGetData does: entries are kept in a per-peer backlog and served until the queue passes 1,000,000 bytes (-maxsendbuffer's default), and the rest as it drains. The peer's next message is not read while a getdata of its is being served or its queue is over the limit (ProcessMessages). -maxuploadtarget is charged only for blocks queued. - The write loop watches the disconnect signal during a write, and drops a peer whose socket takes no bytes for 20 minutes (TIMEOUT_INTERVAL, "socket sending timeout"). Sent bytes are counted as the socket takes them. - A version whose user agent is over MAX_SUBVERSION_LENGTH is dropped, as Core's LIMITED_STRING read does; the user agent and unknown message types are sanitized with Core's SanitizeString before they are stored or logged. - getcfilters/getcfheaders/getcfcheckpt requests that Core's PrepareBlockFilterRequest refuses now disconnect instead of being ignored. Co-Authored-By: Claude Opus 5.5 (1M context) --- node/src/net/connection.rs | 104 +++- node/src/net/connection_bounds_tests.rs | 116 ++++ node/src/net/limits.rs | 60 ++ node/src/net/manager.rs | 642 ++++++++++++++------ node/src/net/manager_p2pbounds_tests.rs | 726 +++++++++++++++++++++++ node/src/net/mod.rs | 2 + node/src/net/peer.rs | 6 +- node/src/net/send_queue.rs | 227 +++++++ node/src/net/send_queue_tests.rs | 110 ++++ node/src/net/v2transport.rs | 39 +- node/src/net/v2transport_bounds_tests.rs | 82 +++ satd/tests/p2p_limits.rs | 221 +++++++ satd/tests/regtest.rs | 49 +- 13 files changed, 2143 insertions(+), 241 deletions(-) create mode 100644 node/src/net/connection_bounds_tests.rs create mode 100644 node/src/net/limits.rs create mode 100644 node/src/net/manager_p2pbounds_tests.rs create mode 100644 node/src/net/send_queue.rs create mode 100644 node/src/net/send_queue_tests.rs create mode 100644 node/src/net/v2transport_bounds_tests.rs create mode 100644 satd/tests/p2p_limits.rs diff --git a/node/src/net/connection.rs b/node/src/net/connection.rs index bec935593..fbc4e67ef 100644 --- a/node/src/net/connection.rs +++ b/node/src/net/connection.rs @@ -6,7 +6,7 @@ use std::io; use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; -use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt, ReadBuf, ReadHalf, WriteHalf}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf, ReadHalf, WriteHalf}; use tokio::net::TcpStream; use crate::net::stats::PeerStats; @@ -225,8 +225,17 @@ pub struct V1Writer { /// Size of a P2P message header: 4 (magic) + 12 (command) + 4 (length) + 4 (checksum). const HEADER_SIZE: usize = 24; -/// Maximum message payload size (32 MB). -const MAX_PAYLOAD_SIZE: usize = 32 * 1024 * 1024; +/// Maximum message payload size: Core's `MAX_PROTOCOL_MESSAGE_LENGTH` +/// (`net.h`). A header announcing more ends the connection +/// (`V1Transport::readHeader`); satd took up to 32 MiB. +const MAX_PAYLOAD_SIZE: usize = 4_000_000; +/// How far ahead of the payload bytes actually received the receive buffer +/// is grown: Core's `V1Transport::readData` ("Allocate up to 256 KiB ahead, +/// but never more than the total message size"). +const PAYLOAD_ALLOC_STEP: usize = 256 * 1024; +/// Receive-buffer capacity kept for the next message. A buffer grown past it +/// by a larger message is released once that message is decoded. +const RECV_BUF_KEEP: usize = 64 * 1024; /// Maximum bytes to scan when resyncing to magic after stream misalignment. const MAX_RESYNC_BYTES: usize = 256 * 1024; @@ -326,15 +335,39 @@ impl V1Writer { pub async fn send(&mut self, msg: NetworkMessage) -> io::Result<()> { let raw = RawNetworkMessage::new(self.magic, msg); let bytes = serialize(&raw); - self.stream.write_all(&bytes).await?; + write_counted(&mut self.stream, &bytes, self.counters.as_ref()).await?; if let Some(c) = &self.counters { - c.record_sent(bytes.len()); c.attribute_sent(raw.cmd(), bytes.len()); } Ok(()) } } +/// Write `bytes` in full, counting each chunk as the socket takes it. +/// +/// Core moves `nSendBytes` and `m_last_send` on every partial send +/// (`SocketSendData`). Counting only whole messages made a large message that +/// a peer reads slowly look like no progress at all, and the write loop's send +/// timeout is measured against these counts. +pub(crate) async fn write_counted( + w: &mut W, + bytes: &[u8], + counters: Option<&Arc>, +) -> io::Result<()> { + let mut written = 0; + while written < bytes.len() { + let n = w.write(&bytes[written..]).await?; + if n == 0 { + return Err(io::ErrorKind::WriteZero.into()); + } + written += n; + if let Some(c) = counters { + c.record_sent(n); + } + } + Ok(()) +} + impl V1Reader { /// Receive the next network message. Skips messages that fail to deserialize. /// @@ -383,10 +416,19 @@ async fn recv_message( )); } - // Read payload - let mut payload = vec![0u8; payload_len]; - if payload_len > 0 { - stream.read_exact(&mut payload).await?; + // Read the payload into `buf`, behind the header, growing it as the + // bytes arrive and never more than PAYLOAD_ALLOC_STEP ahead of them + // (Core's `V1Transport::readData`). Sizing it from the header alone + // let a peer that announced a large payload and sent nothing hold + // that much memory per connection. + buf.clear(); + buf.extend_from_slice(&header); + let frame_len = HEADER_SIZE + payload_len; + while buf.len() < frame_len { + let start = buf.len(); + let end = frame_len.min(start + PAYLOAD_ALLOC_STEP); + buf.resize(end, 0); + stream.read_exact(&mut buf[start..end]).await?; } // Record on-wire bytes for this framed message (counted even if it @@ -399,23 +441,16 @@ async fn recv_message( c.record_recv(wire_len); } - // Combine header + payload and deserialize - buf.clear(); - buf.extend_from_slice(&header); - buf.extend_from_slice(&payload); - - match deserialize::(buf) { + let result = match deserialize::(buf) { + Ok(raw) if *raw.magic() != magic => Some(Err(io::Error::new( + io::ErrorKind::InvalidData, + "wrong network magic", + ))), Ok(raw) => { - if *raw.magic() != magic { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "wrong network magic", - )); - } if let Some(c) = counters { c.attribute_recv(raw.cmd(), wire_len); } - return Ok(raw.payload().clone()); + Some(Ok(raw.into_payload())) } Err(_) if &header[4..16] == b"pong\0\0\0\0\0\0\0\0" && payload_len < 8 => { // Too short to hold a nonce. Core still acts on it -- it @@ -424,20 +459,29 @@ async fn recv_message( if let Some(c) = counters { c.attribute_recv("pong", wire_len); } - return Ok(NetworkMessage::Unknown { + Some(Ok(NetworkMessage::Unknown { command: bitcoin::p2p::message::CommandString::try_from_static("pong") .expect("static command"), - payload, - }); + payload: buf[HEADER_SIZE..].to_vec(), + })) } Err(_) => { if let Some(c) = counters { c.attribute_recv(crate::net::stats::MSG_TYPE_OTHER, wire_len); } - let cmd = String::from_utf8_lossy(&header[4..16]); - tracing::debug!(cmd = %cmd.trim_end_matches('\0'), "Skipping unparseable message"); - continue; + // The peer's own text: sanitized as Core's + // `SanitizeString(msg_type)`, so it cannot break a log line. + let cmd = crate::net::limits::sanitize_string(&String::from_utf8_lossy(&header[4..16])); + tracing::debug!(cmd = %cmd, "Skipping unparseable message"); + None } + }; + // Kept for the next message, but not at the size of a large one. + if buf.capacity() > RECV_BUF_KEEP { + *buf = Vec::new(); + } + if let Some(result) = result { + return result; } } } @@ -500,3 +544,7 @@ async fn resync_to_magic( } } } + +#[cfg(test)] +#[path = "connection_bounds_tests.rs"] +mod bounds_tests; diff --git a/node/src/net/connection_bounds_tests.rs b/node/src/net/connection_bounds_tests.rs new file mode 100644 index 000000000..f76054c4b --- /dev/null +++ b/node/src/net/connection_bounds_tests.rs @@ -0,0 +1,116 @@ +//! The v1 frame limit, the receive buffer's growth and release, and send +//! progress, against Bitcoin Core's `V1Transport` and `SocketSendData`. + +use super::*; +use bitcoin::p2p::message::CommandString; +use std::io::Cursor; +use std::time::Duration; + +/// A reader that records the largest read it was asked for. +struct MaxAsk { + inner: R, + max: usize, +} + +impl AsyncRead for MaxAsk { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + self.max = self.max.max(buf.remaining()); + Pin::new(&mut self.inner).poll_read(cx, buf) + } +} + +fn unknown(len: usize) -> NetworkMessage { + NetworkMessage::Unknown { + command: CommandString::try_from_static("bigmsg").unwrap(), + payload: vec![0x5a; len], + } +} + +fn frame(msg: NetworkMessage) -> Vec { + serialize(&RawNetworkMessage::new(Magic::REGTEST, msg)) +} + +/// Core's `MAX_PROTOCOL_MESSAGE_LENGTH` is 4,000,000 bytes, and a header +/// announcing more ends the connection (`V1Transport::readHeader`). satd +/// took frames up to 32 MiB. +#[tokio::test] +async fn a_frame_over_four_million_bytes_is_refused() { + let mut buf = Vec::new(); + let mut ok = Cursor::new(frame(unknown(4_000_000))); + let got = recv_message(&mut ok, Magic::REGTEST, &mut buf, None).await; + assert!(matches!(got, Ok(NetworkMessage::Unknown { ref payload, .. }) if payload.len() == 4_000_000)); + + let mut over = Cursor::new(frame(unknown(4_000_001))); + let err = recv_message(&mut over, Magic::REGTEST, &mut buf, None) + .await + .expect_err("a 4,000,001-byte payload is refused"); + assert_eq!(err.kind(), io::ErrorKind::InvalidData, "{err}"); +} + +/// Core grows the receive buffer as payload bytes arrive, never more than +/// 256 KiB ahead of them (`V1Transport::readData`). satd allocated the whole +/// announced payload before reading any of it, so a peer could make it hold +/// a full frame per connection by announcing one and sending nothing. +#[tokio::test] +async fn the_payload_buffer_grows_with_the_bytes_received() { + let mut bytes = frame(unknown(4_000_000)); + bytes.truncate(HEADER_SIZE + 1000); + let mut reader = MaxAsk { inner: Cursor::new(bytes), max: 0 }; + let mut buf = Vec::new(); + let err = recv_message(&mut reader, Magic::REGTEST, &mut buf, None) + .await + .expect_err("the payload never arrives"); + assert_eq!(err.kind(), io::ErrorKind::UnexpectedEof); + assert!(reader.max <= PAYLOAD_ALLOC_STEP, "asked for {} bytes at once", reader.max); + assert!(buf.capacity() <= HEADER_SIZE + 2 * PAYLOAD_ALLOC_STEP, "holding {} bytes", buf.capacity()); +} + +/// The receive buffer is reused between messages, but not at the size of +/// the largest message the connection ever carried. +#[tokio::test] +async fn the_receive_buffer_is_released_after_a_large_message() { + let mut bytes = frame(unknown(1_000_000)); + bytes.extend(frame(NetworkMessage::Ping(7))); + let mut reader = Cursor::new(bytes); + let mut buf = Vec::new(); + let big = recv_message(&mut reader, Magic::REGTEST, &mut buf, None).await.unwrap(); + assert!(matches!(big, NetworkMessage::Unknown { .. })); + assert!(buf.capacity() <= RECV_BUF_KEEP, "kept {} bytes after a 1 MB message", buf.capacity()); + let small = recv_message(&mut reader, Magic::REGTEST, &mut buf, None).await.unwrap(); + assert_eq!(small, NetworkMessage::Ping(7)); +} + +/// Bytes are counted as the socket takes them, not once the whole message +/// is out: Core's `m_last_send` and `nSendBytes` move on every partial send +/// (`SocketSendData`), and the send timeout is measured against that. +#[tokio::test] +async fn a_send_counts_its_bytes_as_the_socket_takes_them() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (dialed, accepted) = tokio::join!( + TcpStream::connect(listener.local_addr().unwrap()), + listener.accept() + ); + // The far end never reads. + let (_far, _) = accepted.unwrap(); + let conn = Connection::with_magic(dialed.unwrap(), Magic::REGTEST); + let (_reader, mut writer) = conn.split(); + let stats = PeerStats::new(crate::net::stats::NetTotals::new()); + writer.set_counters(stats.clone()); + + let msg_len = frame(unknown(3_900_000)).len() as u64; + let mut completed = 0u64; + while let Ok(r) = tokio::time::timeout(Duration::from_millis(500), writer.send(unknown(3_900_000))).await { + r.unwrap(); + completed += msg_len; + assert!(completed < 512 * 1024 * 1024, "the socket never filled"); + } + assert!( + stats.bytes_sent() > completed, + "a send the socket only partly took is not counted: {} sent, {completed} in whole messages", + stats.bytes_sent() + ); +} diff --git a/node/src/net/limits.rs b/node/src/net/limits.rs new file mode 100644 index 000000000..283556999 --- /dev/null +++ b/node/src/net/limits.rs @@ -0,0 +1,60 @@ +//! Bitcoin Core's limits on what one P2P message may carry, and its +//! sanitizing of peer-supplied text. +//! +//! rust-bitcoin decodes a vector of any length the frame can hold +//! (`MAX_INV_SIZE` is documented as "not currently enforced"), so these are +//! applied to the decoded message, with Core's reaction to each. + +/// Core's `MAX_HEADERS_RESULTS` (`net_processing.h`): the most headers a +/// `headers` message may carry. More is `Misbehaving` ("headers message +/// size = N"). +pub const MAX_HEADERS_RESULTS: usize = 2000; + +/// Core's `MAX_ADDR_TO_SEND` (`net_processing.cpp`): the most entries an +/// `addr` or `addrv2` message may carry. More is `Misbehaving` ("addr +/// message size = N", `ProcessAddrs`). +pub const MAX_ADDR_TO_SEND: usize = 1000; + +/// The most entries of a `notfound` Core looks at: `MAX_PEER_TX_ANNOUNCEMENTS +/// + MAX_BLOCKS_IN_TRANSIT_PER_PEER` (5000 + 16, `net_processing.cpp` +/// `NOTFOUND`). A longer one is ignored whole, with no penalty. +pub const MAX_NOTFOUND_SZ: usize = 5000 + 16; + +/// Characters Core's `SanitizeString` keeps under `SAFE_CHARS_DEFAULT` +/// (`util/strencodings.cpp`): alphanumerics plus ` .,;-_/:?@()`. +fn is_safe_char(c: char) -> bool { + c.is_ascii_alphanumeric() || " .,;-_/:?@()".contains(c) +} + +/// Core's `SanitizeString(str)`: drop every character outside +/// `SAFE_CHARS_DEFAULT`. Core applies it to a peer's user agent before it +/// stores or logs it (`cleanSubVer`), and to message types in log lines, so +/// nothing a peer sends can put a line break, a control character or an +/// escape sequence into the log or into `getpeerinfo`. +pub fn sanitize_string(s: &str) -> String { + s.chars().filter(|c| is_safe_char(*c)).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Core's `SAFE_CHARS_DEFAULT`, character for character: everything it + /// lists survives, and nothing else in the byte range does. + #[test] + fn sanitize_string_keeps_exactly_cores_safe_chars() { + let core_safe: String = ('a'..='z') + .chain('A'..='Z') + .chain('0'..='9') + .chain(" .,;-_/:?@()".chars()) + .collect(); + assert_eq!(sanitize_string(&core_safe), core_safe); + for b in 0u8..=255 { + let c = char::from(b); + let kept = sanitize_string(&c.to_string()); + assert_eq!(kept.is_empty(), !core_safe.contains(c), "byte {b:#04x}"); + } + assert_eq!(sanitize_string("/Satoshi:27.0.0/\nfake log line\x1b[31m"), "/Satoshi:27.0.0/fake log line31m"); + assert_eq!(sanitize_string("caf\u{e9}"), "caf"); + } +} diff --git a/node/src/net/manager.rs b/node/src/net/manager.rs index aea389ed4..f883cd5e0 100644 --- a/node/src/net/manager.rs +++ b/node/src/net/manager.rs @@ -411,6 +411,12 @@ pub enum NetEvent { id: PeerId, msg: NetworkMessage, }, + /// The peer's queue has drained back under the send buffer while some + /// of its `getdata` entries are still unserved: serve more of them. Sent + /// by the peer's write loop; see [`crate::net::send_queue`]. + GetDataResume { + id: PeerId, + }, } /// A block handed to the block processor: the peer it came from, that peer's @@ -445,7 +451,9 @@ struct RecentBlock { /// Handle for sending messages to a specific peer. struct PeerHandle { info: PeerInfo, - msg_tx: mpsc::Sender, + /// The peer's outbound queue. It counts the bytes queued, which Core's + /// send-buffer limit is measured in; see [`crate::net::send_queue`]. + msg_tx: crate::net::send_queue::PeerSender, /// Last time we sent this peer a `getheaders`, for rate-limiting /// announcement-triggered header discovery (anti-DoS). `None` until the /// first send. @@ -479,10 +487,12 @@ struct PeerHandle { flow: Arc, } -/// A message's type as Core names it in a log line. +/// A message's type as Core names it in a log line. A type satd does not +/// know is the peer's own text, sanitized as Core's `SanitizeString(msg_type)` +/// does, so it cannot break a log line. fn handshake_msg_type(msg: &NetworkMessage) -> String { match msg { - NetworkMessage::Unknown { command, .. } => command.to_string(), + NetworkMessage::Unknown { command, .. } => crate::net::limits::sanitize_string(command.as_ref()), m => m.cmd().to_string(), } } @@ -494,15 +504,16 @@ fn log_received(id: PeerId, msg: &NetworkMessage) { if !tracing::enabled!(tracing::Level::DEBUG) { return; } - let cmd = match msg { - NetworkMessage::Unknown { command, .. } => command.to_string(), - m => m.cmd().to_string(), - }; + let cmd = handshake_msg_type(msg); let raw = bitcoin::p2p::message::RawNetworkMessage::new(bitcoin::p2p::Magic::REGTEST, msg.clone()); let payload = bitcoin::consensus::serialize(&raw).len().saturating_sub(24); tracing::debug!("received: {cmd} ({payload} bytes) peer={id}"); } +#[cfg(test)] +#[path = "manager_p2pbounds_tests.rs"] +mod p2pbounds_tests; + /// How a transaction is announced to a peer: `MSG_WTX` carrying the wtxid /// to a peer that negotiated BIP 339 wtxid relay, the txid to the rest. fn tx_announcement(wtxid_relay: bool, txid: bitcoin::Txid, wtxid: bitcoin::Wtxid) -> Inventory { @@ -2948,7 +2959,7 @@ impl PeerManager { id, PeerHandle { info, - msg_tx, + msg_tx: msg_tx.into(), disconnect: Arc::new(tokio::sync::Notify::new()), flow: Arc::new(crate::net::flow::PeerFlow::new()), last_getheaders_sent: None, @@ -3745,6 +3756,9 @@ impl PeerManager { Ok(NetEvent::PeerDisconnected { id }) => { self.handle_peer_disconnected(id); } + Ok(NetEvent::GetDataResume { id }) => { + self.resume_getdata(id); + } Ok(NetEvent::MessageReceived { id, msg }) => { // The socket task counted this message in as it was // handed over; the guard counts it out once the work @@ -4255,9 +4269,20 @@ impl PeerManager { // against that peer's outstanding ping and does not forward it. NetworkMessage::Pong(_) => {} NetworkMessage::Inv(inventory) => { + // Core's `MAX_INV_SZ` (`net_processing.cpp` INV). + if inventory.len() > MAX_INV_PER_MSG { + self.add_ban_score(id, 100, &format!("inv message size = {}", inventory.len())); + return; + } self.handle_inv(id, inventory); } NetworkMessage::Headers(headers) => { + // Core's `MAX_HEADERS_RESULTS`, checked before any header in + // the message is looked at. + if headers.len() > crate::net::limits::MAX_HEADERS_RESULTS { + self.add_ban_score(id, 100, &format!("headers message size = {}", headers.len())); + return; + } self.handle_headers(id, headers); } NetworkMessage::Block(block) => { @@ -4277,6 +4302,12 @@ impl PeerManager { } NetworkMessage::GetData(inv) => { self.handle_getdata(id, inv); + // The write loop stopped reading this peer when it handed the + // request over. Whatever is left unserved keeps it stopped; + // see `serve_getdata`. + if let Some(sender) = self.peer_sender(id) { + sender.queue().note_getdata_handled(); + } } NetworkMessage::SendCmpct(msg) => { // Only version 2 (witness) compact blocks are spoken; Core @@ -4328,6 +4359,13 @@ impl PeerManager { // Otherwise this is the message that latches the link's // addr relay on, inbound included. let relay_addrs = self.setup_address_relay(id); + // Core's `ProcessAddrs`: past `MAX_ADDR_TO_SEND` entries the + // message is misbehaviour and nothing in it is stored. A link + // that relays no addresses ignores it whatever its size. + if relay_addrs && addrs.len() > crate::net::limits::MAX_ADDR_TO_SEND { + self.add_ban_score(id, 100, &format!("addr message size = {}", addrs.len())); + return; + } let source = self.peer_ip(id); for (_, addr) in &addrs { if relay_addrs @@ -4477,6 +4515,11 @@ impl PeerManager { // As above: a block-relay-only link relays no addresses in // either direction, and anything else latches on. let relay_addrs = self.setup_address_relay(id); + // As for `addr`. + if relay_addrs && addrs.len() > crate::net::limits::MAX_ADDR_TO_SEND { + self.add_ban_score(id, 100, &format!("addrv2 message size = {}", addrs.len())); + return; + } let source = self.peer_ip(id); for addr_msg in addrs.iter().filter(|_| relay_addrs) { match &addr_msg.addr { @@ -4522,6 +4565,13 @@ impl PeerManager { self.disconnect_by_id(id); } NetworkMessage::NotFound(inventory) => { + // Core looks at a `notfound` only up to + // `MAX_PEER_TX_ANNOUNCEMENTS + MAX_BLOCKS_IN_TRANSIT_PER_PEER` + // entries, and ignores a longer one whole, without penalty. + if inventory.len() > crate::net::limits::MAX_NOTFOUND_SZ { + tracing::debug!(id, count = inventory.len(), "ignoring an oversized notfound"); + return; + } // Authoritative "I don't have this" from the peer. Until // this commit we logged at debug and did nothing, so the // height stayed in_flight for the full 60s @@ -4581,24 +4631,20 @@ impl PeerManager { handle.info.prefers_headers = true; } } + // Each request is checked as Core's `PrepareBlockFilterRequest` + // does, including whether filters are served at all. #[cfg(feature = "block-filter-index")] - NetworkMessage::GetCFilters(req) => { - if self.peer_serve_filters_ready() { - self.handle_get_cfilters(id, req); - } - // Silent drop per BIP 157 when not serving. - } + NetworkMessage::GetCFilters(req) => self.handle_get_cfilters(id, req), #[cfg(feature = "block-filter-index")] - NetworkMessage::GetCFHeaders(req) => { - if self.peer_serve_filters_ready() { - self.handle_get_cfheaders(id, req); - } - } + NetworkMessage::GetCFHeaders(req) => self.handle_get_cfheaders(id, req), #[cfg(feature = "block-filter-index")] - NetworkMessage::GetCFCheckpt(req) => { - if self.peer_serve_filters_ready() { - self.handle_get_cfcheckpt(id, req); - } + NetworkMessage::GetCFCheckpt(req) => self.handle_get_cfcheckpt(id, req), + // Built without filters: no filter type is served, and Core + // disconnects a peer that asks for one it does not serve. + #[cfg(not(feature = "block-filter-index"))] + NetworkMessage::GetCFilters(_) | NetworkMessage::GetCFHeaders(_) | NetworkMessage::GetCFCheckpt(_) => { + tracing::debug!("peer requested unsupported block filter type, disconnecting peer={id}"); + self.disconnect_by_id(id); } _ => {} } @@ -7690,42 +7736,104 @@ impl PeerManager { ); } - /// `getcfilters` handler per BIP 157. - /// Validates filter type, range bounds, and active-chain stop hash; - /// silent-drops any violation. On success, replies with one - /// `CFilter` per height in the requested range (BIP 157 specifies - /// per-height responses, not a batched form). + /// Core's `PrepareBlockFilterRequest`: the checks a BIP 157 request + /// passes before it is served. Returns the stop block's height and the + /// index to serve from. + /// + /// A request satd will not answer as asked disconnects the peer, as in + /// Core (`fDisconnect`, no misbehaviour): a filter type it does not serve + /// (any type, while it serves no filters), a stop hash it does not know + /// or would not serve, a start above the stop, or a range of + /// `max_height_diff` heights or more. satd used to ignore them all. + /// + /// A stop hash off the active chain that Core would still serve (a block + /// its `BlockRequestAllowed` admits) is ignored without a disconnect: + /// satd keeps filters for the active chain only, so it has nothing to + /// send, but the request is not one Core would hold against the peer. + #[cfg(feature = "block-filter-index")] + fn prepare_block_filter_request( + &self, + id: PeerId, + filter_type: u8, + start_height: u32, + stop_hash: bitcoin::BlockHash, + max_height_diff: u32, + ) -> Option<(u32, Arc)> { + if filter_type != node_filter_index::FILTER_TYPE_BASIC || !self.peer_serve_filters_ready() { + tracing::debug!( + "peer requested unsupported block filter type: {filter_type}, disconnecting peer={id}" + ); + self.disconnect_by_id(id); + return None; + } + let stop_height = match self.chain_state.get_block_index(&stop_hash) { + Some(e) if self.chain_state.active_chain_contains(&stop_hash, e.height) => e.height, + Some(e) if self.stale_block_request_allowed(&e) => { + tracing::debug!("no filters for block {stop_hash} off the active chain, peer={id}"); + return None; + } + _ => { + tracing::debug!("peer requested invalid block hash: {stop_hash}, disconnecting peer={id}"); + self.disconnect_by_id(id); + return None; + } + }; + if start_height > stop_height { + tracing::debug!( + "peer sent invalid getcfilters/getcfheaders with start height {start_height} and \ + stop height {stop_height}, disconnecting peer={id}" + ); + self.disconnect_by_id(id); + return None; + } + if stop_height - start_height >= max_height_diff { + tracing::debug!( + "peer requested too many cfilters/cfheaders: {} / {max_height_diff}, disconnecting peer={id}", + u64::from(stop_height - start_height) + 1 + ); + self.disconnect_by_id(id); + return None; + } + // Core: "Filter index for supported type not found", no disconnect. + self.filter_index.get().cloned().map(|idx| (stop_height, idx)) + } + + /// The half of Core's `BlockRequestAllowed` that admits a block off the + /// active chain: one this node validated, no more than + /// `STALE_RELAY_AGE_LIMIT` (30 days) older than the best header. + /// `BlockStatus::Valid` is written only by `connect_block` and survives + /// the reorg that takes the block off the active chain, as Core's + /// `BLOCK_VALID_SCRIPTS` does. Core also bounds the block's + /// work-equivalent age; that bound is not applied, so this admits a little + /// more than Core does. + #[cfg(feature = "block-filter-index")] + fn stale_block_request_allowed(&self, entry: &crate::storage::blockindex::BlockIndexEntry) -> bool { + const STALE_RELAY_AGE_LIMIT: i64 = 30 * 24 * 60 * 60; + if entry.status != crate::storage::blockindex::BlockStatus::Valid { + return false; + } + self.chain_state + .get_block_index(&self.chain_state.best_header_hash()) + .is_some_and(|best| { + i64::from(best.header.time) - i64::from(entry.header.time) < STALE_RELAY_AGE_LIMIT + }) + } + + /// `getcfilters` handler per BIP 157, checked by + /// [`Self::prepare_block_filter_request`]. Replies with one `CFilter` per + /// height in the requested range (BIP 157 specifies per-height + /// responses, not a batched form). #[cfg(feature = "block-filter-index")] fn handle_get_cfilters(&self, id: PeerId, req: bitcoin::p2p::message_filter::GetCFilters) { use crate::index::filter::lookups::MAX_GETCFILTERS_SIZE; - use node_filter_index::FILTER_TYPE_BASIC; let bitcoin::p2p::message_filter::GetCFilters { filter_type, start_height, stop_hash, } = req; - // Filter type guard. - if filter_type != FILTER_TYPE_BASIC { - return; - } - // Resolve stop_hash → stop_height via block_index. - let Some(stop_entry) = self.chain_state.get_block_index(&stop_hash) else { - return; - }; - let stop_height = stop_entry.height; - // BIP 157: stop_height ≥ start_height, range < 1000. - if stop_height < start_height { - return; - } - if stop_height - start_height >= MAX_GETCFILTERS_SIZE { - return; - } - // Active-chain check: stop_hash must match the active chain at stop_height. - match self.chain_state.get_block_hash_by_height(stop_height) { - Some(h) if h == stop_hash => {} - _ => return, - } - let Some(idx) = self.filter_index.get() else { + let Some((stop_height, idx)) = + self.prepare_block_filter_request(id, filter_type, start_height, stop_hash, MAX_GETCFILTERS_SIZE) + else { return; }; // Stream responses via an async task with backpressure. The @@ -7743,7 +7851,6 @@ impl PeerManager { return; }; let chain_state = self.chain_state.clone(); - let idx = idx.clone(); tokio::spawn(async move { for h in start_height..=stop_height { let Some(block_hash) = chain_state.get_block_hash_by_height(h) else { @@ -7769,7 +7876,8 @@ impl PeerManager { }); } - /// `getcfheaders` handler per BIP 157. + /// `getcfheaders` handler per BIP 157, checked by + /// [`Self::prepare_block_filter_request`]. /// Replies with a single `CFHeaders` carrying /// `previous_filter_header` plus per-height filter hashes (computed /// on the fly from the stored filter blob — see plan §"Filter-hash @@ -7780,32 +7888,20 @@ impl PeerManager { use crate::index::filter::lookups::MAX_GETCFHEADERS_SIZE; use bitcoin::bip158::FilterHash; use bitcoin::hashes::Hash; - use node_filter_index::FILTER_TYPE_BASIC; let bitcoin::p2p::message_filter::GetCFHeaders { filter_type, start_height, stop_hash, } = req; - if filter_type != FILTER_TYPE_BASIC { - return; - } - let Some(stop_entry) = self.chain_state.get_block_index(&stop_hash) else { - return; - }; - let stop_height = stop_entry.height; - if stop_height < start_height { - return; - } // Bitcoin Core / BIP 157 cap getcfheaders at 2000, not the 1000 // that applies to getcfilters. Review 2026-05-04 M1. - if stop_height - start_height >= MAX_GETCFHEADERS_SIZE { - return; - } - match self.chain_state.get_block_hash_by_height(stop_height) { - Some(h) if h == stop_hash => {} - _ => return, - } - let Some(idx) = self.filter_index.get() else { + let Some((stop_height, idx)) = self.prepare_block_filter_request( + id, + filter_type, + start_height, + stop_hash, + MAX_GETCFHEADERS_SIZE, + ) else { return; }; // previous_filter_header: header at start_height - 1, or all-zeros for height 0. @@ -7839,27 +7935,19 @@ impl PeerManager { /// `getcfcheckpt` handler per BIP 157 — filter headers at every /// 1000-block boundary up to (and including) the highest 1000-block - /// boundary ≤ `stop_height`. + /// boundary ≤ `stop_height`. Checked by + /// [`Self::prepare_block_filter_request`] with no range limit, as Core + /// does. #[cfg(feature = "block-filter-index")] fn handle_get_cfcheckpt(&self, id: PeerId, req: bitcoin::p2p::message_filter::GetCFCheckpt) { use bitcoin::hashes::Hash; - use node_filter_index::FILTER_TYPE_BASIC; let bitcoin::p2p::message_filter::GetCFCheckpt { filter_type, stop_hash, } = req; - if filter_type != FILTER_TYPE_BASIC { - return; - } - let Some(stop_entry) = self.chain_state.get_block_index(&stop_hash) else { - return; - }; - let stop_height = stop_entry.height; - match self.chain_state.get_block_hash_by_height(stop_height) { - Some(h) if h == stop_hash => {} - _ => return, - } - let Some(idx) = self.filter_index.get() else { + let Some((stop_height, idx)) = + self.prepare_block_filter_request(id, filter_type, 0, stop_hash, u32::MAX) + else { return; }; let max_idx = stop_height / 1000; @@ -8023,91 +8111,164 @@ impl PeerManager { } fn handle_getdata(&self, id: PeerId, inventory: Vec) { + // Core's `MAX_INV_SZ` (`net_processing.cpp` GETDATA), checked before + // anything in the request is looked at. + if inventory.len() > MAX_INV_PER_MSG { + self.add_ban_score(id, 100, &format!("getdata message size = {}", inventory.len())); + return; + } // Core's `ProcessMessage` line for every getdata. match inventory.as_slice() { [inv] => tracing::debug!("received getdata for: {} peer={id}", inv_to_string(inv)), invs => tracing::debug!("received getdata ({} invsz) peer={id}", invs.len()), } + let Some(sender) = self.peer_sender(id) else { + return; + }; + // Queued behind anything still unserved and served from there: + // Core's `m_getdata_requests` and `ProcessGetData`. + sender.queue().push_getdata(inventory); + self.serve_getdata(id, &sender); + } + + /// [`NetEvent::GetDataResume`]: the peer's queue has drained enough to + /// serve more of its `getdata` backlog. + fn resume_getdata(&self, id: PeerId) { + let Some(sender) = self.peer_sender(id) else { + return; + }; + // Before looking at the queue: a write that drains it after this + // point asks again rather than finding a request already waiting. + sender.queue().resume_taken(); + self.serve_getdata(id, &sender); + } + + /// The sending end of a peer's queue, while the peer is connected. + fn peer_sender(&self, id: PeerId) -> Option { + self.peers.read().get(&id).map(|h| h.msg_tx.clone()) + } + + /// Core's `ProcessGetData`: serve the peer's unserved `getdata` entries, + /// oldest first, until none are left or its queue is past the send + /// buffer (`fPauseSend`). + /// + /// The rest waits for the queue to drain. The peer's write loop asks for + /// more as it does ([`NetEvent::GetDataResume`]), and takes no other + /// message from the peer until the backlog is empty, which keeps the + /// answers in the order asked and the backlog to one request. + /// + /// satd served every entry at once, reading and deserializing each block + /// and queueing it with a `try_send` whose failure it ignored. One + /// request could hold a peer's whole queue of blocks in memory for as + /// long as the peer did not read them, and the blocks that did not fit + /// were read, dropped, and still charged to `-maxuploadtarget`. + fn serve_getdata(&self, id: PeerId, sender: &crate::net::send_queue::PeerSender) { + let queue = sender.queue(); let mut not_found = Vec::new(); - for inv in inventory { - match inv { - // `MSG_CMPCT_BLOCK` (BIP 152). A block within - // `MAX_CMPCTBLOCK_DEPTH` of the tip is answered with a - // `cmpctblock` (`serve_compact_block`). Anything deeper, or - // anything while we are still syncing, gets the full `block`, - // which BIP 152 permits and Core itself sends; Core accepts - // it against its in-flight compact request. Letting the - // request fall through to `_ => {}` instead would silently - // drop it, and a Core peer never re-requests. - Inventory::Block(hash) - | Inventory::WitnessBlock(hash) - | Inventory::CompactBlock(hash) => { - if matches!(inv, Inventory::CompactBlock(_)) && self.serve_compact_block(id, &hash) { - continue; - } - if let Some(block) = self.chain_state.get_block(&hash) { - // -maxuploadtarget: decline historical blocks once - // the rolling budget is spent (download/noban peers - // and recent blocks are exempt). - if !self.upload_permits_block(id, &block) { - tracing::debug!(id, %hash, "maxuploadtarget reached; declining historical block"); - not_found.push(inv); - continue; - } - self.record_upload(block.total_size() as u64); - self.send_to_peer(id, NetworkMessage::Block(block)); - } else { - not_found.push(inv); - } + while !sender.paused() { + let Some(inv) = queue.front_getdata() else { + break; + }; + if !self.serve_getdata_entry(id, sender, inv, &mut not_found) { + // The queue filled after the check above, or the peer is + // gone. The entry stays first; the write loop asks again once + // there is room. + break; + } + queue.pop_getdata(); + } + if !not_found.is_empty() { + let _ = sender.try_send(NetworkMessage::NotFound(not_found)); + } + if queue.getdata_backlog() == 0 { + queue.wake_reader(); + } + } + + /// Answer one `getdata` entry: queue what it asks for, or add it to the + /// `notfound`. False when nothing could be queued and the entry is still + /// to be served. + fn serve_getdata_entry( + &self, + id: PeerId, + sender: &crate::net::send_queue::PeerSender, + inv: Inventory, + not_found: &mut Vec, + ) -> bool { + match inv { + // `MSG_CMPCT_BLOCK` (BIP 152). A block within + // `MAX_CMPCTBLOCK_DEPTH` of the tip is answered with a + // `cmpctblock` (`serve_compact_block`). Anything deeper, or + // anything while we are still syncing, gets the full `block`, + // which BIP 152 permits and Core itself sends; Core accepts + // it against its in-flight compact request. Letting the + // request fall through to `_ => {}` instead would silently + // drop it, and a Core peer never re-requests. + Inventory::Block(hash) + | Inventory::WitnessBlock(hash) + | Inventory::CompactBlock(hash) => { + if matches!(inv, Inventory::CompactBlock(_)) && self.serve_compact_block(id, &hash) { + return true; + } + let Some(block) = self.chain_state.get_block(&hash) else { + not_found.push(inv); + return true; + }; + // -maxuploadtarget: decline historical blocks once + // the rolling budget is spent (download/noban peers + // and recent blocks are exempt). + if !self.upload_permits_block(id, &block) { + tracing::debug!(id, %hash, "maxuploadtarget reached; declining historical block"); + not_found.push(inv); + return true; + } + let size = block.total_size() as u64; + if sender.try_send(NetworkMessage::Block(block)).is_err() { + return false; } - Inventory::Transaction(txid) | Inventory::WitnessTransaction(txid) => { - // Relay-quarantined txs are not served via `getdata`: - // announce-suppression without serve-suppression would be - // incoherent — a peer that learned the txid elsewhere must - // not be able to pull it from us (design §6.1). Report it - // as not-found, exactly as if we did not hold it. - if let Some(entry) = self - .mempool - .get(&txid) - .filter(|e| e.scope.assists_relay()) - { - // The peer pulled this tx from us. If it's a pending - // local broadcast, that's the primary proof it has - // propagated — count the peer toward stopping - // rebroadcast. Only if the Tx was actually enqueued: - // a try_send drop (slow peer, full channel) means the - // peer never received it, and retiring the tx on a - // dropped send would defeat the exact failure mode - // rebroadcast exists to cover. - if self.send_to_peer(id, NetworkMessage::Tx(entry.tx)) { - self.note_broadcast_witness(id, txid); - } - } else { - not_found.push(inv); - } + // Charged for a block that was queued, not one that was read. + self.record_upload(size); + } + Inventory::Transaction(txid) | Inventory::WitnessTransaction(txid) => { + // Relay-quarantined txs are not served via `getdata`: + // announce-suppression without serve-suppression would be + // incoherent — a peer that learned the txid elsewhere must + // not be able to pull it from us (design §6.1). Report it + // as not-found, exactly as if we did not hold it. + let Some(entry) = self.mempool.get(&txid).filter(|e| e.scope.assists_relay()) else { + not_found.push(inv); + return true; + }; + // The peer pulled this tx from us. If it's a pending + // local broadcast, that's the primary proof it has + // propagated — count the peer toward stopping + // rebroadcast. Only once the Tx is actually enqueued: + // retiring the tx on a send that did not happen would + // defeat the exact failure mode rebroadcast exists to cover. + if sender.try_send(NetworkMessage::Tx(entry.tx)).is_err() { + return false; } - // BIP 339: the same, looked up by wtxid. A resident - // transaction with the same txid and another witness is not - // the one asked for, so the answer is `notfound`. - Inventory::WTx(wtxid) => { - if let Some((txid, entry)) = self - .mempool - .get_by_wtxid(&wtxid) - .filter(|(_, e)| e.scope.assists_relay()) - { - if self.send_to_peer(id, NetworkMessage::Tx(entry.tx)) { - self.note_broadcast_witness(id, txid); - } - } else { - not_found.push(inv); - } + self.note_broadcast_witness(id, txid); + } + // BIP 339: the same, looked up by wtxid. A resident + // transaction with the same txid and another witness is not + // the one asked for, so the answer is `notfound`. + Inventory::WTx(wtxid) => { + let Some((txid, entry)) = + self.mempool.get_by_wtxid(&wtxid).filter(|(_, e)| e.scope.assists_relay()) + else { + not_found.push(inv); + return true; + }; + if sender.try_send(NetworkMessage::Tx(entry.tx)).is_err() { + return false; } - _ => {} + self.note_broadcast_witness(id, txid); } + // An entry of a type that is not served is skipped, as in Core. + _ => {} } - if !not_found.is_empty() { - self.send_to_peer(id, NetworkMessage::NotFound(not_found)); - } + true } /// Accept the header a `cmpctblock` carries, the way a `headers` message @@ -9207,7 +9368,7 @@ impl PeerManager { }; let handle = PeerHandle { info, - msg_tx, + msg_tx: msg_tx.into(), disconnect: Arc::new(tokio::sync::Notify::new()), flow: Arc::new(crate::net::flow::PeerFlow::new()), last_getheaders_sent: None, @@ -9577,13 +9738,16 @@ impl PeerManager { // the connection is torn down anyway. // Also the ping accounting the write loop needs below, so the map is // read once rather than again just before the loop starts. - let (ping_stats, peer_flow, disconnect_signal) = { + let (ping_stats, peer_flow, disconnect_signal, send_queue) = { let peers = self.peers.read(); match peers.get(&id) { - Some(h) => { - (Some(h.stats.clone()), Some(h.flow.clone()), Some(h.disconnect.clone())) - } - None => (None, None, None), + Some(h) => ( + Some(h.stats.clone()), + Some(h.flow.clone()), + Some(h.disconnect.clone()), + Some(h.msg_tx.queue().clone()), + ), + None => (None, None, None, None), } }; if let Some(stats) = &ping_stats { @@ -9732,6 +9896,7 @@ impl PeerManager { peer_flow, Some(self.drain_now.clone()), disconnect_signal, + send_queue, ) .await; @@ -9786,6 +9951,14 @@ impl PeerManager { /// terminate instead of leaving an untracked peer feeding events. /// The earlier `Some(msg) = msg_rx.recv()` pattern silently /// disabled the branch on close — review F2 (PRs #180-#184). + /// - the disconnect signal, or a write that makes no progress for + /// [`crate::net::send_queue::SEND_TIMEOUT`] → exit, even with the + /// write unfinished (`send_watched`). + /// + /// Flow control, after Core's send buffer (`send_queue`): the bytes of + /// every message written are taken off the peer's queue count; while the + /// count is over the limit, or a `getdata` of the peer's is being served, + /// the peer's next message is left unread. #[allow(clippy::too_many_arguments)] async fn peer_write_loop( id: PeerId, @@ -9797,6 +9970,7 @@ impl PeerManager { flow: Option>, drain_now: Option>, disconnect_signal: Option>, + send_queue: Option>, ) -> Result<(), String> { // Bitcoin Core's keepalive cadence (`PING_INTERVAL`, net_processing.h). // The first tick of a tokio interval fires immediately, which is also @@ -9827,6 +10001,12 @@ impl PeerManager { tokio::pin!(disconnect_requested); loop { + // Core's `ProcessMessages` takes nothing more from a peer while a + // `getdata` of its is being served, or while its send buffer is + // full ("Don't bother if send buffer is too full to respond + // anyway"). The message stays in the peer's read queue, and the + // reader task stops reading the socket once that is full. + let reading_paused = send_queue.as_ref().is_some_and(|q| q.reading_paused()); tokio::select! { _ = &mut disconnect_requested => { // Explicit teardown, as opposed to noticing `msg_rx` @@ -9834,6 +10014,14 @@ impl PeerManager { // and closes the socket. return Err("disconnected by manager".to_string()); } + // Woken when the manager has served a `getdata`, to look + // again. The writes below re-check after every message. + _ = async { + match &send_queue { + Some(q) => q.reader_woken().await, + None => std::future::pending::<()>().await, + } + }, if reading_paused => {} _ = ping_timer.tick() => { // A peer that keeps talking but never answers a ping is // dropped here. Returning takes the ordinary teardown @@ -9880,13 +10068,17 @@ impl PeerManager { // it must never go on the wire. let nonce = rand::random::().max(1); stats.ping_sent(nonce); - writer - .send(NetworkMessage::Ping(nonce)) - .await - .map_err(|e| e.to_string())?; + Self::send_watched( + id, + writer, + NetworkMessage::Ping(nonce), + Some(stats), + disconnect_requested.as_mut(), + ) + .await?; } } - msg = read_rx.recv() => { + msg = read_rx.recv(), if !reading_paused => { match msg { // A pong is matched here, on the peer's own task, // rather than forwarded to the manager loop. @@ -9974,10 +10166,14 @@ impl PeerManager { ); } } - writer - .send(NetworkMessage::Pong(nonce)) - .await - .map_err(|e| e.to_string())?; + Self::send_watched( + id, + writer, + NetworkMessage::Pong(nonce), + stats.as_ref(), + disconnect_requested.as_mut(), + ) + .await?; } Some(msg) => { // Counted in here and out wherever the work ends @@ -9987,6 +10183,11 @@ impl PeerManager { if let Some(flow) = &flow { flow.queued(); } + // Nothing more is read until the manager has + // served this request (`reading_paused`). + if let (NetworkMessage::GetData(_), Some(q)) = (&msg, &send_queue) { + q.note_getdata_forwarded(); + } event_tx .send(NetEvent::MessageReceived { id, msg }) .await @@ -10017,7 +10218,23 @@ impl PeerManager { if let (NetworkMessage::Ping(nonce), Some(stats)) = (&msg, &stats) { stats.ping_sent(*nonce); } - writer.send(msg).await.map_err(|e| e.to_string())?; + // What the sender counted the message at. + let size = send_queue + .as_ref() + .map(|_| crate::net::send_queue::queued_size(&msg)); + Self::send_watched(id, writer, msg, stats.as_ref(), disconnect_requested.as_mut()) + .await?; + if let (Some(q), Some(size)) = (&send_queue, size) { + q.sent(size); + // Back under the send buffer with `getdata` + // entries waiting: ask the manager for more. + if q.take_resume() { + event_tx + .send(NetEvent::GetDataResume { id }) + .await + .map_err(|e| e.to_string())?; + } + } } None => { // Manager dropped our handle. Return so the @@ -10034,6 +10251,60 @@ impl PeerManager { } } + /// Write one message to the peer, giving up if the manager drops the peer + /// or the socket takes no bytes for + /// [`crate::net::send_queue::SEND_TIMEOUT`]. + /// + /// A peer that stops reading leaves a write parked for as long as it + /// likes, and the write loop looked at the disconnect signal and the ping + /// deadline only between writes. So `disconnectnode`, a ban and the ping + /// timeout could not reach such a peer, and its socket and everything + /// queued to it stayed. Core drops a peer that has taken no bytes for + /// `TIMEOUT_INTERVAL` (`InactivityCheck`, "socket sending timeout"). + /// Progress is the bytes the socket takes, counted as it takes them, so a + /// slow reader is not mistaken for a stalled one. + async fn send_watched>( + id: PeerId, + writer: &mut ConnectionWriter, + msg: NetworkMessage, + stats: Option<&Arc>, + mut disconnect: std::pin::Pin<&mut F>, + ) -> Result<(), String> { + use crate::net::send_queue::SEND_TIMEOUT; + /// How often a parked write is checked for progress. + const PROGRESS_CHECK: Duration = Duration::from_secs(10); + + let send = writer.send(msg); + tokio::pin!(send); + // Most writes finish at once; only one that has to wait is watched. + tokio::select! { + biased; + res = &mut send => return res.map_err(|e| e.to_string()), + _ = std::future::ready(()) => {} + } + let mut check = tokio::time::interval(PROGRESS_CHECK); + check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut sent = stats.map(|s| s.bytes_sent()); + let mut progress_at = tokio::time::Instant::now(); + loop { + tokio::select! { + res = &mut send => return res.map_err(|e| e.to_string()), + _ = disconnect.as_mut() => return Err("disconnected by manager".to_string()), + _ = check.tick() => { + let now_sent = stats.map(|s| s.bytes_sent()); + if now_sent != sent { + sent = now_sent; + progress_at = tokio::time::Instant::now(); + } else if progress_at.elapsed() >= SEND_TIMEOUT { + let secs = progress_at.elapsed().as_secs(); + tracing::debug!("socket sending timeout: {secs}s, disconnecting peer={id}"); + return Err(format!("socket sending timeout: no bytes taken in {secs}s")); + } + } + } + } + } + /// Receive the next handshake message, dropping the peer when Core's /// `InactivityCheck` would: `-peertimeout` seconds after it connected, on /// the node clock, with the handshake still incomplete. @@ -10144,6 +10415,19 @@ impl PeerManager { // with a line naming it. let their_version = loop { match self.recv_handshake(id, conn, connected_at).await? { + // Core reads the user agent as `LIMITED_STRING(strSubVer, + // MAX_SUBVERSION_LENGTH)`. A longer one throws in + // deserialization: the `version` is dropped, and the peer is + // left without a handshake until a well-formed one arrives or + // the connect timeout ends it (`recv_handshake` applies it + // here too). + NetworkMessage::Version(v) if v.user_agent.len() > crate::MAX_SUBVERSION_LENGTH => { + tracing::debug!( + "ignoring version message with a {}-byte user agent (limit {}) from peer={id}", + v.user_agent.len(), + crate::MAX_SUBVERSION_LENGTH + ) + } NetworkMessage::Version(v) => break v, other => tracing::debug!( "non-version message before version handshake. Message \"{}\" from peer={id}", @@ -10867,7 +11151,7 @@ mod tests { let (tx, _rx) = mpsc::channel::(1); PeerHandle { info, - msg_tx: tx, + msg_tx: tx.into(), disconnect: Arc::new(tokio::sync::Notify::new()), flow: Arc::new(crate::net::flow::PeerFlow::new()), last_getheaders_sent: None, @@ -11069,7 +11353,7 @@ mod tests { ( PeerHandle { info, - msg_tx: tx, + msg_tx: tx.into(), disconnect: Arc::new(tokio::sync::Notify::new()), flow: Arc::new(crate::net::flow::PeerFlow::new()), last_getheaders_sent: None, @@ -13487,7 +13771,7 @@ mod tests { 9, PeerHandle { info, - msg_tx: tx, + msg_tx: tx.into(), disconnect: Arc::new(tokio::sync::Notify::new()), flow: Arc::new(crate::net::flow::PeerFlow::new()), last_getheaders_sent: None, diff --git a/node/src/net/manager_p2pbounds_tests.rs b/node/src/net/manager_p2pbounds_tests.rs new file mode 100644 index 000000000..e14865ac6 --- /dev/null +++ b/node/src/net/manager_p2pbounds_tests.rs @@ -0,0 +1,726 @@ +//! Core's per-message limits, the send-buffer bound on `getdata`, and the +//! BIP 157 request checks, against a live `PeerManager`. + +use super::*; +use crate::net::flow::InFlight; +use crate::net::send_queue::{MAX_SEND_BUFFER_BYTES, queued_size}; + +fn mk_pm() -> (Arc, tempfile::TempDir) { + use crate::chain::state::AssumeValid; + use crate::storage::db::InMemoryStore; + use crate::storage::flatfile::FlatFileManager; + use crate::validation::script::NoopVerifier; + + let dir = tempfile::TempDir::new().unwrap(); + let store = Box::new(InMemoryStore::new()); + let flat_files = FlatFileManager::new(&dir.path().join("blocks")).unwrap(); + let chain_state = Arc::new( + ChainState::new( + store, + flat_files, + Network::Regtest, + Box::new(NoopVerifier), + AssumeValid::Disabled, + 450, + 4, + Default::default(), + Default::default(), + Default::default(), + ) + .unwrap(), + ); + let mempool = Arc::new(Mempool::new(1_000_000, 0)); + let fee_estimator = Arc::new(FeeEstimator::new()); + let (_shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false); + let pm = PeerManager::new(chain_state, mempool, fee_estimator, Network::Regtest, shutdown_rx); + (pm, dir) +} + +/// A connected peer whose queue the test reads, `slots` messages deep. +fn add_peer(pm: &PeerManager, id: PeerId, addr: SocketAddr, slots: usize) -> mpsc::Receiver { + let mut info = PeerInfo::new(id, addr, Direction::Inbound); + info.state = PeerState::Connected; + let (tx, rx) = mpsc::channel::(slots); + pm.peers.write().insert( + id, + PeerHandle { + info, + msg_tx: tx.into(), + disconnect: Arc::new(tokio::sync::Notify::new()), + flow: Arc::new(crate::net::flow::PeerFlow::new()), + last_getheaders_sent: None, + last_mempool_served: None, + fee_filter_sent: None, + stats: PeerStats::new(NetTotals::new()), + }, + ); + rx +} + +fn addr(last: u8) -> SocketAddr { + SocketAddr::from(([10, 61, 0, last], 8333)) +} + +fn deliver(pm: &PeerManager, id: PeerId, msg: NetworkMessage) { + pm.handle_message(id, msg, InFlight::adopt(None)); +} + +fn connected(pm: &PeerManager, id: PeerId) -> bool { + pm.peers.read().contains_key(&id) +} + +/// Mine `n` blocks of roughly `size` bytes each (the coinbase carries an +/// OP_RETURN of that size) and return their hashes in height order. +fn mine_big(pm: &PeerManager, n: usize, size: usize) -> Vec { + let mempool = Mempool::new(1_000_000, 0); + (0..n) + .map(|i| { + let data = bitcoin::script::PushBytesBuf::try_from(vec![i as u8; size]).unwrap(); + let block = crate::mining::miner::build_block_to_script( + &pm.chain_state, + &mempool, + bitcoin::ScriptBuf::new_op_return(&data), + None, + ) + .expect("mine regtest block"); + pm.chain_state.accept_block(&block).expect("accept block"); + block.block_hash() + }) + .collect() +} + +fn drain(rx: &mut mpsc::Receiver) -> Vec { + std::iter::from_fn(|| rx.try_recv().ok()).collect() +} + +// ---- Message count limits ---- + +/// Core's `MAX_INV_SZ`: an `inv` or `getdata` of more than 50,000 entries is +/// `Misbehaving` (`net_processing.cpp` INV / GETDATA). 50,000 is fine. +#[test] +fn inv_and_getdata_past_max_inv_sz_are_misbehaviour() { + let (pm, _dir) = mk_pm(); + let entry = Inventory::Unknown { inv_type: 99, hash: [7u8; 32] }; + for (i, (make, name)) in [ + (NetworkMessage::Inv as fn(Vec) -> NetworkMessage, "inv"), + (NetworkMessage::GetData as fn(Vec) -> NetworkMessage, "getdata"), + ] + .into_iter() + .enumerate() + { + let (ok, over) = (2 * i as u8 + 1, 2 * i as u8 + 2); + let _rx_ok = add_peer(&pm, ok.into(), addr(ok), 8); + deliver(&pm, ok.into(), make(vec![entry; MAX_INV_PER_MSG])); + assert!(connected(&pm, ok.into()), "{name} of {MAX_INV_PER_MSG} entries is allowed"); + + let _rx_over = add_peer(&pm, over.into(), addr(over), 8); + deliver(&pm, over.into(), make(vec![entry; MAX_INV_PER_MSG + 1])); + assert!(!connected(&pm, over.into()), "{name} of {} entries disconnects", MAX_INV_PER_MSG + 1); + assert!(pm.is_addr_banned(&addr(over)), "{name} past MAX_INV_SZ is misbehaviour"); + } +} + +/// Core's `MAX_HEADERS_RESULTS`: a `headers` message of more than 2000 is +/// `Misbehaving`, before any header in it is looked at. +#[test] +fn headers_past_max_headers_results_are_misbehaviour() { + let (pm, _dir) = mk_pm(); + let genesis = pm.chain_state.get_block_index(&pm.chain_state.tip_hash()).unwrap().header; + + let _rx = add_peer(&pm, 1, addr(1), 8); + deliver(&pm, 1, NetworkMessage::Headers(vec![genesis; 2000])); + assert!(connected(&pm, 1), "2000 headers are allowed"); + + let _rx = add_peer(&pm, 2, addr(2), 8); + deliver(&pm, 2, NetworkMessage::Headers(vec![genesis; 2001])); + assert!(!connected(&pm, 2), "2001 headers disconnect"); + assert!(pm.is_addr_banned(&addr(2)), "2001 headers are misbehaviour"); +} + +/// Core's `MAX_ADDR_TO_SEND`: an `addr` or `addrv2` of more than 1000 entries +/// is `Misbehaving` (`ProcessAddrs`), and none of it is stored. +#[test] +fn addr_and_addrv2_past_max_addr_to_send_are_misbehaviour() { + use bitcoin::p2p::address::{AddrV2, AddrV2Message}; + let (pm, _dir) = mk_pm(); + let v1 = |n: usize| { + NetworkMessage::Addr( + (0..n) + .map(|i| { + let a = SocketAddr::from(([44, 1, (i / 256) as u8, (i % 256) as u8], 8333)); + (1_700_000_000, Address::new(&a, ServiceFlags::NETWORK)) + }) + .collect(), + ) + }; + let v2 = |n: usize| { + NetworkMessage::AddrV2( + (0..n) + .map(|i| AddrV2Message { + time: 1_700_000_000, + services: ServiceFlags::NETWORK, + addr: AddrV2::Ipv4(std::net::Ipv4Addr::new(45, 1, (i / 256) as u8, (i % 256) as u8)), + port: 8333, + }) + .collect(), + ) + }; + for (i, make) in [&v1 as &dyn Fn(usize) -> NetworkMessage, &v2].into_iter().enumerate() { + let (ok, over) = (2 * i as u8 + 1, 2 * i as u8 + 2); + let _rx = add_peer(&pm, ok.into(), addr(ok), 8); + deliver(&pm, ok.into(), make(1000)); + assert!(connected(&pm, ok.into()), "case {i}: 1000 addresses are allowed"); + + let _rx = add_peer(&pm, over.into(), addr(over), 8); + deliver(&pm, over.into(), make(1001)); + assert!(!connected(&pm, over.into()), "case {i}: 1001 addresses disconnect"); + assert!(pm.is_addr_banned(&addr(over)), "case {i}: 1001 addresses are misbehaviour"); + } +} + +/// Core looks at a `notfound` only up to `MAX_PEER_TX_ANNOUNCEMENTS + +/// MAX_BLOCKS_IN_TRANSIT_PER_PEER` entries and ignores a longer one whole, +/// with no penalty. satd acts on the block entries of a `notfound` (it +/// releases those heights for another peer), so a longer one releases +/// nothing. +#[test] +fn notfound_past_the_limit_is_ignored() { + use crate::net::limits::MAX_NOTFOUND_SZ; + let (pm, _dir) = mk_pm(); + let hashes = mine_big(&pm, 4, 10); + let filler = Inventory::Unknown { inv_type: 99, hash: [7u8; 32] }; + let notfound = |len: usize| { + let mut inv: Vec = hashes.iter().map(|h| Inventory::WitnessBlock(*h)).collect(); + inv.resize(len, filler); + NetworkMessage::NotFound(inv) + }; + for (len, released) in [(MAX_NOTFOUND_SZ, true), (MAX_NOTFOUND_SZ + 1, false)] { + let _rx = add_peer(&pm, 1, addr(1), 8); + let mut sched = IbdScheduler::new(4, 0, &pm.chain_state, 1024); + let assigned = sched.assign_blocks(1); + assert_eq!(assigned.len(), 4, "every height in flight to peer 1"); + *pm.ibd.write() = Some(sched); + + deliver(&pm, 1, notfound(len)); + let ibd = pm.ibd.read(); + let still = (1..=4).filter(|h| ibd.as_ref().unwrap().in_flight_contains(*h)).count(); + assert_eq!(still == 0, released, "notfound of {len} entries: {still} heights still in flight"); + drop(ibd); + assert!(connected(&pm, 1), "notfound of {len} entries is not misbehaviour"); + *pm.ibd.write() = None; + } +} + +// ---- getdata against the send buffer ---- + +/// Core stops serving a `getdata` once the bytes queued to the peer pass +/// `-maxsendbuffer` (`ProcessGetData` breaks on `fPauseSend`). satd read, +/// deserialized and queued every block asked for, so one request held a +/// peer's whole queue of blocks in memory. +#[test] +fn getdata_stops_serving_past_the_send_buffer() { + let (pm, _dir) = mk_pm(); + let hashes = mine_big(&pm, 30, 100_000); + let mut rx = add_peer(&pm, 1, addr(1), 256); + + pm.handle_getdata(1, hashes.iter().map(|h| Inventory::WitnessBlock(*h)).collect()); + let queued = drain(&mut rx); + let bytes: usize = queued.iter().map(queued_size).sum(); + let largest = queued.iter().map(queued_size).max().unwrap_or(0); + assert!( + bytes <= MAX_SEND_BUFFER_BYTES + largest, + "{} blocks, {bytes} bytes queued at once; the bound is {MAX_SEND_BUFFER_BYTES} plus one block", + queued.len() + ); + assert!(!queued.is_empty(), "the first blocks are served at once"); +} + +/// What the bound holds back is not lost: as the peer's queue drains, the +/// rest of the request is served, in the order asked, and the peer's next +/// message is read only once all of it has been. Each block is charged to +/// `-maxuploadtarget` once, as it is queued. +#[test] +fn getdata_resumes_as_the_queue_drains() { + let (pm, _dir) = mk_pm(); + let hashes = mine_big(&pm, 30, 100_000); + let mut rx = add_peer(&pm, 1, addr(1), 256); + let queue = pm.peer_sender(1).unwrap().queue().clone(); + // A budget large enough never to refuse, so every block is counted. + pm.set_max_upload_target(u64::MAX / 2); + let charged_before = pm.upload_bytes.load(Ordering::Relaxed); + + // As the write loop hands the request over. + queue.note_getdata_forwarded(); + deliver(&pm, 1, NetworkMessage::GetData(hashes.iter().map(|h| Inventory::WitnessBlock(*h)).collect())); + assert!(queue.getdata_backlog() > 0, "the bound held part of the request back"); + assert!(queue.reading_paused(), "nothing more is read from the peer meanwhile"); + + // The write loop, writing and asking for more. + let mut served = Vec::new(); + let mut resumes = 0; + while let Ok(msg) = rx.try_recv() { + queue.sent(queued_size(&msg)); + match msg { + NetworkMessage::Block(b) => served.push(b.block_hash()), + other => panic!("unexpected {other:?}"), + } + if queue.take_resume() { + resumes += 1; + pm.resume_getdata(1); + } + } + assert_eq!(served, hashes, "every block, in the order asked"); + assert!(resumes > 0); + assert_eq!(queue.getdata_backlog(), 0); + assert!(!queue.reading_paused(), "the peer is read again"); + assert_eq!(queue.queued_bytes(), 0); + let charged: usize = served.iter().map(|h| pm.chain_state.get_block(h).unwrap().total_size()).sum(); + assert_eq!(pm.upload_bytes.load(Ordering::Relaxed) - charged_before, charged as u64); +} + +// ---- Peer text ---- + +/// Core keeps a peer's user agent only after `SanitizeString` (`cleanSubVer`) +/// and reports that in `getpeerinfo.subver`; satd logged and reported the raw +/// string, line breaks and escape sequences included. +#[test] +fn peer_user_agent_is_sanitized() { + let mut info = PeerInfo::new(1, addr(1), Direction::Inbound); + let zero: SocketAddr = "0.0.0.0:0".parse().unwrap(); + info.set_version(VersionMessage { + version: 70016, + services: ServiceFlags::NETWORK, + timestamp: 0, + receiver: Address::new(&zero, ServiceFlags::NONE), + sender: Address::new(&zero, ServiceFlags::NONE), + nonce: 1, + user_agent: "/Satoshi:27.0.0/\n2026-01-01T00:00:00Z INFO fake\x1b[0m".into(), + start_height: 0, + relay: true, + }); + assert_eq!(info.user_agent, "/Satoshi:27.0.0/2026-01-01T00:00:00Z INFO fake0m"); +} + +/// Core reads the user agent as `LIMITED_STRING(strSubVer, +/// MAX_SUBVERSION_LENGTH)`. A longer one throws in deserialization, so the +/// `version` is dropped and the peer is left without a handshake until the +/// connect timeout; a later, well-formed `version` is accepted. satd took the +/// oversized one. +#[tokio::test] +async fn version_with_an_oversized_user_agent_is_ignored() { + use crate::net::connection::Connection; + use bitcoin::p2p::Magic; + + fn version(user_agent: String) -> NetworkMessage { + let services = ServiceFlags::NETWORK | ServiceFlags::WITNESS; + let zero: SocketAddr = "0.0.0.0:0".parse().unwrap(); + NetworkMessage::Version(VersionMessage { + version: 70016, + services, + timestamp: crate::time::now_secs() as i64, + receiver: Address::new(&zero, ServiceFlags::NONE), + sender: Address::new(&zero, services), + nonce: 0x5eed, + user_agent, + start_height: 0, + relay: true, + }) + } + + let (pm, _dir) = mk_pm(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let dialer = tokio::net::TcpStream::connect(listener.local_addr().unwrap()); + let (dialed, accepted) = tokio::join!(dialer, listener.accept()); + let (dialed, (accepted, peer_addr)) = (dialed.unwrap(), accepted.unwrap()); + let mut info = PeerInfo::new(1, peer_addr, Direction::Inbound); + info.state = PeerState::Connecting; + let (tx, _rx) = mpsc::channel::(1); + pm.peers.write().insert( + 1, + PeerHandle { + info, + msg_tx: tx.into(), + disconnect: Arc::new(tokio::sync::Notify::new()), + flow: Arc::new(crate::net::flow::PeerFlow::new()), + last_getheaders_sent: None, + last_mempool_served: None, + fee_filter_sent: None, + stats: PeerStats::new(NetTotals::new()), + }, + ); + let mut ours = Connection::with_magic(accepted, Magic::REGTEST); + let mut theirs = Connection::with_magic(dialed, Magic::REGTEST); + + let handshake = pm.perform_handshake(1, &mut ours, Direction::Inbound, crate::time::now_secs()); + let remote = async { + theirs.send(version("a".repeat(crate::MAX_SUBVERSION_LENGTH + 1))).await.unwrap(); + // An inbound node answers a `version` with its own. Nothing comes + // back for the oversized one. + let reply = tokio::time::timeout(Duration::from_millis(500), theirs.recv()).await; + assert!(reply.is_err(), "an oversized user agent must not be answered, got {reply:?}"); + // A user agent of exactly the limit is fine, and completes the handshake. + theirs.send(version("b".repeat(crate::MAX_SUBVERSION_LENGTH))).await.unwrap(); + loop { + if matches!(theirs.recv().await.unwrap(), NetworkMessage::Verack) { + break; + } + } + theirs.send(NetworkMessage::Verack).await.unwrap(); + }; + let (handshake, ()) = tokio::join!(handshake, remote); + let theirs_version = handshake.expect("the well-formed version completes the handshake"); + assert_eq!(theirs_version.user_agent.len(), crate::MAX_SUBVERSION_LENGTH); +} + +// ---- BIP 157 requests (Core's `PrepareBlockFilterRequest`) ---- + +#[cfg(feature = "block-filter-index")] +mod bip157 { + use super::*; + use bitcoin::hashes::Hash; + use bitcoin::p2p::message_filter::{GetCFCheckpt, GetCFHeaders, GetCFilters}; + use node_filter_index::{FilterIndex, IndexError}; + + /// A complete index of empty filters. + struct EmptyFilters; + + impl FilterIndex for EmptyFilters { + fn filter_at(&self, _: u8, _: u32) -> Result, IndexError> { + Ok(Vec::new()) + } + fn header_at(&self, _: u8, _: u32) -> Result<[u8; 32], IndexError> { + Ok([0u8; 32]) + } + fn headers_range(&self, _: u8, _: u32, _: u32) -> Result, IndexError> { + Ok(Vec::new()) + } + fn checkpoints_to(&self, _: u8, _: u32) -> Result, IndexError> { + Ok(Vec::new()) + } + fn is_complete(&self) -> bool { + true + } + } + + fn serving_pm() -> (Arc, tempfile::TempDir) { + let (pm, dir) = mk_pm(); + pm.set_filter_index(Arc::new(EmptyFilters), true); + (pm, dir) + } + + fn cfheaders(start_height: u32, stop_hash: bitcoin::BlockHash) -> NetworkMessage { + NetworkMessage::GetCFHeaders(GetCFHeaders { filter_type: 0, start_height, stop_hash }) + } + + /// The request is answered and the peer kept: the must-succeed half of + /// every refusal below. + #[test] + fn a_valid_request_is_served() { + let (pm, _dir) = serving_pm(); + let chain = mine_big(&pm, 5, 10); + let mut rx = add_peer(&pm, 1, addr(1), 8); + deliver(&pm, 1, cfheaders(1, chain[4])); + assert!(connected(&pm, 1)); + match rx.try_recv() { + Ok(NetworkMessage::CFHeaders(h)) => assert_eq!(h.filter_hashes.len(), 5), + other => panic!("expected cfheaders, got {other:?}"), + } + deliver(&pm, 1, NetworkMessage::GetCFCheckpt(GetCFCheckpt { filter_type: 0, stop_hash: chain[4] })); + assert!(matches!(rx.try_recv(), Ok(NetworkMessage::CFCheckpt(_)))); + assert!(connected(&pm, 1)); + } + + /// Core disconnects (`fDisconnect`, no misbehaviour) a request satd cannot + /// answer as asked: a filter type it does not serve, a stop hash it does + /// not know or would not serve, a start above the stop, or a range over + /// the message's limit. satd ignored every one of them. + #[test] + fn bad_requests_disconnect() { + let (pm, _dir) = serving_pm(); + let chain = mine_big(&pm, 5, 10); + let unknown = bitcoin::BlockHash::from_byte_array([0xab; 32]); + let cases: Vec<(&str, NetworkMessage)> = vec![ + ( + "unsupported filter type", + NetworkMessage::GetCFilters(GetCFilters { filter_type: 0x99, start_height: 0, stop_hash: chain[4] }), + ), + ("unknown stop hash", cfheaders(0, unknown)), + ("start above stop", cfheaders(5, chain[3])), + ( + "unsupported filter type, getcfcheckpt", + NetworkMessage::GetCFCheckpt(GetCFCheckpt { filter_type: 1, stop_hash: chain[4] }), + ), + ("unknown stop hash, getcfcheckpt", NetworkMessage::GetCFCheckpt(GetCFCheckpt { filter_type: 0, stop_hash: unknown })), + ]; + for (i, (case, msg)) in cases.into_iter().enumerate() { + let id = 10 + i as PeerId; + let mut rx = add_peer(&pm, id, addr(10 + i as u8), 8); + deliver(&pm, id, msg); + assert!(!connected(&pm, id), "{case}: the peer is disconnected"); + assert!(!pm.is_addr_banned(&addr(10 + i as u8)), "{case}: disconnected, not banned"); + assert!(rx.try_recv().is_err(), "{case}: nothing is served"); + } + } + + /// A node that does not serve filters disconnects a peer that asks for + /// them: Core's "peer requested unsupported block filter type" when + /// `NODE_COMPACT_FILTERS` is not among the services it offered. + #[test] + fn a_request_to_a_node_not_serving_filters_disconnects() { + let (pm, _dir) = mk_pm(); + let chain = mine_big(&pm, 2, 10); + let _rx = add_peer(&pm, 1, addr(1), 8); + deliver(&pm, 1, cfheaders(0, chain[1])); + assert!(!connected(&pm, 1)); + } + + /// The range limits: 1000 heights for `getcfilters`, 2000 for + /// `getcfheaders` (Core's `MAX_GETCFILTERS_SIZE`, `MAX_GETCFHEADERS_SIZE`). + #[tokio::test] + async fn a_range_past_the_limit_disconnects() { + let (pm, _dir) = serving_pm(); + let chain = mine_big(&pm, 1001, 1); + let _rx = add_peer(&pm, 1, addr(1), 2048); + deliver(&pm, 1, NetworkMessage::GetCFilters(GetCFilters { filter_type: 0, start_height: 2, stop_hash: chain[1000] })); + assert!(connected(&pm, 1), "1000 filters are allowed"); + let _rx = add_peer(&pm, 2, addr(2), 8); + deliver(&pm, 2, NetworkMessage::GetCFilters(GetCFilters { filter_type: 0, start_height: 1, stop_hash: chain[1000] })); + assert!(!connected(&pm, 2), "1001 filters disconnect"); + } + + /// A stop hash off the active chain: Core serves one it would relay (a + /// block it validated, no more than a month older than the best header, + /// `BlockRequestAllowed`) and disconnects for anything else. satd keeps + /// filters for the active chain only, so it cannot serve the first kind; + /// it ignores the request and keeps the peer. A header it has not + /// validated is the second kind. + #[test] + fn an_off_chain_stop_hash() { + let (pm, _dir) = serving_pm(); + let chain = mine_big(&pm, 3, 10); + // A heavier branch from height 2, built on a second node and handed + // over, so the block at height 3 is connected and then reorged out: + // validated, and stale. + let (other, _other_dir) = mk_pm(); + for h in &chain[..2] { + other.chain_state.accept_block(&pm.chain_state.get_block(h).unwrap()).unwrap(); + } + let branch = mine_big(&other, 2, 12); + for h in &branch { + pm.chain_state.accept_block(&other.chain_state.get_block(h).unwrap()).unwrap(); + } + assert_eq!(pm.chain_state.tip_hash(), branch[1], "the heavier branch is active"); + let stale = pm.chain_state.get_block_index(&chain[2]).unwrap(); + assert_eq!(stale.status, crate::storage::blockindex::BlockStatus::Valid); + + let mut rx = add_peer(&pm, 1, addr(1), 8); + deliver(&pm, 1, cfheaders(1, chain[2])); + assert!(connected(&pm, 1), "a recent, validated stale block is not grounds to disconnect"); + assert!(rx.try_recv().is_err(), "but satd has no filter for it"); + + // A header with no block behind it. + let tip = pm.chain_state.tip_hash(); + let mut header = pm.chain_state.get_block_index(&tip).unwrap().header; + header.prev_blockhash = tip; + header.time += 1; + while header.validate_pow(header.target()).is_err() { + header.nonce += 1; + } + pm.chain_state.accept_headers(&[header]).1.map_or(Ok(()), Err).expect("header accepted"); + let _rx = add_peer(&pm, 2, addr(2), 8); + deliver(&pm, 2, cfheaders(0, header.block_hash())); + assert!(!connected(&pm, 2), "a header satd never validated is not a block it would serve"); + } +} + +// ---- The write loop ---- + +/// A connected socket pair whose far end the test controls, and the near +/// end's halves with counters attached. +async fn socket_pair() -> ( + crate::net::connection::ConnectionReader, + ConnectionWriter, + tokio::net::TcpStream, + Arc, +) { + use crate::net::connection::Connection; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (dialed, accepted) = + tokio::join!(tokio::net::TcpStream::connect(listener.local_addr().unwrap()), listener.accept()); + let (far, _) = accepted.unwrap(); + let conn = Connection::with_magic(dialed.unwrap(), bitcoin::p2p::Magic::REGTEST); + let (reader, mut writer) = conn.split(); + let stats = PeerStats::new(NetTotals::new()); + writer.set_counters(stats.clone()); + (reader, writer, far, stats) +} + +fn big(len: usize) -> NetworkMessage { + NetworkMessage::Unknown { + command: bitcoin::p2p::message::CommandString::try_from_static("bigmsg").unwrap(), + payload: vec![0x5a; len], + } +} + +/// Everything a write loop needs, for a peer whose far end the test holds. +struct Loop { + id: PeerId, + event_rx: mpsc::Receiver, + msg_tx: crate::net::send_queue::PeerSender, + read_tx: mpsc::Sender, + disconnect: Arc, + stats: Arc, + far: tokio::net::TcpStream, + task: tokio::task::JoinHandle>, +} + +async fn run_loop() -> Loop { + let (_reader, mut writer, far, stats) = socket_pair().await; + let (event_tx, event_rx) = mpsc::channel(64); + let (msg_tx, mut msg_rx) = mpsc::channel::(256); + let msg_tx = crate::net::send_queue::PeerSender::from(msg_tx); + let (read_tx, mut read_rx) = mpsc::channel::(64); + let disconnect = Arc::new(tokio::sync::Notify::new()); + let queue = msg_tx.queue().clone(); + let (s, d) = (stats.clone(), disconnect.clone()); + let task = tokio::spawn(async move { + let _keep = _reader; + PeerManager::peer_write_loop( + 7, + &event_tx, + &mut writer, + &mut msg_rx, + &mut read_rx, + Some(s), + Some(Arc::new(crate::net::flow::PeerFlow::new())), + None, + Some(d), + Some(queue), + ) + .await + }); + Loop { id: 7, event_rx, msg_tx, read_tx, disconnect, stats, far, task } +} + +/// Queue big messages until the socket stops taking bytes: the far end is +/// not reading and the write loop is parked in a write. +async fn park(l: &Loop) { + // More than loopback socket buffers hold. + for _ in 0..8 { + l.msg_tx.try_send(big(3_900_000)).unwrap(); + } + let mut last = 0; + loop { + tokio::time::sleep(Duration::from_millis(200)).await; + let now = l.stats.bytes_sent(); + if now == last && now > 0 { + return; + } + last = now; + } +} + +/// `disconnectnode`, a ban and the ping timeout all reach the write loop +/// through the disconnect signal, which it only looked at between writes. A +/// peer that stopped reading kept the loop parked in a write, so the socket +/// and everything queued to it outlived the disconnect. +#[tokio::test] +async fn a_parked_write_ends_on_disconnect() { + let l = run_loop().await; + park(&l).await; + l.disconnect.notify_one(); + let ended = tokio::time::timeout(Duration::from_secs(5), l.task) + .await + .expect("the write loop ends while its write is parked") + .unwrap(); + assert_eq!(ended, Err("disconnected by manager".to_string())); + drop(l.far); +} + +/// Core drops a peer whose socket has taken nothing for `TIMEOUT_INTERVAL` +/// (20 minutes; `InactivityCheck`, "socket sending timeout"). A slow reader +/// is not a stalled one: the clock restarts whenever the socket takes bytes. +#[tokio::test(start_paused = true)] +async fn a_write_that_makes_no_progress_times_out() { + use crate::net::send_queue::SEND_TIMEOUT; + let l = run_loop().await; + park(&l).await; + let parked_at = tokio::time::Instant::now(); + let ended = tokio::time::timeout(SEND_TIMEOUT * 2, l.task) + .await + .expect("a write with no progress ends") + .unwrap(); + let waited = parked_at.elapsed(); + let err = ended.expect_err("the loop ends with an error"); + assert!(err.contains("sending timeout"), "{err}"); + assert!( + waited >= SEND_TIMEOUT - Duration::from_secs(60) && waited <= SEND_TIMEOUT + Duration::from_secs(60), + "dropped after {waited:?}" + ); + drop(l.far); +} + +/// Core takes no further message from a peer while a `getdata` of its is +/// being served (`ProcessMessages`). The loop hands the manager the +/// `getdata` and then reads nothing until the manager has taken it in and +/// served all of it. +#[tokio::test] +async fn reading_waits_for_a_getdata_to_be_served() { + let mut l = run_loop().await; + let queue = l.msg_tx.queue().clone(); + l.read_tx.send(NetworkMessage::GetData(vec![])).await.unwrap(); + l.read_tx.send(NetworkMessage::SendHeaders).await.unwrap(); + let first = tokio::time::timeout(Duration::from_secs(5), l.event_rx.recv()).await.unwrap(); + assert!(matches!(first, Some(NetEvent::MessageReceived { msg: NetworkMessage::GetData(_), .. }))); + let early = tokio::time::timeout(Duration::from_millis(300), l.event_rx.recv()).await; + assert!(early.is_err(), "nothing after the getdata is read before it is served"); + + // Taken in with one entry still to serve: still waiting. + queue.push_getdata(vec![Inventory::Unknown { inv_type: 99, hash: [1; 32] }]); + queue.note_getdata_handled(); + let early = tokio::time::timeout(Duration::from_millis(300), l.event_rx.recv()).await; + assert!(early.is_err(), "an unserved entry still holds the peer's next message"); + + queue.pop_getdata(); + queue.wake_reader(); + let next = tokio::time::timeout(Duration::from_secs(5), l.event_rx.recv()).await.unwrap(); + assert!(matches!(next, Some(NetEvent::MessageReceived { msg: NetworkMessage::SendHeaders, .. }))); + l.disconnect.notify_one(); + let _ = l.task.await; +} + +/// The loop asks the manager to serve more of a `getdata` backlog once the +/// queue is back under the send buffer, and holds the peer's messages while +/// it is over. +#[tokio::test] +async fn the_loop_asks_for_more_once_the_queue_drains() { + use tokio::io::AsyncReadExt; + let mut l = run_loop().await; + let queue = l.msg_tx.queue().clone(); + queue.push_getdata(vec![Inventory::Unknown { inv_type: 99, hash: [1; 32] }]); + // Over the buffer: two messages of 600 kB. + l.msg_tx.try_send(big(600_000)).unwrap(); + l.msg_tx.try_send(big(600_000)).unwrap(); + // Let the far end read, so the queue drains. + let mut far = l.far; + let reader = tokio::spawn(async move { + let mut sink = vec![0u8; 1 << 16]; + while far.read(&mut sink).await.is_ok_and(|n| n > 0) {} + }); + let ev = tokio::time::timeout(Duration::from_secs(10), async { + loop { + match l.event_rx.recv().await { + Some(NetEvent::GetDataResume { id }) => return id, + Some(_) => continue, + None => panic!("the loop ended"), + } + } + }) + .await + .expect("a resume is asked for"); + assert_eq!(ev, l.id); + assert_eq!(queue.queued_bytes(), 0); + l.disconnect.notify_one(); + let _ = l.task.await; + reader.abort(); +} diff --git a/node/src/net/mod.rs b/node/src/net/mod.rs index 5069ddd6e..2e3c41835 100644 --- a/node/src/net/mod.rs +++ b/node/src/net/mod.rs @@ -7,10 +7,12 @@ pub mod connection; pub mod dns; pub mod flow; pub mod ibd; +pub mod limits; pub mod manager; pub mod peer; pub mod permissions; pub mod proxy; +pub mod send_queue; pub mod stats; pub mod tor; pub mod sync; diff --git a/node/src/net/peer.rs b/node/src/net/peer.rs index 1f558d584..9cae1f83e 100644 --- a/node/src/net/peer.rs +++ b/node/src/net/peer.rs @@ -338,9 +338,13 @@ impl PeerInfo { } /// Update peer info after receiving their version message. - pub fn set_version(&mut self, version: VersionMessage) { + pub fn set_version(&mut self, mut version: VersionMessage) { self.services = version.services; self.best_height = version.start_height; + // Core keeps the user agent only as `SanitizeString(strSubVer)` + // (`cleanSubVer`): it is logged and reported in `getpeerinfo.subver`, + // and the raw string can carry line breaks and escape sequences. + version.user_agent = crate::net::limits::sanitize_string(&version.user_agent); self.user_agent = version.user_agent.clone(); // The peer's own `fRelay` is kept on the stored `version` message; // `relays_txs()` is the answer to "do *we* relay to them", which diff --git a/node/src/net/send_queue.rs b/node/src/net/send_queue.rs new file mode 100644 index 000000000..97a6f3291 --- /dev/null +++ b/node/src/net/send_queue.rs @@ -0,0 +1,227 @@ +//! Per-peer outbound queue accounting, behind Bitcoin Core's send-buffer +//! limit. +//! +//! Core stops working for a peer once the bytes queued to it pass +//! `-maxsendbuffer` (1,000,000 by default). `fPauseSend` is set +//! (`net.cpp` `SocketSendData` and `PushMessage`), `ProcessGetData` stops +//! serving (`net_processing.cpp` `ProcessGetData`), and `ProcessMessages` +//! takes no further message from that peer until the buffer drains. The +//! unserved rest of a `getdata` stays in `m_getdata_requests` and is served +//! as the buffer empties; until it has all been served, nothing else the peer +//! sent is processed either ("this maintains the order of responses and +//! prevents m_getdata_requests to grow unbounded"). +//! +//! satd's queue to a peer is a bounded channel that the peer's write loop +//! drains. [`PeerSender`] is the sending end: it counts the bytes each +//! message adds. [`SendQueue`] is the state the manager and the write loop +//! share: those bytes, the unserved `getdata` entries, and whether the write +//! loop may hand the manager the peer's next message. + +use std::collections::VecDeque; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::time::Duration; + +use bitcoin::consensus::Encodable; +use bitcoin::p2p::message::NetworkMessage; +use bitcoin::p2p::message_blockdata::Inventory; +use tokio::sync::{Notify, mpsc}; + +/// Core's default `-maxsendbuffer`, `DEFAULT_MAXSENDBUFFER` (1000 kB, +/// `net.h`), in bytes. Past it a peer is not served and not read. +pub const MAX_SEND_BUFFER_BYTES: usize = 1_000_000; + +/// Core's `TIMEOUT_INTERVAL` (`net.h`). A peer whose socket has taken no +/// bytes for this long while a message waits to go out is dropped (Core's +/// `InactivityCheck`, "socket sending timeout"). +pub const SEND_TIMEOUT: Duration = Duration::from_secs(20 * 60); + +/// Size of a v1 message header: magic, command, length and checksum. +const MESSAGE_HEADER_SIZE: usize = 24; + +/// What a message counts for while it waits in a peer's queue: its v1 wire +/// size, header and payload. +/// +/// The sender adds it and the write loop takes it off again, both from the +/// message itself, so the two always agree. A v2 packet is a few bytes +/// longer or shorter on the wire; the budget does not need to know. +pub fn queued_size(msg: &NetworkMessage) -> usize { + // Encoding into a sink cannot fail. + MESSAGE_HEADER_SIZE + msg.consensus_encode(&mut bitcoin::io::sink()).unwrap_or(0) +} + +/// State shared by the manager and one peer's write loop. +#[derive(Debug, Default)] +pub struct SendQueue { + /// Bytes queued to the peer and not yet written to its socket. + queued_bytes: AtomicUsize, + /// `getdata` entries not yet served, oldest first: Core's + /// `m_getdata_requests`. + getdata: parking_lot::Mutex>, + /// `getdata.len()`, readable without the lock. + getdata_len: AtomicUsize, + /// `getdata` messages the write loop has handed the manager that the + /// manager has not yet taken in. + getdata_unhandled: AtomicUsize, + /// A [`crate::net::manager::NetEvent::GetDataResume`] for this peer is + /// waiting in the manager's queue. + resume_queued: AtomicBool, + /// Wakes the write loop when the peer's messages may be read again. + reader_wake: Notify, +} + +impl SendQueue { + pub fn new() -> Self { + Self::default() + } + + /// Bytes queued and not yet written. + pub fn queued_bytes(&self) -> usize { + self.queued_bytes.load(Ordering::Acquire) + } + + /// Core's `fPauseSend`: more than [`MAX_SEND_BUFFER_BYTES`] is queued. + pub fn over_limit(&self) -> bool { + self.queued_bytes() > MAX_SEND_BUFFER_BYTES + } + + fn add(&self, n: usize) { + self.queued_bytes.fetch_add(n, Ordering::AcqRel); + } + + /// The write loop has written (or given up on) a message counted at `n`. + pub fn sent(&self, n: usize) { + let _ = self + .queued_bytes + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |v| Some(v.saturating_sub(n))); + } + + /// Append a `getdata` request's entries to the unserved ones. + pub fn push_getdata(&self, entries: Vec) { + let mut q = self.getdata.lock(); + q.extend(entries); + self.getdata_len.store(q.len(), Ordering::Release); + } + + /// The oldest unserved `getdata` entry, left in place. It stays counted + /// as unserved while the manager works on it, so the write loop never + /// sees an empty backlog and resumes reading before the entry is done. + pub fn front_getdata(&self) -> Option { + self.getdata.lock().front().copied() + } + + /// Drop the oldest unserved `getdata` entry: it has been answered. + pub fn pop_getdata(&self) { + let mut q = self.getdata.lock(); + q.pop_front(); + self.getdata_len.store(q.len(), Ordering::Release); + } + + /// How many `getdata` entries are waiting to be served. + pub fn getdata_backlog(&self) -> usize { + self.getdata_len.load(Ordering::Acquire) + } + + /// The write loop has handed the manager a `getdata`. Until the manager + /// has taken it in and served it, the loop reads nothing more. + pub fn note_getdata_forwarded(&self) { + self.getdata_unhandled.fetch_add(1, Ordering::AcqRel); + } + + /// The manager has taken in a `getdata` (whatever it made of it). + pub fn note_getdata_handled(&self) { + let _ = self + .getdata_unhandled + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |v| Some(v.saturating_sub(1))); + self.wake_reader(); + } + + /// Whether the write loop must leave the peer's next message unread: + /// a `getdata` is still being served, or the queue is over + /// [`MAX_SEND_BUFFER_BYTES`] (Core's `ProcessMessages`). + pub fn reading_paused(&self) -> bool { + self.getdata_unhandled.load(Ordering::Acquire) > 0 + || self.getdata_backlog() > 0 + || self.over_limit() + } + + /// Called by the write loop after a write. True when the manager should + /// be asked to serve more of the backlog: there is some, the queue has + /// room, and no request to serve is already waiting. + pub fn take_resume(&self) -> bool { + self.getdata_backlog() > 0 && !self.over_limit() && !self.resume_queued.swap(true, Ordering::AcqRel) + } + + /// Called by the manager as it starts on a resume request, before it + /// looks at the queue, so a write that drains the queue after that look + /// asks again rather than being lost. + pub fn resume_taken(&self) { + self.resume_queued.store(false, Ordering::Release); + } + + /// Tell the write loop the peer's messages may be readable again. + pub fn wake_reader(&self) { + // `notify_one` stores a permit when the loop is not waiting, so a + // wake sent between its check and its wait is not lost. + self.reader_wake.notify_one(); + } + + /// Resolves after [`Self::wake_reader`]. + pub async fn reader_woken(&self) { + self.reader_wake.notified().await; + } +} + +/// The sending end of a peer's queue. Counts what it queues in the shared +/// [`SendQueue`]. +/// +/// Holds the channel's sender: the write loop learns that the manager has +/// dropped the peer when the last one goes. The write loop itself holds only +/// the [`SendQueue`]. +#[derive(Debug, Clone)] +pub struct PeerSender { + tx: mpsc::Sender, + queue: Arc, +} + +impl From> for PeerSender { + fn from(tx: mpsc::Sender) -> Self { + Self { tx, queue: Arc::new(SendQueue::new()) } + } +} + +impl PeerSender { + /// The state this sender shares with the peer's write loop. + pub fn queue(&self) -> &Arc { + &self.queue + } + + /// Queue a message if there is room, counting its bytes. The signature + /// is the tokio sender's, which this replaces at every call site. + #[allow(clippy::result_large_err)] + pub fn try_send(&self, msg: NetworkMessage) -> Result<(), mpsc::error::TrySendError> { + let n = queued_size(&msg); + // Counted before the message is visible to the write loop, so the + // loop can never take it off before it was put on. + self.queue.add(n); + self.tx.try_send(msg).inspect_err(|_| self.queue.sent(n)) + } + + /// Queue a message, waiting for room, counting its bytes. + #[allow(clippy::result_large_err)] + pub async fn send(&self, msg: NetworkMessage) -> Result<(), mpsc::error::SendError> { + let n = queued_size(&msg); + self.queue.add(n); + self.tx.send(msg).await.inspect_err(|_| self.queue.sent(n)) + } + + /// Whether `getdata` serving must stop for now: the queue is over + /// [`MAX_SEND_BUFFER_BYTES`], or the channel has no free slot. + pub fn paused(&self) -> bool { + self.queue.over_limit() || self.tx.capacity() == 0 + } +} + +#[cfg(test)] +#[path = "send_queue_tests.rs"] +mod tests; diff --git a/node/src/net/send_queue_tests.rs b/node/src/net/send_queue_tests.rs new file mode 100644 index 000000000..fbbc043d0 --- /dev/null +++ b/node/src/net/send_queue_tests.rs @@ -0,0 +1,110 @@ +use super::*; + +fn block_of(size: usize) -> NetworkMessage { + NetworkMessage::Unknown { + command: bitcoin::p2p::message::CommandString::try_from_static("bigmsg").unwrap(), + payload: vec![0; size], + } +} + +/// The count is the message's v1 wire size, exactly. +#[test] +fn queued_size_is_the_v1_wire_size() { + for msg in [NetworkMessage::Verack, NetworkMessage::Ping(9), block_of(123_456)] { + let wire = bitcoin::consensus::serialize(&bitcoin::p2p::message::RawNetworkMessage::new( + bitcoin::p2p::Magic::REGTEST, + msg.clone(), + )); + assert_eq!(queued_size(&msg), wire.len(), "{}", msg.cmd()); + } +} + +/// What goes in is counted, what the write loop takes off is subtracted, +/// and a send that does not happen leaves no trace. +#[test] +fn bytes_are_counted_in_and_out() { + let (tx, mut rx) = mpsc::channel(2); + let sender = PeerSender::from(tx); + let q = sender.queue().clone(); + let a = block_of(600_000); + let b = block_of(500_000); + sender.try_send(a.clone()).unwrap(); + assert_eq!(q.queued_bytes(), queued_size(&a)); + assert!(!q.over_limit(), "600 kB is within the buffer"); + sender.try_send(b.clone()).unwrap(); + assert!(q.over_limit(), "1.1 MB is past it"); + assert!(sender.paused()); + + // The channel is full: the refused message is not counted. + let before = q.queued_bytes(); + assert!(sender.try_send(NetworkMessage::Verack).is_err()); + assert_eq!(q.queued_bytes(), before); + + let first = rx.try_recv().unwrap(); + q.sent(queued_size(&first)); + assert_eq!(q.queued_bytes(), queued_size(&b)); + assert!(!q.over_limit()); + assert!(!sender.paused(), "a free slot and room in the buffer"); +} + +/// A full channel pauses serving even when the bytes are few. +#[test] +fn a_full_channel_pauses() { + let (tx, _rx) = mpsc::channel(1); + let sender = PeerSender::from(tx); + assert!(!sender.paused()); + sender.try_send(NetworkMessage::Verack).unwrap(); + assert!(!sender.queue().over_limit()); + assert!(sender.paused()); +} + +/// Reading stops from the moment a `getdata` is handed over until the +/// manager has taken it in and served all of it, and while the buffer is +/// over the limit. +#[test] +fn reading_pauses_for_an_unserved_getdata() { + let q = SendQueue::new(); + assert!(!q.reading_paused()); + q.note_getdata_forwarded(); + assert!(q.reading_paused(), "forwarded, not yet taken in"); + q.push_getdata(vec![Inventory::Block(bitcoin::BlockHash::from_byte_array([1; 32]))]); + q.note_getdata_handled(); + assert!(q.reading_paused(), "taken in, one entry unserved"); + assert_eq!(q.getdata_backlog(), 1); + assert!(q.front_getdata().is_some()); + assert!(q.reading_paused(), "still unserved while it is being worked on"); + q.pop_getdata(); + assert!(!q.reading_paused()); + + q.add(MAX_SEND_BUFFER_BYTES + 1); + assert!(q.reading_paused(), "over the buffer"); + q.sent(2); + assert!(!q.reading_paused()); +} + +/// One resume request at a time, and only when there is a backlog and room. +#[test] +fn a_resume_is_requested_once_until_taken() { + let q = SendQueue::new(); + assert!(!q.take_resume(), "nothing to resume"); + q.push_getdata(vec![Inventory::Block(bitcoin::BlockHash::from_byte_array([1; 32]))]); + q.add(MAX_SEND_BUFFER_BYTES + 1); + assert!(!q.take_resume(), "no room yet"); + q.sent(MAX_SEND_BUFFER_BYTES + 1); + assert!(q.take_resume()); + assert!(!q.take_resume(), "one is already waiting"); + q.resume_taken(); + assert!(q.take_resume()); +} + +/// A wake sent while nobody waits is kept for the next wait. +#[tokio::test] +async fn a_wake_is_not_lost() { + let q = SendQueue::new(); + q.wake_reader(); + tokio::time::timeout(Duration::from_secs(1), q.reader_woken()) + .await + .expect("the stored wake resolves the next wait"); +} + +use bitcoin::hashes::Hash as _; diff --git a/node/src/net/v2transport.rs b/node/src/net/v2transport.rs index cfb704af4..f635a970b 100644 --- a/node/src/net/v2transport.rs +++ b/node/src/net/v2transport.rs @@ -50,6 +50,9 @@ const MIN_PACKET_LEN: usize = 1 + 16; /// `MAX_PROTOCOL_MESSAGE_LENGTH` (4 MB) payload; `decrypt_packet_len` also /// counts the header byte and the 16-byte tag. const MAX_V2_PACKET_LEN: usize = 1 + 12 + 4_000_000 + 1 + 16; +/// How far a packet buffer is grown ahead of the bytes received: Core's +/// `MAX_RESERVE_AHEAD` (`V2Transport::ReceivedBytes`). +const MAX_RESERVE_AHEAD: usize = 256 * 1024; /// Errors from the v2 message codec. #[derive(Debug, thiserror::Error)] @@ -154,14 +157,19 @@ async fn read_exact_buffered( leftover: &mut Vec, n: usize, ) -> io::Result> { - let mut out = vec![0u8; n]; let from_leftover = leftover.len().min(n); - if from_leftover > 0 { - out[..from_leftover].copy_from_slice(&leftover[..from_leftover]); - leftover.drain(..from_leftover); - } - if from_leftover < n { - stream.read_exact(&mut out[from_leftover..]).await?; + let mut out = Vec::with_capacity(n.min(from_leftover + MAX_RESERVE_AHEAD)); + out.extend_from_slice(&leftover[..from_leftover]); + leftover.drain(..from_leftover); + // Grown as the bytes arrive and never more than MAX_RESERVE_AHEAD past + // them, as Core's `V2Transport` does: sizing it from the length alone let + // a peer that announced a large packet and sent nothing hold that much + // memory per connection. + while out.len() < n { + let start = out.len(); + let end = n.min(start + MAX_RESERVE_AHEAD); + out.resize(end, 0); + stream.read_exact(&mut out[start..end]).await?; } Ok(out) } @@ -230,15 +238,18 @@ async fn recv_v2( } /// Encode, encrypt, and write a single `NetworkMessage` as a genuine v2 -/// packet. Returns the number of on-wire (ciphertext-framed) bytes written. +/// packet, counting the bytes as the socket takes them (see +/// [`crate::net::connection::write_counted`]). Returns the number of on-wire +/// (ciphertext-framed) bytes written. async fn send_v2( stream: &mut S, cipher: &mut OutboundCipher, msg: NetworkMessage, + counters: Option<&Arc>, ) -> io::Result { let contents = encode_message(msg); let packet = cipher.encrypt_to_vec(&contents, PacketType::Genuine, None); - stream.write_all(&packet).await?; + crate::net::connection::write_counted(stream, &packet, counters).await?; Ok(packet.len()) } @@ -457,9 +468,8 @@ impl V2Connection { /// (handshake) path, counted once counters are attached. pub async fn send(&mut self, msg: NetworkMessage) -> io::Result<()> { let cmd = msg.cmd(); - let n = send_v2(&mut self.stream, self.cipher.outbound(), msg).await?; + let n = send_v2(&mut self.stream, self.cipher.outbound(), msg, self.counters.as_ref()).await?; if let Some(c) = &self.counters { - c.record_sent(n); c.attribute_sent(cmd, n); } Ok(()) @@ -493,9 +503,8 @@ impl V2Writer { /// (ciphertext-framed) bytes if counters are set. pub async fn send(&mut self, msg: NetworkMessage) -> io::Result<()> { let cmd = msg.cmd(); - let n = send_v2(&mut self.stream, &mut self.cipher, msg).await?; + let n = send_v2(&mut self.stream, &mut self.cipher, msg, self.counters.as_ref()).await?; if let Some(c) = &self.counters { - c.record_sent(n); c.attribute_sent(cmd, n); } Ok(()) @@ -700,3 +709,7 @@ mod tests { let _ = server.await; } } + +#[cfg(test)] +#[path = "v2transport_bounds_tests.rs"] +mod bounds_tests; diff --git a/node/src/net/v2transport_bounds_tests.rs b/node/src/net/v2transport_bounds_tests.rs new file mode 100644 index 000000000..30ffe3697 --- /dev/null +++ b/node/src/net/v2transport_bounds_tests.rs @@ -0,0 +1,82 @@ +//! The v2 receive buffer's growth and send progress, against Bitcoin Core's +//! `V2Transport::ReceivedBytes` and `SocketSendData`. + +use super::*; +use std::time::Duration; +use tokio::net::TcpListener; + +/// A reader that records the largest read it was asked for. +struct MaxAsk { + inner: R, + max: usize, +} + +impl AsyncRead for MaxAsk { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + self.max = self.max.max(buf.remaining()); + Pin::new(&mut self.inner).poll_read(cx, buf) + } +} + +/// Core grows a v2 packet buffer as the bytes arrive, never more than +/// `MAX_RESERVE_AHEAD` (256 KiB) past them. satd allocated the whole +/// announced packet before reading any of it. +#[tokio::test] +async fn a_packet_buffer_grows_with_the_bytes_received() { + let mut reader = MaxAsk { inner: std::io::Cursor::new(vec![7u8; 1000]), max: 0 }; + let mut leftover = vec![1u8; 10]; + let err = read_exact_buffered(&mut reader, &mut leftover, 4_000_000) + .await + .expect_err("the packet never arrives"); + assert_eq!(err.kind(), io::ErrorKind::UnexpectedEof); + assert!(reader.max <= MAX_RESERVE_AHEAD, "asked for {} bytes at once", reader.max); + + // And a packet that does arrive is read whole, leftover first. + let mut reader = std::io::Cursor::new((0..600_000u32).map(|i| i as u8).collect::>()); + let mut leftover = vec![0xee; 3]; + let got = read_exact_buffered(&mut reader, &mut leftover, 500_003).await.unwrap(); + assert_eq!(got.len(), 500_003); + assert_eq!(&got[..3], &[0xee; 3]); + assert_eq!(got[3..], (0..500_000u32).map(|i| i as u8).collect::>()[..]); + assert!(leftover.is_empty()); +} + +/// Bytes are counted as the socket takes them, as on v1. +#[tokio::test] +async fn a_send_counts_its_bytes_as_the_socket_takes_them() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (mut sock, _) = listener.accept().await.unwrap(); + let (cipher, leftover) = responder_handshake(&mut sock, Network::Regtest, &[], 1, None).await.unwrap(); + V2Connection::new(sock, cipher, leftover) + }); + let mut client = TcpStream::connect(addr).await.unwrap(); + let (cipher, leftover) = initiator_handshake(&mut client, Network::Regtest, 2, None).await.unwrap(); + let client = V2Connection::new(client, cipher, leftover); + // The far end never reads. + let _server = server.await.unwrap(); + let (_reader, mut writer) = client.split(); + let stats = PeerStats::new(crate::net::stats::NetTotals::new()); + writer.set_counters(stats.clone()); + + let msg = || NetworkMessage::Unknown { + command: bitcoin::p2p::message::CommandString::try_from_static("bigmsg").unwrap(), + payload: vec![0x5a; 3_900_000], + }; + let mut completed = 0u64; + while let Ok(r) = tokio::time::timeout(Duration::from_millis(500), writer.send(msg())).await { + r.unwrap(); + completed = stats.bytes_sent(); + assert!(completed < 512 * 1024 * 1024, "the socket never filled"); + } + assert!( + stats.bytes_sent() > completed, + "a send the socket only partly took is not counted: {} sent, {completed} in whole messages", + stats.bytes_sent() + ); +} diff --git a/satd/tests/p2p_limits.rs b/satd/tests/p2p_limits.rs new file mode 100644 index 000000000..83edd2c67 --- /dev/null +++ b/satd/tests/p2p_limits.rs @@ -0,0 +1,221 @@ +//! Bitcoin Core's limits on what one P2P message may carry, and its send +//! buffer, exercised against a live regtest node over raw connections. + +mod common; + +use bitcoin::consensus::{deserialize, serialize}; +use bitcoin::hashes::Hash as _; +use bitcoin::p2p::message::{CommandString, NetworkMessage, RawNetworkMessage}; +use bitcoin::p2p::message_blockdata::Inventory; +use bitcoin::p2p::message_network::VersionMessage; +use bitcoin::p2p::{Address, Magic, ServiceFlags}; +use bitcoin::BlockHash; +use common::TestNode; +use std::io::{self, Read, Write}; +use std::net::{SocketAddr, TcpStream}; +use std::time::{Duration, Instant}; + +const ADDR: &str = "bcrt1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqdku202"; + +/// A raw inbound peer, read on the test's own thread. +struct Peer { + stream: TcpStream, +} + +impl Peer { + fn connect(port: u16) -> Self { + let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); + let deadline = Instant::now() + Duration::from_secs(30); + let stream = loop { + match TcpStream::connect_timeout(&addr, Duration::from_secs(2)) { + Ok(s) => break s, + Err(_) if Instant::now() < deadline => std::thread::sleep(Duration::from_millis(100)), + Err(e) => panic!("connect to {addr}: {e}"), + } + }; + stream.set_nodelay(true).unwrap(); + stream.set_write_timeout(Some(Duration::from_secs(30))).unwrap(); + let mut peer = Peer { stream }; + peer.send(NetworkMessage::Version(version())); + let mut got_version = false; + loop { + match peer.recv(Duration::from_secs(30)) { + Ok(NetworkMessage::Version(_)) => got_version = true, + Ok(NetworkMessage::Verack) if got_version => break, + Ok(_) => {} + Err(e) => panic!("handshake: {e}"), + } + } + peer.send(NetworkMessage::Verack); + peer + } + + fn send(&mut self, msg: NetworkMessage) { + self.send_raw(&serialize(&RawNetworkMessage::new(Magic::REGTEST, msg))); + } + + fn send_raw(&mut self, bytes: &[u8]) { + // The node may close the socket part way through. + let _ = self.stream.write_all(bytes); + } + + fn recv(&mut self, timeout: Duration) -> io::Result { + self.stream.set_read_timeout(Some(timeout)).unwrap(); + let mut header = [0u8; 24]; + self.stream.read_exact(&mut header)?; + let len = u32::from_le_bytes(header[16..20].try_into().unwrap()) as usize; + let mut buf = header.to_vec(); + buf.resize(24 + len, 0); + self.stream.read_exact(&mut buf[24..])?; + let raw: RawNetworkMessage = + deserialize(&buf).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?; + Ok(raw.into_payload()) + } + + /// Answer pings until the node sends a message `pred` accepts. + fn recv_until(&mut self, timeout: Duration, pred: impl Fn(&NetworkMessage) -> bool) -> NetworkMessage { + let deadline = Instant::now() + timeout; + loop { + let left = deadline.saturating_duration_since(Instant::now()).max(Duration::from_millis(1)); + match self.recv(left) { + Ok(NetworkMessage::Ping(n)) => self.send(NetworkMessage::Pong(n)), + Ok(m) if pred(&m) => return m, + Ok(_) => {} + Err(e) => panic!("waiting for a message: {e}"), + } + } + } + + /// The connection is still up: a ping gets its pong. + fn alive(&mut self) -> bool { + self.send(NetworkMessage::Ping(0x1234)); + let deadline = Instant::now() + Duration::from_secs(30); + loop { + let left = deadline.saturating_duration_since(Instant::now()).max(Duration::from_millis(1)); + match self.recv(left) { + Ok(NetworkMessage::Pong(0x1234)) => return true, + Ok(NetworkMessage::Ping(n)) => self.send(NetworkMessage::Pong(n)), + Ok(_) => {} + Err(_) => return false, + } + } + } + + /// The node closes the connection within `timeout`. + fn closed_within(&mut self, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + loop { + let left = deadline.saturating_duration_since(Instant::now()); + if left.is_zero() { + return false; + } + match self.recv(left) { + Ok(_) => {} + Err(e) if matches!(e.kind(), io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut) => return false, + Err(_) => return true, + } + } + } +} + +fn version() -> VersionMessage { + let services = ServiceFlags::NETWORK | ServiceFlags::WITNESS; + let zero: SocketAddr = "0.0.0.0:0".parse().unwrap(); + VersionMessage { + version: 70016, + services, + timestamp: 1_760_000_000, + receiver: Address::new(&zero, ServiceFlags::NONE), + sender: Address::new(&zero, services), + nonce: rand_nonce(), + user_agent: "/p2p-limits-test/".into(), + start_height: 0, + relay: true, + } +} + +fn rand_nonce() -> u64 { + std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos() as u64 +} + +fn unknown(len: usize) -> NetworkMessage { + NetworkMessage::Unknown { command: CommandString::try_from_static("bigmsg").unwrap(), payload: vec![0x5a; len] } +} + +/// Core's `MAX_PROTOCOL_MESSAGE_LENGTH`: a message of 4,000,000 bytes is +/// taken (and, of an unknown type, ignored); one byte more and the node +/// drops the connection. satd took frames up to 32 MiB. +#[test] +fn a_frame_over_four_million_bytes_ends_the_connection() { + let mut node = TestNode::start(&[]); + let port = node.p2p_port.unwrap(); + + let mut peer = Peer::connect(port); + peer.send(unknown(4_000_000)); + assert!(peer.alive(), "a 4,000,000-byte message is within the limit"); + + let mut peer = Peer::connect(port); + peer.send(unknown(4_000_001)); + assert!(peer.closed_within(Duration::from_secs(30)), "a 4,000,001-byte message ends the connection"); + node.stop(); +} + +/// Core's `MAX_INV_SZ`: an `inv` of 50,000 entries is fine, one of 50,001 is +/// misbehaviour and the peer is dropped. +#[test] +fn an_inv_over_max_inv_sz_ends_the_connection() { + let mut node = TestNode::start(&[]); + let port = node.p2p_port.unwrap(); + let entry = Inventory::Unknown { inv_type: 99, hash: [7u8; 32] }; + + let mut peer = Peer::connect(port); + peer.send(NetworkMessage::Inv(vec![entry; 50_000])); + assert!(peer.alive(), "50,000 entries are within the limit"); + // Misbehaviour bans the address, and every peer here is 127.0.0.1, so + // the refusal goes last. + peer.send(NetworkMessage::Inv(vec![entry; 50_001])); + assert!(peer.closed_within(Duration::from_secs(30)), "50,001 entries end the connection"); + node.stop(); +} + +/// One `getdata` for more data than the socket and the peer's queue can +/// hold is served in full and in order, as the peer reads (Core keeps the +/// unserved rest in `m_getdata_requests`). satd queued blocks with a +/// `try_send` that dropped whatever did not fit, so a peer that read a +/// little late never got the rest of its request. +#[test] +fn a_large_getdata_is_served_in_full_as_the_peer_reads() { + let mut node = TestNode::start(&[]); + let port = node.p2p_port.unwrap(); + const N: u64 = 100; + node.mine_blocks(N, ADDR); + let hashes: Vec = (1..=N) + .map(|h| { + let hex = node.rpc_ok("getblockhash", vec![serde_json::json!(h)]); + let mut bytes = hex::decode(hex.as_str().unwrap()).unwrap(); + bytes.reverse(); + BlockHash::from_byte_array(bytes.try_into().unwrap()) + }) + .collect(); + // A full-size request, each block asked for many times over: about + // 12 MB of blocks, more than loopback socket buffers take. + let asked: Vec = hashes.iter().copied().cycle().take(50_000).collect(); + + let mut peer = Peer::connect(port); + peer.send(NetworkMessage::GetData(asked.iter().map(|h| Inventory::WitnessBlock(*h)).collect())); + // Read nothing for a while: the node fills the socket and its queue, and + // must hold the rest of the request rather than drop it. + std::thread::sleep(Duration::from_secs(3)); + let mut got = Vec::with_capacity(asked.len()); + while got.len() < asked.len() { + match peer.recv_until(Duration::from_secs(60), |m| { + matches!(m, NetworkMessage::Block(_) | NetworkMessage::NotFound(_)) + }) { + NetworkMessage::Block(b) => got.push(b.block_hash()), + other => panic!("after {} blocks: {other:?}", got.len()), + } + } + assert!(got == asked, "every block, in the order asked"); + assert!(peer.alive()); + node.stop(); +} diff --git a/satd/tests/regtest.rs b/satd/tests/regtest.rs index c6c7f42f5..14d272c9c 100644 --- a/satd/tests/regtest.rs +++ b/satd/tests/regtest.rs @@ -15586,12 +15586,17 @@ mod cf_client { None } - /// Assert that no CFilter / CFHeaders / CFCheckpt arrives in the - /// window — the silent-drop contract from BIP 157. - pub fn assert_silent(&mut self, window: Duration) { + /// Assert that the node closes the connection within the window + /// without sending a CFilter / CFHeaders / CFCheckpt: Core's + /// `PrepareBlockFilterRequest` disconnects a peer whose request it + /// will not serve. + pub fn assert_disconnected(&mut self, window: Duration) { let deadline = Instant::now() + window; - while Instant::now() < deadline { + loop { let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + panic!("assert_disconnected: still connected after {window:?}"); + } self.stream .set_read_timeout(Some(remaining.max(Duration::from_millis(100)))) .unwrap(); @@ -15599,10 +15604,13 @@ mod cf_client { Ok(NetworkMessage::CFilter(_)) | Ok(NetworkMessage::CFHeaders(_)) | Ok(NetworkMessage::CFCheckpt(_)) => { - panic!("assert_silent: peer sent a filter response unexpectedly"); + panic!("assert_disconnected: peer sent a filter response unexpectedly"); } Ok(_) => continue, - Err(_) => break, + Err(e) if matches!(e.kind(), io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut) => { + panic!("assert_disconnected: still connected after {window:?}"); + } + Err(_) => return, } } } @@ -15857,13 +15865,14 @@ fn test_p2p_getcfcheckpt_thousand_block_intervals() { } #[test] -fn test_p2p_silent_drop_when_peer_serve_disabled() { +fn test_p2p_disconnect_when_peer_serve_disabled() { use bitcoin::hashes::Hash as _; use bitcoin::p2p::message_filter::GetCFHeaders; - // blockfilterindex=basic but peerblockfilters off: handlers must - // silent-drop and the version handshake must NOT advertise - // NODE_COMPACT_FILTERS. + // blockfilterindex=basic but peerblockfilters off: the version + // handshake must NOT advertise NODE_COMPACT_FILTERS, and a request for + // filters disconnects, as Core's does ("peer requested unsupported block + // filter type"). let p2p_port = find_available_port(); let rpcport = find_available_port(); let datadir = fresh_test_datadir("satd-p2p-silent-test"); @@ -15909,13 +15918,13 @@ fn test_p2p_silent_drop_when_peer_serve_disabled() { start_height: 0, stop_hash, }); - client.assert_silent(Duration::from_secs(2)); + client.assert_disconnected(Duration::from_secs(10)); node.stop(); let _ = std::fs::remove_dir_all(&datadir); } #[test] -fn test_p2p_silent_drop_invalid_filter_type() { +fn test_p2p_disconnect_invalid_filter_type() { use bitcoin::hashes::Hash as _; use bitcoin::p2p::message_filter::GetCFilters; @@ -15956,7 +15965,7 @@ fn test_p2p_silent_drop_invalid_filter_type() { start_height: 0, stop_hash, }); - client.assert_silent(Duration::from_secs(2)); + client.assert_disconnected(Duration::from_secs(10)); node.stop(); let _ = std::fs::remove_dir_all(&datadir); } @@ -16088,7 +16097,7 @@ fn test_p2p_getcfheaders_accepts_2000_headers() { resp.filter_hashes.len() ); - // 2001 headers must be silent-dropped (start=0, stop=2000 = 2001). + // 2001 headers disconnect, as in Core (start=0, stop=2000 = 2001). let stop_hex_genesis = node .rpc_call_with_params("getblockhash", vec![serde_json::json!(0)]) .unwrap()["result"] @@ -16098,20 +16107,20 @@ fn test_p2p_getcfheaders_accepts_2000_headers() { let mut _consensus_g = <[u8; 32]>::try_from(hex::decode(&stop_hex_genesis).expect("hex").as_slice()) .expect("32 bytes"); - let _ = tip_hash; // suppress unused var warnings on the silent-drop path + let _ = tip_hash; // suppress unused var warnings on the refusal path client.send_get_cfheaders(GetCFHeaders { filter_type: 0, start_height: 0, stop_hash: stop_hash_2000, }); - client.assert_silent(Duration::from_secs(2)); + client.assert_disconnected(Duration::from_secs(10)); node.stop(); let _ = std::fs::remove_dir_all(&datadir); } #[test] -fn test_p2p_silent_drop_oversized_range() { +fn test_p2p_disconnect_oversized_range() { use bitcoin::hashes::Hash as _; use bitcoin::p2p::message_filter::GetCFilters; @@ -16150,13 +16159,13 @@ fn test_p2p_silent_drop_oversized_range() { start_height: 0, stop_hash, }); - client.assert_silent(Duration::from_secs(2)); + client.assert_disconnected(Duration::from_secs(10)); node.stop(); let _ = std::fs::remove_dir_all(&datadir); } #[test] -fn test_p2p_silent_drop_off_chain_stop_hash() { +fn test_p2p_disconnect_unknown_stop_hash() { use bitcoin::hashes::Hash as _; use bitcoin::p2p::message_filter::GetCFHeaders; @@ -16190,7 +16199,7 @@ fn test_p2p_silent_drop_off_chain_stop_hash() { start_height: 0, stop_hash, }); - client.assert_silent(Duration::from_secs(2)); + client.assert_disconnected(Duration::from_secs(10)); node.stop(); let _ = std::fs::remove_dir_all(&datadir); } From a33a2666c0c7c2b3b93302bbea73402ba20bbeef Mon Sep 17 00:00:00 2001 From: Benjamen Keroack Date: Wed, 7 Oct 2026 17:13:14 -0600 Subject: [PATCH 2/3] p2p: wake the manager when a getdata or a getdata resume is handed over The write loop reads nothing more from a peer until its getdata has been served, so the manager's 500 ms drain tick bounded a peer fetching blocks to about one getdata, or one send buffer's worth of a large one, per tick. Co-Authored-By: Claude Opus 5.5 (1M context) --- node/src/net/manager.rs | 14 ++++++++++++-- node/src/net/manager_p2pbounds_tests.rs | 15 ++++++++++++--- 2 files changed, 24 insertions(+), 5 deletions(-) diff --git a/node/src/net/manager.rs b/node/src/net/manager.rs index f883cd5e0..1da8eeed9 100644 --- a/node/src/net/manager.rs +++ b/node/src/net/manager.rs @@ -10184,14 +10184,21 @@ impl PeerManager { flow.queued(); } // Nothing more is read until the manager has - // served this request (`reading_paused`). - if let (NetworkMessage::GetData(_), Some(q)) = (&msg, &send_queue) { + // served this request (`reading_paused`), so it + // is served now rather than at the manager's next + // tick: a peer fetching blocks sends one `getdata` + // after another. + let getdata = matches!(msg, NetworkMessage::GetData(_)); + if let (true, Some(q)) = (getdata, &send_queue) { q.note_getdata_forwarded(); } event_tx .send(NetEvent::MessageReceived { id, msg }) .await .map_err(|e| e.to_string())?; + if let (true, Some(wake)) = (getdata, &drain_now) { + wake.notify_one(); + } } None => { // Reader task ended (error or timeout) @@ -10233,6 +10240,9 @@ impl PeerManager { .send(NetEvent::GetDataResume { id }) .await .map_err(|e| e.to_string())?; + if let Some(wake) = &drain_now { + wake.notify_one(); + } } } } diff --git a/node/src/net/manager_p2pbounds_tests.rs b/node/src/net/manager_p2pbounds_tests.rs index e14865ac6..73586a9f6 100644 --- a/node/src/net/manager_p2pbounds_tests.rs +++ b/node/src/net/manager_p2pbounds_tests.rs @@ -569,6 +569,8 @@ struct Loop { msg_tx: crate::net::send_queue::PeerSender, read_tx: mpsc::Sender, disconnect: Arc, + /// The manager's drain wake-up. + drain_now: Arc, stats: Arc, far: tokio::net::TcpStream, task: tokio::task::JoinHandle>, @@ -581,8 +583,9 @@ async fn run_loop() -> Loop { let msg_tx = crate::net::send_queue::PeerSender::from(msg_tx); let (read_tx, mut read_rx) = mpsc::channel::(64); let disconnect = Arc::new(tokio::sync::Notify::new()); + let drain_now = Arc::new(tokio::sync::Notify::new()); let queue = msg_tx.queue().clone(); - let (s, d) = (stats.clone(), disconnect.clone()); + let (s, d, w) = (stats.clone(), disconnect.clone(), drain_now.clone()); let task = tokio::spawn(async move { let _keep = _reader; PeerManager::peer_write_loop( @@ -593,13 +596,13 @@ async fn run_loop() -> Loop { &mut read_rx, Some(s), Some(Arc::new(crate::net::flow::PeerFlow::new())), - None, + Some(w), Some(d), Some(queue), ) .await }); - Loop { id: 7, event_rx, msg_tx, read_tx, disconnect, stats, far, task } + Loop { id: 7, event_rx, msg_tx, read_tx, disconnect, drain_now, stats, far, task } } /// Queue big messages until the socket stops taking bytes: the far end is @@ -672,6 +675,9 @@ async fn reading_waits_for_a_getdata_to_be_served() { l.read_tx.send(NetworkMessage::SendHeaders).await.unwrap(); let first = tokio::time::timeout(Duration::from_secs(5), l.event_rx.recv()).await.unwrap(); assert!(matches!(first, Some(NetEvent::MessageReceived { msg: NetworkMessage::GetData(_), .. }))); + tokio::time::timeout(Duration::from_secs(1), l.drain_now.notified()) + .await + .expect("the manager is woken to serve the getdata, not left to its next tick"); let early = tokio::time::timeout(Duration::from_millis(300), l.event_rx.recv()).await; assert!(early.is_err(), "nothing after the getdata is read before it is served"); @@ -719,6 +725,9 @@ async fn the_loop_asks_for_more_once_the_queue_drains() { .await .expect("a resume is asked for"); assert_eq!(ev, l.id); + tokio::time::timeout(Duration::from_secs(1), l.drain_now.notified()) + .await + .expect("the manager is woken to serve it"); assert_eq!(queue.queued_bytes(), 0); l.disconnect.notify_one(); let _ = l.task.await; From 02bdb2da1b9b40eaeb94bb121f0544e6893dd070 Mon Sep 17 00:00:00 2001 From: Benjamen Keroack Date: Wed, 7 Oct 2026 21:56:36 -0600 Subject: [PATCH 3/3] p2p: keep a getdata cmpctblock entry that could not be queued serve_compact_block reported a MSG_CMPCT_BLOCK entry as answered even when its cmpctblock could not be queued, so the entry left the getdata backlog with nothing sent. It now says whether the message was queued, and an entry that was not stays first, as every other entry does. Co-Authored-By: Claude Opus 5.5 (1M context) --- node/src/net/manager.rs | 41 +++++++++++++------------ node/src/net/manager_p2pbounds_tests.rs | 27 ++++++++++++++++ 2 files changed, 48 insertions(+), 20 deletions(-) diff --git a/node/src/net/manager.rs b/node/src/net/manager.rs index 1da8eeed9..6eb4b0291 100644 --- a/node/src/net/manager.rs +++ b/node/src/net/manager.rs @@ -8076,38 +8076,32 @@ impl PeerManager { /// Answer a `MSG_CMPCT_BLOCK` getdata with a `cmpctblock`, if the block /// is within [`compact::MAX_CMPCTBLOCK_DEPTH`] of the tip and we are not /// syncing (Core: `can_direct_fetch && pindex->nHeight >= tip->nHeight - - /// MAX_CMPCTBLOCK_DEPTH`). Returns false when the full block should be - /// sent instead. - fn serve_compact_block(&self, id: PeerId, hash: &bitcoin::BlockHash) -> bool { + /// MAX_CMPCTBLOCK_DEPTH`). `None` when the full block should be sent + /// instead; otherwise whether the `cmpctblock` was queued. + fn serve_compact_block(&self, id: PeerId, hash: &bitcoin::BlockHash) -> Option { if self.ibd.read().is_some() || self.is_ibd() { - return false; + return None; } - let Some(entry) = self.chain_state.get_block_index(hash) else { - return false; - }; + let entry = self.chain_state.get_block_index(hash)?; if entry.height.saturating_add(compact::MAX_CMPCTBLOCK_DEPTH) < self.chain_state.tip_height() { - return false; + return None; } let compact = match self.cached_compact(hash) { Some(c) => c, None => { - let Some(block) = self.chain_state.get_block(hash) else { - return false; - }; - match self.compact_for(&block, entry.height, false) { - Some(c) => c, - None => return false, - } + let block = self.chain_state.get_block(hash)?; + self.compact_for(&block, entry.height, false)? } }; let msg = NetworkMessage::CmpctBlock(bitcoin::p2p::message_compact_blocks::CmpctBlock { compact_block: (*compact).clone(), }); - if self.send_to_peer(id, msg) { + let queued = self.send_to_peer(id, msg); + if queued { self.compact_stats.sent_getdata.fetch_add(1, Ordering::Relaxed); self.note_peer_has_block(id, *hash); } - true + Some(queued) } fn handle_getdata(&self, id: PeerId, inventory: Vec) { @@ -8207,8 +8201,12 @@ impl PeerManager { Inventory::Block(hash) | Inventory::WitnessBlock(hash) | Inventory::CompactBlock(hash) => { - if matches!(inv, Inventory::CompactBlock(_)) && self.serve_compact_block(id, &hash) { - return true; + // A `cmpctblock` that could not be queued leaves the entry to + // be served again once the queue drains, as below. + if matches!(inv, Inventory::CompactBlock(_)) + && let Some(queued) = self.serve_compact_block(id, &hash) + { + return queued; } let Some(block) = self.chain_state.get_block(&hash) else { not_found.push(inv); @@ -10272,7 +10270,10 @@ impl PeerManager { /// queued to it stayed. Core drops a peer that has taken no bytes for /// `TIMEOUT_INTERVAL` (`InactivityCheck`, "socket sending timeout"). /// Progress is the bytes the socket takes, counted as it takes them, so a - /// slow reader is not mistaken for a stalled one. + /// slow reader is not mistaken for a stalled one. Without counters, which + /// only a loop whose peer handle was already gone when it started lacks, + /// progress cannot be seen and the deadline runs from the start of the + /// write. async fn send_watched>( id: PeerId, writer: &mut ConnectionWriter, diff --git a/node/src/net/manager_p2pbounds_tests.rs b/node/src/net/manager_p2pbounds_tests.rs index 73586a9f6..4094d5630 100644 --- a/node/src/net/manager_p2pbounds_tests.rs +++ b/node/src/net/manager_p2pbounds_tests.rs @@ -278,6 +278,33 @@ fn getdata_resumes_as_the_queue_drains() { assert_eq!(pm.upload_bytes.load(Ordering::Relaxed) - charged_before, charged as u64); } +/// A `MSG_CMPCT_BLOCK` entry whose `cmpctblock` could not be queued stays +/// first in the backlog, like every other entry, to be served once the +/// queue drains. It was popped as answered with nothing sent. +#[test] +fn a_compact_block_entry_that_cannot_be_queued_is_kept() { + let (pm, _dir) = mk_pm(); + let chain = mine_big(&pm, 3, 10); + let tip = *chain.last().unwrap(); + let mut rx = add_peer(&pm, 1, addr(1), 1); + let sender = pm.peer_sender(1).unwrap(); + sender.try_send(NetworkMessage::Verack).unwrap(); + + let mut not_found = Vec::new(); + assert!( + !pm.serve_getdata_entry(1, &sender, Inventory::CompactBlock(tip), &mut not_found), + "nothing could be queued, so the entry is not answered" + ); + assert!(not_found.is_empty()); + + assert_eq!(rx.try_recv().unwrap(), NetworkMessage::Verack); + assert!(pm.serve_getdata_entry(1, &sender, Inventory::CompactBlock(tip), &mut not_found)); + match rx.try_recv() { + Ok(NetworkMessage::CmpctBlock(c)) => assert_eq!(c.compact_block.header.block_hash(), tip), + other => panic!("expected the cmpctblock, got {other:?}"), + } +} + // ---- Peer text ---- /// Core keeps a peer's user agent only after `SanitizeString` (`cleanSubVer`)