GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Merge RUDP stray-packet hardening across all four runtimes

Browse source 
A stray, malformed, foreign-id or refused datagram is dropped and counted,
never fatal; a refused reliable frame ends the peer's session instead of
being acked and lost; flush fails once its session ends; one reliable
Accept per session; the TS operation service expires idle sessions and
survives a throwing handler. Four Soul passes; session-generation and
reconnect semantics move to the ack cut.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 22:05:30 +02:00
commit 3bf1c0cea5
14 changed files with 2933 additions and 128 deletions
Download patch file Download diff file Expand all files Collapse all files

104
packages/cultmesh-rs/src/rudp_document_server.rs
View file

@ -207,6 +207,7 @@ pub struct CultMeshRudpDocumentServer<S, Q, C> {
snapshot_source: Q,
clock: C,
options: CultMeshRudpDocumentServerOptions,
packets_dropped: u64,
}
impl<S, Q, C> CultMeshRudpDocumentServer<S, Q, C>
@ -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<SocketAddr> {
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<bool> {
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

100
packages/cultmesh-rs/tests/rudp_document_server.rs
View file

@ -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(())
}

111
packages/cultnet-py/src/cultnet_py/transport.py
View file

@ -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")

238
packages/cultnet-py/tests/test_cultnet.py
View file

@ -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()

193
packages/cultnet-rs/src/rudp.rs
View file

@ -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<usize>,
max_payload_bytes: Option<usize>,
max_pending_fragment_sets: usize,
initial_sequence: u32,
next_sequence: u32,
next_sequenced_by_channel: BTreeMap<String, u32>,
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<u8>) -> Result<CultNetRudpPacket> {
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<CultNetRudpPacket> {
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<Self> {
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<Self> {
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<u32>,
) -> 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<String>,

5
packages/cultnet-rs/src/transport.rs
View file

@ -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)]

17
packages/cultnet-rs/tests/rudp_server_hub.rs
View file

@ -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]

752
packages/cultnet-rs/tests/rudp_stray_packets.rs Normal file
View file

@ -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<UdpSocket> {
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<CultNetRudpPacket> {
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<u8>) {
(64, vec![7_u8; 100])
}
fn client_with_fast_resend(server_addr: SocketAddr) -> Result<CultNetRudpSocketTransportConnection> {
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<std::collections::BTreeSet<u32>> {
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<Vec<CultNetRudpPacket>> {
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(())
}

112
packages/cultnet-ts/src/operation-service.ts
View file

@ -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<CultNetOperationResponseMessage>;
}
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<void>;
}
@ -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<string, RemoteSession>,
connectionId: number,
options: CultNetOperationServerOptions,
wire: Buffer,
remote: RemoteInfo,
sendPacket: (remote: RemoteInfo, packet: CultNetRudpPacket) => void,
): Promise<void> {
): Promise<boolean> {
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<CultNetRudpSession["receive"]>;
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<string, RemoteSession>,
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 } {

149
packages/cultnet-ts/src/rudp.ts
View file

@ -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<void> {
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<CultNetRudpSession["receive"]>;
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;

3
packages/cultnet-ts/src/transport.ts
View file

@ -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) {

623
packages/cultnet-ts/test/cultnet.test.ts
View file

@ -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();
}
});

281
src/GameCult.Networking/CultNetTransport.cs
View file

@ -33,6 +33,12 @@ namespace GameCult.Networking
/// Gets the number of bytes received, including transport framing.
/// </summary>
public long BytesReceived { get; internal set; }
/// <summary>
/// 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.
/// </summary>
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");
/// <summary>
/// A session cannot admit a Connect when its sequence space starts exhausted or its reliable
/// queue holds nothing, so neither is a usable configuration.
/// </summary>
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.");
}
}
/// <summary>
/// Gets the connection/session binding id.
/// </summary>
@ -999,6 +1025,36 @@ namespace GameCult.Networking
return response;
}
/// <summary>
/// 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.
/// </summary>
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();
}
/// <summary>
/// Creates a data packet.
/// </summary>
@ -1230,6 +1286,38 @@ namespace GameCult.Networking
return CreatePacket(CultNetRudpPacketType.Ping, "control", payload ?? Array.Empty<byte>(), reliable: false, ordered: false, sequenced: false);
}
/// <summary>
/// Forgets everything learned from the peer; sequence numbers already issued stay issued.
/// </summary>
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();
}
/// <summary>
/// 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.
/// </summary>
internal CultNetRudpPacket EndRefused()
{
ResetPeerState();
return CreateDisconnect(RefusedPacketReason);
}
/// <summary>
/// Creates a packet carrying a transport-level disconnect reason.
/// </summary>
@ -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<byte>()));
}
@ -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
/// </summary>
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<byte>();
_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;

373
tests/GameCult.Networking.Tests/NetworkingTests.cs
View file

@ -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<CultNetRudpPacket> DrainPackets(Socket socket)
{
var packets = new List<CultNetRudpPacket>();
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;
}
/// <summary>A reliable frame no session accepts: a fragment set with no fragment id.</summary>
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<InvalidOperationException>(() => 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<CultNetRudpSocketServerPeer>();
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<InvalidOperationException>(() => 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<CultNetRudpSocketServerPeer>();
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<ArgumentOutOfRangeException>(() =>
new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 1, InitialSequence = uint.MaxValue }));
Assert.Throws<ArgumentOutOfRangeException>(() =>
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<ArgumentOutOfRangeException>(() =>
new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions
{
RuntimeId = "csharp-rudp-listener",
Socket = socket,
ConnectionId = 1,
InitialSequence = uint.MaxValue
}));
Assert.Throws<ArgumentOutOfRangeException>(() =>
new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions
{
RuntimeId = "csharp-rudp-listener",
Socket = socket,
ConnectionId = 1,
MaxPendingReliablePackets = 0
}));
}
[Test]
public void RudpSocketTransport_CarriesCultNetSchemaMessages()
{