From b6b1d9c66049bbdb9c4dbca8ab7cbac1e69d0f11 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 10 Sep 2026 11:42:17 +0200 Subject: [PATCH] Bound the Python fragment reassembly map cultcache-py removed a fragment buffer only on successful reassembly, the same shape cultnet-ts had: a set that loses one fragment is kept for the life of the session and the map grows under any loss on a fragmenting channel. Cap at 64 to match the other runtimes, evict the oldest stranded set (dict insertion order) rather than refuse the payload, and expose fragment_sets_evicted so the drop is observable. The fragment id already wrapped correctly here; only the Rust allocator saturated. PYTHONPATH=src python -m unittest tests.test_cultcache -> 142 run, 1 pre-existing error (interop fixture path is cwd-relative, same on main) Co-Authored-By: Claude Fable 5.1 --- .../cultcache-py/src/cultnet_py/transport.py | 15 ++++++++ packages/cultcache-py/tests/test_cultcache.py | 35 +++++++++++++++++++ 2 files changed, 50 insertions(+) diff --git a/packages/cultcache-py/src/cultnet_py/transport.py b/packages/cultcache-py/src/cultnet_py/transport.py index 2b9fe462..73bd1e62 100644 --- a/packages/cultcache-py/src/cultnet_py/transport.py +++ b/packages/cultcache-py/src/cultnet_py/transport.py @@ -418,11 +418,21 @@ 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]] = {} + # 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. + self._max_pending_fragment_sets = 64 + self._fragment_sets_evicted = 0 @property def connected(self) -> bool: return self._connected + @property + def fragment_sets_evicted(self) -> int: + """Incomplete fragment sets dropped to admit newer ones; one lost payload each.""" + return self._fragment_sets_evicted + @property def last_received_at_ms(self) -> int | None: return self._last_received_at_ms @@ -785,6 +795,11 @@ class CultNetRudpSession: raise ValueError("RUDP fragment index must be lower than fragment count") 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))] + self._fragment_sets_evicted += 1 buffer = self._fragment_buffers.setdefault( key, { diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 29fdefef..639b5508 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -6654,5 +6654,40 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(streams.latest_frame("mimir:kiyo-pro").sequence, 42) +class CultNetRudpFragmentBoundTests(unittest.TestCase): + def _pair(self) -> tuple[CultNetRudpSession, CultNetRudpSession]: + sender = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=311, initial_sequence=1)) + receiver = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=311, initial_sequence=100)) + sender.receive(CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 311, 90, 0, 0, "control")) + receiver.receive(CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 311, 91, 0, 0, "control")) + return sender, receiver + + def test_stranded_fragment_sets_are_bounded_and_evicted_oldest_first(self) -> None: + sender, receiver = self._pair() + receiver._max_pending_fragment_sets = 4 + + # Eight payloads each lose their last fragment; the map must not grow past four. + stranded = [] + for fill in range(1, 9): + packets = sender.send_many("media", bytes([fill]) * 2500, max_fragment_bytes=1000) + self.assertEqual(len(packets), 3) + for packet in packets[:2]: + self.assertEqual(receiver.receive(packet).delivered, ()) + stranded.append(packets[2]) + self.assertEqual(len(receiver._fragment_buffers), 4) + self.assertEqual(receiver.fragment_sets_evicted, 4) + + # A complete payload still lands. + delivered = [] + for packet in sender.send_many("media", bytes([9]) * 2500, max_fragment_bytes=1000): + delivered.extend(receiver.receive(packet).delivered) + self.assertEqual([frame.payload for frame in delivered], [bytes([9]) * 2500]) + self.assertEqual(receiver.fragment_sets_evicted, 5) + + # Oldest-first: set 8 still pends and completes late; set 1 was evicted. + self.assertEqual(receiver.receive(stranded[7]).delivered[0].payload, bytes([8]) * 2500) + self.assertEqual(receiver.receive(stranded[0]).delivered, ()) + + if __name__ == "__main__": unittest.main()