diff --git a/src/transport/udp/peer_drain.rs b/src/transport/udp/peer_drain.rs index 49ab08e..fe8df5b 100644 --- a/src/transport/udp/peer_drain.rs +++ b/src/transport/udp/peer_drain.rs @@ -187,7 +187,29 @@ fn drain_loop( break; } } - if pfds[0].revents & libc::POLLIN == 0 { + let socket_revents = pfds[0].revents; + if socket_revents & libc::POLLNVAL != 0 { + warn!("fips-peer-drain: socket fd became invalid; exiting"); + break; + } + if socket_revents & libc::POLLHUP != 0 { + debug!("fips-peer-drain: socket hung up; exiting"); + break; + } + if socket_revents & libc::POLLERR != 0 { + match take_socket_error(socket_fd) { + Ok(Some(err)) => { + debug!(error = %err, "fips-peer-drain: consumed socket error"); + } + Ok(None) => { + debug!("fips-peer-drain: poll reported socket error with SO_ERROR=0"); + } + Err(err) => { + debug!(error = %err, "fips-peer-drain: failed to read socket error"); + } + } + } + if socket_revents & libc::POLLIN == 0 { continue; } @@ -230,6 +252,28 @@ fn drain_loop( ); } +fn take_socket_error(fd: RawFd) -> io::Result> { + let mut value: libc::c_int = 0; + let mut len = std::mem::size_of::() as libc::socklen_t; + let r = unsafe { + libc::getsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_ERROR, + &mut value as *mut _ as *mut libc::c_void, + &mut len, + ) + }; + if r < 0 { + return Err(io::Error::last_os_error()); + } + if value == 0 { + Ok(None) + } else { + Ok(Some(io::Error::from_raw_os_error(value))) + } +} + fn make_pipe() -> io::Result<(RawFd, RawFd)> { let mut pipe_fds = [0i32; 2]; #[cfg(target_os = "linux")] @@ -454,4 +498,56 @@ mod tests { // Drop the drain handle — should stop the thread within one // poll iteration. } + + #[cfg(target_os = "linux")] + #[test] + fn socket_error_is_consumed_so_poll_does_not_spin() { + let closed_peer = UdpSocket::bind("127.0.0.1:0").expect("bind closed peer"); + let peer_addr = closed_peer.local_addr().expect("closed peer local_addr"); + drop(closed_peer); + + let socket = UdpSocket::bind("127.0.0.1:0").expect("bind connected socket"); + socket.connect(peer_addr).expect("connect to closed peer"); + socket + .set_nonblocking(true) + .expect("set connected socket nonblocking"); + socket.send(&[0xA5]).expect("send to closed peer"); + + let fd = socket.as_raw_fd(); + let mut pfd = libc::pollfd { + fd, + events: libc::POLLIN, + revents: 0, + }; + let mut saw_error = false; + for _ in 0..100 { + pfd.revents = 0; + let r = unsafe { libc::poll(&mut pfd, 1, 10) }; + assert!(r >= 0, "poll failed: {}", io::Error::last_os_error()); + if pfd.revents & libc::POLLERR != 0 { + saw_error = true; + break; + } + } + assert!(saw_error, "connected UDP socket should report POLLERR"); + assert_eq!( + pfd.revents & libc::POLLIN, + 0, + "regression setup expects socket error without readable data" + ); + + let err = take_socket_error(fd) + .expect("take socket error") + .expect("pending socket error"); + assert_eq!(err.raw_os_error(), Some(libc::ECONNREFUSED)); + + pfd.revents = 0; + let r = unsafe { libc::poll(&mut pfd, 1, 0) }; + assert!(r >= 0, "poll after SO_ERROR failed"); + assert_eq!( + pfd.revents & libc::POLLERR, + 0, + "SO_ERROR must be consumed so poll stops waking in a tight loop" + ); + } }