From 47b9fc4ee3a4824327623d7efc9d7614d96b5912 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 11 Sep 2026 01:21:37 +0200 Subject: [PATCH] A reader returns "nothing" only when nothing is waiting Both RUDP socket readers returned None after consuming a packet that carried no frame or event for the caller: an acknowledgement, a ping, a fragment that completed nothing yet. A caller that polls once per send therefore read one incoming datagram per outgoing one, and fell behind its peer's acknowledgements as soon as a reliable channel shared the session; every unacknowledged packet was then resent, and the resends starved the sends. Measured on the media path 2026-09-10: 7 video frames in 9 s with reliable audio on, 266 without. Readers now consume no-event packets and keep reading; None means the socket is empty. The client no longer acknowledges unreliable frames. They carry no sequence, so the ACK told the sender nothing and cost it one datagram to read for every one it sent. The fragment-set bound test now asserts eviction, which is what the bound has done since the fuse was cut earlier today; the old assertion was hiding behind a truncated test run. Co-Authored-By: Claude Fable 5.1 --- packages/cultnet-rs/src/rudp.rs | 66 ++++++++++++++------ packages/cultnet-rs/tests/rudp_server_hub.rs | 63 ++++++++++++++++++- 2 files changed, 109 insertions(+), 20 deletions(-) diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index f5a31e76..174739be 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -1538,11 +1538,26 @@ impl CultNetRudpServerHub { Ok(true) } + /// The next event, or `None` only when nothing is waiting on the socket. + /// + /// Packets that carry no event for the caller — acknowledgements, pings, + /// a repeated Connect, a fragment that completes nothing yet — are + /// consumed here and reading continues. Returning `None` after one of + /// those made a caller that polls once per send fall behind its own + /// peers' acknowledgements, and a reliable channel then resent everything. pub fn receive_event_once(&mut self) -> Result> { - if let Some(event) = self.pending_events.pop_front() { - return Ok(Some(event)); + loop { + if let Some(event) = self.pending_events.pop_front() { + return Ok(Some(event)); + } + if !self.receive_packet_once()? { + return Ok(None); + } } + } + /// Reads one datagram and applies it. `false` when the socket is empty. + fn receive_packet_once(&mut self) -> Result { let mut wire = vec![0_u8; 65_535]; let (received, remote_addr) = match self.socket.recv_from(&mut wire) { Ok(value) => value, @@ -1551,7 +1566,7 @@ impl CultNetRudpServerHub { || error.kind() == ErrorKind::TimedOut || error.kind() == ErrorKind::ConnectionReset => { - return Ok(None); + return Ok(false); } Err(error) => return Err(error.into()), }; @@ -1559,7 +1574,7 @@ impl CultNetRudpServerHub { self.stats.bytes_received += received as u64; let packet = decode_rudp_packet(&wire)?; if packet.connection_id != self.connection_id { - return Ok(None); + return Ok(true); } if packet.packet_type == CultNetRudpPacketType::Connect { @@ -1572,7 +1587,7 @@ impl CultNetRudpServerHub { .pending_accept_for_resend(now_ms()) .unwrap_or_else(|| peer.session.create_ack()); self.send_packet(remote_addr, &reply)?; - return Ok(None); + return Ok(true); } if !self.peers.contains_key(&remote_addr) && self.peers.len() >= self.max_peers { return Err(anyhow!("RUDP server hub peer limit reached")); @@ -1613,11 +1628,11 @@ impl CultNetRudpServerHub { } self.pending_events .push_back(CultNetRudpServerEvent::Connected { session: context }); - return Ok(self.pending_events.pop_front()); + return Ok(true); } let Some(peer) = self.peers.get_mut(&remote_addr) else { - return Ok(None); + return Ok(true); }; let result = peer.session.receive(&packet, now_ms())?; let context = peer.context.clone(); @@ -1660,7 +1675,7 @@ impl CultNetRudpServerHub { }); self.peers.remove(&remote_addr); } - Ok(self.pending_events.pop_front()) + Ok(true) } pub fn poll_resends(&mut self) -> Result<()> { @@ -1923,11 +1938,23 @@ impl CultNetRudpSocketTransportConnection { } } + /// The next delivered frame, or `None` only when nothing is waiting on + /// the socket. Acknowledgements, pongs and incomplete fragments are + /// consumed and reading continues; see the hub's `receive_event_once`. pub fn receive_once(&mut self) -> Result> { - if let Some(frame) = self.delivered_frames.pop_front() { - return Ok(Some(frame)); + loop { + if let Some(frame) = self.delivered_frames.pop_front() { + return Ok(Some(frame)); + } + if !self.receive_packet_once()? { + return Ok(None); + } } + } + /// Reads one datagram and applies it. `false` when the socket is empty + /// or the session has been disconnected. + fn receive_packet_once(&mut self) -> Result { let mut wire = vec![0_u8; 65_535]; let (received, remote_addr) = match self.socket.recv_from(&mut wire) { Ok(value) => value, @@ -1936,7 +1963,7 @@ impl CultNetRudpSocketTransportConnection { || error.kind() == ErrorKind::TimedOut || error.kind() == ErrorKind::ConnectionReset => { - return Ok(None); + return Ok(false); } Err(error) => return Err(error.into()), }; @@ -1951,14 +1978,14 @@ impl CultNetRudpSocketTransportConnection { { self.remote_addr = Some(remote_addr); } else { - return Ok(None); + return Ok(true); } } } else { if self.mode == CultNetRudpSocketMode::Server && packet.packet_type != CultNetRudpPacketType::Connect { - return Ok(None); + return Ok(true); } self.remote_addr = Some(remote_addr); } @@ -1968,7 +1995,7 @@ impl CultNetRudpSocketTransportConnection { self.session.reset_peer_state(); let accept = self.session.accept_connect(&packet, now_ms(), Vec::new())?; self.send_packet(&accept)?; - return Ok(None); + return Ok(true); } let result = self.session.receive(&packet, now_ms())?; @@ -1983,7 +2010,7 @@ impl CultNetRudpSocketTransportConnection { } if result.disconnected { self.disconnect_reason = Some(result.disconnect_reason); - return Ok(None); + return Ok(false); } for frame in result.delivered { @@ -1993,13 +2020,14 @@ impl CultNetRudpSocketTransportConnection { }); self.stats.frames_received += 1; } - let frame = self.delivered_frames.pop_front(); - if packet.reliable || packet.packet_type == CultNetRudpPacketType::Accept || frame.is_some() - { + // Only what the sender is waiting on is acknowledged. An unreliable + // frame carries no sequence, so an ACK for it tells the sender + // nothing and costs it one datagram to read for every one it sent. + if packet.reliable || packet.packet_type == CultNetRudpPacketType::Accept { let ack = self.session.create_ack_for_received(packet.sequence); self.send_packet(&ack)?; } - Ok(frame) + Ok(true) } pub fn poll_resends(&mut self) -> Result<()> { diff --git a/packages/cultnet-rs/tests/rudp_server_hub.rs b/packages/cultnet-rs/tests/rudp_server_hub.rs index cb765a33..201cf3b4 100644 --- a/packages/cultnet-rs/tests/rudp_server_hub.rs +++ b/packages/cultnet-rs/tests/rudp_server_hub.rs @@ -399,7 +399,12 @@ fn session_and_hub_reject_configured_memory_bounds() -> Result<()> { )?; let second_set = sender.send_many("schema", vec![5, 6, 7, 8], fragment_options, Some(2))?; receiver.receive(&first_set[0], 2)?; - assert!(receiver.receive(&second_set[0], 2).is_err()); + // The fragment-set bound evicts the stalest incomplete set rather than + // refusing the packet: a receiver that dies at its bound under loss is a + // fuse, not a limit (measured 2026-09-10, see tests/cultnet.rs). + assert_eq!(receiver.fragment_sets_evicted(), 0); + receiver.receive(&second_set[0], 2)?; + assert_eq!(receiver.fragment_sets_evicted(), 1, "the first set was evicted for the second"); let server_socket = socket()?; let server_addr = server_socket.local_addr()?; @@ -511,3 +516,59 @@ fn replay_history_is_bounded_and_idle_hub_sessions_expire() -> Result<()> { assert!(hub.sessions().is_empty()); Ok(()) } + +/// A packet that carries no event — an acknowledgement, a ping — must not +/// make the reader report "nothing waiting" while frames sit behind it. +/// A caller that polls once per send fell behind its peer's acknowledgements +/// this way and a reliable channel resent everything. +#[test] +fn a_reader_does_not_stop_at_a_packet_that_carries_no_event() -> Result<()> { + let mut hub = CultNetRudpServerHub::new(CultNetRudpServerHubOptions::new( + "hub", + socket()?, + CONNECTION_ID, + ))?; + let mut peer = client(hub.local_addr()?)?; + let session = connect(&mut hub, &mut peer, b"evidence")?; + + // Hub side: a ping (reply, no event) ahead of a reliable frame. + peer.ping(b"keepalive".to_vec())?; + peer.send("schema", b"after the ping".to_vec())?; + std::thread::sleep(Duration::from_millis(20)); + match hub.receive_event_once()? { + Some(CultNetRudpServerEvent::Frame { frame, .. }) => { + assert_eq!(frame.payload, b"after the ping"); + } + other => panic!("the first read must reach the frame behind the ping, got {other:?}"), + } + + // Client side: the hub's acknowledgement of that frame is queued ahead of + // two unreliable frames; the client must reach both without a "nothing" + // in between, and must not acknowledge unreliable frames back. + let mut unreliable_hub_options = + CultNetRudpServerHubOptions::new("hub-unreliable", socket()?, CONNECTION_ID); + unreliable_hub_options.media_delivery = Some(CultNetTransportDelivery::Unreliable); + let mut media_hub = CultNetRudpServerHub::new(unreliable_hub_options)?; + let mut media_client = client(media_hub.local_addr()?)?; + let media_session = connect(&mut media_hub, &mut media_client, b"media")?; + media_client.send("schema", b"reliable, will be acked".to_vec())?; + std::thread::sleep(Duration::from_millis(20)); + while media_hub.receive_event_once()?.is_some() {} + media_hub.send(&media_session, "media", b"one".to_vec())?; + media_hub.send(&media_session, "media", b"two".to_vec())?; + std::thread::sleep(Duration::from_millis(20)); + let before = media_hub.stats().bytes_received; + let first = media_client.receive_once()?.expect("the first read reaches the frame behind the ack"); + let second = media_client.receive_once()?.expect("the second frame follows"); + assert_eq!((first.payload, second.payload), (b"one".to_vec(), b"two".to_vec())); + assert!(media_client.receive_once()?.is_none()); + std::thread::sleep(Duration::from_millis(20)); + while media_hub.receive_event_once()?.is_some() {} + assert_eq!( + media_hub.stats().bytes_received, + before, + "unreliable frames are not acknowledged; the hub read nothing back" + ); + let _ = session; + Ok(()) +}