From f722c213e9b32d09cf5a06bb724911ecc3970967 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 23:57:43 +0200 Subject: [PATCH] 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);