From f9be6db8ab5297c9f1188dd90507b391d4abd5df Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 23:51:52 +0200 Subject: [PATCH 1/4] Ack Cut 2 tests: a reliable fragment set is never evicted (P5) Rust, TypeScript and Python receivers evict the stalest fragment set when the pending bound is reached, including sets whose fragments were already acknowledged. These tests fail until eviction skips reliable sets and a bound full of reliable sets refuses the packet. Co-Authored-By: Claude Sonnet 5.5 --- packages/cultnet-py/tests/test_cultnet.py | 31 +++++++++++ packages/cultnet-rs/tests/cultnet.rs | 65 +++++++++++++++++++++++ packages/cultnet-ts/test/cultnet.test.ts | 34 ++++++++++++ 3 files changed, 130 insertions(+) diff --git a/packages/cultnet-py/tests/test_cultnet.py b/packages/cultnet-py/tests/test_cultnet.py index 7e78c48f..db89dd1f 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/packages/cultnet-py/tests/test_cultnet.py @@ -2308,5 +2308,36 @@ class CultNetRudpFragmentBoundTests(unittest.TestCase): self.assertEqual(receiver.receive(stranded[0]).delivered, ()) + def test_a_reliable_fragment_set_is_never_evicted_and_a_full_reliable_bound_refuses(self) -> None: + sender, receiver = self._pair() + receiver._max_pending_fragment_sets = 4 + reliable = CultNetRudpSendOptions(reliable=True) + + # Two of three reliable fragments arrive and are acknowledged. + held = sender.send_many("media", bytes([7]) * 2500, reliable, max_fragment_bytes=1000) + for packet in held[:2]: + receiver.receive(packet) + # Lossy sets pile up past the bound. + for fill in range(1, 9): + for packet in sender.send_many("media", bytes([fill]) * 2500, max_fragment_bytes=1000)[:2]: + receiver.receive(packet) + self.assertGreaterEqual(receiver.fragment_sets_evicted, 5) + delivered = receiver.receive(held[2]).delivered + self.assertEqual([frame.payload for frame in delivered], [bytes([7]) * 2500]) + + # Four pending reliable sets fill the bound; a fifth fragment is refused. + sender, receiver = self._pair() + receiver._max_pending_fragment_sets = 4 + for fill in range(1, 5): + receiver.receive(sender.send_many("media", bytes([fill]) * 2500, reliable, max_fragment_bytes=1000)[0]) + fifth = sender.send_many("media", bytes([5]) * 2500, reliable, max_fragment_bytes=1000) + with self.assertRaises(ValueError): + receiver.receive(fifth[0]) + unreliable = sender.send_many("media", bytes([6]) * 2500, max_fragment_bytes=1000) + with self.assertRaises(ValueError): + receiver.receive(unreliable[0]) + self.assertEqual(receiver.fragment_sets_evicted, 0) + + if __name__ == "__main__": unittest.main() diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 426e2ed9..ace118ce 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -3102,6 +3102,71 @@ fn rudp_evicts_stalest_incomplete_fragment_set_instead_of_dying() -> Result<()> Ok(()) } +#[test] +fn rudp_never_evicts_a_reliable_fragment_set_whose_fragments_were_acknowledged() -> Result<()> { + let (mut client, mut server) = connected_rudp_pair()?; + client.set_max_pending_fragment_sets(4)?; + let reliable = CultNetRudpSendOptions { + reliable: true, + ..CultNetRudpSendOptions::default() + }; + + // A reliable frame in three fragments: two arrive and are acknowledged. + let held = server.send_many("media", vec![7u8; 2500], reliable.clone(), Some(1000))?; + assert_eq!(held.len(), 3); + for packet in &held[..2] { + client.receive(packet, 10)?; + } + + // Lossy fragment sets pile up far past the bound. + for fill in 1..=8u8 { + let packets = server.send_many( + "media", + vec![fill; 2500], + CultNetRudpSendOptions::default(), + Some(1000), + )?; + for packet in &packets[..2] { + client.receive(packet, 11)?; + } + } + assert!(client.fragment_sets_evicted() >= 5); + + // The sender will never resend the acknowledged fragments, so the last one + // must still complete the frame. + let delivered = client.receive(&held[2], 12)?.delivered; + assert_eq!(delivered.len(), 1); + assert_eq!(delivered[0].payload, vec![7u8; 2500]); + Ok(()) +} + +#[test] +fn rudp_refuses_a_fragment_when_every_pending_set_holds_reliable_fragments() -> Result<()> { + let (mut client, mut server) = connected_rudp_pair()?; + client.set_max_pending_fragment_sets(4)?; + let reliable = CultNetRudpSendOptions { + reliable: true, + ..CultNetRudpSendOptions::default() + }; + for fill in 1..=4u8 { + let packets = server.send_many("media", vec![fill; 2500], reliable.clone(), Some(1000))?; + client.receive(&packets[0], 10)?; + } + assert_eq!(client.fragment_sets_evicted(), 0); + + let fifth = server.send_many("media", vec![5u8; 2500], reliable.clone(), Some(1000))?; + assert!(client.receive(&fifth[0], 11).is_err()); + let unreliable = server.send_many( + "media", + vec![6u8; 2500], + CultNetRudpSendOptions::default(), + Some(1000), + )?; + assert!(client.receive(&unreliable[0], 12).is_err()); + assert_eq!(client.fragment_sets_evicted(), 0); + Ok(()) +} + #[test] fn rudp_fragment_ids_wrap_instead_of_saturating() -> Result<()> { let (_client, mut server) = connected_rudp_pair()?; diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 35cba767..c97ab66e 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -1872,6 +1872,40 @@ test("rudp receiver evicts stranded fragment sets instead of refusing new ones", assert.equal(Buffer.from(second.delivered[0]!.payload).toString(), "hello"); }); +test("rudp receiver never evicts a reliable fragment set, and refuses when every pending set is reliable", () => { + const fragment = (sequence: number, id: number, index: number, reliable: boolean, count = 3) => ({ + packetType: "data" as const, connectionId: 9, sequence, ack: 0, ackMask: 0, + channelId: "schema", reliable, fragmentId: id, fragmentIndex: index, fragmentCount: count, + payload: Buffer.from([id & 0xff, index]), + }); + const open = () => { + const receiver = new CultNetRudpSession({ connectionId: 9, initialSequence: 1, resendDelayMs: 50 }); + receiver.receive({ packetType: "accept", connectionId: 9, sequence: 1, ack: 0, ackMask: 0, channelId: "control" }); + return receiver; + }; + + // Two of three reliable fragments are acknowledged; then far more lossy sets than the bound. + let receiver = open(); + receiver.receive(fragment(2, 1, 0, true)); + receiver.receive(fragment(3, 1, 1, true)); + for (let id = 2; id <= 200; id += 1) { + receiver.receive(fragment(2 + id, id, 0, false)); + } + assert.ok(receiver.fragmentSetsEvicted > 0, "nothing was evicted"); + const delivered = receiver.receive(fragment(500, 1, 2, true)).delivered; + assert.equal(delivered.length, 1, "the reliable set was evicted after its fragments were acknowledged"); + + // A bound full of reliable sets refuses the next fragment, reliable or not. + receiver = open(); + for (let id = 1; id <= 64; id += 1) { + receiver.receive(fragment(1 + id, id, 0, true)); + } + assert.equal(receiver.fragmentSetsEvicted, 0); + assert.throws(() => receiver.receive(fragment(100, 65, 0, true))); + assert.throws(() => receiver.receive(fragment(101, 66, 0, false))); + assert.equal(receiver.fragmentSetsEvicted, 0); +}); + test("rudp socket transport drops strays without an error event and keeps serving its peer", async () => { const serverSocket = await bindUdpSocket(); const clientSocket = await bindUdpSocket(); From f722c213e9b32d09cf5a06bb724911ecc3970967 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 23:57:43 +0200 Subject: [PATCH 2/4] Ack Cut 2: fragment-set eviction skips reliable sets (P5) Rust, TypeScript and Python receivers evicted the stalest fragment set when the pending bound was reached, including sets whose fragments had already been acknowledged. The sender never resends an acknowledged fragment, so the frame was lost while the sender saw it delivered. Eviction now picks the stalest set holding no reliable fragment. When every pending set is reliable the packet is refused, as every other refusal is. A conforming sender keeps at most 32 reliable packets in flight against a bound of 64 sets, so it cannot reach the refusal. Co-Authored-By: Claude Sonnet 5.5 --- .../cultnet-py/src/cultnet_py/transport.py | 16 ++++++++-- packages/cultnet-rs/src/rudp.rs | 18 ++++++++---- packages/cultnet-ts/src/rudp.ts | 29 ++++++++++++------- 3 files changed, 43 insertions(+), 20 deletions(-) diff --git a/packages/cultnet-py/src/cultnet_py/transport.py b/packages/cultnet-py/src/cultnet_py/transport.py index 0f378613..4d92857f 100644 --- a/packages/cultnet-py/src/cultnet_py/transport.py +++ b/packages/cultnet-py/src/cultnet_py/transport.py @@ -836,15 +836,25 @@ class CultNetRudpSession: key = (packet.channel_id, packet.fragment_id) if key not in self._fragment_buffers and len(self._fragment_buffers) >= self._max_pending_fragment_sets: - # Evict the oldest stranded set rather than refusing the payload; dicts - # iterate in insertion order, so the first key is the oldest. - del self._fragment_buffers[next(iter(self._fragment_buffers))] + # Evict the oldest stranded set of unreliable fragments rather than + # refusing the payload; dicts iterate in insertion order. A set holding + # reliable fragments is never evicted, because its fragments were + # acknowledged and will not be resent; when every set is reliable the + # packet is refused. + oldest = next( + (pending_key for pending_key, pending in self._fragment_buffers.items() if not pending["reliable"]), + None, + ) + if oldest is None: + raise ValueError("RUDP pending reliable fragment sets exceed the bound") + del self._fragment_buffers[oldest] self._fragment_sets_evicted += 1 buffer = self._fragment_buffers.setdefault( key, { "fragment_count": packet.fragment_count, "ordered": packet.ordered, + "reliable": packet.reliable, "payloads": {}, "sequences": {}, }, diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 3b348fe3..26dada3f 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -166,6 +166,9 @@ struct PendingOrderedFrame { struct FragmentBuffer { channel_id: String, ordered: bool, + /// Fragments of a reliable set are acknowledged on arrival, so dropping + /// the set would lose data the sender will not resend. + reliable: bool, fragment_count: u16, payloads: BTreeMap>, sequences: BTreeMap, @@ -1029,17 +1032,19 @@ impl CultNetRudpSession { if !self.fragment_buffers.contains_key(&key) && self.fragment_buffers.len() >= self.max_pending_fragment_sets { - // A set that is still incomplete when the bound is reached has - // almost certainly lost a fragment; its siblings were delivered - // long ago. Drop the one untouched longest so the session keeps - // receiving. Erroring here made any lossy fragmenting channel a - // countdown to a dead receiver. + // A set of unreliable fragments that is still incomplete when the + // bound is reached has almost certainly lost one; drop the one + // untouched longest so the session keeps receiving. A set holding + // reliable fragments is never dropped: its fragments were + // acknowledged, so the sender will not resend them. If every set + // is reliable the packet is refused instead. let stalest = self .fragment_buffers .iter() + .filter(|(_, buffer)| !buffer.reliable) .min_by_key(|(_, buffer)| buffer.last_touched) .map(|(key, _)| key.clone()) - .expect("bound reached implies a pending set exists"); + .ok_or_else(|| anyhow!("RUDP pending reliable fragment sets exceed the bound"))?; self.fragment_buffers.remove(&stalest); self.fragment_sets_evicted += 1; } @@ -1065,6 +1070,7 @@ impl CultNetRudpSession { .or_insert_with(|| FragmentBuffer { channel_id: packet.channel_id.clone(), ordered: packet.ordered, + reliable: packet.reliable, fragment_count: packet.fragment_count, payloads: BTreeMap::new(), sequences: BTreeMap::new(), diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index ae7ea838..94e5f895 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -676,18 +676,25 @@ export class CultNetRudpSession { payloads: new Map(), sequences: new Map(), }; - // Evict the oldest stranded set rather than refusing the payload. A - // receiver that stops accepting fragmented traffic after a bounded - // number of losses is a denial of service delivered by the network; - // dropping the least recent incomplete set costs one payload that was - // already incomplete. Map iteration is insertion-ordered, so the first - // key is the oldest. - while (this.#fragmentBuffers.size >= this.#maxPendingFragmentSets) { - const oldest = this.#fragmentBuffers.keys().next(); - if (oldest.done) { - break; + // Evict the oldest stranded set of unreliable fragments rather than + // refusing the payload: a receiver that stops accepting fragmented + // traffic after a bounded number of losses is a denial of service + // delivered by the network. A set holding reliable fragments is never + // evicted, because its fragments were acknowledged and will not be + // resent; when every set is reliable the packet is refused. Map + // iteration is insertion-ordered, so the first match is the oldest. + if (this.#fragmentBuffers.size >= this.#maxPendingFragmentSets) { + let oldest: string | undefined; + for (const [pendingKey, pending] of this.#fragmentBuffers) { + if (!pending.reliable) { + oldest = pendingKey; + break; + } } - this.#fragmentBuffers.delete(oldest.value); + if (oldest === undefined) { + throw new Error("RUDP pending reliable fragment sets exceed the bound."); + } + this.#fragmentBuffers.delete(oldest); this.#fragmentSetsEvicted += 1; } this.#fragmentBuffers.set(key, buffer); From c5026aa14d111c09c66c8bf8fcbda1696105a721 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 23:59:35 +0200 Subject: [PATCH 3/4] Ack Cut 2: hub bounds test follows the reliable-set rule session_and_hub_reject_configured_memory_bounds expected a reliable fragment set to be evicted at a bound of one pending set. That is the loss the cut removes: the second reliable set is now refused and nothing is evicted. Unreliable eviction stays covered in tests/cultnet.rs. Co-Authored-By: Claude Sonnet 5.5 --- packages/cultnet-rs/tests/rudp_server_hub.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/packages/cultnet-rs/tests/rudp_server_hub.rs b/packages/cultnet-rs/tests/rudp_server_hub.rs index 80cfde3b..69435846 100644 --- a/packages/cultnet-rs/tests/rudp_server_hub.rs +++ b/packages/cultnet-rs/tests/rudp_server_hub.rs @@ -399,12 +399,12 @@ fn session_and_hub_reject_configured_memory_bounds() -> Result<()> { )?; let second_set = sender.send_many("schema", vec![5, 6, 7, 8], fragment_options, Some(2))?; receiver.receive(&first_set[0], 2)?; - // The fragment-set bound evicts the stalest incomplete set rather than - // refusing the packet: a receiver that dies at its bound under loss is a - // fuse, not a limit (measured 2026-09-10, see tests/cultnet.rs). + // A reliable set is never evicted: its fragments were acknowledged and the + // sender will not resend them. With every pending set reliable, the bound + // refuses the next set instead (unreliable sets are evicted; see + // tests/cultnet.rs). + assert!(receiver.receive(&second_set[0], 2).is_err()); assert_eq!(receiver.fragment_sets_evicted(), 0); - receiver.receive(&second_set[0], 2)?; - assert_eq!(receiver.fragment_sets_evicted(), 1, "the first set was evicted for the second"); let server_socket = socket()?; let server_addr = server_socket.local_addr()?; From 848b423d9c4eaeba28db6b0483361c19e2a7143e Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 00:27:41 +0200 Subject: [PATCH 4/4] RUDP: evict the least recently touched fragment set in every runtime; document the bare receive refusal The pending-fragment bound evicted the first non-reliable set in insertion order in TypeScript and Python, so a set actively receiving could be the victim. Rust already took the least recently touched set. TS and Python now track a per-set touch counter and match Rust. A test in each runtime pins the scenario: bound full, the oldest set receives another fragment, a new set forces one eviction, and the oldest set still completes. Session receive in all three runtimes now documents that a refusal error ends the session and the caller must end it, since the refused sequence is already recorded and would be acknowledged. Co-Authored-By: Claude Sonnet 5.5 --- .../cultnet-py/src/cultnet_py/transport.py | 31 ++++++++++++----- packages/cultnet-py/tests/test_cultnet.py | 16 +++++++++ packages/cultnet-rs/src/rudp.rs | 6 ++++ packages/cultnet-rs/tests/cultnet.rs | 33 +++++++++++++++++++ packages/cultnet-ts/src/rudp.ts | 33 ++++++++++++++----- packages/cultnet-ts/test/cultnet.test.ts | 21 ++++++++++++ 6 files changed, 122 insertions(+), 18 deletions(-) diff --git a/packages/cultnet-py/src/cultnet_py/transport.py b/packages/cultnet-py/src/cultnet_py/transport.py index 4d92857f..84d3a8e0 100644 --- a/packages/cultnet-py/src/cultnet_py/transport.py +++ b/packages/cultnet-py/src/cultnet_py/transport.py @@ -424,6 +424,7 @@ class CultNetRudpSession: self._ordered_next_sequence_by_channel: dict[str, int] = {} self._ordered_buffers: dict[str, dict[int, tuple[CultNetRudpDeliveredFrame, int]]] = {} self._fragment_buffers: dict[tuple[str, int], dict[str, Any]] = {} + self._fragment_touches = 0 # Matches cultnet-rs max_pending_fragment_sets. A set that loses one # fragment is never completed; without a bound it is kept for the life # of the session and the map grows under any loss on a fragmenting channel. @@ -588,6 +589,14 @@ class CultNetRudpSession: return self._admit_reliable_packets(tuple(packets), options.now_ms) if options.reliable else tuple(packets) def receive(self, packet: CultNetRudpPacket, now_ms: int = 0) -> CultNetRudpReceiveResult: + """Feed one packet to the session. + + A raised refusal error (an oversized payload, a full reliable fragment + bound) ends the session: the packet's sequence is already recorded, so a + caller that swallows the error would acknowledge the refused packet and + drop its retransmit as a duplicate. The socket transports end the session + on such an error; a caller driving the session directly must do the same. + """ self._require_connection(packet) self._apply_acknowledgements(packet) ready_to_send = self._promote_queued_reliable(now_ms) @@ -836,18 +845,19 @@ class CultNetRudpSession: key = (packet.channel_id, packet.fragment_id) if key not in self._fragment_buffers and len(self._fragment_buffers) >= self._max_pending_fragment_sets: - # Evict the oldest stranded set of unreliable fragments rather than - # refusing the payload; dicts iterate in insertion order. A set holding - # reliable fragments is never evicted, because its fragments were - # acknowledged and will not be resent; when every set is reliable the - # packet is refused. - oldest = next( + # Evict the stalest stranded set of unreliable fragments (least recently + # touched by a fragment, so a set still receiving is never the victim) + # rather than refusing the payload. A set holding reliable fragments is + # never evicted, because its fragments were acknowledged and will not be + # resent; when every set is reliable the packet is refused. + stalest = min( (pending_key for pending_key, pending in self._fragment_buffers.items() if not pending["reliable"]), - None, + key=lambda pending_key: self._fragment_buffers[pending_key]["last_touched"], + default=None, ) - if oldest is None: + if stalest is None: raise ValueError("RUDP pending reliable fragment sets exceed the bound") - del self._fragment_buffers[oldest] + del self._fragment_buffers[stalest] self._fragment_sets_evicted += 1 buffer = self._fragment_buffers.setdefault( key, @@ -857,10 +867,13 @@ class CultNetRudpSession: "reliable": packet.reliable, "payloads": {}, "sequences": {}, + "last_touched": 0, }, ) if buffer["fragment_count"] != packet.fragment_count or buffer["ordered"] != packet.ordered: raise ValueError("RUDP fragment metadata changed within a fragment set") + self._fragment_touches += 1 + buffer["last_touched"] = self._fragment_touches buffer["payloads"][packet.fragment_index] = bytes(packet.payload) buffer["sequences"][packet.fragment_index] = packet.sequence if len(buffer["payloads"]) < packet.fragment_count: diff --git a/packages/cultnet-py/tests/test_cultnet.py b/packages/cultnet-py/tests/test_cultnet.py index db89dd1f..5f1d6b2d 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/packages/cultnet-py/tests/test_cultnet.py @@ -2308,6 +2308,22 @@ class CultNetRudpFragmentBoundTests(unittest.TestCase): self.assertEqual(receiver.receive(stranded[0]).delivered, ()) + def test_eviction_takes_the_least_recently_touched_set_not_the_actively_receiving_one(self) -> None: + sender, receiver = self._pair() + receiver._max_pending_fragment_sets = 4 + sets = [] + for fill in range(1, 5): + packets = sender.send_many("media", bytes([fill]) * 2500, max_fragment_bytes=1000) + self.assertEqual(receiver.receive(packets[0]).delivered, ()) + sets.append(packets) + # Set 1 is the oldest but is still receiving. + self.assertEqual(receiver.receive(sets[0][1]).delivered, ()) + # A fifth set forces an eviction: the stalled set 2, not set 1. + fifth = sender.send_many("media", bytes([5]) * 2500, max_fragment_bytes=1000) + self.assertEqual(receiver.receive(fifth[0]).delivered, ()) + self.assertEqual(receiver.fragment_sets_evicted, 1) + self.assertEqual(receiver.receive(sets[0][2]).delivered[0].payload, bytes([1]) * 2500) + def test_a_reliable_fragment_set_is_never_evicted_and_a_full_reliable_bound_refuses(self) -> None: sender, receiver = self._pair() receiver._max_pending_fragment_sets = 4 diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 26dada3f..2b60e95b 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -530,6 +530,12 @@ impl CultNetRudpSession { Ok(vec![packet]) } + /// Feeds one packet to the session. An `Err` refusal (an oversized payload, + /// a full reliable fragment bound) ends the session: the packet's sequence + /// is already recorded, so a caller that ignores the error would acknowledge + /// the refused packet and drop its retransmit as a duplicate. The socket + /// transports end the session on such an error; a caller driving the session + /// directly must do the same. pub fn receive( &mut self, packet: &CultNetRudpPacket, diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index ace118ce..7d512898 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -3102,6 +3102,39 @@ fn rudp_evicts_stalest_incomplete_fragment_set_instead_of_dying() -> Result<()> Ok(()) } +#[test] +fn rudp_eviction_takes_the_least_recently_touched_set_not_the_actively_receiving_one() -> Result<()> { + let (mut client, mut server) = connected_rudp_pair()?; + client.set_max_pending_fragment_sets(4)?; + let mut sets = Vec::new(); + for fill in 1..=4u8 { + let packets = server.send_many( + "media", + vec![fill; 2500], + CultNetRudpSendOptions::default(), + Some(1000), + )?; + assert!(client.receive(&packets[0], 10)?.delivered.is_empty()); + sets.push(packets); + } + // Set 1 is the oldest but is still receiving. + assert!(client.receive(&sets[0][1], 11)?.delivered.is_empty()); + // A fifth set forces an eviction: the stalled set 2, not set 1. + let fifth = server.send_many( + "media", + vec![5u8; 2500], + CultNetRudpSendOptions::default(), + Some(1000), + )?; + assert!(client.receive(&fifth[0], 12)?.delivered.is_empty()); + assert_eq!(client.fragment_sets_evicted(), 1); + assert_eq!( + client.receive(&sets[0][2], 13)?.delivered[0].payload, + vec![1u8; 2500] + ); + Ok(()) +} + #[test] fn rudp_never_evicts_a_reliable_fragment_set_whose_fragments_were_acknowledged() -> Result<()> { let (mut client, mut server) = connected_rudp_pair()?; diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index 94e5f895..e82c948f 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -85,6 +85,7 @@ type FragmentBuffer = { fragmentCount: number; payloads: Map; sequences: Map; + lastTouched: number; }; export interface RudpTransportProfileOptions { @@ -160,6 +161,7 @@ export class CultNetRudpSession { readonly #orderedNextSequenceByChannel = new Map(); readonly #orderedBuffers = new Map>(); readonly #fragmentBuffers = new Map(); + #fragmentTouches = 0; /// Matches cultnet-rs's `max_pending_fragment_sets`. A fragment set is only /// removed on successful reassembly, so a set that loses one fragment is /// stranded for the life of the session. Without a bound the map grows @@ -367,6 +369,14 @@ export class CultNetRudpSession { : packets; } + /** + * Feeds one packet to the session. A thrown refusal error (an oversized + * payload, a full reliable fragment bound) ends the session: the packet's + * sequence is already recorded, so a caller that swallows the error would + * acknowledge the refused packet and drop its retransmit as a duplicate. + * The socket transports end the session on such an error; a caller driving + * the session directly must do the same. + */ receive(packet: CultNetRudpPacket, nowMs = 0): CultNetRudpReceiveResult { this.#requireConnection(packet); this.#applyAcknowledgements(packet); @@ -675,26 +685,29 @@ export class CultNetRudpSession { fragmentCount, payloads: new Map(), sequences: new Map(), + lastTouched: 0, }; - // Evict the oldest stranded set of unreliable fragments rather than + // Evict the stalest stranded set of unreliable fragments rather than // refusing the payload: a receiver that stops accepting fragmented // traffic after a bounded number of losses is a denial of service // delivered by the network. A set holding reliable fragments is never // evicted, because its fragments were acknowledged and will not be - // resent; when every set is reliable the packet is refused. Map - // iteration is insertion-ordered, so the first match is the oldest. + // resent; when every set is reliable the packet is refused. Stalest + // means least recently touched by a fragment, so a set that is still + // receiving is never the victim. if (this.#fragmentBuffers.size >= this.#maxPendingFragmentSets) { - let oldest: string | undefined; + let stalest: string | undefined; + let stalestTouched = Number.POSITIVE_INFINITY; for (const [pendingKey, pending] of this.#fragmentBuffers) { - if (!pending.reliable) { - oldest = pendingKey; - break; + if (!pending.reliable && pending.lastTouched < stalestTouched) { + stalest = pendingKey; + stalestTouched = pending.lastTouched; } } - if (oldest === undefined) { + if (stalest === undefined) { throw new Error("RUDP pending reliable fragment sets exceed the bound."); } - this.#fragmentBuffers.delete(oldest); + this.#fragmentBuffers.delete(stalest); this.#fragmentSetsEvicted += 1; } this.#fragmentBuffers.set(key, buffer); @@ -703,6 +716,8 @@ export class CultNetRudpSession { throw new Error("RUDP fragment metadata changed within a fragment set."); } + this.#fragmentTouches += 1; + buffer.lastTouched = this.#fragmentTouches; buffer.payloads.set(fragmentIndex, payload); buffer.sequences.set(fragmentIndex, packet.sequence); if (buffer.payloads.size < fragmentCount) { diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index c97ab66e..fb249773 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -1872,6 +1872,27 @@ test("rudp receiver evicts stranded fragment sets instead of refusing new ones", assert.equal(Buffer.from(second.delivered[0]!.payload).toString(), "hello"); }); +test("rudp eviction takes the least recently touched set, not the one still receiving", () => { + const receiver = new CultNetRudpSession({ connectionId: 9, initialSequence: 1, resendDelayMs: 50 }); + receiver.receive({ packetType: "accept", connectionId: 9, sequence: 1, ack: 0, ackMask: 0, channelId: "control" }); + const fragment = (id: number, index: number, sequence: number) => receiver.receive({ + packetType: "data", connectionId: 9, sequence, ack: 0, ackMask: 0, + channelId: "schema", fragmentId: id, fragmentIndex: index, fragmentCount: 3, + payload: Buffer.from([id & 0xff, index]), + }); + let sequence = 100; + for (let id = 1; id <= 64; id += 1) { + assert.equal(fragment(id, 0, sequence++).delivered.length, 0); + } + assert.equal(receiver.fragmentSetsEvicted, 0); + // Set 1 is the oldest but is still receiving. + assert.equal(fragment(1, 1, sequence++).delivered.length, 0); + // A new set forces one eviction: the stalled set 2, not set 1. + assert.equal(fragment(1000, 0, sequence++).delivered.length, 0); + assert.equal(receiver.fragmentSetsEvicted, 1); + assert.equal(fragment(1, 2, sequence++).delivered.length, 1, "the actively receiving set was evicted"); +}); + test("rudp receiver never evicts a reliable fragment set, and refuses when every pending set is reliable", () => { const fragment = (sequence: number, id: number, index: number, reliable: boolean, count = 3) => ({ packetType: "data" as const, connectionId: 9, sequence, ack: 0, ackMask: 0,