diff --git a/packages/cultnet-py/src/cultnet_py/transport.py b/packages/cultnet-py/src/cultnet_py/transport.py index 0f378613..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,21 +845,35 @@ 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 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"]), + key=lambda pending_key: self._fragment_buffers[pending_key]["last_touched"], + default=None, + ) + if stalest is None: + raise ValueError("RUDP pending reliable fragment sets exceed the bound") + del self._fragment_buffers[stalest] self._fragment_sets_evicted += 1 buffer = self._fragment_buffers.setdefault( key, { "fragment_count": packet.fragment_count, "ordered": packet.ordered, + "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 7e78c48f..5f1d6b2d 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/packages/cultnet-py/tests/test_cultnet.py @@ -2308,5 +2308,52 @@ 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 + 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/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 3b348fe3..2b60e95b 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, @@ -527,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, @@ -1029,17 +1038,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 +1076,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-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 426e2ed9..7d512898 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -3102,6 +3102,104 @@ 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()?; + 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-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()?; diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index ae7ea838..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,19 +685,29 @@ export class CultNetRudpSession { fragmentCount, payloads: new Map(), sequences: new Map(), + lastTouched: 0, }; - // 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 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. 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 stalest: string | undefined; + let stalestTouched = Number.POSITIVE_INFINITY; + for (const [pendingKey, pending] of this.#fragmentBuffers) { + if (!pending.reliable && pending.lastTouched < stalestTouched) { + stalest = pendingKey; + stalestTouched = pending.lastTouched; + } } - this.#fragmentBuffers.delete(oldest.value); + if (stalest === undefined) { + throw new Error("RUDP pending reliable fragment sets exceed the bound."); + } + this.#fragmentBuffers.delete(stalest); this.#fragmentSetsEvicted += 1; } this.#fragmentBuffers.set(key, buffer); @@ -696,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 35cba767..fb249773 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -1872,6 +1872,61 @@ 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, + 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();