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 <noreply@anthropic.com>
This commit is contained in:
parent
2041a0e0ed
commit
b6b1d9c660
2 changed files with 50 additions and 0 deletions
|
|
@ -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,
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Reference in a new issue