Ack Cut D (Rust): pin the receive reset, the ack after an unsendable reply, and a refusal at every fragment

- Document server: a reset reported by a receive is idle, not an error
  (Windows only, where the system reports it; run on Starfire).
- Document server: no ack follows a reply that can never be sent. The
  guard is not decoration: without it the ack's success overwrote the
  failure and the session that owed the Pong was kept.
- Hub: a send refused at any fragment index delivers nothing of it; at
  index 0 the next send arrives whole (Soul's probe, committed).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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"]);
+}

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<Vec<u8>> {
+    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<u8> = (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<u8> = (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,
+                    }
+                }
+            }
+        }
+    }
+}