fix(lib): parse what a peer sent before acting on its close

Draining several buffers a tick meant a reply and the close after it
could land in one call, and the close was acted on first: the reply sat
unparsed in the reader while the actor disconnected. Only the first read
of a drain reports an error now; the close is seen again on the next
tick, after the reply has been handled.
This commit is contained in:
Michel de Graaf
2026-09-08 01:23:19 +02:00
parent 2ab7902d1f
commit bcf71dabec
2 changed files with 32 additions and 5 deletions
+6 -5
View File
@@ -556,17 +556,18 @@ impl PeerActor {
/// Read what the socket holds, up to `READS_PER_TICK` buffers, so a big /// Read what the socket holds, up to `READS_PER_TICK` buffers, so a big
/// listing lands in a few ticks rather than a few minutes; the cap keeps /// listing lands in a few ticks rather than a few minutes; the cap keeps
/// a peer that never stops sending from starving the mailbox. /// a peer that never stops sending from starving the mailbox. Only the
/// first read reports an error: a peer that sends its reply and hangs up
/// has still sent the reply, which must be parsed before the close is
/// acted on, and the socket says so again on the next call.
fn drain( fn drain(
reader: &mut MessageReader, reader: &mut MessageReader,
stream: &mut TcpStream, stream: &mut TcpStream,
) -> io::Result<()> { ) -> io::Result<()> {
reader.read_from_socket(stream)?; reader.read_from_socket(stream)?;
for _ in 1..READS_PER_TICK { for _ in 1..READS_PER_TICK {
match reader.read_from_socket(stream) { if reader.read_from_socket(stream).is_err() {
Ok(()) => {} break;
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
Err(e) => return Err(e),
} }
} }
Ok(()) Ok(())
@@ -166,6 +166,32 @@ fn a_retired_actor_delivers_what_is_on_the_wire_then_stops_itself() {
); );
} }
// The peer that replaced the connection considers the old one done too: it
// writes its reply and hangs up in the same breath. The reply still counts.
#[test]
fn a_reply_sent_right_before_the_peer_hangs_up_is_still_delivered() {
use std::io::Write;
let (mut actor, rx, mut far_end) = connected_actor();
actor.handle_message(PeerMessage::Retire);
far_end.write_all(&big_listing()).unwrap();
drop(far_end);
std::thread::sleep(Duration::from_millis(50));
actor.tick();
actor.tick();
match rx.try_recv() {
Ok(ClientOperation::BrowseResult { username, .. }) => {
assert_eq!(username, "bob");
}
other => panic!("expected the listing before the close, got {other:?}"),
}
match rx.try_recv() {
Ok(ClientOperation::PeerDisconnected(7, _, None)) => {}
other => panic!("expected a clean PeerDisconnected, got {other:?}"),
}
}
#[test] #[test]
fn a_retired_actor_reporting_an_error_reports_a_clean_close() { fn a_retired_actor_reporting_an_error_reports_a_clean_close() {
let (mut actor, rx, _far_end) = connected_actor(); let (mut actor, rx, _far_end) = connected_actor();