diff --git a/packages/cultmesh-rs/src/rudp_document_server/send_failure_tests.rs b/packages/cultmesh-rs/src/rudp_document_server/send_failure_tests.rs index 7320c2bf..68351b35 100644 --- a/packages/cultmesh-rs/src/rudp_document_server/send_failure_tests.rs +++ b/packages/cultmesh-rs/src/rudp_document_server/send_failure_tests.rs @@ -366,3 +366,81 @@ fn a_session_ended_in_the_resend_loop_is_sent_nothing_more_from_it() { assert_eq!(server.send_failures() - before, 1); assert_eq!(server.session_count(), 1); } + +#[test] +fn no_ack_follows_a_reply_that_can_never_be_sent() { + let (mut server, _clock, _received, _x, _y) = two_peers(); + let target = server.local_addr().unwrap(); + // A peer driven packet by packet: a client never sends a reliable Ping, but a + // peer may, and the server both answers it and acknowledges it. + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_millis(20))) + .unwrap(); + let addr = socket.local_addr().unwrap(); + let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 7, + ..Default::default() + }); + let connect = session.create_connect(0, Vec::new()).unwrap(); + socket + .send_to(&encode_rudp_packet(&connect).unwrap(), target) + .unwrap(); + let mut wire = vec![0_u8; MAX_UDP_DATAGRAM_BYTES]; + for _ in 0..10 { + server.poll_once().unwrap(); + } + let (read, _) = socket.recv_from(&mut wire).unwrap(); + let accept = decode_rudp_packet(&wire[..read]).unwrap(); + assert_eq!(accept.packet_type, CultNetRudpPacketType::Accept); + session.receive(&accept, 0).unwrap(); + assert_eq!(server.session_count(), 3); + + let mut ping = session.create_ping(b"p".to_vec()); + ping.reliable = true; + ping.sequence = connect.sequence.wrapping_add(1); + // The Pong can never be sent; the ack after it could. + server.unsendable_after.insert(addr, 0); + socket + .send_to(&encode_rudp_packet(&ping).unwrap(), target) + .unwrap(); + for _ in 0..10 { + server.poll_once().unwrap(); + } + + // The session ends over the Pong: the peer gets its goodbye and nothing else. + assert_eq!(server.session_count(), 2); + let mut arrived = Vec::new(); + while let Ok((read, _)) = socket.recv_from(&mut wire) { + arrived.push(decode_rudp_packet(&wire[..read]).unwrap()); + } + assert_eq!(arrived.len(), 1, "{arrived:?}"); + assert_eq!(arrived[0].packet_type, CultNetRudpPacketType::Disconnect); + assert!(is_unsendable_reason(&arrived[0].payload)); +} + +/// Windows reports an ICMP port-unreachable for an earlier datagram as +/// `ConnectionReset` on the next receive. It names no datagram of this poll, so +/// the poll is idle and the server goes on serving. Only Windows reports it, so +/// only a Windows run can see the rule break. +#[cfg(windows)] +#[test] +fn a_reset_reported_by_a_receive_is_idle_not_an_error() { + let (mut server, _clock, received, (mut x, _), (mut y, _)) = two_peers(); + // X asks for a snapshot and goes away before the answer. + send(&mut x, &snapshot("x-snapshot")); + drop(x); + for _ in 0..20 { + server + .poll_once() + .expect("a reset from X's closed port is not an error"); + thread::sleep(Duration::from_millis(1)); + } + + send(&mut y, &put("y-put")); + for _ in 0..20 { + server.poll_once().unwrap(); + thread::sleep(Duration::from_millis(1)); + } + assert_eq!(received.lock().unwrap().as_slice(), ["y-put"]); +} diff --git a/packages/cultnet-rs/src/rudp/send_failure_tests.rs b/packages/cultnet-rs/src/rudp/send_failure_tests.rs index 4e5b7125..4bff8cf9 100644 --- a/packages/cultnet-rs/src/rudp/send_failure_tests.rs +++ b/packages/cultnet-rs/src/rudp/send_failure_tests.rs @@ -436,3 +436,67 @@ fn a_refused_send_withdraws_the_fragments_the_window_queued_as_well() { "nothing of the refused send stays queued" ); } + +fn receive_frame( + peer: &mut CultNetRudpSocketTransportConnection, + hub: &mut CultNetRudpServerHub, +) -> Option> { + for _ in 0..600 { + if let Some(frame) = peer.receive_once().unwrap() { + return Some(frame.payload); + } + let _ = hub.receive_event_once().unwrap(); + hub.poll_resends().unwrap(); + } + None +} + +/// Wherever in a fragmented send the datagram that can never be sent falls, the +/// peer never delivers a frame built from it. At the first fragment nothing left, +/// the send is withdrawn, and the next send is delivered whole; later, the peer has +/// seen part of it, and the session ends with nothing delivered. +#[test] +fn a_send_refused_at_any_fragment_delivers_nothing_of_it() { + for fragments in [5_usize, 40] { + for failed in 0..fragments { + let (mut hub, mut peer, session) = one_peer(Some(1000)); + hub.send(&session, "schema", b"warm".to_vec()).unwrap(); + assert_eq!(receive_frame(&mut peer, &mut hub).as_deref(), Some(&b"warm"[..])); + let refused: Vec = (0..fragments * 1000).map(|i| (i % 251) as u8).collect(); + hub.unsendable_after.insert(session.remote_addr, failed); + let result = hub.send(&session, "schema", refused); + if hub.unsendable_after.remove(&session.remote_addr).is_some() { + // The flow window queued that fragment rather than sending it now. + assert!(result.is_ok(), "{fragments}/{failed}: not reached, yet refused"); + continue; + } + assert!(result.is_err(), "{fragments}/{failed}"); + if failed == 0 { + assert_eq!(hub.sessions().len(), 1, "{fragments}/{failed}"); + let next: Vec = (0..fragments * 1000).map(|i| ((i % 13) + 100) as u8).collect(); + hub.send(&session, "schema", next.clone()).unwrap(); + let delivered = receive_frame(&mut peer, &mut hub).expect("the next send arrives"); + assert!(delivered == next, "{fragments}/{failed}: delivered something else"); + for _ in 0..30 { + if let Some(frame) = peer.receive_once().unwrap() { + panic!("{fragments}/{failed}: a stray frame of {} bytes", frame.payload.len()); + } + let _ = hub.receive_event_once().unwrap(); + hub.poll_resends().unwrap(); + } + } else { + assert!(hub.sessions().is_empty(), "{fragments}/{failed}"); + for _ in 0..30 { + match peer.receive_once() { + Ok(Some(frame)) => panic!( + "{fragments}/{failed}: the peer delivered {} bytes", + frame.payload.len() + ), + Ok(None) => {} + Err(_) => break, + } + } + } + } + } +}