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 <noreply@anthropic.com>

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:

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

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<u16, Vec<u8>>,
     sequences: BTreeMap<u16, u32>,
@@ -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(),

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

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

packages/cultnet-ts/src/rudp.ts

@@ -85,6 +85,7 @@ type FragmentBuffer = {
   fragmentCount: number;
   payloads: Map<number, Uint8Array>;
   sequences: Map<number, number>;
+  lastTouched: number;
 };
 
 export interface RudpTransportProfileOptions {
@@ -160,6 +161,7 @@ export class CultNetRudpSession {
   readonly #orderedNextSequenceByChannel = new Map<string, number>();
   readonly #orderedBuffers = new Map<string, Map<number, PendingOrderedFrame>>();
   readonly #fragmentBuffers = new Map<string, FragmentBuffer>();
+  #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) {

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