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);