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>
This commit is contained in:
parent
c5026aa14d
commit
848b423d9c
6 changed files with 122 additions and 18 deletions
|
|
@ -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,18 +845,19 @@ 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 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(
|
||||
# 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"]),
|
||||
None,
|
||||
key=lambda pending_key: self._fragment_buffers[pending_key]["last_touched"],
|
||||
default=None,
|
||||
)
|
||||
if oldest is None:
|
||||
if stalest is None:
|
||||
raise ValueError("RUDP pending reliable fragment sets exceed the bound")
|
||||
del self._fragment_buffers[oldest]
|
||||
del self._fragment_buffers[stalest]
|
||||
self._fragment_sets_evicted += 1
|
||||
buffer = self._fragment_buffers.setdefault(
|
||||
key,
|
||||
|
|
@ -857,10 +867,13 @@ class CultNetRudpSession:
|
|||
"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:
|
||||
|
|
|
|||
|
|
@ -2308,6 +2308,22 @@ 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
|
||||
|
|
|
|||
|
|
@ -530,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,
|
||||
|
|
|
|||
|
|
@ -3102,6 +3102,39 @@ 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()?;
|
||||
|
|
|
|||
|
|
@ -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,26 +685,29 @@ export class CultNetRudpSession {
|
|||
fragmentCount,
|
||||
payloads: new Map(),
|
||||
sequences: new Map(),
|
||||
lastTouched: 0,
|
||||
};
|
||||
// Evict the oldest stranded set of unreliable fragments rather than
|
||||
// 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. Map
|
||||
// iteration is insertion-ordered, so the first match is the oldest.
|
||||
// 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 oldest: string | undefined;
|
||||
let stalest: string | undefined;
|
||||
let stalestTouched = Number.POSITIVE_INFINITY;
|
||||
for (const [pendingKey, pending] of this.#fragmentBuffers) {
|
||||
if (!pending.reliable) {
|
||||
oldest = pendingKey;
|
||||
break;
|
||||
if (!pending.reliable && pending.lastTouched < stalestTouched) {
|
||||
stalest = pendingKey;
|
||||
stalestTouched = pending.lastTouched;
|
||||
}
|
||||
}
|
||||
if (oldest === undefined) {
|
||||
if (stalest === undefined) {
|
||||
throw new Error("RUDP pending reliable fragment sets exceed the bound.");
|
||||
}
|
||||
this.#fragmentBuffers.delete(oldest);
|
||||
this.#fragmentBuffers.delete(stalest);
|
||||
this.#fragmentSetsEvicted += 1;
|
||||
}
|
||||
this.#fragmentBuffers.set(key, buffer);
|
||||
|
|
@ -703,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) {
|
||||
|
|
|
|||
|
|
@ -1872,6 +1872,27 @@ 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,
|
||||
|
|
|
|||
Loading…
Reference in a new issue