diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index bebeb6c3..b99c7145 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -207,6 +207,7 @@ pub struct CultMeshRudpDocumentServer { snapshot_source: Q, clock: C, options: CultMeshRudpDocumentServerOptions, + packets_dropped: u64, } impl CultMeshRudpDocumentServer @@ -234,9 +235,18 @@ where snapshot_source, clock, options, + packets_dropped: 0, }) } + /// Datagrams read and discarded because they belong to no admissible + /// session: malformed frames, unknown sessions, a Connect the session cap + /// refuses, a packet that fails its session. Never an + /// error: a moved flow or a scanner must not end the daemon loop. + pub fn packets_dropped(&self) -> u64 { + self.packets_dropped + } + pub fn local_addr(&self) -> Result { Ok(self.socket.local_addr()?) } @@ -264,7 +274,10 @@ where let packet = match decode_rudp_packet(&wire) { Ok(packet) => packet, - Err(_) => return Ok(CultMeshRudpPollOutcome::Handled), + Err(_) => { + self.packets_dropped += 1; + return Ok(CultMeshRudpPollOutcome::Handled); + } }; let now_unix = self.clock.now_unix_millis(); let now_monotonic = self.clock.now_monotonic_millis(); @@ -274,11 +287,14 @@ where }; if packet.packet_type == CultNetRudpPacketType::Connect { - self.accept_connection(key, &packet, now_monotonic)?; + if !self.accept_connection(key, &packet, now_monotonic)? { + self.packets_dropped += 1; + } return Ok(CultMeshRudpPollOutcome::Handled); } if !self.sessions.contains_key(&key) { + self.packets_dropped += 1; return Ok(CultMeshRudpPollOutcome::Handled); } if packet.packet_type == CultNetRudpPacketType::Data @@ -309,7 +325,8 @@ where let result = match result { Ok(result) => result, Err(_) => { - self.sessions.remove(&key); + self.end_refused_session(key)?; + self.packets_dropped += 1; return Ok(CultMeshRudpPollOutcome::Handled); } }; @@ -385,45 +402,54 @@ where }) } + /// `Ok(false)` when the Connect was refused because the session cap is full; + /// only a socket failure is an `Err`. fn accept_connection( &mut self, key: CultMeshRudpSessionKey, packet: &CultNetRudpPacket, now: u64, - ) -> Result<()> { - if !self.sessions.contains_key(&key) { - if self.sessions.len() >= self.options.max_sessions { - return Ok(()); - } - self.sessions.insert( - key, - SessionEntry { - session: CultNetRudpSession::new(CultNetRudpSessionOptions { - connection_id: key.connection_id, - initial_sequence: 1, - resend_delay_ms: duration_millis(self.options.resend_delay), - max_pending_reliable_packets: Some( - self.options.max_pending_reliable_packets_per_session, - ), - }), - created_at_monotonic_millis: now, - last_activity_monotonic_millis: now, - admitted_payload_bytes: 0, - }, - ); - } - - let accept = { - let entry = self - .sessions - .get_mut(&key) - .expect("accepted session must be present"); + ) -> Result { + let reply = match self.sessions.get_mut(&key) { // The same key is a retransmitted Connect for the existing epoch. // A genuinely fresh client incarnation must choose a fresh id. - entry.last_activity_monotonic_millis = now; - entry.session.accept_connect(packet, now, Vec::new())? + Some(entry) => { + entry.last_activity_monotonic_millis = now; + match entry.session.answer_repeated_connect(packet, now) { + Ok(reply) => reply, + Err(_) => { + self.end_refused_session(key)?; + return Ok(false); + } + } + } + None => { + if self.sessions.len() >= self.options.max_sessions { + return Ok(false); + } + let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: key.connection_id, + initial_sequence: 1, + resend_delay_ms: duration_millis(self.options.resend_delay), + max_pending_reliable_packets: Some( + self.options.max_pending_reliable_packets_per_session, + ), + }); + let accept = session.accept_connect(packet, now, Vec::new())?; + self.sessions.insert( + key, + SessionEntry { + session, + created_at_monotonic_millis: now, + last_activity_monotonic_millis: now, + admitted_payload_bytes: 0, + }, + ); + accept + } }; - self.send_packet(key.remote_addr, &accept) + self.send_packet(key.remote_addr, &reply)?; + Ok(true) } fn deliver_application_message( @@ -566,6 +592,16 @@ where Ok(None) } + /// The session refused a packet: it ends, and the client is told so it does + /// not wait on a session the server no longer holds. + fn end_refused_session(&mut self, key: CultMeshRudpSessionKey) -> Result<()> { + let Some(mut entry) = self.sessions.remove(&key) else { + return Ok(()); + }; + let goodbye = entry.session.end_refused_session(); + self.send_packet(key.remote_addr, &goodbye) + } + fn payload_budget_allows(&self, key: CultMeshRudpSessionKey, bytes: usize) -> bool { let Some(session_bytes) = self .sessions diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index df14b4ea..fa741827 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -553,3 +553,103 @@ fn session_cap_and_expiry_use_monotonic_time() -> Result<()> { assert_eq!(server.session_count(), 0); Ok(()) } + +#[test] +fn a_connect_storm_and_stray_frames_are_dropped_not_fatal() -> Result<()> { + use cultnet_rs::{ + CultNetRudpSendOptions, CultNetRudpSession, CultNetRudpSessionOptions, + encode_rudp_packet, + }; + let clock = Clock::new(60_000); + let sink = Sink::default(); + let options = CultMeshRudpDocumentServerOptions { + max_pending_reliable_packets_per_session: 4, + ..Default::default() + }; + let mut server = server(options, clock.clone(), sink.clone(), Source::default())?; + let target = server.local_addr()?; + let mut real = client(target, 101)?; + connect(&mut server, &mut [&mut real])?; + assert_eq!(server.packets_dropped(), 0); + + let raw = UdpSocket::bind("127.0.0.1:0")?; + let session = |id| { + CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: id, + initial_sequence: 1, + resend_delay_ms: 10, + max_pending_reliable_packets: None, + }) + }; + let send_raw = |server: &mut Server, packet: &cultnet_rs::CultNetRudpPacket| -> Result<()> { + raw.send_to(&encode_rudp_packet(packet)?, target)?; + server.poll_once()?; + Ok(()) + }; + + // A malformed frame. + raw.send_to(b"not a rudp packet", target)?; + server.poll_once()?; + assert_eq!(server.packets_dropped(), 1); + + // Data for a session no Connect admitted. + let mut unadmitted = session(300); + unadmitted.assume_connected(0); + let reliable = CultNetRudpSendOptions { + reliable: true, + ..Default::default() + }; + send_raw(&mut server, &unadmitted.send("schema", vec![1], reliable.clone())?)?; + assert_eq!(server.packets_dropped(), 2); + + // A moved flow re-sends Connect from a socket that never hears the Accept. + // Each repeat re-sends the Accept still awaiting acknowledgement and queues + // nothing, so the storm neither errors nor grows the session's queue. + let mut stormer = session(102); + let connect_packet = stormer.create_connect(0, Vec::new())?; + for _ in 0..200 { + send_raw(&mut server, &connect_packet)?; + } + assert_eq!(server.packets_dropped(), 2); + assert_eq!(server.session_count(), 2); + clock.set(60_000 + 1_000); + assert_eq!( + server.maintain()?.packets_resent, + 1, + "one Accept awaits acknowledgement, however many Connects repeated" + ); + + // A packet its own admitted session refuses ends that session, not the loop. + stormer.assume_connected(0); + let mut poison = stormer.send("schema", vec![1], reliable)?; + poison.fragment_count = 2; + poison.fragment_id = 0; + send_raw(&mut server, &poison)?; + assert_eq!(server.packets_dropped(), 3); + assert_eq!(server.session_count(), 1); + // ...and the refused client is told, after the Accepts its storm drew. + raw.set_read_timeout(Some(Duration::from_millis(200)))?; + let mut buffer = vec![0_u8; 65_535]; + let mut told = false; + while let Ok(received) = raw.recv(&mut buffer) { + told |= cultnet_rs::decode_rudp_packet(&buffer[..received])?.packet_type + == cultnet_rs::CultNetRudpPacketType::Disconnect; + } + assert!(told, "the refused client must be sent a goodbye"); + + send( + &mut real, + &CultNetMessage::DocumentPutRaw { + message_id: "after-storm".into(), + document: document("real", vec![9]), + }, + )?; + for _ in 0..20 { + server.poll_once()?; + if sink.0.lock().unwrap().receipts.len() == 1 { + break; + } + } + assert_eq!(sink.0.lock().unwrap().receipts.len(), 1); + Ok(()) +} diff --git a/packages/cultnet-py/src/cultnet_py/transport.py b/packages/cultnet-py/src/cultnet_py/transport.py index 73bd1e62..0f378613 100644 --- a/packages/cultnet-py/src/cultnet_py/transport.py +++ b/packages/cultnet-py/src/cultnet_py/transport.py @@ -23,6 +23,10 @@ class CultNetTransportStats: bytes_sent: int = 0 frames_received: int = 0 frames_sent: int = 0 + # Datagrams read and discarded because they belong to no session on this + # transport: malformed frames, another session's connection id, a sender + # that is not the peer, or a packet the session refuses. + packets_dropped: int = 0 @dataclass(frozen=True) @@ -406,6 +410,8 @@ class CultNetRudpSession: raise ValueError("RUDP max_pending_reliable_packets must be greater than zero") self.max_pending_reliable_packets = options.max_pending_reliable_packets self._next_sequence = _uint32(options.initial_sequence, "initial_sequence") + if self._next_sequence == 0xFFFFFFFF: + raise ValueError("RUDP initial_sequence must leave room for a reliable packet") self._next_sequenced_by_channel: dict[str, int] = {} self._next_fragment_id = 1 self._connected = False @@ -449,6 +455,21 @@ class CultNetRudpSession: def outstanding_reliable_packet_count(self) -> int: return len(self._pending_reliable) + len(self._queued_reliable) + def reset_peer_state(self) -> None: + """Forgets everything learned from the peer; sequence numbers already issued stay issued.""" + self._connected = False + self._last_received_at_ms = None + self._highest_received_sequence = None + self._received_sequences.clear() + self._next_sequenced_by_channel.clear() + self._latest_sequenced_by_channel.clear() + self._pending_reliable.clear() + self._queued_reliable.clear() + self._ordered_next_sequence_by_channel.clear() + self._ordered_buffers.clear() + self._fragment_buffers.clear() + self._fragment_sets_evicted = 0 + def create_connect(self, now_ms: int = 0, payload: bytes = b"") -> CultNetRudpPacket: self._ensure_reliable_capacity(1) packet = self._create_packet( @@ -484,6 +505,25 @@ class CultNetRudpSession: self._track_reliable(response, now_ms) return response + def answer_repeated_connect(self, packet: CultNetRudpPacket, now_ms: int = 0) -> CultNetRudpPacket: + """Answers a Connect from a peer this session already accepted. + + The packet's sequence is remembered and nothing is queued, so a Connect + storm cannot grow the reliable queue. The reply is the Accept still awaiting + acknowledgement, or an Ack once it was acknowledged. + """ + self._require_connection(packet) + if packet.packet_type != CultNetRudpPacketType.CONNECT: + raise ValueError(f"Expected RUDP connect packet, got {packet.packet_type.value}") + self._apply_acknowledgements(packet) + self._remember_received(packet.sequence) + self._last_received_at_ms = now_ms + for pending in self._pending_reliable.values(): + if pending.packet.packet_type == CultNetRudpPacketType.ACCEPT: + pending.last_sent_at_ms = now_ms + return pending.packet + return self.create_ack() + def send( self, channel_id: str, @@ -921,6 +961,7 @@ class CultNetRudpSocketTransportConnection: self._bytes_sent = 0 self._frames_received = 0 self._frames_sent = 0 + self._packets_dropped = 0 self._delivered_frames: deque[CultNetTransportFrame] = deque() self._closed = False self.disconnect_reason: bytes | None = None @@ -943,6 +984,7 @@ class CultNetRudpSocketTransportConnection: bytes_sent=self._bytes_sent, frames_received=self._frames_received, frames_sent=self._frames_sent, + packets_dropped=self._packets_dropped, ) @property @@ -952,6 +994,7 @@ class CultNetRudpSocketTransportConnection: def connect(self, payload: bytes = b"") -> None: if self.mode != CultNetRudpSocketMode.CLIENT: raise ValueError("Only a client RUDP socket transport can initiate connect") + self.disconnect_reason = None self._send_packet(self.session.create_connect(_now_ms(), payload)) def send(self, channel_id: str, payload: bytes) -> None: @@ -977,17 +1020,53 @@ class CultNetRudpSocketTransportConnection: return None self._bytes_received += len(wire) + # What a datagram carries is the sender's business, not a fault of this + # process: a malformed frame, another session's connection id, a sender + # that is not the peer and a packet the session refuses are dropped and + # counted. Only the socket ends a loop. The id is checked before the peer + # endpoint is claimed, so a stray cannot become the peer. + try: + packet = decode_rudp_packet(wire) + except ValueError: + self._packets_dropped += 1 + return None + if packet.connection_id != self.session.connection_id: + self._packets_dropped += 1 + return None if self.remote_addr is None: + # Only a Connect claims the endpoint of a server-mode transport. + if self.mode == CultNetRudpSocketMode.SERVER and packet.packet_type != CultNetRudpPacketType.CONNECT: + self._packets_dropped += 1 + return None self.remote_addr = remote_addr elif remote_addr != self.remote_addr: + self._packets_dropped += 1 return None - packet = decode_rudp_packet(wire) if self.mode == CultNetRudpSocketMode.SERVER and packet.packet_type == CultNetRudpPacketType.CONNECT: - self._send_packet(self.session.accept_connect(packet, _now_ms())) + # A Connect from the peer this transport already accepted repeats: + # answer it with the Accept already owed and queue nothing. + try: + if self.session.connected: + reply = self.session.answer_repeated_connect(packet, _now_ms()) + else: + reply = self.session.accept_connect(packet, _now_ms()) + self.disconnect_reason = None + except ValueError: + self._packets_dropped += 1 + return None + self._send_packet(reply) return None - result = self.session.receive(packet, _now_ms()) + try: + result = self.session.receive(packet, _now_ms()) + except ValueError: + # receive() has already recorded the packet's reliable sequence, so the + # session cannot be kept: a retransmit would be acknowledged and the + # frame silently lost. End it and tell the peer. + self._packets_dropped += 1 + self._end_refused_session() + return None if result.reply is not None: self._send_packet(result.reply) for ready in result.ready_to_send: @@ -1027,7 +1106,16 @@ class CultNetRudpSocketTransportConnection: self._delivered_frames.clear() self.socket.settimeout(poll_timeout) try: - while self.session.outstanding_reliable_packet_count > 0: + while True: + # An ended session forgot its unacknowledged writes; reporting + # them flushed would be a lie. + if self.disconnect_reason is not None: + raise ConnectionError( + "RUDP session ended before its reliable writes were acknowledged: " + + self.disconnect_reason.decode("utf-8", errors="replace") + ) + if self.session.outstanding_reliable_packet_count == 0: + return if time.monotonic() >= deadline: raise TimeoutError( "RUDP reliable flush timed out with " @@ -1061,6 +1149,21 @@ class CultNetRudpSocketTransportConnection: self._closed = True self.socket.close() + def _end_refused_session(self) -> None: + reason = b"session refused a packet" + # The goodbye is built after the reset, or its ack field would acknowledge + # the very frame the session refused. + self.session.reset_peer_state() + goodbye = self.session.create_disconnect(reason) + self.disconnect_reason = reason + try: + self._send_packet(goodbye) + except OSError: + pass # Best effort: the session ends whether or not the peer hears it. + if self.mode == CultNetRudpSocketMode.SERVER: + # Only a Connect can claim the endpoint again. + self.remote_addr = None + def _send_packet(self, packet: CultNetRudpPacket) -> None: if self.remote_addr is None: raise ValueError("RUDP socket transport does not have a remote endpoint") diff --git a/packages/cultnet-py/tests/test_cultnet.py b/packages/cultnet-py/tests/test_cultnet.py index 475305d3..7e78c48f 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/packages/cultnet-py/tests/test_cultnet.py @@ -8,7 +8,7 @@ import unittest import hashlib import hmac import io -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import UTC, datetime from pathlib import Path from typing import Callable @@ -695,6 +695,15 @@ class CultNetTests(unittest.TestCase): self.assertTrue(client.check_timeout(91, 50)) self.assertFalse(client.connected) + def test_a_repeated_connect_acknowledges_what_it_carries(self) -> None: + client = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=103, initial_sequence=1)) + server = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=103, initial_sequence=100)) + connect = client.create_connect(0, b"join") + accept = server.accept_connect(connect, 0) + reply = server.answer_repeated_connect(replace(connect, ack=accept.sequence), 1) + self.assertEqual(reply.packet_type, CultNetRudpPacketType.ACK) + self.assertEqual(server.outstanding_reliable_packet_count, 0) + def test_cultnet_rudp_session_bounds_pending_reliable_packets_before_enqueue(self) -> None: session = CultNetRudpSession( CultNetRudpSessionOptions(connection_id=102, initial_sequence=1, max_pending_reliable_packets=2) @@ -918,6 +927,233 @@ class CultNetTests(unittest.TestCase): client.close() server.close() + def test_cultnet_rudp_socket_transport_drops_strays_and_keeps_serving_its_peer(self) -> None: + server_socket = bind_udp_socket() + client_socket = bind_udp_socket() + stray_socket = bind_udp_socket() + connection_id = 0x10203050 + server = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-server", + socket=server_socket, + mode=CultNetRudpSocketMode.SERVER, + connection_id=connection_id, + ) + ) + client = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-client", + socket=client_socket, + mode=CultNetRudpSocketMode.CLIENT, + remote_addr=server_socket.getsockname(), + connection_id=connection_id, + ) + ) + foreign = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=0x0BADF10E)) + strays = [ + b"not a rudp packet", + b"", + encode_rudp_packet(foreign.create_connect(0, b"foreign")), + ] + + try: + # A stray that arrives before the peer must not become the peer. + for stray in strays: + stray_socket.sendto(stray, server_socket.getsockname()) + for _ in strays: + self.assertIsNone(server.receive_once()) + self.assertEqual(server.stats.packets_dropped, len(strays)) + + client.connect(b"join") + pump_rudp_handshake(client, server) + + # A moved flow answers from the peer's own address with someone else's session. + for stray in strays: + server_socket.sendto(stray, client_socket.getsockname()) + client_socket.sendto(stray, server_socket.getsockname()) + for _ in strays: + self.assertIsNone(client.receive_once()) + self.assertIsNone(server.receive_once()) + self.assertEqual(client.stats.packets_dropped, len(strays)) + self.assertEqual(server.stats.packets_dropped, 2 * len(strays)) + + client.send("schema", b"still here") + self.assertEqual(receive_rudp_frame(server).payload, b"still here") + server.send("schema", b"server still here") + self.assertEqual(receive_rudp_frame(client).payload, b"server still here") + finally: + stray_socket.close() + client.close() + server.close() + + def test_cultnet_rudp_server_mode_is_claimed_only_by_a_connect(self) -> None: + server_socket = bind_udp_socket() + orphan_socket = bind_udp_socket() + client_socket = bind_udp_socket() + connection_id = 0x10203052 + server = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-server", + socket=server_socket, + mode=CultNetRudpSocketMode.SERVER, + connection_id=connection_id, + ) + ) + client = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-client", + socket=client_socket, + mode=CultNetRudpSocketMode.CLIENT, + remote_addr=server_socket.getsockname(), + connection_id=connection_id, + ) + ) + orphan = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id)) + try: + # A right-id packet that is not a Connect must claim nothing. + orphan_socket.sendto(encode_rudp_packet(orphan.create_ack()), server_socket.getsockname()) + self.assertIsNone(server.receive_once()) + self.assertEqual(server.stats.packets_dropped, 1) + self.assertIsNone(server.remote_addr) + + client.connect(b"join") + pump_rudp_handshake(client, server) + self.assertTrue(client.connected) + self.assertEqual(server.remote_addr, client_socket.getsockname()) + finally: + orphan_socket.close() + client.close() + server.close() + + def test_cultnet_rudp_socket_transport_ends_the_session_that_refuses_a_reliable_frame(self) -> None: + server_socket = bind_udp_socket() + peer_socket = bind_udp_socket() + connection_id = 0x10203053 + server = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-server", + socket=server_socket, + mode=CultNetRudpSocketMode.SERVER, + connection_id=connection_id, + ) + ) + peer = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id)) + + def to_server(packet) -> None: + peer_socket.sendto(encode_rudp_packet(packet), server_socket.getsockname()) + + def drain_peer() -> list: + packets = [] + while True: + try: + wire, _ = peer_socket.recvfrom(65535) + except TimeoutError: + return packets + packets.append(decode_rudp_packet(wire)) + + try: + to_server(peer.create_connect(0, b"join")) + server.receive_once() + accept = [p for p in drain_peer() if p.packet_type == CultNetRudpPacketType.ACCEPT][0] + peer.receive(accept, 0) + + (frame,) = peer.send_many("schema", b"poison", CultNetRudpSendOptions(reliable=True, ordered=True)) + poison = replace(frame, fragment_count=2, fragment_id=0) + to_server(poison) + self.assertIsNone(server.receive_once()) + self.assertEqual(server.stats.packets_dropped, 1) + self.assertFalse(server.connected) + self.assertEqual(server.disconnect_reason, b"session refused a packet") + + # The peer is told, and the goodbye does not acknowledge the refused frame. + (goodbye,) = [p for p in drain_peer() if p.packet_type == CultNetRudpPacketType.DISCONNECT] + self.assertTrue(peer.receive(goodbye, 1).disconnected) + self.assertIn(poison.sequence, peer.pending_reliable_sequences) + + # A retransmit is dropped, not acknowledged. + to_server(poison) + self.assertIsNone(server.receive_once()) + self.assertEqual(server.stats.packets_dropped, 2) + self.assertEqual(drain_peer(), []) + self.assertIn(poison.sequence, peer.pending_reliable_sequences) + + # A Connect claims the endpoint again. + again = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id)) + to_server(again.create_connect(0, b"again")) + server.receive_once() + self.assertEqual( + [p.packet_type for p in drain_peer()], [CultNetRudpPacketType.ACCEPT] + ) + finally: + peer_socket.close() + server.close() + + def _server_mode_pair(self, connection_id: int): + server_socket = bind_udp_socket() + peer_socket = bind_udp_socket() + server = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-server", + socket=server_socket, + mode=CultNetRudpSocketMode.SERVER, + connection_id=connection_id, + ) + ) + peer = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id)) + + def to_server(packet) -> None: + peer_socket.sendto(encode_rudp_packet(packet), server_socket.getsockname()) + + def drain_peer() -> list: + packets = [] + while True: + try: + wire, _ = peer_socket.recvfrom(65535) + except TimeoutError: + return packets + packets.append(decode_rudp_packet(wire)) + + return server, server_socket, peer_socket, peer, to_server, drain_peer + + def test_cultnet_rudp_flush_fails_for_a_write_the_ended_session_forgot(self) -> None: + server, _, peer_socket, peer, to_server, drain_peer = self._server_mode_pair(0x10203054) + try: + to_server(peer.create_connect(0, b"join")) + server.receive_once() + peer.receive([p for p in drain_peer() if p.packet_type == CultNetRudpPacketType.ACCEPT][0], 0) + server.send("schema", b"never acknowledged") + (frame,) = peer.send_many("schema", b"poison", CultNetRudpSendOptions(reliable=True, ordered=True)) + to_server(replace(frame, fragment_count=2, fragment_id=0)) + self.assertIsNone(server.receive_once()) + with self.assertRaises(ConnectionError): + server.flush_reliable(0.2) + finally: + peer_socket.close() + server.close() + + def test_cultnet_rudp_server_mode_owes_one_reliable_accept_however_many_connects_repeat(self) -> None: + server, _, peer_socket, peer, to_server, drain_peer = self._server_mode_pair(0x10203055) + try: + connect = peer.create_connect(0, b"join") + for _ in range(20): + to_server(connect) + server.receive_once() + accepts = {p.sequence for p in drain_peer() if p.packet_type == CultNetRudpPacketType.ACCEPT} + self.assertEqual(len(accepts), 1) + self.assertLessEqual(server.outstanding_reliable_packet_count, 1) + finally: + peer_socket.close() + server.close() + + def test_cultnet_rudp_session_rejects_limits_that_cannot_admit_a_connect(self) -> None: + with self.assertRaises(ValueError): + CultNetRudpSession(CultNetRudpSessionOptions(connection_id=1, initial_sequence=0xFFFFFFFF)) + with self.assertRaises(ValueError): + CultNetRudpSession(CultNetRudpSessionOptions(connection_id=1, max_pending_reliable_packets=0)) + CultNetRudpSession( + CultNetRudpSessionOptions(connection_id=1, initial_sequence=0xFFFFFFFE, max_pending_reliable_packets=1) + ) + def test_cultnet_rudp_socket_transport_carries_fragmented_reliable_ordered_schema_frames(self) -> None: server_socket = bind_udp_socket() client_socket = bind_udp_socket() diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 174739be..3b348fe3 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -126,8 +126,10 @@ pub struct CultNetRudpSendOptions { } /// Identifies every transport packet belonging to one non-expiring reliable -/// send. Receipts are local to the session that issued them and cannot survive -/// a peer reset. +/// send. Receipts are local to the session that issued them: a peer reset or +/// a Disconnect received from the peer invalidates them for good. A local +/// `disconnect()` or a timeout does not; the writes it leaves owed carry into +/// the next session. #[derive(Clone, Debug, PartialEq, Eq)] pub struct CultNetRudpReliableSendReceipt { session_scope: Uuid, @@ -180,7 +182,6 @@ pub struct CultNetRudpSession { max_pending_reliable_packets: Option, max_payload_bytes: Option, max_pending_fragment_sets: usize, - initial_sequence: u32, next_sequence: u32, next_sequenced_by_channel: BTreeMap, next_fragment_id: u16, @@ -208,7 +209,6 @@ impl CultNetRudpSession { max_pending_reliable_packets: options.max_pending_reliable_packets, max_payload_bytes: None, max_pending_fragment_sets: 64, - initial_sequence: options.initial_sequence, next_sequence: options.initial_sequence, next_sequenced_by_channel: BTreeMap::new(), next_fragment_id: 1, @@ -280,6 +280,10 @@ impl CultNetRudpSession { Some(pending.packet.clone()) } + fn session_scope(&self) -> Uuid { + self.session_scope + } + fn reliable_send_status( &self, receipt: &CultNetRudpReliableSendReceipt, @@ -334,11 +338,12 @@ impl CultNetRudpSession { Ok(()) } + /// Forgets everything learned from the peer. Sequence numbers and fragment + /// ids already issued stay issued: a peer that still remembers them must + /// never see one reused. pub fn reset_peer_state(&mut self) { self.session_scope = Uuid::new_v4(); - self.next_sequence = self.initial_sequence; self.next_sequenced_by_channel.clear(); - self.next_fragment_id = 1; self.connected = false; self.last_received_at_ms = None; self.highest_received_sequence = None; @@ -351,6 +356,16 @@ impl CultNetRudpSession { self.fragment_buffers.clear(); } + /// Ends a session that refused a packet. `receive` has already recorded the + /// refused packet's reliable sequence, so the session cannot be kept: a + /// retransmit would be acknowledged and the frame silently lost. Returns the + /// goodbye for the peer, built after the reset or its ack field would + /// acknowledge the very frame that was refused. + pub fn end_refused_session(&mut self) -> CultNetRudpPacket { + self.reset_peer_state(); + self.create_disconnect(RUDP_REFUSED_PACKET_REASON.to_vec()) + } + pub fn create_connect(&mut self, now_ms: u64, payload: Vec) -> Result { self.ensure_reliable_capacity(1)?; let packet = self.create_packet( @@ -395,6 +410,30 @@ impl CultNetRudpSession { Ok(response) } + /// Answers a Connect from a peer this session has already accepted. The + /// packet's sequence is remembered and nothing is queued, so a Connect storm + /// cannot grow the reliable queue. The reply is the Accept still awaiting + /// acknowledgement, or an Ack once it was acknowledged. + pub fn answer_repeated_connect( + &mut self, + packet: &CultNetRudpPacket, + now_ms: u64, + ) -> Result { + self.require_connection(packet)?; + if packet.packet_type != CultNetRudpPacketType::Connect { + return Err(anyhow!( + "Expected RUDP connect packet, got {:?}", + packet.packet_type + )); + } + self.apply_acknowledgements(packet); + self.remember_received(packet.sequence); + self.last_received_at_ms = Some(now_ms); + Ok(self + .pending_accept_for_resend(now_ms) + .unwrap_or_else(|| self.create_ack())) + } + pub fn send( &mut self, channel_id: &str, @@ -555,7 +594,13 @@ impl CultNetRudpSession { } if packet.packet_type == CultNetRudpPacketType::Disconnect { + // The session is over. What it still owed the peer dies with it, + // and receipts it issued can never become Acknowledged in a later + // session. self.connected = false; + self.session_scope = Uuid::new_v4(); + self.pending_reliable.clear(); + self.queued_reliable.clear(); return Ok(CultNetRudpReceiveResult { delivered: Vec::new(), ready_to_send, @@ -1399,6 +1444,10 @@ pub struct CultNetRudpServerHub { impl CultNetRudpServerHub { pub fn new(options: CultNetRudpServerHubOptions) -> Result { + validate_rudp_session_limits( + options.initial_sequence, + options.max_pending_reliable_packets, + )?; let local_addr = options.socket.local_addr()?; if options.max_peers == 0 { return Err(anyhow!( @@ -1572,8 +1621,15 @@ impl CultNetRudpServerHub { }; wire.truncate(received); self.stats.bytes_received += received as u64; - let packet = decode_rudp_packet(&wire)?; + // What a datagram carries is the sender's business, not a fault of this + // process: a malformed frame, another session's id and a sender no + // Connect admitted are dropped and counted. Only the socket ends a loop. + let Ok(packet) = decode_rudp_packet(&wire) else { + self.stats.packets_dropped += 1; + return Ok(true); + }; if packet.connection_id != self.connection_id { + self.stats.packets_dropped += 1; return Ok(true); } @@ -1581,16 +1637,16 @@ impl CultNetRudpServerHub { if let Some(peer) = self.peers.get_mut(&remote_addr) && peer.context.connect_payload == packet.payload { - let _ = peer.session.receive(&packet, now_ms())?; - let reply = peer - .session - .pending_accept_for_resend(now_ms()) - .unwrap_or_else(|| peer.session.create_ack()); + let Ok(reply) = peer.session.answer_repeated_connect(&packet, now_ms()) else { + self.end_refusing_peer(remote_addr)?; + return Ok(true); + }; self.send_packet(remote_addr, &reply)?; 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")); + self.stats.packets_dropped += 1; + return Ok(true); } let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions { connection_id: self.connection_id, @@ -1610,7 +1666,10 @@ impl CultNetRudpServerHub { session_generation: generation, connect_payload: packet.payload.clone(), }; - let accept = session.accept_connect(&packet, now_ms(), Vec::new())?; + let Ok(accept) = session.accept_connect(&packet, now_ms(), Vec::new()) else { + self.stats.packets_dropped += 1; + return Ok(true); + }; self.send_packet(remote_addr, &accept)?; let replaced = self.peers.insert( remote_addr, @@ -1632,9 +1691,13 @@ impl CultNetRudpServerHub { } let Some(peer) = self.peers.get_mut(&remote_addr) else { + self.stats.packets_dropped += 1; + return Ok(true); + }; + let Ok(result) = peer.session.receive(&packet, now_ms()) else { + self.end_refusing_peer(remote_addr)?; return Ok(true); }; - let result = peer.session.receive(&packet, now_ms())?; let context = peer.context.clone(); let ack = if packet.reliable { Some(peer.session.create_ack_for_received(packet.sequence)) @@ -1678,6 +1741,24 @@ impl CultNetRudpServerHub { Ok(true) } + /// An admitted peer sent a packet its session refuses. `receive` has + /// already recorded the packet's reliable sequence, so the session cannot be + /// kept: a retransmit would be acknowledged and the frame silently lost. + /// The session ends, the peer is told, and the caller sees `Disconnected`. + fn end_refusing_peer(&mut self, remote_addr: SocketAddr) -> Result<()> { + self.stats.packets_dropped += 1; + let Some(mut peer) = self.peers.remove(&remote_addr) else { + return Ok(()); + }; + let goodbye = peer.session.end_refused_session(); + self.pending_events + .push_back(CultNetRudpServerEvent::Disconnected { + session: peer.context, + reason: goodbye.payload.clone(), + }); + self.send_packet(remote_addr, &goodbye) + } + pub fn poll_resends(&mut self) -> Result<()> { let now = now_ms(); let mut packets = Vec::new(); @@ -1725,6 +1806,10 @@ impl CultNetRudpServerHub { impl CultNetRudpSocketTransportConnection { pub fn new(options: CultNetRudpSocketTransportOptions) -> Result { + validate_rudp_session_limits( + options.initial_sequence, + options.max_pending_reliable_packets, + )?; let local_addr = options.socket.local_addr()?; let max_payload_bytes = options .max_payload_bytes @@ -1793,6 +1878,7 @@ impl CultNetRudpSocketTransportConnection { )); } let packet = self.session.create_connect(now_ms(), payload)?; + self.disconnect_reason = None; self.send_packet(&packet) } @@ -1911,9 +1997,23 @@ impl CultNetRudpSocketTransportConnection { self.socket.set_read_timeout(Some(poll_timeout))?; let deadline = Instant::now() + timeout; + // A flush belongs to the session it started in: a Connect that + // replaces the peer mid-wait forgets the writes being waited on. + let scope = self.session.session_scope(); let mut preserved_frames = VecDeque::new(); let result = (|| { - while self.outstanding_reliable_packet_count() > 0 { + loop { + // An ended session forgot its unacknowledged writes; reporting + // them flushed would be a lie. + if self.session.session_scope() != scope || self.disconnect_reason.is_some() { + return Err(anyhow!( + "RUDP session ended before its reliable writes were acknowledged: {}", + String::from_utf8_lossy(self.disconnect_reason.as_deref().unwrap_or(b"replaced by a new Connect")) + )); + } + if self.outstanding_reliable_packet_count() == 0 { + return Ok(()); + } if Instant::now() >= deadline { return Err(anyhow!( "RUDP reliable flush timed out with {} packets outstanding", @@ -1925,7 +2025,6 @@ impl CultNetRudpSocketTransportConnection { } self.poll_resends()?; } - Ok(()) })(); preserved_frames.append(&mut self.delivered_frames); @@ -1970,7 +2069,20 @@ impl CultNetRudpSocketTransportConnection { wire.truncate(received); self.stats.bytes_received += received as u64; - let packet = decode_rudp_packet(&wire)?; + // What a datagram carries is the sender's business, not a fault of this + // process: a malformed frame, another session's id, a sender that is not + // this transport's peer and a packet the session refuses are dropped and + // counted. Only the socket ends a loop. The id is checked before the + // peer endpoint or session state is touched, so a stray Connect can + // neither move the endpoint nor reset the session. + let Ok(packet) = decode_rudp_packet(&wire) else { + self.stats.packets_dropped += 1; + return Ok(true); + }; + if packet.connection_id != self.session.connection_id() { + self.stats.packets_dropped += 1; + return Ok(true); + } if let Some(expected) = self.remote_addr { if expected != remote_addr { if self.mode == CultNetRudpSocketMode::Server @@ -1978,6 +2090,7 @@ impl CultNetRudpSocketTransportConnection { { self.remote_addr = Some(remote_addr); } else { + self.stats.packets_dropped += 1; return Ok(true); } } @@ -1985,6 +2098,7 @@ impl CultNetRudpSocketTransportConnection { if self.mode == CultNetRudpSocketMode::Server && packet.packet_type != CultNetRudpPacketType::Connect { + self.stats.packets_dropped += 1; return Ok(true); } self.remote_addr = Some(remote_addr); @@ -1993,12 +2107,28 @@ impl CultNetRudpSocketTransportConnection { && packet.packet_type == CultNetRudpPacketType::Connect { self.session.reset_peer_state(); - let accept = self.session.accept_connect(&packet, now_ms(), Vec::new())?; + self.disconnect_reason = None; + let Ok(accept) = self.session.accept_connect(&packet, now_ms(), Vec::new()) else { + self.stats.packets_dropped += 1; + return Ok(true); + }; self.send_packet(&accept)?; return Ok(true); } - let result = self.session.receive(&packet, now_ms())?; + let Ok(result) = self.session.receive(&packet, now_ms()) else { + // End the session and tell the peer. A server-mode transport also + // forgets its peer: only a Connect can claim it again. + self.stats.packets_dropped += 1; + let goodbye = self.session.end_refused_session(); + self.disconnect_reason = Some(goodbye.payload.clone()); + let sent = self.send_packet(&goodbye); + if self.mode == CultNetRudpSocketMode::Server { + self.remote_addr = None; + } + sent?; + return Ok(false); + }; if let Some(reply) = result.reply { self.send_packet(&reply)?; } @@ -2142,6 +2272,29 @@ where } } +/// Sent to a peer whose session refused one of its packets, and reported to +/// the local caller as the disconnect reason. +const RUDP_REFUSED_PACKET_REASON: &[u8] = b"session refused a packet"; + +/// A session cannot admit a Connect when its sequence space starts exhausted or +/// its reliable queue holds nothing, so neither is a usable configuration. +fn validate_rudp_session_limits( + initial_sequence: u32, + max_pending_reliable_packets: Option, +) -> Result<()> { + if initial_sequence == u32::MAX { + return Err(anyhow!( + "RUDP initial_sequence must leave room for a reliable packet" + )); + } + if max_pending_reliable_packets == Some(0) { + return Err(anyhow!( + "RUDP max_pending_reliable_packets must be greater than zero" + )); + } + Ok(()) +} + #[derive(Clone, Debug, Default)] pub struct RudpTransportProfileOptions { pub transport_id: Option, diff --git a/packages/cultnet-rs/src/transport.rs b/packages/cultnet-rs/src/transport.rs index d097aa3c..71e714c4 100644 --- a/packages/cultnet-rs/src/transport.rs +++ b/packages/cultnet-rs/src/transport.rs @@ -26,6 +26,11 @@ pub struct CultNetTransportStats { /// payload that lost a fragment on the wire. Zero for non-fragmenting /// transports. pub fragment_sets_evicted: u64, + /// Datagrams read and discarded because they belong to no session on this + /// transport: malformed frames, another session's connection id, an + /// unadmitted sender, or a packet the session refused. A steady rate means + /// a flow is being moved or scanned, not that this transport is failing. + pub packets_dropped: u64, } #[derive(Clone, Debug, PartialEq, Eq)] diff --git a/packages/cultnet-rs/tests/rudp_server_hub.rs b/packages/cultnet-rs/tests/rudp_server_hub.rs index 201cf3b4..80cfde3b 100644 --- a/packages/cultnet-rs/tests/rudp_server_hub.rs +++ b/packages/cultnet-rs/tests/rudp_server_hub.rs @@ -415,16 +415,13 @@ fn session_and_hub_reject_configured_memory_bounds() -> Result<()> { let mut second = client(server_addr)?; connect(&mut hub, &mut first, b"first")?; second.connect(b"second".to_vec())?; - for _ in 0..20 { - match hub.receive_event_once() { - Err(error) => { - assert!(error.to_string().contains("peer limit")); - return Ok(()); - } - Ok(_) => continue, - } - } - Err(anyhow!("hub did not enforce peer limit")) + // The peer limit refuses the second Connect without ending the hub loop: + // a full table is not a fault of the process that holds it. + std::thread::sleep(Duration::from_millis(20)); + while hub.receive_event_once()?.is_some() {} + assert_eq!(hub.sessions().len(), 1); + assert_eq!(hub.stats().packets_dropped, 1); + Ok(()) } #[test] diff --git a/packages/cultnet-rs/tests/rudp_stray_packets.rs b/packages/cultnet-rs/tests/rudp_stray_packets.rs new file mode 100644 index 00000000..33f02794 --- /dev/null +++ b/packages/cultnet-rs/tests/rudp_stray_packets.rs @@ -0,0 +1,752 @@ +//! A packet that does not belong to a session is dropped; only socket failures +//! end a transport loop. Each test puts one class of stray in front of a real +//! session and requires the real session to keep being served. + +use std::net::{SocketAddr, UdpSocket}; +use std::time::Duration; + +use anyhow::Result; +use cultnet_rs::*; + +const CONNECTION_ID: u32 = 0x4355_4c54; +const FOREIGN_ID: u32 = 0x0BAD_F10E; + +#[derive(Clone, Copy)] +enum Stray { + /// Bytes that are not a RUDP frame. + Malformed, + /// A well-formed session that carries someone else's connection id: what a + /// moved UDP flow delivers. + ForeignConnectionId, + /// The right connection id from a sender no Connect ever admitted, or one + /// that already closed. + UnadmittedSender, +} + +fn socket() -> Result { + let socket = UdpSocket::bind("127.0.0.1:0")?; + socket.set_read_timeout(Some(Duration::from_millis(20)))?; + Ok(socket) +} + +fn raw_session(connection_id: u32) -> CultNetRudpSession { + CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id, + initial_sequence: 1, + resend_delay_ms: 250, + max_pending_reliable_packets: None, + }) +} + +fn reliable() -> CultNetRudpSendOptions { + CultNetRudpSendOptions { + reliable: true, + ..Default::default() + } +} + +fn send_to(socket: &UdpSocket, to: SocketAddr, packet: &CultNetRudpPacket) -> Result<()> { + socket.send_to(&encode_rudp_packet(packet)?, to)?; + Ok(()) +} + +/// How many datagrams `send_stray` puts on the wire for `kind`. +fn datagrams(kind: Stray) -> u64 { + match kind { + Stray::Malformed => 2, + Stray::ForeignConnectionId => 4, + Stray::UnadmittedSender => 3, + } +} + +/// A packet the sender's own session accepted for sending and its receiver's +/// session refuses: a fragment with no fragment id. +fn poisoned(session: &mut CultNetRudpSession) -> Result { + let mut packet = session.send("schema", b"poison".to_vec(), reliable())?; + packet.fragment_count = 2; + packet.fragment_id = 0; + Ok(packet) +} + +fn send_stray(kind: Stray, from: &UdpSocket, to: SocketAddr) -> Result<()> { + match kind { + Stray::Malformed => { + from.send_to(b"not a rudp packet", to)?; + from.send_to(&[], to)?; + } + Stray::ForeignConnectionId => { + let mut foreign = raw_session(FOREIGN_ID); + send_to(from, to, &foreign.create_connect(0, b"foreign".to_vec())?)?; + foreign.assume_connected(0); + send_to(from, to, &foreign.send("schema", b"foreign".to_vec(), reliable())?)?; + send_to(from, to, &foreign.create_ack())?; + send_to(from, to, &foreign.create_disconnect(b"foreign".to_vec()))?; + } + Stray::UnadmittedSender => { + let mut orphan = raw_session(CONNECTION_ID); + orphan.assume_connected(0); + send_to(from, to, &orphan.send("schema", b"orphan".to_vec(), reliable())?)?; + send_to(from, to, &orphan.create_ack())?; + send_to(from, to, &orphan.create_disconnect(b"orphan".to_vec()))?; + } + } + Ok(()) +} + +fn hub_keeps_serving_its_session(kind: Stray) -> Result<()> { + let mut options = CultNetRudpServerHubOptions::new("hub", socket()?, CONNECTION_ID); + options.max_peers = 1; + let mut hub = CultNetRudpServerHub::new(options)?; + let hub_addr = hub.local_addr()?; + let mut real = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::client("real", socket()?, hub_addr, CONNECTION_ID), + )?; + real.connect(b"real".to_vec())?; + let mut admitted = false; + for _ in 0..20 { + if let Some(CultNetRudpServerEvent::Connected { .. }) = hub.receive_event_once()? { + admitted = true; + break; + } + } + assert!(admitted); + let _ = real.receive_once()?; + + let stray = socket()?; + send_stray(kind, &stray, hub_addr)?; + std::thread::sleep(Duration::from_millis(20)); + while hub.receive_event_once()?.is_some() {} + assert_eq!(hub.stats().packets_dropped, datagrams(kind), "every stray is counted once"); + + real.send("schema", b"still here".to_vec())?; + std::thread::sleep(Duration::from_millis(20)); + match hub.receive_event_once()? { + Some(CultNetRudpServerEvent::Frame { frame, .. }) => { + assert_eq!(frame.payload, b"still here"); + } + other => panic!("the real session must still be served, got {other:?}"), + } + Ok(()) +} + +#[test] +fn hub_drops_a_connect_that_arrives_with_the_peer_table_full() -> Result<()> { + let mut options = CultNetRudpServerHubOptions::new("hub", socket()?, CONNECTION_ID); + options.max_peers = 1; + let mut hub = CultNetRudpServerHub::new(options)?; + let hub_addr = hub.local_addr()?; + let mut real = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::client("real", socket()?, hub_addr, CONNECTION_ID), + )?; + real.connect(b"real".to_vec())?; + while !matches!( + hub.receive_event_once()?, + Some(CultNetRudpServerEvent::Connected { .. }) + ) {} + let _ = real.receive_once()?; + + send_to( + &socket()?, + hub_addr, + &raw_session(CONNECTION_ID).create_connect(0, b"late".to_vec())?, + )?; + std::thread::sleep(Duration::from_millis(20)); + while hub.receive_event_once()?.is_some() {} + assert_eq!(hub.sessions().len(), 1); + assert_eq!(hub.stats().packets_dropped, 1); + + real.send("schema", b"still here".to_vec())?; + std::thread::sleep(Duration::from_millis(20)); + assert!(matches!( + hub.receive_event_once()?, + Some(CultNetRudpServerEvent::Frame { .. }) + )); + Ok(()) +} + +/// A moved flow answers a client from its server address with someone else's +/// session. +fn client_keeps_its_session(kind: Stray) -> Result<()> { + let server = socket()?; + let server_addr = server.local_addr()?; + let mut client = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::client("client", socket()?, server_addr, CONNECTION_ID), + )?; + client.connect(b"hello".to_vec())?; + let mut buffer = vec![0_u8; 65_535]; + let (received, client_addr) = server.recv_from(&mut buffer)?; + let connect = decode_rudp_packet(&buffer[..received])?; + let mut server_session = raw_session(CONNECTION_ID); + let accept = server_session.accept_connect(&connect, 0, Vec::new())?; + send_to(&server, client_addr, &accept)?; + let _ = client.receive_once()?; + assert!(client.connected()); + + send_stray(kind, &server, client_addr)?; + send_to( + &server, + client_addr, + &server_session.send("schema", b"real".to_vec(), reliable())?, + )?; + let frame = client + .receive_once()? + .expect("the real frame behind the strays"); + assert_eq!(frame.payload, b"real"); + assert_eq!(client.stats().packets_dropped, datagrams(kind), "every stray is counted once"); + Ok(()) +} + +/// A server-mode transport owns one peer. A stray, including a foreign Connect +/// from another address, neither resets that peer's session nor moves the +/// endpoint the transport answers. +fn server_mode_keeps_its_peer(kind: Stray) -> Result<()> { + let server_socket = socket()?; + let server_addr = server_socket.local_addr()?; + let mut server = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID), + )?; + let peer = socket()?; + let mut peer_session = raw_session(CONNECTION_ID); + send_to(&peer, server_addr, &peer_session.create_connect(0, b"peer".to_vec())?)?; + let _ = server.receive_once()?; + let mut buffer = vec![0_u8; 65_535]; + let (received, _) = peer.recv_from(&mut buffer)?; + peer_session.receive(&decode_rudp_packet(&buffer[..received])?, 0)?; + + send_stray(kind, &socket()?, server_addr)?; + send_stray(kind, &peer, server_addr)?; + send_to( + &peer, + server_addr, + &peer_session.send("schema", b"real".to_vec(), reliable())?, + )?; + let frame = server + .receive_once()? + .expect("the real frame behind the strays"); + assert_eq!(frame.payload, b"real"); + // Once from an unrelated address, once from the peer's own. + assert_eq!(server.stats().packets_dropped, 2 * datagrams(kind), "every stray is counted once"); + Ok(()) +} + +macro_rules! per_stray { + ($body:ident, $malformed:ident, $foreign:ident, $unadmitted:ident) => { + #[test] + fn $malformed() -> Result<()> { + $body(Stray::Malformed) + } + #[test] + fn $foreign() -> Result<()> { + $body(Stray::ForeignConnectionId) + } + #[test] + fn $unadmitted() -> Result<()> { + $body(Stray::UnadmittedSender) + } + }; + ($body:ident, $malformed:ident, $foreign:ident) => { + #[test] + fn $malformed() -> Result<()> { + $body(Stray::Malformed) + } + #[test] + fn $foreign() -> Result<()> { + $body(Stray::ForeignConnectionId) + } + }; +} + +per_stray!( + hub_keeps_serving_its_session, + hub_survives_malformed_frames, + hub_survives_foreign_connection_ids, + hub_survives_unadmitted_senders +); +per_stray!( + client_keeps_its_session, + client_survives_malformed_frames, + client_survives_foreign_connection_ids +); +per_stray!( + server_mode_keeps_its_peer, + server_mode_survives_malformed_frames, + server_mode_survives_foreign_connection_ids +); + +/// A session that refuses a packet has already recorded its reliable sequence, +/// so keeping it would acknowledge the sender's retransmit and lose the frame. +/// The session ends and the sender is told. +fn refused_frame_options() -> (u32, Vec) { + (64, vec![7_u8; 100]) +} + +fn client_with_fast_resend(server_addr: SocketAddr) -> Result { + let mut options = + CultNetRudpSocketTransportOptions::client("client", socket()?, server_addr, CONNECTION_ID); + options.resend_delay_ms = 1; + CultNetRudpSocketTransportConnection::new(options) +} + +/// The client's view after the server refused its frame: told the session is +/// over, and never shown the frame as acknowledged, even after it retransmits. +fn assert_client_sees_refusal( + client: &mut CultNetRudpSocketTransportConnection, + receipt: &CultNetRudpReliableSendReceipt, + mut serve: impl FnMut() -> Result<()>, +) -> Result<()> { + serve()?; + let _ = client.receive_once()?; + assert!( + client.disconnect_reason().is_some(), + "the client must be told the session ended" + ); + std::thread::sleep(Duration::from_millis(10)); + client.poll_resends()?; + serve()?; + let _ = client.receive_once()?; + assert_ne!( + client.reliable_send_status(receipt), + CultNetRudpReliableSendStatus::Acknowledged, + "a refused frame must never be acknowledged" + ); + Ok(()) +} + +#[test] +fn hub_ends_the_session_of_a_peer_whose_reliable_frame_it_refuses() -> Result<()> { + let (limit, payload) = refused_frame_options(); + let mut options = CultNetRudpServerHubOptions::new("hub", socket()?, CONNECTION_ID); + options.max_payload_bytes = Some(limit); + let mut hub = CultNetRudpServerHub::new(options)?; + let hub_addr = hub.local_addr()?; + let mut client = client_with_fast_resend(hub_addr)?; + client.connect(b"peer".to_vec())?; + assert!(matches!( + hub.receive_event_once()?, + Some(CultNetRudpServerEvent::Connected { .. }) + )); + let _ = client.receive_once()?; + assert!(client.connected()); + + let receipt = client.send_reliable("schema", payload)?; + let mut ended = false; + assert_client_sees_refusal(&mut client, &receipt, || { + while let Some(event) = hub.receive_event_once()? { + ended |= matches!(event, CultNetRudpServerEvent::Disconnected { .. }); + } + Ok(()) + })?; + assert!(ended, "the hub must report the ended session"); + assert_eq!(hub.sessions().len(), 0); + assert!(hub.stats().packets_dropped >= 1); + Ok(()) +} + +#[test] +fn server_mode_ends_the_session_of_a_peer_whose_reliable_frame_it_refuses() -> Result<()> { + let (limit, payload) = refused_frame_options(); + let server_socket = socket()?; + let server_addr = server_socket.local_addr()?; + let mut options = CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID); + options.max_payload_bytes = Some(limit); + let mut server = CultNetRudpSocketTransportConnection::new(options)?; + let mut client = client_with_fast_resend(server_addr)?; + client.connect(b"peer".to_vec())?; + let _ = server.receive_once()?; + let _ = client.receive_once()?; + assert!(client.connected()); + + let receipt = client.send_reliable("schema", payload)?; + assert_client_sees_refusal(&mut client, &receipt, || { + let _ = server.receive_once()?; + Ok(()) + })?; + assert!(server.disconnect_reason().is_some()); + assert!(server.stats().packets_dropped >= 1); + Ok(()) +} + +#[test] +fn client_ends_its_session_when_it_refuses_a_packet() -> Result<()> { + let server = socket()?; + let server_addr = server.local_addr()?; + let mut client = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::client("client", socket()?, server_addr, CONNECTION_ID), + )?; + client.connect(b"hello".to_vec())?; + let mut buffer = vec![0_u8; 65_535]; + let (received, client_addr) = server.recv_from(&mut buffer)?; + let mut server_session = raw_session(CONNECTION_ID); + let accept = + server_session.accept_connect(&decode_rudp_packet(&buffer[..received])?, 0, Vec::new())?; + send_to(&server, client_addr, &accept)?; + let _ = client.receive_once()?; + + send_to(&server, client_addr, &poisoned(&mut server_session)?)?; + assert!(client.receive_once()?.is_none()); + assert_eq!(client.stats().packets_dropped, 1); + assert!(client.disconnect_reason().is_some()); + + // The server is told, not left holding a live session. + let mut told = false; + while let Ok((received, _)) = server.recv_from(&mut buffer) { + told |= decode_rudp_packet(&buffer[..received])?.packet_type + == CultNetRudpPacketType::Disconnect; + } + assert!(told, "the server must be told the session ended"); + Ok(()) +} + +#[test] +fn server_mode_ends_the_session_when_the_session_refuses_a_packet() -> Result<()> { + let server_socket = socket()?; + let server_addr = server_socket.local_addr()?; + let mut server = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID), + )?; + let stray = socket()?; + + // Before any peer: a right-id packet that is not a Connect claims nothing. + send_stray(Stray::UnadmittedSender, &stray, server_addr)?; + assert!(server.receive_once()?.is_none()); + assert_eq!(server.stats().packets_dropped, 3); + + let peer = socket()?; + let mut peer_session = raw_session(CONNECTION_ID); + send_to(&peer, server_addr, &peer_session.create_connect(0, b"peer".to_vec())?)?; + let _ = server.receive_once()?; + let mut buffer = vec![0_u8; 65_535]; + let (received, _) = peer.recv_from(&mut buffer)?; + peer_session.receive(&decode_rudp_packet(&buffer[..received])?, 0)?; + assert_eq!(server.stats().packets_dropped, 3); + + // After the peer is known: right-id packets from any other address. + send_stray(Stray::UnadmittedSender, &stray, server_addr)?; + assert!(server.receive_once()?.is_none()); + assert_eq!(server.stats().packets_dropped, 6); + + // A packet from the peer that the session itself refuses ends the session + // and the peer is told. + send_to(&peer, server_addr, &poisoned(&mut peer_session)?)?; + assert!(server.receive_once()?.is_none()); + assert_eq!(server.stats().packets_dropped, 7); + assert!(server.disconnect_reason().is_some()); + let (received, _) = peer.recv_from(&mut buffer)?; + assert_eq!( + decode_rudp_packet(&buffer[..received])?.packet_type, + CultNetRudpPacketType::Disconnect + ); + Ok(()) +} + +#[test] +fn constructors_reject_limits_that_make_a_connect_unadmittable() -> Result<()> { + let mut hub = CultNetRudpServerHubOptions::new("hub", socket()?, CONNECTION_ID); + hub.initial_sequence = u32::MAX; + assert!(CultNetRudpServerHub::new(hub).is_err()); + let mut hub = CultNetRudpServerHubOptions::new("hub", socket()?, CONNECTION_ID); + hub.max_pending_reliable_packets = Some(0); + assert!(CultNetRudpServerHub::new(hub).is_err()); + + for build in [ + (|s, a| CultNetRudpSocketTransportOptions::client("c", s, a, CONNECTION_ID)) + as fn(UdpSocket, SocketAddr) -> CultNetRudpSocketTransportOptions, + |s, _| CultNetRudpSocketTransportOptions::server("s", s, CONNECTION_ID), + ] { + let addr = "127.0.0.1:9".parse()?; + let mut options = build(socket()?, addr); + options.initial_sequence = u32::MAX; + assert!(CultNetRudpSocketTransportConnection::new(options).is_err()); + let mut options = build(socket()?, addr); + options.max_pending_reliable_packets = Some(0); + assert!(CultNetRudpSocketTransportConnection::new(options).is_err()); + let mut options = build(socket()?, addr); + options.initial_sequence = u32::MAX - 1; + options.max_pending_reliable_packets = Some(1); + assert!(CultNetRudpSocketTransportConnection::new(options).is_ok()); + } + Ok(()) +} + +/// The distinct reliable Accept sequences a peer has been sent so far. +fn accept_sequences(peer: &UdpSocket) -> Result> { + let mut buffer = vec![0_u8; 65_535]; + let mut sequences = std::collections::BTreeSet::new(); + while let Ok((received, _)) = peer.recv_from(&mut buffer) { + let packet = decode_rudp_packet(&buffer[..received])?; + if packet.packet_type == CultNetRudpPacketType::Accept && packet.reliable { + sequences.insert(packet.sequence); + } + } + Ok(sequences) +} + +/// A Connect that repeats (a moved flow, a retransmit, a storm) is answered +/// with the Accept already owed, never a fresh reliable Accept per Connect. +#[test] +fn hub_owes_one_reliable_accept_however_many_connects_repeat() -> Result<()> { + let mut hub = + CultNetRudpServerHub::new(CultNetRudpServerHubOptions::new("hub", socket()?, CONNECTION_ID))?; + let hub_addr = hub.local_addr()?; + let peer = socket()?; + let mut peer_session = raw_session(CONNECTION_ID); + let connect = peer_session.create_connect(0, b"peer".to_vec())?; + let mut connected = 0; + for _ in 0..20 { + send_to(&peer, hub_addr, &connect)?; + while let Some(event) = hub.receive_event_once()? { + assert!( + matches!(event, CultNetRudpServerEvent::Connected { .. }), + "a repeated Connect must not replace the admitted session" + ); + connected += 1; + } + } + assert_eq!(connected, 1); + assert_eq!(accept_sequences(&peer)?.len(), 1); + Ok(()) +} + +#[test] +fn server_mode_owes_one_reliable_accept_however_many_connects_repeat() -> Result<()> { + let server_socket = socket()?; + let server_addr = server_socket.local_addr()?; + let mut server = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID), + )?; + let peer = socket()?; + let mut peer_session = raw_session(CONNECTION_ID); + let connect = peer_session.create_connect(0, b"peer".to_vec())?; + for _ in 0..20 { + send_to(&peer, server_addr, &connect)?; + let _ = server.receive_once()?; + } + assert!(server.outstanding_reliable_packet_count() <= 1); + Ok(()) +} + +/// A late duplicate Connect resets what the server learned of the peer, never +/// what it already issued: a reused sequence is dropped by the peer as a +/// duplicate. +#[test] +fn server_mode_never_reuses_a_sequence_after_a_late_duplicate_connect() -> Result<()> { + let server_socket = socket()?; + let server_addr = server_socket.local_addr()?; + let mut server = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID), + )?; + let peer = socket()?; + let mut peer_session = raw_session(CONNECTION_ID); + let connect = peer_session.create_connect(0, b"peer".to_vec())?; + send_to(&peer, server_addr, &connect)?; + let _ = server.receive_once()?; + server.send_reliable("schema", b"before".to_vec())?; + send_to(&peer, server_addr, &connect)?; + let _ = server.receive_once()?; + server.send_reliable("schema", b"after".to_vec())?; + + let mut buffer = vec![0_u8; 65_535]; + let mut delivered = Vec::new(); + while let Ok((received, _)) = peer.recv_from(&mut buffer) { + let result = peer_session.receive(&decode_rudp_packet(&buffer[..received])?, 0)?; + delivered.extend(result.delivered.into_iter().map(|frame| frame.payload)); + } + assert!(delivered.contains(&b"before".to_vec())); + assert!( + delivered.contains(&b"after".to_vec()), + "the frame sent after the duplicate Connect was dropped as a reused sequence" + ); + Ok(()) +} + +/// A session that ended forgot its unacknowledged writes; a flush that then +/// reports success would claim writes the peer never acknowledged. +#[test] +fn a_write_unacknowledged_when_the_session_is_refused_fails_its_flush() -> Result<()> { + let server = socket()?; + let server_addr = server.local_addr()?; + let mut client = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::client("client", socket()?, server_addr, CONNECTION_ID), + )?; + client.connect(b"hello".to_vec())?; + let mut buffer = vec![0_u8; 65_535]; + let (received, client_addr) = server.recv_from(&mut buffer)?; + let mut server_session = raw_session(CONNECTION_ID); + let accept = + server_session.accept_connect(&decode_rudp_packet(&buffer[..received])?, 0, Vec::new())?; + send_to(&server, client_addr, &accept)?; + let _ = client.receive_once()?; + + let receipt = client.send_reliable("schema", b"never acknowledged".to_vec())?; + send_to(&server, client_addr, &poisoned(&mut server_session)?)?; + let _ = client.receive_once()?; + assert!(client.disconnect_reason().is_some()); + + let flushed = client.flush_reliable(Duration::from_millis(200)); + assert!(flushed.is_err(), "the write was never acknowledged"); + assert_eq!( + client.reliable_send_status(&receipt), + CultNetRudpReliableSendStatus::Invalidated + ); + Ok(()) +} + +/// A peer's Disconnect ends the session as surely as a refusal: writes it never +/// acknowledged are lost, and the transport says so. +#[test] +fn a_write_unacknowledged_when_the_peer_disconnects_fails_its_flush() -> Result<()> { + let server = socket()?; + let server_addr = server.local_addr()?; + let mut client = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::client("client", socket()?, server_addr, CONNECTION_ID), + )?; + client.connect(b"hello".to_vec())?; + let mut buffer = vec![0_u8; 65_535]; + let (received, client_addr) = server.recv_from(&mut buffer)?; + let mut server_session = raw_session(CONNECTION_ID); + let accept = + server_session.accept_connect(&decode_rudp_packet(&buffer[..received])?, 0, Vec::new())?; + send_to(&server, client_addr, &accept)?; + let _ = client.receive_once()?; + + let receipt = client.send_reliable("schema", b"never acknowledged".to_vec())?; + send_to(&server, client_addr, &server_session.create_disconnect(b"bye".to_vec()))?; + let _ = client.receive_once()?; + + assert!(client.flush_reliable(Duration::from_millis(200)).is_err()); + assert_eq!( + client.reliable_send_status(&receipt), + CultNetRudpReliableSendStatus::Invalidated + ); + Ok(()) +} + +/// Drains what a raw peer socket has been sent. +fn drain(socket: &UdpSocket) -> Result> { + let mut buffer = vec![0_u8; 65_535]; + let mut packets = Vec::new(); + while let Ok((received, _)) = socket.recv_from(&mut buffer) { + packets.push(decode_rudp_packet(&buffer[..received])?); + } + Ok(packets) +} + +/// A receipt belongs to the session it was issued in. A peer's Disconnect ends +/// that session for good: reconnecting must not bring the write back to Pending, +/// and an ack the old peer sent late must not make it Acknowledged. +#[test] +fn a_receipt_invalidated_by_a_peer_disconnect_never_becomes_acknowledged() -> Result<()> { + let server = socket()?; + let server_addr = server.local_addr()?; + let mut client = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::client("client", socket()?, server_addr, CONNECTION_ID), + )?; + client.connect(b"hello".to_vec())?; + let mut buffer = vec![0_u8; 65_535]; + let (received, client_addr) = server.recv_from(&mut buffer)?; + let mut old_peer = raw_session(CONNECTION_ID); + let accept = old_peer.accept_connect(&decode_rudp_packet(&buffer[..received])?, 0, Vec::new())?; + send_to(&server, client_addr, &accept)?; + let _ = client.receive_once()?; + + let receipt = client.send_reliable("schema", b"written before the disconnect".to_vec())?; + let data = drain(&server)? + .into_iter() + .find(|packet| packet.packet_type == CultNetRudpPacketType::Data) + .expect("the write reached the peer"); + old_peer.receive(&data, 0)?; + let late_ack = old_peer.create_ack_for_received(data.sequence); + send_to(&server, client_addr, &old_peer.create_disconnect(b"bye".to_vec()))?; + let _ = client.receive_once()?; + assert_eq!(client.reliable_send_status(&receipt), CultNetRudpReliableSendStatus::Invalidated); + + client.connect(b"hello again".to_vec())?; + assert_eq!( + client.reliable_send_status(&receipt), + CultNetRudpReliableSendStatus::Invalidated, + "reconnecting brought the write back to Pending" + ); + let connect = drain(&server)? + .into_iter() + .find(|packet| packet.packet_type == CultNetRudpPacketType::Connect) + .expect("the reconnect reached the peer"); + let mut new_peer = raw_session(CONNECTION_ID); + let new_accept = new_peer.accept_connect(&connect, 0, Vec::new())?; + send_to(&server, client_addr, &new_accept)?; + send_to(&server, client_addr, &late_ack)?; + while client.receive_once()?.is_some() {} + assert_eq!( + client.reliable_send_status(&receipt), + CultNetRudpReliableSendStatus::Invalidated, + "the old session's write was acknowledged in the new session" + ); + // The old write is not retransmitted into the new session either: the + // Accept acknowledged the reconnect's Connect, and nothing else is owed. + assert_eq!(client.outstanding_reliable_packet_count(), 0, "the old write survived into the new session"); + Ok(()) +} + +/// A server-mode transport that accepts a new Connect has a live session again; +/// the reason the previous one ended is history, not state. +#[test] +fn server_mode_forgets_the_end_reason_once_a_new_connect_is_accepted() -> Result<()> { + let server_socket = socket()?; + let server_addr = server_socket.local_addr()?; + let mut server = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID), + )?; + let peer = socket()?; + let mut peer_session = raw_session(CONNECTION_ID); + send_to(&peer, server_addr, &peer_session.create_connect(0, b"peer".to_vec())?)?; + let _ = server.receive_once()?; + send_to(&peer, server_addr, &peer_session.create_disconnect(b"bye".to_vec()))?; + let _ = server.receive_once()?; + assert_eq!(server.disconnect_reason(), Some(&b"bye"[..])); + + let mut next_peer = raw_session(CONNECTION_ID); + send_to(&peer, server_addr, &next_peer.create_connect(0, b"peer".to_vec())?)?; + let _ = server.receive_once()?; + assert!(server.connected()); + assert_eq!(server.disconnect_reason(), None, "a stale reason survives the new session"); + Ok(()) +} + +/// A flush waits on one session. A Connect from another endpoint replaces the +/// peer and forgets the writes being waited on; the flush must say so. +#[test] +fn a_flush_fails_when_a_new_endpoint_replaces_the_peer_mid_wait() -> Result<()> { + let server_socket = socket()?; + let server_addr = server_socket.local_addr()?; + let mut server = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID), + )?; + let peer_a = socket()?; + let peer_b = socket()?; + send_to(&peer_a, server_addr, &raw_session(CONNECTION_ID).create_connect(0, b"a".to_vec())?)?; + let _ = server.receive_once()?; + server.send_reliable("schema", b"for A".to_vec())?; + send_to(&peer_b, server_addr, &raw_session(CONNECTION_ID).create_connect(0, b"b".to_vec())?)?; + + let error = server.flush_reliable(Duration::from_millis(300)).expect_err("A's write was forgotten"); + assert!( + error.to_string().contains("ended before its reliable writes were acknowledged"), + "{error}" + ); + Ok(()) +} + +/// A repeated Connect still carries an acknowledgement field, and the session +/// honours it: the Accept it acknowledges is no longer owed. +#[test] +fn a_repeated_connect_acknowledges_what_it_carries() -> Result<()> { + let mut client = raw_session(CONNECTION_ID); + let mut server = raw_session(CONNECTION_ID); + let connect = client.create_connect(0, Vec::new())?; + let accept = server.accept_connect(&connect, 0, Vec::new())?; + let mut repeat = connect.clone(); + repeat.ack = accept.sequence; + let reply = server.answer_repeated_connect(&repeat, 1)?; + assert_eq!(reply.packet_type, CultNetRudpPacketType::Ack); + assert_eq!(server.outstanding_reliable_packet_count(), 0); + Ok(()) +} diff --git a/packages/cultnet-ts/src/operation-service.ts b/packages/cultnet-ts/src/operation-service.ts index ee13457f..606cd418 100644 --- a/packages/cultnet-ts/src/operation-service.ts +++ b/packages/cultnet-ts/src/operation-service.ts @@ -17,6 +17,10 @@ import { import { CultNetPeer } from "./peer"; const DEFAULT_CONNECTION_ID = 0x43554c54; +/** Matches the document server's default session cap. A Connect past it is dropped. */ +const MAX_OPERATION_SESSIONS = 64; +/** Matches the document server's default `session_idle_timeout` (30 s). */ +const DEFAULT_SESSION_IDLE_TIMEOUT_MS = 30_000; export interface CultNetOperationServerOptions { runtimeId: string; @@ -24,12 +28,16 @@ export interface CultNetOperationServerOptions { port?: number; connectionId?: number; maxFragmentBytes?: number; + /** A session that has sent nothing for this long is dropped, freeing its slot. Default 30 000. */ + sessionIdleTimeoutMs?: number; handler: (request: CultNetOperationRequestMessage) => CultNetOperationResponseMessage | Promise; } export interface CultNetOperationServer { readonly endpoint: string; + /** Datagrams read and discarded: malformed, unadmitted, refused by their session, or failed in handling. */ + readonly packetsDropped: number; close(): Promise; } @@ -43,6 +51,10 @@ export interface CultNetOperationClientOptions { interface RemoteSession { session: CultNetRudpSession; remote: RemoteInfo; + /** Handlers running for this session; a session serving a request is not idle. */ + handling: number; + /** When the last handler finished: the response still needs its acknowledgement. */ + lastHandledAtMs: number; } export async function startCultNetOperationServer( @@ -57,11 +69,26 @@ export async function startCultNetOperationServer( const wire = encodeRudpPacket(packet); socket.send(wire, remote.port, remote.address); }; + let packetsDropped = 0; socket.on("message", (wire, remote) => { - void handleServerDatagram(socket, sessions, connectionId, options, wire, remote, sendPacket); + // What a datagram carries is the sender's business: a rejection here must + // drop and count the packet, never become an unhandled rejection that ends + // the process. + handleServerDatagram(sessions, connectionId, options, wire, remote, sendPacket).then( + admitted => { if (!admitted) packetsDropped += 1; }, + () => { packetsDropped += 1; }, + ); }); await bindSocket(socket, options.port ?? 0, options.host ?? "127.0.0.1"); + const idleTimeoutMs = options.sessionIdleTimeoutMs ?? DEFAULT_SESSION_IDLE_TIMEOUT_MS; const resendTimer = setInterval(() => { + const now = Date.now(); + for (const [key, peer] of sessions) { + // An abandoned client must not hold a slot forever, or the cap locks + // every new client out. + if (peer.handling > 0 || now - peer.lastHandledAtMs < idleTimeoutMs) continue; + if (peer.session.checkTimeout(now, idleTimeoutMs)) sessions.delete(key); + } for (const peer of sessions.values()) { for (const packet of peer.session.dueResends(Date.now())) sendPacket(peer.remote, packet); } @@ -71,6 +98,7 @@ export async function startCultNetOperationServer( const endpoint = `rudp://${address.address}:${address.port}`; return { endpoint, + get packetsDropped() { return packetsDropped; }, close: async () => { clearInterval(resendTimer); await closeSocket(socket); @@ -118,54 +146,96 @@ export async function invokeCultNetOperation( } } +/** Resolves false when the datagram was dropped without an error. */ async function handleServerDatagram( - socket: Socket, sessions: Map, connectionId: number, options: CultNetOperationServerOptions, wire: Buffer, remote: RemoteInfo, sendPacket: (remote: RemoteInfo, packet: CultNetRudpPacket) => void, -): Promise { +): Promise { let packet: CultNetRudpPacket; try { packet = decodeRudpPacket(wire); } catch { - return; + return false; } - if (packet.connectionId !== connectionId) return; + if (packet.connectionId !== connectionId) return false; const key = `${remote.address}:${remote.port}`; let peer = sessions.get(key); if (packet.packetType === "connect") { - peer = { session: new CultNetRudpSession({ connectionId, resendDelayMs: 25 }), remote }; - sessions.set(key, peer); + if (!peer) { + if (sessions.size >= MAX_OPERATION_SESSIONS) return false; + peer = { session: new CultNetRudpSession({ connectionId, resendDelayMs: 25 }), remote, handling: 0, lastHandledAtMs: 0 }; + sessions.set(key, peer); + } + // A Connect from an admitted peer repeats: the session answers with the + // Accept already owed and queues nothing. sendPacket(remote, peer.session.acceptConnect(packet, Date.now(), encode("cultnet-operation-service"))); - return; + return true; + } + if (!peer) return false; + let result: ReturnType; + try { + result = peer.session.receive(packet, Date.now()); + } catch { + endSession(sessions, key, peer, sendPacket); + return false; } - if (!peer) return; - const result = peer.session.receive(packet, Date.now()); if (result.reply) sendPacket(remote, result.reply); for (const ready of result.readyToSend ?? []) sendPacket(remote, ready); if (result.disconnected) { sessions.delete(key); - return; + return true; } for (const frame of result.delivered) { if (frame.channelId !== "schema") continue; - const message = parseCultNetMessage(decode(frame.payload)); - if (message.schemaVersion !== "cultnet.operation_request.v0") continue; - const response = await options.handler(message); - const payload = encode(encodeCultNetMessageForWire(response, "cultnet.schema.v0")); - for (const responsePacket of peer.session.sendMany("schema", payload, { - reliable: true, - ordered: true, - nowMs: Date.now(), - maxFragmentBytes: options.maxFragmentBytes ?? 2048, - })) sendPacket(remote, responsePacket); + try { + const message = parseCultNetMessage(decode(frame.payload)); + if (message.schemaVersion !== "cultnet.operation_request.v0") continue; + peer.handling += 1; + let response: CultNetOperationResponseMessage; + try { + response = await options.handler(message); + } finally { + peer.handling -= 1; + peer.lastHandledAtMs = Date.now(); + } + const payload = encode(encodeCultNetMessageForWire(response, "cultnet.schema.v0")); + for (const responsePacket of peer.session.sendMany("schema", payload, { + reliable: true, + ordered: true, + nowMs: Date.now(), + maxFragmentBytes: options.maxFragmentBytes ?? 2048, + })) sendPacket(remote, responsePacket); + } catch { + // The session recorded this request's sequence, so keeping it would + // acknowledge the retransmit of a request that was never handled. + endSession(sessions, key, peer, sendPacket); + return false; + } } if (packet.packetType === "data" || result.delivered.length > 0) { sendPacket(remote, peer.session.createAckForReceived(packet.sequence)); } + return true; +} + +/** + * Ends a session that took a packet it could not serve, and tells the peer. The + * reset comes first, or the goodbye's ack field would acknowledge the very + * packet that was refused. A session already replaced under the same key is left. + */ +function endSession( + sessions: Map, + key: string, + peer: RemoteSession, + sendPacket: (remote: RemoteInfo, packet: CultNetRudpPacket) => void, +): void { + if (sessions.get(key) === peer) sessions.delete(key); + peer.session.resetPeerState(); + sendPacket(peer.remote, peer.session.createDisconnect(Buffer.from("session refused a packet", "utf8"))); } function parseRudpEndpoint(endpoint: string): { host: string; port: number } { diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index 56dc67fa..ae7ea838 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -170,6 +170,9 @@ export class CultNetRudpSession { constructor(options: CultNetRudpSessionOptions) { this.connectionId = toUint32(options.connectionId, "connectionId"); this.#nextSequence = toUint32(options.initialSequence ?? 1, "initialSequence"); + if (this.#nextSequence === 0xffff_ffff) { + throw new Error("RUDP initialSequence must leave room for a reliable packet."); + } this.resendDelayMs = options.resendDelayMs ?? 250; if (options.maxPendingReliablePackets !== undefined && options.maxPendingReliablePackets <= 0) { throw new Error("RUDP maxPendingReliablePackets must be greater than zero."); @@ -242,16 +245,9 @@ export class CultNetRudpSession { throw new Error(`Expected RUDP connect packet, got ${packet.packetType}.`); } - const pendingAccept = [...this.#pendingReliable.values()] - .find(pending => pending.packet.packetType === "accept"); - if (pendingAccept) { - this.#rememberReceived(packet.sequence); - this.#lastReceivedAtMs = nowMs; - pendingAccept.lastSentAtMs = nowMs; - return { - ...pendingAccept.packet, - payload: new Uint8Array(pendingAccept.packet.payload ?? new Uint8Array()), - }; + // A Connect from a peer this session already accepted repeats. + if (this.#connected) { + return this.answerRepeatedConnect(packet, nowMs); } this.#ensureReliableCapacity(1); this.#rememberReceived(packet.sequence); @@ -268,6 +264,32 @@ export class CultNetRudpSession { return response; } + /** + * Answers a Connect from a peer this session has already accepted. The packet's + * sequence is remembered and nothing is queued, so a Connect storm cannot grow + * the reliable queue. The reply is the Accept still awaiting acknowledgement, + * or an Ack once it was acknowledged. + */ + answerRepeatedConnect(packet: CultNetRudpPacket, nowMs = 0): CultNetRudpPacket { + this.#requireConnection(packet); + if (packet.packetType !== "connect") { + throw new Error(`Expected RUDP connect packet, got ${packet.packetType}.`); + } + this.#applyAcknowledgements(packet); + this.#rememberReceived(packet.sequence); + this.#lastReceivedAtMs = nowMs; + const pendingAccept = [...this.#pendingReliable.values()] + .find(pending => pending.packet.packetType === "accept"); + if (!pendingAccept) { + return this.createAck(); + } + pendingAccept.lastSentAtMs = nowMs; + return { + ...pendingAccept.packet, + payload: new Uint8Array(pendingAccept.packet.payload ?? new Uint8Array()), + }; + } + send( channelId: string, payload: Uint8Array, @@ -802,11 +824,19 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement #remoteHost: string | undefined; #remotePort: number | undefined; #closed = false; + // A flush belongs to the session it started in. Each session start advances + // the generation; each session end records the generation it ended and why. + // A reconnect from a disconnect listener starts a new generation but cannot + // un-end the old one. + #generation = 0; + #endedGeneration = -1; + #endedReason: Uint8Array | undefined; readonly #stats: CultNetTransportStats = { bytesReceived: 0, bytesSent: 0, framesReceived: 0, framesSent: 0, + packetsDropped: 0, }; constructor(options: CultNetRudpSocketTransportOptions) { @@ -862,6 +892,7 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement if (this.#mode !== "client") { throw new Error("Only a client RUDP socket transport can initiate connect."); } + this.#generation += 1; this.#sendPacket(this.#session.createConnect(Date.now(), payload)); } @@ -879,7 +910,18 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement async flush(timeoutMs = 30_000): Promise { const deadline = Date.now() + Math.max(0, timeoutMs); - while (this.#session.outstandingReliablePacketCount > 0) { + const generation = this.#generation; + for (;;) { + // An ended session forgot its unacknowledged writes; reporting them + // flushed would be a lie, even if a new session has since begun. + if (this.#endedGeneration >= generation) { + throw new Error( + `RUDP session ended before its reliable writes were acknowledged: ${Buffer.from(this.#endedReason ?? new Uint8Array()).toString("utf8")}`, + ); + } + if (this.#session.outstandingReliablePacketCount === 0) { + return; + } if (this.#closed) { throw new Error("RUDP transport closed before reliable packets were acknowledged."); } @@ -892,6 +934,11 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement } } + #endGeneration(reason: Uint8Array): void { + this.#endedGeneration = this.#generation; + this.#endedReason = reason; + } + ping(payload = new Uint8Array()): void { this.#sendPacket(this.#session.createPing(payload)); } @@ -899,6 +946,7 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement checkTimeout(timeoutMs: number, nowMs = Date.now()): boolean { const timedOut = this.#session.checkTimeout(nowMs, timeoutMs); if (timedOut) { + this.#endGeneration(Buffer.from("session timed out", "utf8")); this.emit("timeout"); this.emit("close"); } @@ -920,36 +968,102 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement } } + #endRefusedSession(): void { + const reason = Buffer.from("session refused a packet", "utf8"); + // The goodbye is built after the reset, or its ack field would acknowledge + // the very frame the session refused. + this.#session.resetPeerState(); + this.#endGeneration(reason); + try { + this.#sendPacket(this.#session.createDisconnect(reason)); + } catch { + // Best-effort: the session ends whether or not the peer hears it. + } + if (this.#mode === "server") { + // Only a Connect can claim the endpoint again. + this.#remoteHost = undefined; + this.#remotePort = undefined; + } + this.emit("disconnect", { reason }); + this.emit("close"); + } + #receiveDatagram(wire: Buffer, remote: RemoteInfo): void { this.#stats.bytesReceived += wire.length; + // What a datagram carries is the sender's business, not a fault of this + // process: a malformed frame, another session's connection id and a packet + // the session refuses are dropped and counted, never emitted as "error" + // (an "error" with no listener throws and ends the process). The id is + // checked before the peer endpoint or session state is touched, so a stray + // Connect can neither move the endpoint nor reset the session. let packet: CultNetRudpPacket; try { packet = decodeRudpPacket(wire); - } catch (error) { - this.emit("error", error instanceof Error ? error : new Error(String(error))); + } catch { + this.#stats.packetsDropped += 1; + return; + } + if (packet.connectionId !== this.#session.connectionId) { + this.#stats.packetsDropped += 1; return; } + let movedEndpoint = false; if (!this.#remoteHost || this.#remotePort === undefined) { + if (this.#mode === "server" && packet.packetType !== "connect") { + this.#stats.packetsDropped += 1; + return; + } this.#remoteHost = remote.address; this.#remotePort = remote.port; } else if (remote.address !== this.#remoteHost || remote.port !== this.#remotePort) { if (this.#mode === "server" && packet.packetType === "connect") { + movedEndpoint = true; this.#remoteHost = remote.address; this.#remotePort = remote.port; } else { + this.#stats.packetsDropped += 1; return; } } - try { - if (this.#mode === "server" && packet.packetType === "connect") { + if (this.#mode === "server" && packet.packetType === "connect") { + // Only a Connect from the endpoint already accepted repeats, and the + // session answers it with the Accept owed. A Connect from another + // endpoint replaces the peer: the old session's writes die with it. + const repeated = this.#session.connected && !movedEndpoint; + if (!repeated) { + // A timed-out session already ended its generation; any other + // predecessor ends here, connected or not. + if (this.#endedGeneration < this.#generation) { + this.#endGeneration(Buffer.from("replaced by a new Connect", "utf8")); + } this.#session.resetPeerState(); - this.#sendPacket(this.#session.acceptConnect(packet, Date.now())); + } + let accept: CultNetRudpPacket; + try { + accept = this.#session.acceptConnect(packet, Date.now()); + if (!repeated) this.#generation += 1; + } catch { + this.#stats.packetsDropped += 1; return; } + this.#sendPacket(accept); + return; + } - const result = this.#session.receive(packet, Date.now()); + let result: ReturnType; + try { + result = this.#session.receive(packet, Date.now()); + } catch { + // receive() has already recorded the packet's reliable sequence, so the + // session cannot be kept: a retransmit would be acknowledged and the + // frame silently lost. End it and tell the peer. + this.#stats.packetsDropped += 1; + this.#endRefusedSession(); + return; + } + try { if (result.reply) { this.#sendPacket(result.reply); } @@ -967,6 +1081,7 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement } satisfies CultNetTransportFrame); } if (result.disconnected) { + this.#endGeneration(result.disconnectReason ?? new Uint8Array()); this.emit("disconnect", { reason: result.disconnectReason ?? new Uint8Array() }); this.emit("close"); return; diff --git a/packages/cultnet-ts/src/transport.ts b/packages/cultnet-ts/src/transport.ts index a1b9b3ce..f900aa23 100644 --- a/packages/cultnet-ts/src/transport.ts +++ b/packages/cultnet-ts/src/transport.ts @@ -14,6 +14,8 @@ export interface CultNetTransportStats { bytesSent: number; framesReceived: number; framesSent: number; + /** Datagrams read and discarded because they belong to no session on this transport. */ + packetsDropped: number; } export interface CultNetReconnectPolicy { @@ -192,6 +194,7 @@ export class TcpFramedTransportConnection extends EventEmitter implements CultNe bytesSent: 0, framesReceived: 0, framesSent: 0, + packetsDropped: 0, }; constructor(stream: Duplex, profile: CultNetTransportProfile) { diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 2f540a86..35cba767 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -8,6 +8,7 @@ import { join } from "node:path"; import { tmpdir } from "node:os"; import { z } from "zod"; +import { encode } from "@msgpack/msgpack"; import { CultCache, SingleFileMessagePackBackingStore, @@ -38,6 +39,9 @@ import { createRudpTransportProfile, defineCultNetDocumentBinding, decodeRudpPacket, + type CultNetOperationRequestMessage, + type CultNetOperationResponseMessage, + type CultNetRudpPacket, encodeCultNetMessageForWire, encodeRudpPacket, ghostlightAgentStateGeneratedContract, @@ -1867,3 +1871,622 @@ test("rudp receiver evicts stranded fragment sets instead of refusing new ones", assert.equal(second.delivered.length, 1); assert.equal(Buffer.from(second.delivered[0]!.payload).toString(), "hello"); }); + +test("rudp socket transport drops strays without an error event and keeps serving its peer", async () => { + const serverSocket = await bindUdpSocket(); + const clientSocket = await bindUdpSocket(); + const strayHost = await bindUdpSocket(); + const connectionId = 0x10203050; + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-server", + socket: serverSocket, + mode: "server", + connectionId, + resendPollMs: 5, + }); + const client = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-client", + socket: clientSocket, + mode: "client", + remoteHost: "127.0.0.1", + remotePort: udpPort(serverSocket), + connectionId, + resendPollMs: 5, + }); + + try { + // No "error" listener on either side: an unhandled "error" event throws + // and would end the process, which is what a stray must never do. + const frames: string[] = []; + server.on("frame", (frame) => frames.push(Buffer.from(frame.payload).toString("utf8"))); + client.connect(); + await waitFor(() => client.connected && server.connected, "RUDP socket handshake"); + + const foreign = new CultNetRudpSession({ connectionId: 0x0badf10e }); + const strays = [ + Buffer.from("not a rudp packet"), + Buffer.alloc(0), + Buffer.from(encodeRudpPacket(foreign.createConnect(0, Buffer.from("foreign")))), + ]; + for (const stray of strays) { + // From an unrelated address, and from the peer's own address. + strayHost.send(stray, udpPort(serverSocket), "127.0.0.1"); + clientSocket.send(stray, udpPort(serverSocket), "127.0.0.1"); + } + await waitFor(() => server.stats.packetsDropped >= strays.length * 2, "strays counted"); + + client.send("schema", Buffer.from("still here", "utf8")); + await waitFor(() => frames.length === 1, "the real frame"); + assert.deepEqual(frames, ["still here"]); + + // The client side too: a moved flow answers from the server address. + for (const stray of strays) { + serverSocket.send(stray, udpPort(clientSocket), "127.0.0.1"); + } + await waitFor(() => client.stats.packetsDropped >= strays.length, "client strays counted"); + const clientFrames: string[] = []; + client.on("frame", (frame) => clientFrames.push(Buffer.from(frame.payload).toString("utf8"))); + server.send("schema", Buffer.from("server still here", "utf8")); + await waitFor(() => clientFrames.length === 1, "the real server frame"); + } finally { + strayHost.close(); + client.close(); + server.close(); + } +}); + +/** A reliable frame no session accepts: a fragment set with no fragment id. */ +function poisonedFrame(session: CultNetRudpSession): CultNetRudpPacket { + const [packet] = session.sendMany("schema", Buffer.from("poison"), { reliable: true, ordered: true, nowMs: 0 }); + return { ...packet!, fragmentCount: 2, fragmentId: 0, fragmentIndex: 0 }; +} + +test("rudp socket transport ends the session that refuses a reliable frame and never acknowledges it", async () => { + const serverSocket = await bindUdpSocket(); + const peerSocket = await bindUdpSocket(); + const connectionId = 0x10203051; + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-server", + socket: serverSocket, + mode: "server", + connectionId, + resendPollMs: 1000, + }); + const received: CultNetRudpPacket[] = []; + peerSocket.on("message", (wire) => received.push(decodeRudpPacket(wire))); + const peer = new CultNetRudpSession({ connectionId }); + const toServer = (packet: CultNetRudpPacket) => + peerSocket.send(encodeRudpPacket(packet), udpPort(serverSocket), "127.0.0.1"); + const disconnects: string[] = []; + server.on("disconnect", ({ reason }) => disconnects.push(Buffer.from(reason).toString("utf8"))); + + try { + toServer(peer.createConnect(0)); + await waitFor(() => received.some((p) => p.packetType === "accept"), "the Accept"); + peer.receive(received.find((p) => p.packetType === "accept")!, 0); + + const poison = poisonedFrame(peer); + toServer(poison); + await waitFor(() => disconnects.length === 1, "the server ending the session"); + await waitFor(() => received.some((p) => p.packetType === "disconnect"), "the goodbye"); + assert.equal(server.stats.packetsDropped, 1); + assert.equal(server.connected, false); + + // The peer sees the session end and never sees the frame acknowledged, + // neither in the goodbye nor after it retransmits. + const goodbye = received.find((p) => p.packetType === "disconnect")!; + assert.equal(peer.receive(goodbye, 1).disconnected, true); + assert.ok(peer.pendingReliableSequences.includes(poison.sequence), "the refused frame was acknowledged"); + const before = received.length; + toServer(poison); + await waitFor(() => server.stats.packetsDropped === 2, "the retransmit dropped"); + await new Promise((resolve) => setTimeout(resolve, 30)); + assert.equal(received.length, before, "the server must not answer a retransmit for an ended session"); + + // Only a Connect claims the endpoint again. + const again = new CultNetRudpSession({ connectionId }); + toServer(again.createConnect(0)); + await waitFor(() => received.length > before, "the new Accept"); + assert.ok(received.slice(before).some((p) => p.packetType === "accept")); + } finally { + peerSocket.close(); + server.close(); + } +}); + +test("rudp sessions reject sequence spaces and queues that cannot admit a Connect", () => { + assert.throws(() => new CultNetRudpSession({ connectionId: 1, initialSequence: 0xffff_ffff }), /initialSequence/); + assert.throws(() => new CultNetRudpSession({ connectionId: 1, maxPendingReliablePackets: 0 }), /greater than zero/); + assert.doesNotThrow(() => new CultNetRudpSession({ connectionId: 1, initialSequence: 0xffff_fffe, maxPendingReliablePackets: 1 })); +}); + +test("operation service drops and counts a packet its session refuses and keeps serving", async () => { + const handler = (request: CultNetOperationRequestMessage): CultNetOperationResponseMessage => ({ + schemaVersion: "cultnet.operation_response.v0", + messageId: request.messageId, + serviceId: request.serviceId, + operation: request.operation, + status: "ok", + payloadSchema: "gamecult.eve.plugin_abi.response.v1", + payloadEncoding: "messagepack-base64", + payload: request.payload, + diagnostics: [], + sourceRuntimeId: "sai-sidecar", + }); + const server = await startCultNetOperationServer({ runtimeId: "sai-sidecar", handler }); + const raw = await bindUdpSocket(); + const port = Number(new URL(server.endpoint).port); + const connectionId = 0x43554c54; + const received: CultNetRudpPacket[] = []; + raw.on("message", (wire) => received.push(decodeRudpPacket(wire))); + const peer = new CultNetRudpSession({ connectionId }); + const toServer = (packet: CultNetRudpPacket) => raw.send(encodeRudpPacket(packet), port, "127.0.0.1"); + try { + toServer(peer.createConnect(0)); + await waitFor(() => received.some((p) => p.packetType === "accept"), "the Accept"); + peer.receive(received[0]!, 0); + toServer(poisonedFrame(peer)); + await waitFor(() => server.packetsDropped === 1, "the poisoned frame dropped"); + await waitFor(() => received.some((p) => p.packetType === "disconnect"), "the session ended"); + + // A frame the session accepts but the service cannot handle rejects inside + // the handler; that too is a dropped packet, not a dead process. + const garbage = new CultNetRudpSession({ connectionId }); + received.length = 0; + toServer(garbage.createConnect(0)); + await waitFor(() => received.some((p) => p.packetType === "accept"), "the second Accept"); + garbage.receive(received[0]!, 0); + const [notMessagePack] = garbage.sendMany("schema", Buffer.from([0xc1]), { reliable: true, ordered: true, nowMs: 0 }); + toServer(notMessagePack!); + await waitFor(() => server.packetsDropped === 2, "the unhandled frame dropped"); + // The session that could not serve it ends and the client is told, so its + // retransmit is not acknowledged for a request that was never handled. + await waitFor(() => received.some((p) => p.packetType === "disconnect"), "the goodbye"); + const beforeRetransmit = received.length; + toServer(notMessagePack!); + await waitFor(() => server.packetsDropped === 3, "the retransmit dropped"); + await new Promise((resolve) => setTimeout(resolve, 30)); + assert.equal(received.length, beforeRetransmit, "the retransmit of an unhandled request was answered"); + + const response = await invokeCultNetOperation(server.endpoint, { + schemaVersion: "cultnet.operation_request.v0", + messageId: "after-poison", + serviceId: "sai.vn", + operation: "describe", + payloadSchema: "gamecult.eve.plugin_abi.request.v1", + payloadEncoding: "messagepack-base64", + payload: "gaZzY2hlbWE=", + }, { runtimeId: "eve-test" }); + assert.equal(response.messageId, "after-poison"); + } finally { + raw.close(); + await server.close(); + } +}); + +test("operation service drops a Connect once its session table is full", async () => { + const server = await startCultNetOperationServer({ + runtimeId: "sai-sidecar", + handler: () => { throw new Error("no requests expected"); }, + }); + const port = Number(new URL(server.endpoint).port); + const sockets: Socket[] = []; + try { + for (let index = 0; index < 66; index += 1) { + const socket = await bindUdpSocket(); + sockets.push(socket); + socket.send(encodeRudpPacket(new CultNetRudpSession({ connectionId: 0x43554c54 }).createConnect(0)), port, "127.0.0.1"); + } + await waitFor(() => server.packetsDropped === 2, "the Connects past the cap dropped"); + await new Promise((resolve) => setTimeout(resolve, 30)); + assert.equal(server.packetsDropped, 2); + } finally { + for (const socket of sockets) socket.close(); + await server.close(); + } +}); + +test("rudp flush fails for a write the ended session forgot", async () => { + const serverSocket = await bindUdpSocket(); + const peerSocket = await bindUdpSocket(); + const connectionId = 0x10203056; + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-server", + socket: serverSocket, + mode: "server", + connectionId, + resendPollMs: 1000, + }); + const received: CultNetRudpPacket[] = []; + peerSocket.on("message", (wire) => received.push(decodeRudpPacket(wire))); + const peer = new CultNetRudpSession({ connectionId }); + const toServer = (packet: CultNetRudpPacket) => + peerSocket.send(encodeRudpPacket(packet), udpPort(serverSocket), "127.0.0.1"); + try { + toServer(peer.createConnect(0)); + await waitFor(() => received.some((p) => p.packetType === "accept"), "the Accept"); + peer.receive(received.find((p) => p.packetType === "accept")!, 0); + server.send("schema", Buffer.from("never acknowledged")); + toServer(poisonedFrame(peer)); + await waitFor(() => server.stats.packetsDropped === 1, "the refusal"); + await assert.rejects(server.flush(200), /ended before its reliable writes were acknowledged/); + } finally { + peerSocket.close(); + server.close(); + } +}); + +test("a Connect that repeats after its Accept was acknowledged is answered with an Ack and queues nothing", () => { + const connectionId = 0x10203057; + const client = new CultNetRudpSession({ connectionId }); + const server = new CultNetRudpSession({ connectionId }); + const connect = client.createConnect(0); + const accept = server.acceptConnect(connect, 10); + client.receive(accept, 11); + server.receive(client.createAckForReceived(accept.sequence), 12); + assert.equal(server.outstandingReliablePacketCount, 0); + for (let attempt = 0; attempt < 20; attempt += 1) { + assert.equal(server.acceptConnect(connect, 20 + attempt).packetType, "ack"); + } + assert.equal(server.outstandingReliablePacketCount, 0); +}); + +test("server-mode transport and operation service owe one reliable Accept however many Connects repeat", async () => { + const connectionId = 0x43554c54; + const serverSocket = await bindUdpSocket(); + const transport = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-server", + socket: serverSocket, + mode: "server", + connectionId, + resendPollMs: 1000, + }); + const service = await startCultNetOperationServer({ + runtimeId: "sai-sidecar", + handler: () => { throw new Error("no requests expected"); }, + }); + const raw = await bindUdpSocket(); + let received: CultNetRudpPacket[] = []; + raw.on("message", (wire) => received.push(decodeRudpPacket(wire))); + const settle = () => new Promise((resolve) => setTimeout(resolve, 100)); + try { + for (const port of [udpPort(serverSocket), Number(new URL(service.endpoint).port)]) { + const peer = new CultNetRudpSession({ connectionId }); + const connect = peer.createConnect(0); + received = []; + for (let attempt = 0; attempt < 20; attempt += 1) raw.send(encodeRudpPacket(connect), port, "127.0.0.1"); + await settle(); + const accepts = received.filter((p) => p.packetType === "accept"); + assert.equal(new Set(accepts.map((p) => p.sequence)).size, 1, `reliable Accepts sent by port ${port}`); + + // Once the Accept is acknowledged, a repeated Connect is answered with an Ack, never a new Accept. + peer.receive(accepts[0]!, 0); + raw.send(encodeRudpPacket(peer.createAckForReceived(accepts[0]!.sequence)), port, "127.0.0.1"); + await settle(); + received = []; + for (let attempt = 0; attempt < 20; attempt += 1) raw.send(encodeRudpPacket(connect), port, "127.0.0.1"); + await settle(); + assert.equal(received.filter((p) => p.packetType === "accept").length, 0, `Accepts sent by port ${port} after the ack`); + assert.ok(received.some((p) => p.packetType === "ack"), `an Ack from port ${port}`); + } + assert.equal(transport.outstandingReliablePacketCount, 0); + } finally { + raw.close(); + transport.close(); + await service.close(); + } +}); + +test("operation service frees the slots of abandoned sessions once they idle out", async () => { + const server = await startCultNetOperationServer({ + runtimeId: "sai-sidecar", + sessionIdleTimeoutMs: 300, + handler: (request): CultNetOperationResponseMessage => ({ + schemaVersion: "cultnet.operation_response.v0", + messageId: request.messageId, + serviceId: request.serviceId, + operation: request.operation, + status: "ok", + payloadSchema: "gamecult.eve.plugin_abi.response.v1", + payloadEncoding: "messagepack-base64", + payload: request.payload, + diagnostics: [], + sourceRuntimeId: "sai-sidecar", + }), + }); + const port = Number(new URL(server.endpoint).port); + const sockets: Socket[] = []; + const request: CultNetOperationRequestMessage = { + schemaVersion: "cultnet.operation_request.v0", + messageId: "after-abandonment", + serviceId: "sai.vn", + operation: "describe", + payloadSchema: "gamecult.eve.plugin_abi.request.v1", + payloadEncoding: "messagepack-base64", + payload: "gaZzY2hlbWE=", + }; + try { + for (let index = 0; index < 64; index += 1) { + const socket = await bindUdpSocket(); + sockets.push(socket); + socket.send(encodeRudpPacket(new CultNetRudpSession({ connectionId: 0x43554c54 }).createConnect(0)), port, "127.0.0.1"); + } + await new Promise((resolve) => setTimeout(resolve, 60)); + await assert.rejects(invokeCultNetOperation(server.endpoint, request, { runtimeId: "eve-test", timeoutMs: 100 })); + await new Promise((resolve) => setTimeout(resolve, 400)); + const response = await invokeCultNetOperation(server.endpoint, request, { runtimeId: "eve-test" }); + assert.equal(response.messageId, "after-abandonment"); + } finally { + for (const socket of sockets) socket.close(); + await server.close(); + } +}); + +test("server-mode transport replaces its peer when a Connect arrives from another endpoint", async () => { + const serverSocket = await bindUdpSocket(); + const socketA = await bindUdpSocket(); + const socketB = await bindUdpSocket(); + const connectionId = 0x10203058; + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-server", + socket: serverSocket, + mode: "server", + connectionId, + resendPollMs: 1000, + }); + const receivedByA: CultNetRudpPacket[] = []; + socketA.on("message", (wire) => receivedByA.push(decodeRudpPacket(wire))); + const receivedByB: CultNetRudpPacket[] = []; + socketB.on("message", (wire) => receivedByB.push(decodeRudpPacket(wire))); + const frames: string[] = []; + server.on("frame", (frame: { payload: Uint8Array }) => frames.push(Buffer.from(frame.payload).toString("utf8"))); + const peerA = new CultNetRudpSession({ connectionId }); + const peerB = new CultNetRudpSession({ connectionId }); + const send = (socket: Socket, packet: CultNetRudpPacket) => + socket.send(encodeRudpPacket(packet), udpPort(serverSocket), "127.0.0.1"); + try { + send(socketA, peerA.createConnect(0)); + await waitFor(() => receivedByA.some((p) => p.packetType === "accept"), "A's Accept"); + const acceptA = receivedByA.find((p) => p.packetType === "accept")!; + peerA.receive(acceptA, 0); + send(socketA, peerA.createAckForReceived(acceptA.sequence)); + await waitFor(() => server.outstandingReliablePacketCount === 0, "A's Accept acknowledged"); + send(socketB, peerB.createConnect(0)); + await waitFor(() => receivedByB.some((p) => p.packetType === "accept"), "B's Accept"); + peerB.receive(receivedByB.find((p) => p.packetType === "accept")!, 0); + + // B owns the session; A's traffic is a stranger's and is dropped. + const [fromA] = peerA.sendMany("schema", Buffer.from("from-A"), { reliable: true, ordered: true, nowMs: 0 }); + send(socketA, fromA!); + const [fromB] = peerB.sendMany("schema", Buffer.from("from-B"), { reliable: true, ordered: true, nowMs: 0 }); + send(socketB, fromB!); + await waitFor(() => frames.length > 0, "B's frame"); + await new Promise((resolve) => setTimeout(resolve, 30)); + assert.deepEqual(frames, ["from-B"]); + assert.equal(server.stats.packetsDropped, 1); + } finally { + socketA.close(); + socketB.close(); + assert.doesNotThrow(() => server.close()); + } +}); + +test("operation service does not idle out a session while its handler is running", async () => { + const respond = (request: CultNetOperationRequestMessage): CultNetOperationResponseMessage => ({ + schemaVersion: "cultnet.operation_response.v0", + messageId: request.messageId, + serviceId: request.serviceId, + operation: request.operation, + status: "ok", + payloadSchema: "gamecult.eve.plugin_abi.response.v1", + payloadEncoding: "messagepack-base64", + payload: request.payload, + diagnostics: [], + sourceRuntimeId: "sai-sidecar", + }); + const server = await startCultNetOperationServer({ + runtimeId: "sai-sidecar", + sessionIdleTimeoutMs: 200, + handler: async (request) => { + await new Promise((resolve) => setTimeout(resolve, 700)); + return respond(request); + }, + }); + try { + const response = await invokeCultNetOperation(server.endpoint, { + schemaVersion: "cultnet.operation_request.v0", + messageId: "slow", + serviceId: "sai.vn", + operation: "describe", + payloadSchema: "gamecult.eve.plugin_abi.request.v1", + payloadEncoding: "messagepack-base64", + payload: "gaZzY2hlbWE=", + }, { runtimeId: "eve-test", timeoutMs: 3000 }); + assert.equal(response.messageId, "slow"); + } finally { + await server.close(); + } +}); + +test("a flush is bound to the session it started in, even when a disconnect listener reconnects", async () => { + const peerSocket = await bindUdpSocket(); + const clientSocket = await bindUdpSocket(); + const connectionId = 0x10203059; + const client = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-client", + socket: clientSocket, + mode: "client", + remoteHost: "127.0.0.1", + remotePort: udpPort(peerSocket), + connectionId, + resendDelayMs: 1000, + resendPollMs: 1000, + }); + const received: CultNetRudpPacket[] = []; + let clientPort = 0; + peerSocket.on("message", (wire, remote) => { + clientPort = remote.port; + received.push(decodeRudpPacket(wire)); + }); + const peer = new CultNetRudpSession({ connectionId }); + try { + client.connect(); + await waitFor(() => received.some((p) => p.packetType === "connect"), "the Connect"); + const accept = peer.acceptConnect(received.find((p) => p.packetType === "connect")!, 0); + peerSocket.send(encodeRudpPacket(accept), clientPort, "127.0.0.1"); + await waitFor(() => client.connected, "the client connected"); + client.send("schema", Buffer.from("never acknowledged")); + client.on("disconnect", () => client.connect()); + const flushed = client.flush(500); + const outcome = assert.rejects(flushed, /ended before its reliable writes were acknowledged/); + peerSocket.send(encodeRudpPacket(peer.createDisconnect(Buffer.from("bye"))), clientPort, "127.0.0.1"); + await outcome; + } finally { + peerSocket.close(); + client.close(); + } +}); + +test("a repeated Connect acknowledges what it carries", () => { + const connectionId = 0x1020305a; + const client = new CultNetRudpSession({ connectionId }); + const server = new CultNetRudpSession({ connectionId }); + const connect = client.createConnect(0); + const accept = server.acceptConnect(connect, 0); + const reply = server.answerRepeatedConnect({ ...connect, ack: accept.sequence }, 1); + assert.equal(reply.packetType, "ack"); + assert.equal(server.outstandingReliablePacketCount, 0); +}); + +async function timedOutServerWithUnackedWrite(connectionId: number) { + const serverSocket = await bindUdpSocket(); + const socketA = await bindUdpSocket(); + const socketB = await bindUdpSocket(); + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-server", + socket: serverSocket, + mode: "server", + connectionId, + resendDelayMs: 1000, + resendPollMs: 1000, + }); + const receivedByA: CultNetRudpPacket[] = []; + socketA.on("message", (wire) => receivedByA.push(decodeRudpPacket(wire))); + const peerA = new CultNetRudpSession({ connectionId }); + const send = (socket: Socket, packet: CultNetRudpPacket) => + socket.send(encodeRudpPacket(packet), udpPort(serverSocket), "127.0.0.1"); + send(socketA, peerA.createConnect(0)); + await waitFor(() => receivedByA.some((p) => p.packetType === "accept"), "A's Accept"); + const acceptA = receivedByA.find((p) => p.packetType === "accept")!; + peerA.receive(acceptA, 0); + send(socketA, peerA.createAckForReceived(acceptA.sequence)); + await waitFor(() => server.outstandingReliablePacketCount === 0, "A's Accept acknowledged"); + server.send("schema", Buffer.from("never acknowledged")); + assert.ok(server.outstandingReliablePacketCount > 0); + const flushed = server.flush(2000); + const outcome = assert.rejects(flushed, /ended before its reliable writes were acknowledged/); + // The flush must have started before the timeout fires. + await new Promise((resolve) => setTimeout(resolve, 20)); + assert.equal(server.checkTimeout(50, Date.now() + 10_000), true); + return { + server, socketA, socketB, outcome, send, + close: () => { + socketA.close(); + socketB.close(); + server.close(); + }, + }; +} + +test("a flush waiting on an unacknowledged write fails as soon as the session times out", async () => { + const t = await timedOutServerWithUnackedWrite(0x1020305a); + try { + await t.outcome; + } finally { + t.close(); + } +}); + +test("a flush fails when a Connect from another endpoint follows a timeout", async () => { + const t = await timedOutServerWithUnackedWrite(0x1020305b); + try { + t.send(t.socketB, new CultNetRudpSession({ connectionId: 0x1020305b }).createConnect(0)); + await t.outcome; + } finally { + t.close(); + } +}); + +test("a flush fails when a Connect from the same endpoint follows a timeout", async () => { + const t = await timedOutServerWithUnackedWrite(0x1020305c); + try { + t.send(t.socketA, new CultNetRudpSession({ connectionId: 0x1020305c }).createConnect(0)); + await t.outcome; + } finally { + t.close(); + } +}); + +test("operation service sweeps a session that ran a handler once it idles out", async () => { + const server = await startCultNetOperationServer({ + runtimeId: "sai-sidecar", + sessionIdleTimeoutMs: 300, + handler: (request): CultNetOperationResponseMessage => ({ + schemaVersion: "cultnet.operation_response.v0", + messageId: request.messageId, + serviceId: request.serviceId, + operation: request.operation, + status: "ok", + payloadSchema: "gamecult.eve.plugin_abi.response.v1", + payloadEncoding: "messagepack-base64", + payload: request.payload, + diagnostics: [], + sourceRuntimeId: "sai-sidecar", + }), + }); + const port = Number(new URL(server.endpoint).port); + const sockets: Socket[] = []; + const request: CultNetOperationRequestMessage = { + schemaVersion: "cultnet.operation_request.v0", + messageId: "after-handling", + serviceId: "sai.vn", + operation: "describe", + payloadSchema: "gamecult.eve.plugin_abi.request.v1", + payloadEncoding: "messagepack-base64", + payload: "gaZzY2hlbWE=", + }; + const connectionId = 0x43554c54; + try { + // Sixty-four clients each run a handler and then go silent without a + // Disconnect. Only the idle sweep can free their slots. + const clients: { socket: Socket; peer: CultNetRudpSession; handled: () => boolean }[] = []; + for (let index = 0; index < 64; index += 1) { + const socket = await bindUdpSocket(); + sockets.push(socket); + const peer = new CultNetRudpSession({ connectionId }); + let handled = false; + let accepted = false; + socket.on("message", (wire) => { + const packet = decodeRudpPacket(wire); + if (packet.packetType === "accept" && !accepted) { + accepted = true; + peer.receive(packet, 0); + for (const out of peer.sendMany("schema", Buffer.from(encode(encodeCultNetMessageForWire(request, "cultnet.schema.v0"))), { reliable: true, ordered: true, nowMs: 0 })) { + socket.send(encodeRudpPacket(out), port, "127.0.0.1"); + } + } else if (packet.channelId === "schema") { + handled = true; + } + }); + socket.send(encodeRudpPacket(peer.createConnect(0)), port, "127.0.0.1"); + clients.push({ socket, peer, handled: () => handled }); + } + await waitFor(() => clients.every((c) => c.handled()), "every handler ran"); + await new Promise((resolve) => setTimeout(resolve, 700)); + const response = await invokeCultNetOperation(server.endpoint, request, { runtimeId: "eve-test", timeoutMs: 1000 }); + assert.equal(response.messageId, "after-handling"); + } finally { + for (const socket of sockets) socket.close(); + await server.close(); + } +}); diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index f549f735..e98fdc03 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -33,6 +33,12 @@ namespace GameCult.Networking /// Gets the number of bytes received, including transport framing. /// public long BytesReceived { get; internal set; } + /// + /// Gets the number of datagrams read and discarded because they belong to no session on + /// this transport: malformed frames, another session's connection id, a sender that is not + /// the peer, or a packet the session refuses. + /// + public long PacketsDropped { get; internal set; } internal CultNetTransportStats Snapshot() { @@ -41,7 +47,8 @@ namespace GameCult.Networking FramesSent = FramesSent, FramesReceived = FramesReceived, BytesSent = BytesSent, - BytesReceived = BytesReceived + BytesReceived = BytesReceived, + PacketsDropped = PacketsDropped }; } } @@ -657,12 +664,15 @@ namespace GameCult.Networking throw new InvalidOperationException("CultNet RUDP packet header length does not match the channel id length."); } - var payloadLength = checked((int)BinaryPrimitives.ReadUInt32BigEndian(wire.AsSpan(30, 4))); - if (wire.Length != headerBytes + payloadLength) + // Compared as a long: a declared length past int.MaxValue is a malformed packet, not an overflow. + var declaredPayloadLength = BinaryPrimitives.ReadUInt32BigEndian(wire.AsSpan(30, 4)); + if (declaredPayloadLength != (long)wire.Length - headerBytes) { throw new InvalidOperationException("CultNet RUDP packet payload length does not match the packet size."); } + var payloadLength = (int)declaredPayloadLength; + var flags = wire[6]; var payload = new byte[payloadLength]; Array.Copy(wire, headerBytes, payload, 0, payload.Length); @@ -916,10 +926,7 @@ namespace GameCult.Networking public CultNetRudpSession(CultNetRudpSessionOptions options) { if (options == null) throw new ArgumentNullException(nameof(options)); - if (options.MaxPendingReliablePackets.HasValue && options.MaxPendingReliablePackets.Value <= 0) - { - throw new ArgumentOutOfRangeException(nameof(options), "RUDP MaxPendingReliablePackets must be greater than zero."); - } + ValidateLimits(options.InitialSequence, options.MaxPendingReliablePackets); ConnectionId = options.ConnectionId; _nextSequence = options.InitialSequence; @@ -927,6 +934,25 @@ namespace GameCult.Networking _maxPendingReliablePackets = options.MaxPendingReliablePackets; } + internal static readonly byte[] RefusedPacketReason = Encoding.UTF8.GetBytes("session refused a packet"); + + /// + /// A session cannot admit a Connect when its sequence space starts exhausted or its reliable + /// queue holds nothing, so neither is a usable configuration. + /// + internal static void ValidateLimits(uint initialSequence, int? maxPendingReliablePackets) + { + if (initialSequence == uint.MaxValue) + { + throw new ArgumentOutOfRangeException(nameof(initialSequence), "RUDP InitialSequence must leave room for a reliable packet."); + } + + if (maxPendingReliablePackets.HasValue && maxPendingReliablePackets.Value <= 0) + { + throw new ArgumentOutOfRangeException(nameof(maxPendingReliablePackets), "RUDP MaxPendingReliablePackets must be greater than zero."); + } + } + /// /// Gets the connection/session binding id. /// @@ -999,6 +1025,36 @@ namespace GameCult.Networking return response; } + /// + /// Answers a connect from a peer this session has already accepted. The packet's sequence is + /// remembered and nothing is queued, so a connect storm cannot grow the reliable queue. The reply + /// is the accept still awaiting acknowledgement, or an ack once it was acknowledged. + /// + public CultNetRudpPacket AnswerRepeatedConnect(CultNetRudpPacket packet, long nowMs = 0) + { + RequireConnection(packet); + if (packet.PacketType != CultNetRudpPacketType.Connect) + { + throw new InvalidOperationException($"Expected RUDP connect packet, got {packet.PacketType}."); + } + + ApplyAcknowledgements(packet); + RememberReceived(packet.Sequence); + _lastReceivedAtMs = nowMs; + lock (_pendingReliableGate) + { + foreach (var pending in _pendingReliable.Values) + { + if (pending.Packet.PacketType != CultNetRudpPacketType.Accept) + continue; + pending.LastSentAtMs = nowMs; + return ClonePacket(pending.Packet); + } + } + + return CreateAck(); + } + /// /// Creates a data packet. /// @@ -1230,6 +1286,38 @@ namespace GameCult.Networking return CreatePacket(CultNetRudpPacketType.Ping, "control", payload ?? Array.Empty(), reliable: false, ordered: false, sequenced: false); } + /// + /// Forgets everything learned from the peer; sequence numbers already issued stay issued. + /// + public void ResetPeerState() + { + _connected = false; + _lastReceivedAtMs = null; + _highestReceivedSequence = null; + _receivedSequences.Clear(); + _nextSequencedByChannel.Clear(); + _latestSequencedByChannel.Clear(); + lock (_pendingReliableGate) + { + _pendingReliable.Clear(); + _queuedReliable.Clear(); + } + _orderedNextSequenceByChannel.Clear(); + _orderedBuffers.Clear(); + _fragmentBuffers.Clear(); + } + + /// + /// Ends a session that refused a packet. Receive has already recorded the packet's reliable + /// sequence, so the peer's sequences are forgotten before the goodbye is built: its ack field + /// would otherwise acknowledge the very frame the session refused. + /// + internal CultNetRudpPacket EndRefused() + { + ResetPeerState(); + return CreateDisconnect(RefusedPacketReason); + } + /// /// Creates a packet carrying a transport-level disconnect reason. /// @@ -1752,6 +1840,7 @@ namespace GameCult.Networking throw new InvalidOperationException("Only a client RUDP socket transport can initiate connect."); } + DisconnectReason = null; lock (_sessionGate) SendPacket(_session.CreateConnect(NowMs(), payload ?? Array.Empty())); } @@ -1832,6 +1921,12 @@ namespace GameCult.Networking { while (true) { + // An ended session forgot its unacknowledged writes; reporting them + // flushed would be a lie. + var ended = DisconnectReason; + if (ended != null) + throw new InvalidOperationException( + "RUDP session ended before its reliable writes were acknowledged: " + Encoding.UTF8.GetString(ended)); lock (_sessionGate) { if (_session.OutstandingReliablePacketCount == 0) @@ -2081,23 +2176,72 @@ namespace GameCult.Networking } _stats.BytesReceived += received; + + // What a datagram carries is the sender's business, not a fault of this process: a + // malformed frame, another session's connection id, a sender that is not the peer and + // a packet the session refuses are dropped and counted. Only the socket ends a loop. + // The id is checked before the peer endpoint is claimed, so a stray cannot become it. + var wire = new byte[received]; + Array.Copy(buffer, wire, received); + CultNetRudpPacket packet; + try + { + packet = CultNetRudpPacketCodec.Decode(wire); + } + catch (Exception error) when (error is InvalidOperationException || error is ArgumentException) + { + _stats.PacketsDropped++; + return true; + } + if (packet.ConnectionId != _session.ConnectionId) + { + _stats.PacketsDropped++; + return true; + } if (_remoteEndPoint == null) { + // Only a Connect claims the endpoint of a server-mode transport. + if (_mode == CultNetRudpSocketMode.Server && packet.PacketType != CultNetRudpPacketType.Connect) + { + _stats.PacketsDropped++; + return true; + } + _remoteEndPoint = remote; } else if (!_remoteEndPoint.Equals(remote)) { + _stats.PacketsDropped++; return true; } - var wire = new byte[received]; - Array.Copy(buffer, wire, received); - var packet = CultNetRudpPacketCodec.Decode(wire); TracePacket("rx", packet, received, remote); if (_mode == CultNetRudpSocketMode.Server && packet.PacketType == CultNetRudpPacketType.Connect) { lock (_sessionGate) - SendPacket(_session.AcceptConnect(packet, NowMs())); + { + CultNetRudpPacket accept; + try + { + // A connect from the peer this transport already accepted repeats: answer it + // with the accept already owed and queue nothing. + if (_session.Connected) + { + accept = _session.AnswerRepeatedConnect(packet, NowMs()); + } + else + { + accept = _session.AcceptConnect(packet, NowMs()); + DisconnectReason = null; + } + } + catch (InvalidOperationException) + { + _stats.PacketsDropped++; + return true; + } + SendPacket(accept); + } return true; } @@ -2105,7 +2249,32 @@ namespace GameCult.Networking CultNetRudpPacket? acknowledgement = null; lock (_sessionGate) { - result = _session.Receive(packet, NowMs()); + try + { + result = _session.Receive(packet, NowMs()); + } + catch (InvalidOperationException) + { + // Receive has already recorded the packet's reliable sequence, so the session + // cannot be kept: a retransmit would be acknowledged and the frame silently lost. + // End it and tell the peer; a server-mode transport also forgets its endpoint, so + // only a Connect can claim it again. + _stats.PacketsDropped++; + var goodbye = _session.EndRefused(); + DisconnectReason = CultNetRudpSession.RefusedPacketReason; + try + { + SendPacket(goodbye); + } + catch (SocketException) + { + // Best effort: the session ends whether or not the peer hears it. + } + + if (_mode == CultNetRudpSocketMode.Server) + _remoteEndPoint = null; + return true; + } if (result.Reply != null) SendPacket(result.Reply); SendPackets(result.ReadyToSend); @@ -2268,12 +2437,14 @@ namespace GameCult.Networking /// public sealed class CultNetRudpSocketServerPeer { - internal CultNetRudpSocketServerPeer(EndPoint remoteEndPoint, CultNetRudpSession session) + internal CultNetRudpSocketServerPeer(EndPoint remoteEndPoint, byte[] connectPayload, CultNetRudpSession session) { RemoteEndPoint = remoteEndPoint; + ConnectPayload = connectPayload; Session = session; } + internal byte[] ConnectPayload { get; } internal CultNetRudpSession Session { get; } internal object SessionGate { get; } = new object(); @@ -2316,6 +2487,7 @@ namespace GameCult.Networking public sealed class CultNetRudpSocketTransportServer : IDisposable { private const int MinimumReceiveBufferBytes = 4 * 1024 * 1024; + private static readonly byte[] ReplacedPeerReason = Encoding.UTF8.GetBytes("replaced by a new Connect generation"); // Keep server response pacing aligned with the 32-packet ACK mask. Large // content responses otherwise pay one Windows scheduler quantum per 1 KiB. private const int WireBurstPackets = 32; @@ -2341,6 +2513,7 @@ namespace GameCult.Networking _socket = options.Socket ?? throw new ArgumentNullException(nameof(options.Socket)); if (_socket.ReceiveBufferSize < MinimumReceiveBufferBytes) _socket.ReceiveBufferSize = MinimumReceiveBufferBytes; + CultNetRudpSession.ValidateLimits(options.InitialSequence, options.MaxPendingReliablePackets); _connectionId = options.ConnectionId; _initialSequence = options.InitialSequence; _resendDelayMs = options.ResendDelayMs; @@ -2478,21 +2651,47 @@ namespace GameCult.Networking { packet = CultNetRudpPacketCodec.Decode(wire); } - catch (InvalidOperationException) + catch (Exception error) when (error is InvalidOperationException || error is ArgumentException) { + _stats.PacketsDropped++; return true; } if (packet.ConnectionId != _connectionId) { + _stats.PacketsDropped++; return true; } var peerKey = RemoteKey(remote); if (packet.PacketType == CultNetRudpPacketType.Connect) { + var connectPayload = packet.Payload ?? Array.Empty(); + _peers.TryGetValue(peerKey, out var admitted); + // The same payload from an admitted peer is a repeated Connect: answer it with the + // Accept already owed and queue nothing. A different payload is a new generation. + if (admitted != null && admitted.ConnectPayload.AsSpan().SequenceEqual(connectPayload)) + { + lock (admitted.SessionGate) + { + CultNetRudpPacket reply; + try + { + reply = admitted.Session.AnswerRepeatedConnect(packet, NowMs()); + } + catch (InvalidOperationException) + { + _stats.PacketsDropped++; + return true; + } + SendPacket(admitted.RemoteEndPoint, reply); + } + return true; + } + var peer = new CultNetRudpSocketServerPeer( CloneEndPoint(remote), + connectPayload, new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = _connectionId, @@ -2503,25 +2702,65 @@ namespace GameCult.Networking _peers[peerKey] = peer; lock (peer.SessionGate) SendPacket(peer.RemoteEndPoint, peer.Session.AcceptConnect(packet, NowMs(), _acceptPayload)); + if (admitted != null) + { + admitted.DisconnectReason = ReplacedPeerReason; + PeerDisconnected?.Invoke(admitted); + } return true; } if (!_peers.TryGetValue(peerKey, out var existingPeer)) { + _stats.PacketsDropped++; return true; } - CultNetRudpReceiveResult result; + CultNetRudpReceiveResult? outcome; CultNetRudpPacket? acknowledgement = null; lock (existingPeer.SessionGate) { - result = existingPeer.Session.Receive(packet, NowMs()); - if (result.Reply != null) - SendPacket(existingPeer.RemoteEndPoint, result.Reply); - SendPackets(existingPeer, result.ReadyToSend); - if (packet.PacketType == CultNetRudpPacketType.Data) - acknowledgement = existingPeer.Session.CreateAckForReceived(packet.Sequence); + try + { + outcome = existingPeer.Session.Receive(packet, NowMs()); + } + catch (InvalidOperationException) + { + // Receive has already recorded the packet's reliable sequence, so the session + // cannot be kept: a retransmit would be acknowledged and the frame silently lost. + _stats.PacketsDropped++; + outcome = null; + } + + if (outcome == null) + { + try + { + SendPacket(existingPeer.RemoteEndPoint, existingPeer.Session.EndRefused()); + } + catch (SocketException) + { + // Best effort: the session ends whether or not the peer hears it. + } + } + else + { + if (outcome.Reply != null) + SendPacket(existingPeer.RemoteEndPoint, outcome.Reply); + SendPackets(existingPeer, outcome.ReadyToSend); + if (packet.PacketType == CultNetRudpPacketType.Data) + acknowledgement = existingPeer.Session.CreateAckForReceived(packet.Sequence); + } } + if (outcome == null) + { + existingPeer.DisconnectReason = CultNetRudpSession.RefusedPacketReason; + _peers.Remove(peerKey); + PeerDisconnected?.Invoke(existingPeer); + return true; + } + + var result = outcome; if (result.Disconnected) { existingPeer.DisconnectReason = result.DisconnectReason; diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 40bf43eb..e32ca91a 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -900,6 +900,20 @@ namespace GameCult.Networking.Tests Assert.That(client.Connected, Is.False); } + [Test] + public void RudpSession_RepeatedConnectAcknowledgesWhatItCarries() + { + var client = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 103, InitialSequence = 1 }); + var server = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 103, InitialSequence = 100 }); + var connect = client.CreateConnect(0, Encoding.UTF8.GetBytes("join")); + var accept = server.AcceptConnect(connect, 0); + var repeat = CultNetRudpPacketCodec.Decode(CultNetRudpPacketCodec.Encode(connect)); + repeat.Ack = accept.Sequence; + var reply = server.AnswerRepeatedConnect(repeat, 1); + Assert.That(reply.PacketType, Is.EqualTo(CultNetRudpPacketType.Ack)); + Assert.That(server.OutstandingReliablePacketCount, Is.EqualTo(0)); + } + [Test] public void RudpSession_LossyPacketsCannotCreateReliableOrderedGaps() { @@ -1193,6 +1207,365 @@ namespace GameCult.Networking.Tests } + [Test] + public void RudpSocketTransport_DropsStraysAndKeepsServingItsPeer() + { + using var serverSocket = BindUdpSocket(); + using var clientSocket = BindUdpSocket(); + using var strayHost = BindUdpSocket(); + var serverEndPoint = serverSocket.LocalEndPoint!; + var clientEndPoint = clientSocket.LocalEndPoint!; + const uint connectionId = 0x10203050; + using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "csharp-rudp-server", + Socket = serverSocket, + Mode = CultNetRudpSocketMode.Server, + ConnectionId = connectionId + }); + using var client = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "csharp-rudp-client", + Socket = clientSocket, + Mode = CultNetRudpSocketMode.Client, + RemoteEndPoint = serverEndPoint, + ConnectionId = connectionId + }); + var foreign = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 0x0BADF10E }); + var strays = new[] + { + Encoding.UTF8.GetBytes("not a rudp packet"), + new byte[] { 1 }, + CultNetRudpPacketCodec.Encode(foreign.CreateConnect(0, Encoding.UTF8.GetBytes("foreign"))) + }; + + // A stray that arrives before the peer must not become the peer. + foreach (var stray in strays) + strayHost.SendTo(stray, serverEndPoint); + for (var index = 0; index < strays.Length; index++) + Assert.That(server.ReceiveOnce(), Is.Null); + Assert.That(server.Stats.PacketsDropped, Is.EqualTo(strays.Length)); + + client.Connect(Encoding.UTF8.GetBytes("join")); + PumpRudpHandshake(client, server); + + // A moved flow answers from the peer's own address with someone else's session. + foreach (var stray in strays) + { + serverSocket.SendTo(stray, clientEndPoint); + clientSocket.SendTo(stray, serverEndPoint); + } + for (var index = 0; index < strays.Length; index++) + { + Assert.That(client.ReceiveOnce(), Is.Null); + Assert.That(server.ReceiveOnce(), Is.Null); + } + Assert.That(client.Stats.PacketsDropped, Is.EqualTo(strays.Length)); + Assert.That(server.Stats.PacketsDropped, Is.EqualTo(2 * strays.Length)); + + client.Send("schema", Encoding.UTF8.GetBytes("still here")); + Assert.That(Encoding.UTF8.GetString(ReceiveRudpFrame(server).Payload), Is.EqualTo("still here")); + server.Send("schema", Encoding.UTF8.GetBytes("server still here")); + Assert.That(Encoding.UTF8.GetString(ReceiveRudpFrame(client).Payload), Is.EqualTo("server still here")); + } + + private static List DrainPackets(Socket socket) + { + var packets = new List(); + var buffer = new byte[65535]; + EndPoint from = new IPEndPoint(IPAddress.Any, 0); + while (socket.Poll(20_000, SelectMode.SelectRead)) + { + var received = socket.ReceiveFrom(buffer, ref from); + packets.Add(CultNetRudpPacketCodec.Decode(buffer.AsSpan(0, received).ToArray())); + } + return packets; + } + + /// A reliable frame no session accepts: a fragment set with no fragment id. + private static CultNetRudpPacket PoisonedFrame(CultNetRudpSession session) + { + var packet = session.Send("schema", Encoding.UTF8.GetBytes("poison"), new CultNetRudpSendOptions { Reliable = true, Ordered = true }); + packet.FragmentCount = 2; + packet.FragmentId = 0; + return packet; + } + + [Test] + public void RudpPacketCodec_TreatsADeclaredLengthPastInt32AsMalformed() + { + var wire = CultNetRudpPacketCodec.Encode(new CultNetRudpPacket + { + PacketType = CultNetRudpPacketType.Ack, + ConnectionId = 0x10203054, + ChannelId = string.Empty + }); + Assert.That(wire, Has.Length.EqualTo(36)); + wire[30] = 0x80; + wire[31] = 0; + wire[32] = 0; + wire[33] = 0; + Assert.Throws(() => CultNetRudpPacketCodec.Decode(wire)); + + using var connectionSocket = BindUdpSocket(); + using var listenerSocket = BindUdpSocket(); + using var senderSocket = BindUdpSocket(); + using var connection = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "csharp-rudp-connection", + Socket = connectionSocket, + Mode = CultNetRudpSocketMode.Server, + ConnectionId = 0x10203054 + }); + using var listener = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions + { + RuntimeId = "csharp-rudp-listener", + Socket = listenerSocket, + ConnectionId = 0x10203054 + }); + senderSocket.SendTo(wire, connectionSocket.LocalEndPoint!); + senderSocket.SendTo(wire, listenerSocket.LocalEndPoint!); + Assert.That(connection.ReceiveOnce(), Is.Null); + Assert.That(listener.ReceiveOnce(), Is.Null); + Assert.That(connection.Stats.PacketsDropped, Is.EqualTo(1)); + Assert.That(listener.Stats.PacketsDropped, Is.EqualTo(1)); + } + + [Test] + public void RudpServerMode_IsClaimedOnlyByAConnect() + { + using var serverSocket = BindUdpSocket(); + using var orphanSocket = BindUdpSocket(); + using var clientSocket = BindUdpSocket(); + const uint connectionId = 0x10203055; + using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "csharp-rudp-server", + Socket = serverSocket, + Mode = CultNetRudpSocketMode.Server, + ConnectionId = connectionId + }); + using var client = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "csharp-rudp-client", + Socket = clientSocket, + Mode = CultNetRudpSocketMode.Client, + RemoteEndPoint = serverSocket.LocalEndPoint!, + ConnectionId = connectionId + }); + var orphan = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }); + + // A right-id packet that is not a Connect must claim nothing. + orphanSocket.SendTo(CultNetRudpPacketCodec.Encode(orphan.CreateAck()), serverSocket.LocalEndPoint!); + Assert.That(server.ReceiveOnce(), Is.Null); + Assert.That(server.Stats.PacketsDropped, Is.EqualTo(1)); + + client.Connect(Encoding.UTF8.GetBytes("join")); + PumpRudpHandshake(client, server); + Assert.That(client.Connected, Is.True); + Assert.That(server.Connected, Is.True); + } + + [Test] + public void RudpSocketTransport_EndsTheSessionThatRefusesAReliableFrame() + { + using var serverSocket = BindUdpSocket(); + using var peerSocket = BindUdpSocket(); + var serverEndPoint = serverSocket.LocalEndPoint!; + const uint connectionId = 0x10203056; + using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "csharp-rudp-server", + Socket = serverSocket, + Mode = CultNetRudpSocketMode.Server, + ConnectionId = connectionId + }); + var peer = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }); + + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(peer.CreateConnect(0)), serverEndPoint); + server.ReceiveOnce(); + peer.Receive(DrainPackets(peerSocket).Single(p => p.PacketType == CultNetRudpPacketType.Accept), 0); + + var poison = PoisonedFrame(peer); + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(poison), serverEndPoint); + Assert.That(server.ReceiveOnce(), Is.Null); + Assert.That(server.Stats.PacketsDropped, Is.EqualTo(1)); + Assert.That(server.Connected, Is.False); + Assert.That(server.DisconnectReason, Is.Not.Null); + + // The peer is told, and the goodbye does not acknowledge the refused frame. + var goodbye = DrainPackets(peerSocket).Single(); + Assert.That(goodbye.PacketType, Is.EqualTo(CultNetRudpPacketType.Disconnect)); + Assert.That(peer.Receive(goodbye, 1).Disconnected, Is.True); + Assert.That(peer.PendingReliableSequences, Does.Contain(poison.Sequence)); + + // A retransmit is dropped, not acknowledged. + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(poison), serverEndPoint); + Assert.That(server.ReceiveOnce(), Is.Null); + Assert.That(server.Stats.PacketsDropped, Is.EqualTo(2)); + Assert.That(DrainPackets(peerSocket), Is.Empty); + + // A Connect claims the endpoint again. + var again = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }); + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(again.CreateConnect(0)), serverEndPoint); + server.ReceiveOnce(); + Assert.That(DrainPackets(peerSocket).Select(p => p.PacketType), Is.EqualTo(new[] { CultNetRudpPacketType.Accept })); + } + + [Test] + public void RudpSocketTransportServer_EndsThePeerWhoseReliableFrameItRefuses() + { + using var serverSocket = BindUdpSocket(); + using var peerSocket = BindUdpSocket(); + var serverEndPoint = serverSocket.LocalEndPoint!; + const uint connectionId = 0x10203057; + using var server = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions + { + RuntimeId = "csharp-rudp-listener", + Socket = serverSocket, + ConnectionId = connectionId + }); + var ended = new List(); + server.PeerDisconnected += ended.Add; + var peer = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }); + + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(peer.CreateConnect(0)), serverEndPoint); + server.ReceiveOnce(); + peer.Receive(DrainPackets(peerSocket).Single(p => p.PacketType == CultNetRudpPacketType.Accept), 0); + Assert.That(server.Peers, Has.Count.EqualTo(1)); + + var poison = PoisonedFrame(peer); + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(poison), serverEndPoint); + Assert.That(server.ReceiveOnce(), Is.Null); + Assert.That(server.Stats.PacketsDropped, Is.EqualTo(1)); + Assert.That(server.Peers, Is.Empty); + Assert.That(ended, Has.Count.EqualTo(1)); + + var goodbye = DrainPackets(peerSocket).Single(); + Assert.That(goodbye.PacketType, Is.EqualTo(CultNetRudpPacketType.Disconnect)); + Assert.That(peer.Receive(goodbye, 1).Disconnected, Is.True); + Assert.That(peer.PendingReliableSequences, Does.Contain(poison.Sequence)); + + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(poison), serverEndPoint); + Assert.That(server.ReceiveOnce(), Is.Null); + Assert.That(server.Stats.PacketsDropped, Is.EqualTo(2)); + Assert.That(DrainPackets(peerSocket), Is.Empty); + } + + [Test] + public void RudpFlushReliable_FailsForAWriteTheEndedSessionForgot() + { + using var serverSocket = BindUdpSocket(); + using var peerSocket = BindUdpSocket(); + var serverEndPoint = serverSocket.LocalEndPoint!; + const uint connectionId = 0x10203058; + using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "csharp-rudp-server", + Socket = serverSocket, + Mode = CultNetRudpSocketMode.Server, + ConnectionId = connectionId + }); + var peer = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }); + + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(peer.CreateConnect(0)), serverEndPoint); + server.ReceiveOnce(); + peer.Receive(DrainPackets(peerSocket).Single(p => p.PacketType == CultNetRudpPacketType.Accept), 0); + server.Send("schema", Encoding.UTF8.GetBytes("never acknowledged")); + + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(PoisonedFrame(peer)), serverEndPoint); + Assert.That(server.ReceiveOnce(), Is.Null); + Assert.That(server.DisconnectReason, Is.Not.Null); + Assert.Throws(() => server.FlushReliable(TimeSpan.FromMilliseconds(200))); + } + + [Test] + public void RudpServerMode_OwesOneReliableAcceptHoweverManyConnectsRepeat() + { + using var serverSocket = BindUdpSocket(); + using var peerSocket = BindUdpSocket(); + const uint connectionId = 0x10203059; + using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "csharp-rudp-server", + Socket = serverSocket, + Mode = CultNetRudpSocketMode.Server, + ConnectionId = connectionId + }); + var connect = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }).CreateConnect(0); + for (var attempt = 0; attempt < 20; attempt++) + { + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(connect), serverSocket.LocalEndPoint!); + server.ReceiveOnce(); + } + + var accepts = DrainPackets(peerSocket).Where(p => p.PacketType == CultNetRudpPacketType.Accept && p.Reliable); + Assert.That(accepts.Select(p => p.Sequence).Distinct().Count(), Is.EqualTo(1)); + } + + [Test] + public void RudpListener_OwesOneReliableAcceptPerGenerationAndReportsAReplacedPeer() + { + using var serverSocket = BindUdpSocket(); + using var peerSocket = BindUdpSocket(); + const uint connectionId = 0x1020305a; + using var server = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions + { + RuntimeId = "csharp-rudp-listener", + Socket = serverSocket, + ConnectionId = connectionId + }); + var ended = new List(); + server.PeerDisconnected += ended.Add; + var connect = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }).CreateConnect(0, Encoding.UTF8.GetBytes("join")); + for (var attempt = 0; attempt < 20; attempt++) + { + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(connect), serverSocket.LocalEndPoint!); + server.ReceiveOnce(); + } + + var accepts = DrainPackets(peerSocket).Where(p => p.PacketType == CultNetRudpPacketType.Accept && p.Reliable); + Assert.That(accepts.Select(p => p.Sequence).Distinct().Count(), Is.EqualTo(1)); + Assert.That(server.Peers, Has.Count.EqualTo(1)); + Assert.That(ended, Is.Empty); + + // A different payload from the same endpoint is a new generation, and the old one is reported. + var other = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }).CreateConnect(0, Encoding.UTF8.GetBytes("rejoin")); + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(other), serverSocket.LocalEndPoint!); + server.ReceiveOnce(); + Assert.That(server.Peers, Has.Count.EqualTo(1)); + Assert.That(ended, Has.Count.EqualTo(1)); + } + + [Test] + public void RudpConfiguration_RejectsLimitsThatCannotAdmitAConnect() + { + Assert.Throws(() => + new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 1, InitialSequence = uint.MaxValue })); + Assert.Throws(() => + new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 1, MaxPendingReliablePackets = 0 })); + Assert.DoesNotThrow(() => + new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 1, InitialSequence = uint.MaxValue - 1, MaxPendingReliablePackets = 1 })); + + using var socket = BindUdpSocket(); + Assert.Throws(() => + new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions + { + RuntimeId = "csharp-rudp-listener", + Socket = socket, + ConnectionId = 1, + InitialSequence = uint.MaxValue + })); + Assert.Throws(() => + new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions + { + RuntimeId = "csharp-rudp-listener", + Socket = socket, + ConnectionId = 1, + MaxPendingReliablePackets = 0 + })); + } + [Test] public void RudpSocketTransport_CarriesCultNetSchemaMessages() {