← Watch feed
Only pause read in `PeerManager` `send_data` not `read_event`
Public commit record
95/100 · Strong
What the developer wrote
Authored by Matt Corallo
Only pause read in `PeerManager` `send_data` not `read_event`
We recently ran into a race condition on macOS where `read_event`
would return `Ok(true)` (implying reads should be paused) due to
many queued outbound messages but before the caller was able to
set the read-pause flag, the `send_data` calls to flush the
buffered messages completed. Thus, when the `read_event` caller got
scheduled again, the buffer was empty and we should be reading, but
it is finally processing the read-pause flag and we end up hanging,
unwilling to read messages and unable to learn that we should start
reading again as there are no messages to `send_data` for.
This should be fairly rare, but not unheard of - the `pause_read`
flag in `read_event` is calculated before handling the last
message, so there's some time between when its calculated and when
its returned. However, that has to race with multiple calls to
`send_data` to send all the pending messages, which all have to
complete before the `read_event` return happens. We've (as far as I
recall) never hit this in prod, but a benchmark HTLC-flood test
managed to hit it somewhat reliably within a few minutes on macOS
and when a synthetic few-ms sleep was added to each message
handling call.
Ultimately this is an issue with the API - we pause reads via a
returned flag but unpause them via a called method, creating two
independent "stream"s of pause/unpauses which can get out of sync.
Thus, here, we stick to a single "stream" of pause-read events from
`PeerManager` to user code via `send_data` calls, dropping the
read-pause flag return from `read_event` entirely.
Technically this adds risk that someone can flood us with enough
messages fast enough to bloat our outbound buffer for a peer before
`PeerManager::process_events` gets called and can flush the pause
flag via `read_event` calls to all descriptors. This isn't ideal
but it should still be relatively hard to do as `process_events`
calls are pretty quick and should be triggered immediately after
each `read_event` call completes.
✓ Specific, descriptive subject✓ Names a concrete action or component✓ Provides detailed explanatory context✓ Explains rationale or failure mode✓ Mentions testing or verification
What changed, and why it matters
This commit fixes a race condition in how the Lightning networking code pauses and resumes reading from a peer's connection. Previously, the code could tell a caller to pause reading, but by the time the caller acted on it, the reason to pause had already gone away. This could leave a peer connection stuck: no new messages would be read, and because nothing was being sent either, there was no later signal to resume reading. The fix moves the pause/resume signal into the same path used to send data, so the two cannot get out of sync. The commit also notes a small new risk: a very fast message flood could briefly bloat memory before the next processing cycle pauses reads.
Recommended action
Review custom `SocketDescriptor` implementations outside the repository for compatibility with the new `send_data` signature and `continue_read` semantics. Ensure `process_events` is called promptly after `read_event` to limit memory bloat from queued outbound messages. Monitor for any hang regressions in high-throughput or macOS environments.
Security signals we found
01
Race condition between read pause signal and outbound buffer drain could cause connection hang/DoS
02
Read pause/resume logic moved into single `send_data` stream to prevent desynchronization
03
New DoS consideration noted: fast inbound message flood could bloat outbound buffer before `process_events` flushes pause flag
04
API-breaking change in `SocketDescriptor::send_data` semantics and `read_event` return type
Evidence from the diff
The patch removes the pause_read boolean return value from PeerManager::read_event and instead controls read pausing through the continue_read flag on SocketDescriptor::send_data. This unifies pause/unpause signaling into a single stream of send_data calls, eliminating a race where read_event returned Ok(true) (pause reads) but the outbound queue was drained before the caller could set its read-paused flag, causing a hang. The Tokio net adapter now updates read_paused directly from send_data’s continue_read flag and removes the old manual pause logic in the Poll::Pending write path. write_buffer_space_avail now forces at least one write attempt to clear potentially stale pause state. The public read_event API changes from Result<bool, PeerHandleError> to Result<(), PeerHandleError>, and SocketDescriptor::send_data parameter is renamed from resume_read to continue_read with inverted semantics (false means pause).
Changed components
lightning/src/ln/peer_handler.rslightning-net-tokio/src/lib.rslightning-background-processor/src/lib.rsfuzz/src/full_stack.rsInspect captured patch +53 / −70
diff --git a/fuzz/src/full_stack.rs b/fuzz/src/full_stack.rs
index 674be81..97a7487 100644
--- a/fuzz/src/full_stack.rs
+++ b/fuzz/src/full_stack.rs
@@ -195,7 +195,7 @@ struct Peer<'a> {
peers_connected: &'a RefCell<[bool; 256]>,
}
impl<'a> SocketDescriptor for Peer<'a> {
- fn send_data(&mut self, data: &[u8], _resume_read: bool) -> usize {
+ fn send_data(&mut self, data: &[u8], _continue_read: bool) -> usize {
data.len()
}
fn disconnect_socket(&mut self) {
@@ -695,7 +695,7 @@ pub fn do_test(mut data: &[u8], logger: &Arc<dyn Logger>) {
}
let mut peer = Peer { id: peer_id, peers_connected: &peers };
match loss_detector.handler.read_event(&mut peer, get_slice!(get_slice!(1)[0])) {
- Ok(res) => assert!(!res),
+ Ok(()) => {},
Err(_) => {
peers.borrow_mut()[peer_id as usize] = false;
},
diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs
index 2b8bfb0..4d13925 100644
--- a/lightning-background-processor/src/lib.rs
+++ b/lightning-background-processor/src/lib.rs
@@ -774,7 +774,7 @@ use futures_util::{dummy_waker, Joiner, OptionalSelector, Selector, SelectorOutp
/// # #[derive(Eq, PartialEq, Clone, Hash)]
/// # struct SocketDescriptor {}
/// # impl lightning::ln::peer_handler::SocketDescriptor for SocketDescriptor {
-/// # fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize { 0 }
+/// # fn send_data(&mut self, _data: &[u8], _continue_read: bool) -> usize { 0 }
/// # fn disconnect_socket(&mut self) {}
/// # }
/// # type ChainMonitor<B, F, FE> = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<F>, Arc<B>, Arc<FE>, Arc<Logger>, Arc<StoreSync>, Arc<lightning::sign::KeysManager>>;
@@ -1878,7 +1878,7 @@ mod tests {
#[derive(Clone, Hash, PartialEq, Eq)]
struct TestDescriptor {}
impl SocketDescriptor for TestDescriptor {
- fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize {
+ fn send_data(&mut self, _data: &[u8], _continue_read: bool) -> usize {
0
}
diff --git a/lightning-net-tokio/src/lib.rs b/lightning-net-tokio/src/lib.rs
index 2ec69de..068f77a 100644
--- a/lightning-net-tokio/src/lib.rs
+++ b/lightning-net-tokio/src/lib.rs
@@ -243,13 +243,8 @@ impl Connection {
Ok(len) => {
let read_res =
peer_manager.as_ref().read_event(&mut our_descriptor, &buf[0..len]);
- let mut us_lock = us.lock().unwrap();
match read_res {
- Ok(pause_read) => {
- if pause_read {
- us_lock.read_paused = true;
- }
- },
+ Ok(()) => {},
Err(_) => break Disconnect::CloseConnection,
}
},
@@ -533,7 +528,7 @@ impl SocketDescriptor {
}
}
impl peer_handler::SocketDescriptor for SocketDescriptor {
- fn send_data(&mut self, data: &[u8], resume_read: bool) -> usize {
+ fn send_data(&mut self, data: &[u8], continue_read: bool) -> usize {
// To send data, we take a lock on our Connection to access the TcpStream, writing to it if
// there's room in the kernel buffer, or otherwise create a new Waker with a
// SocketDescriptor in it which can wake up the write_avail Sender, waking up the
@@ -544,13 +539,16 @@ impl peer_handler::SocketDescriptor for SocketDescriptor {
return 0;
}
- if resume_read && us.read_paused {
+ let read_was_paused = us.read_paused;
+ us.read_paused = !continue_read;
+
+ if continue_read && read_was_paused {
// The schedule_read future may go to lock up but end up getting woken up by there
// being more room in the write buffer, dropping the other end of this Sender
// before we get here, so we ignore any failures to wake it up.
- us.read_paused = false;
let _ = us.read_waker.try_send(());
}
+
if data.is_empty() {
return 0;
}
@@ -576,16 +574,7 @@ impl peer_handler::SocketDescriptor for SocketDescriptor {
}
},
task::Poll::Ready(Err(_)) => return written_len,
- task::Poll::Pending => {
- // We're queued up for a write event now, but we need to make sure we also
- // pause read given we're now waiting on the remote end to ACK (and in
- // accordance with the send_data() docs).
- us.read_paused = true;
- // Further, to avoid any current pending read causing a `read_event` call, wake
- // up the read_waker and restart its loop.
- let _ = us.read_waker.try_send(());
- return written_len;
- },
+ task::Poll::Pending => return written_len,
}
}
}
diff --git a/lightning/src/ln/peer_handler.rs b/lightning/src/ln/peer_handler.rs
index 3cf6c6c..a3e7107 100644
--- a/lightning/src/ln/peer_handler.rs
+++ b/lightning/src/ln/peer_handler.rs
@@ -632,16 +632,15 @@ pub trait SocketDescriptor: cmp::Eq + hash::Hash + Clone {
///
/// If the returned size is smaller than `data.len()`, a
/// [`PeerManager::write_buffer_space_avail`] call must be made the next time more data can be
- /// written. Additionally, until a `send_data` event completes fully, no further
- /// [`PeerManager::read_event`] calls should be made for the same peer! Because this is to
- /// prevent denial-of-service issues, you should not read or buffer any data from the socket
- /// until then.
+ /// written.
///
- /// If a [`PeerManager::read_event`] call on this descriptor had previously returned true
- /// (indicating that read events should be paused to prevent DoS in the send buffer),
- /// `resume_read` may be set indicating that read events on this descriptor should resume. A
- /// `resume_read` of false carries no meaning, and should not cause any action.
- fn send_data(&mut self, data: &[u8], resume_read: bool) -> usize;
+ /// If `continue_read` is *not* set, further [`PeerManager::read_event`] calls should be
+ /// avoided until another call is made with it set. This allows us to pause read if there are
+ /// too many outgoing messages queued for a peer to avoid DoS issues where a peer fills our
+ /// buffer by sending us messages that need response without reading the responses.
+ ///
+ /// Note that calls may be made with an empty `data` to update the `continue_read` flag.
+ fn send_data(&mut self, data: &[u8], continue_read: bool) -> usize;
/// Disconnect the socket pointed to by this SocketDescriptor.
///
/// You do *not* need to call [`PeerManager::socket_disconnected`] with this socket after this
@@ -1664,7 +1663,10 @@ where
Some(peer_mutex) => {
let mut peer = peer_mutex.lock().unwrap();
peer.awaiting_write_event = false;
- self.do_attempt_write_data(descriptor, &mut peer, false);
+ // We go ahead and force at least one write here, because if we don't have any
+ // messages to send and the net driver thought we did that's weird, so they might
+ // also have a confused read-paused state that we should go ahead and clear.
+ self.do_attempt_write_data(descriptor, &mut peer, true);
},
};
Ok(())
@@ -1676,11 +1678,9 @@ where
///
/// Will *not* call back into [`send_data`] on any descriptors to avoid reentrancy complexity.
/// Thus, however, you should call [`process_events`] after any `read_event` to generate
- /// [`send_data`] calls to handle responses.
- ///
- /// If `Ok(true)` is returned, further read_events should not be triggered until a
- /// [`send_data`] call on this descriptor has `resume_read` set (preventing DoS issues in the
- /// send buffer).
+ /// [`send_data`] calls to handle responses. This is also important to give [`send_data`] calls
+ /// a chance to pause reads if too many messages have been queued in response allowing a peer
+ /// to bloat our memory.
///
/// In order to avoid processing too many messages at once per peer, `data` should be on the
/// order of 4KiB.
@@ -1689,7 +1689,7 @@ where
/// [`process_events`]: PeerManager::process_events
pub fn read_event(
&self, peer_descriptor: &mut Descriptor, data: &[u8],
- ) -> Result<bool, PeerHandleError> {
+ ) -> Result<(), PeerHandleError> {
match self.do_read_event(peer_descriptor, data) {
Ok(res) => Ok(res),
Err(e) => {
@@ -1718,8 +1718,7 @@ where
fn do_read_event(
&self, peer_descriptor: &mut Descriptor, data: &[u8],
- ) -> Result<bool, PeerHandleError> {
- let mut pause_read = false;
+ ) -> Result<(), PeerHandleError> {
let peers = self.peers.read().unwrap();
let mut msgs_to_forward = Vec::new();
let mut peer_node_id = None;
@@ -1994,7 +1993,6 @@ where
},
}
}
- pause_read = !self.peer_should_read(peer);
if let Some(message) = msg_to_handle {
match self.handle_message(&peer_mutex, peer_lock, message) {
@@ -2027,7 +2025,7 @@ where
);
}
- Ok(pause_read)
+ Ok(())
}
/// Process an incoming message and return a decision (ok, lightning error, peer handling error) regarding the next action with the peer
@@ -3725,7 +3723,7 @@ mod tests {
}
impl SocketDescriptor for FileDescriptor {
- fn send_data(&mut self, data: &[u8], _resume_read: bool) -> usize {
+ fn send_data(&mut self, data: &[u8], _continue_read: bool) -> usize {
if self.hang_writes.load(Ordering::Acquire) {
0
} else {
@@ -3939,12 +3937,8 @@ mod tests {
fn try_establish_connection<'a>(
peer_a: &TestPeer<'a>, peer_b: &TestPeer<'a>,
- ) -> (
- FileDescriptor,
- FileDescriptor,
- Result<bool, PeerHandleError>,
- Result<bool, PeerHandleError>,
- ) {
+ ) -> (FileDescriptor, FileDescriptor, Result<(), PeerHandleError>, Result<(), PeerHandleError>)
+ {
let addr_a = SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port: 1000 };
let addr_b = SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port: 1001 };
@@ -3958,11 +3952,11 @@ mod tests {
let initial_data =
peer_b.new_outbound_connection(id_a, fd_b.clone(), Some(addr_a.clone())).unwrap();
peer_a.new_inbound_connection(fd_a.clone(), Some(addr_b.clone())).unwrap();
- assert_eq!(peer_a.read_event(&mut fd_a, &initial_data).unwrap(), false);
+ peer_a.read_event(&mut fd_a, &initial_data).unwrap();
peer_a.process_events();
let a_data = fd_a.outbound_data.lock().unwrap().split_off(0);
- assert_eq!(peer_b.read_event(&mut fd_b, &a_data).unwrap(), false);
+ peer_b.read_event(&mut fd_b, &a_data).unwrap();
peer_b.process_events();
let b_data = fd_b.outbound_data.lock().unwrap().split_off(0);
@@ -3989,8 +3983,8 @@ mod tests {
let (fd_a, fd_b, a_refused, b_refused) = try_establish_connection(peer_a, peer_b);
- assert_eq!(a_refused.unwrap(), false);
- assert_eq!(b_refused.unwrap(), false);
+ a_refused.unwrap();
+ b_refused.unwrap();
assert_eq!(peer_a.peer_by_node_id(&id_b).unwrap().counterparty_node_id, id_b);
assert_eq!(peer_a.peer_by_node_id(&id_b).unwrap().socket_address, Some(addr_b));
@@ -4113,11 +4107,11 @@ mod tests {
let initial_data =
peer_b.new_outbound_connection(id_a, fd_b.clone(), Some(addr_a.clone())).unwrap();
peer_a.new_inbound_connection(fd_a.clone(), Some(addr_b.clone())).unwrap();
- assert_eq!(peer_a.read_event(&mut fd_a, &initial_data).unwrap(), false);
+ peer_a.read_event(&mut fd_a, &initial_data).unwrap();
peer_a.process_events();
let a_data = fd_a.outbound_data.lock().unwrap().split_off(0);
- assert_eq!(peer_b.read_event(&mut fd_b, &a_data).unwrap(), false);
+ peer_b.read_event(&mut fd_b, &a_data).unwrap();
peer_b.process_events();
let b_data = fd_b.outbound_data.lock().unwrap().split_off(0);
@@ -4144,11 +4138,11 @@ mod tests {
let initial_data =
peer_b.new_outbound_connection(id_a, fd_b.clone(), Some(addr_a.clone())).unwrap();
peer_a.new_inbound_connection(fd_a.clone(), Some(addr_b.clone())).unwrap();
- assert_eq!(peer_a.read_event(&mut fd_a, &initial_data).unwrap(), false);
+ peer_a.read_event(&mut fd_a, &initial_data).unwrap();
peer_a.process_events();
let a_data = fd_a.outbound_data.lock().unwrap().split_off(0);
- assert_eq!(peer_b.read_event(&mut fd_b, &a_data).unwrap(), false);
+ peer_b.read_event(&mut fd_b, &a_data).unwrap();
peer_b.process_events();
let b_data = fd_b.outbound_data.lock().unwrap().split_off(0);
@@ -4220,7 +4214,7 @@ mod tests {
peers[0].process_events();
let a_data = fd_a.outbound_data.lock().unwrap().split_off(0);
- assert_eq!(peers[1].read_event(&mut fd_b, &a_data).unwrap(), false);
+ peers[1].read_event(&mut fd_b, &a_data).unwrap();
}
#[test]
@@ -4240,13 +4234,13 @@ mod tests {
let mut dup_encryptor =
PeerChannelEncryptor::new_outbound(id_a, SecretKey::from_slice(&[42; 32]).unwrap());
let initial_data = dup_encryptor.get_act_one(&peers[1].secp_ctx);
- assert_eq!(peers[0].read_event(&mut fd_dup, &initial_data).unwrap(), false);
+ peers[0].read_event(&mut fd_dup, &initial_data).unwrap();
peers[0].process_events();
let a_data = fd_dup.outbound_data.lock().unwrap().split_off(0);
let (act_three, _) =
dup_encryptor.process_act_two(&a_data[..], &&cfgs[1].node_signer).unwrap();
- assert_eq!(peers[0].read_event(&mut fd_dup, &act_three).unwrap(), false);
+ peers[0].read_event(&mut fd_dup, &act_three).unwrap();
let not_init_msg = msgs::Ping { ponglen: 4, byteslen: 0 };
let msg_bytes = dup_encryptor.encrypt_message(¬_init_msg);
@@ -4504,10 +4498,10 @@ mod tests {
assert_eq!(peers_len, 1);
}
- assert_eq!(peers[0].read_event(&mut fd_a, &initial_data).unwrap(), false);
+ peers[0].read_event(&mut fd_a, &initial_data).unwrap();
peers[0].process_events();
let a_data = fd_a.outbound_data.lock().unwrap().split_off(0);
- assert_eq!(peers[1].read_event(&mut fd_b, &a_data).unwrap(), false);
+ peers[1].read_event(&mut fd_b, &a_data).unwrap();
peers[1].process_events();
// ...but if we get a second timer tick, we should disconnect the peer
@@ -4557,11 +4551,11 @@ mod tests {
let act_one = peer_b.new_outbound_connection(a_id, fd_b.clone(), None).unwrap();
peer_a.new_inbound_connection(fd_a.clone(), None).unwrap();
- assert_eq!(peer_a.read_event(&mut fd_a, &act_one).unwrap(), false);
+ peer_a.read_event(&mut fd_a, &act_one).unwrap();
peer_a.process_events();
let act_two = fd_a.outbound_data.lock().unwrap().split_off(0);
- assert_eq!(peer_b.read_event(&mut fd_b, &act_two).unwrap(), false);
+ peer_b.read_event(&mut fd_b, &act_two).unwrap();
peer_b.process_events();
// Calling this here triggers the race on inbound connections.
@@ -4575,7 +4569,7 @@ mod tests {
assert!(!handshake_complete);
}
- assert_eq!(peer_a.read_event(&mut fd_a, &act_three_with_init_b).unwrap(), false);
+ peer_a.read_event(&mut fd_a, &act_three_with_init_b).unwrap();
peer_a.process_events();
{
@@ -4595,7 +4589,7 @@ mod tests {
assert!(!handshake_complete);
}
- assert_eq!(peer_b.read_event(&mut fd_b, &init_a).unwrap(), false);
+ peer_b.read_event(&mut fd_b, &init_a).unwrap();
peer_b.process_events();
{
@@ -4632,7 +4626,7 @@ mod tests {
peer_a.process_events();
let msg = fd_a.outbound_data.lock().unwrap().split_off(0);
assert!(!msg.is_empty());
- assert_eq!(peer_b.read_event(&mut fd_b, &msg).unwrap(), false);
+ peer_b.read_event(&mut fd_b, &msg).unwrap();
peer_b.process_events();
};
@@ -4675,12 +4669,12 @@ mod tests {
let msg = fd_a.outbound_data.lock().unwrap().split_off(0);
if !msg.is_empty() {
- assert_eq!(peers[1].read_event(&mut fd_b, &msg).unwrap(), false);
+ peers[1].read_event(&mut fd_b, &msg).unwrap();
continue;
}
let msg = fd_b.outbound_data.lock().unwrap().split_off(0);
if !msg.is_empty() {
- assert_eq!(peers[0].read_event(&mut fd_a, &msg).unwrap(), false);
+ peers[0].read_event(&mut fd_a, &msg).unwrap();
continue;
}
break;
Risk score
Our methodology →Why this scored 48/100
Human-validated context
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
No validated notes yet.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.