Bound the TypeScript fragment reassembly map
cultnet-ts removed a fragment buffer only on successful reassembly, with no cap, no eviction and no timeout. A fragment set that loses one fragment was stranded for the life of the session, so the map grew without bound under any loss on a fragmenting channel. cultnet-rs at least fails closed at 64 pending sets; TypeScript just consumes memory. This is live, not theoretical. Ghostlight sends Heimdall sealed envelopes over CultMesh RUDP with max_fragment_bytes = 2048 (Ghostlight/crates/ghostlight-dungeon/src/heimdall.rs:697), and Heimdall's private command plane reassembles them through this path. Every lost fragment on that link left a buffer nothing would ever collect, on a long-running auth service. Caps pending sets at 64 to match cultnet-rs's max_pending_fragment_sets, and evicts 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. Map iteration is insertion-ordered, so the first key is the oldest. fragmentSetsEvicted and pendingFragmentSetCount make it observable. Found by the Muninn media load test, which hit the Rust half of the same defect; cultcache-py has the same unbounded map and is not touched here. npm run test --workspace=packages/cultnet-ts -> pass 54, fail 0 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
05e0925143
commit
2041a0e0ed
2 changed files with 78 additions and 0 deletions
|
|
@ -160,6 +160,12 @@ export class CultNetRudpSession {
|
|||
readonly #orderedNextSequenceByChannel = new Map<string, number>();
|
||||
readonly #orderedBuffers = new Map<string, Map<number, PendingOrderedFrame>>();
|
||||
readonly #fragmentBuffers = new Map<string, FragmentBuffer>();
|
||||
/// 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
|
||||
/// forever under any loss on a fragmenting channel.
|
||||
#maxPendingFragmentSets = 64;
|
||||
#fragmentSetsEvicted = 0;
|
||||
|
||||
constructor(options: CultNetRudpSessionOptions) {
|
||||
this.connectionId = toUint32(options.connectionId, "connectionId");
|
||||
|
|
@ -187,6 +193,17 @@ export class CultNetRudpSession {
|
|||
return this.#pendingReliable.size + this.#queuedReliable.length;
|
||||
}
|
||||
|
||||
/// Incomplete fragment sets dropped to stay within the pending bound.
|
||||
/// Non-zero means the peer is losing fragments: the receiver is still
|
||||
/// serving, and something upstream is not.
|
||||
get fragmentSetsEvicted(): number {
|
||||
return this.#fragmentSetsEvicted;
|
||||
}
|
||||
|
||||
get pendingFragmentSetCount(): number {
|
||||
return this.#fragmentBuffers.size;
|
||||
}
|
||||
|
||||
get lastReceivedAtMs(): number | undefined {
|
||||
return this.#lastReceivedAtMs;
|
||||
}
|
||||
|
|
@ -203,6 +220,7 @@ export class CultNetRudpSession {
|
|||
this.#orderedNextSequenceByChannel.clear();
|
||||
this.#orderedBuffers.clear();
|
||||
this.#fragmentBuffers.clear();
|
||||
this.#fragmentSetsEvicted = 0;
|
||||
}
|
||||
|
||||
createConnect(nowMs = 0, payload = new Uint8Array()): CultNetRudpPacket {
|
||||
|
|
@ -636,6 +654,20 @@ 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;
|
||||
}
|
||||
this.#fragmentBuffers.delete(oldest.value);
|
||||
this.#fragmentSetsEvicted += 1;
|
||||
}
|
||||
this.#fragmentBuffers.set(key, buffer);
|
||||
}
|
||||
if (buffer.fragmentCount !== fragmentCount || buffer.ordered !== (packet.ordered ?? false)) {
|
||||
|
|
|
|||
|
|
@ -1821,3 +1821,49 @@ test("Generated Ghostlight contracts can feed CultCacheTS directly without a Zod
|
|||
rmSync(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("rudp receiver evicts stranded fragment sets instead of refusing new ones", () => {
|
||||
// A fragment set is removed only on successful reassembly, so every set that
|
||||
// loses a fragment is stranded for the life of the session. Ghostlight sends
|
||||
// Heimdall fragmented envelopes (max_fragment_bytes = 2048), so on the live
|
||||
// auth path this map grew for every lost fragment and was never swept.
|
||||
const receiver = new CultNetRudpSession({ connectionId: 9, initialSequence: 1, resendDelayMs: 50 });
|
||||
receiver.receive({ packetType: "accept", connectionId: 9, sequence: 1, ack: 0, ackMask: 0, channelId: "control" });
|
||||
|
||||
// 200 fragment sets that each lose their second fragment.
|
||||
for (let id = 1; id <= 200; id += 1) {
|
||||
const stranded = receiver.receive({
|
||||
packetType: "data",
|
||||
connectionId: 9,
|
||||
sequence: 100 + id,
|
||||
ack: 0,
|
||||
ackMask: 0,
|
||||
channelId: "schema",
|
||||
fragmentId: id,
|
||||
fragmentIndex: 0,
|
||||
fragmentCount: 2,
|
||||
payload: Buffer.from([id & 0xff]),
|
||||
});
|
||||
assert.equal(stranded.delivered.length, 0, "an incomplete set must not reassemble");
|
||||
}
|
||||
|
||||
// Bounded rather than unbounded, and still serving.
|
||||
assert.ok(receiver.pendingFragmentSetCount <= 64, `pending ${receiver.pendingFragmentSetCount}`);
|
||||
assert.ok(receiver.fragmentSetsEvicted > 0, "nothing was evicted");
|
||||
|
||||
// The receiver still completes a fresh fragmented payload, which is what
|
||||
// erroring at the cap took away: any nonzero loss killed the channel.
|
||||
const first = receiver.receive({
|
||||
packetType: "data", connectionId: 9, sequence: 900, ack: 0, ackMask: 0,
|
||||
channelId: "schema", fragmentId: 1000, fragmentIndex: 0, fragmentCount: 2,
|
||||
payload: Buffer.from("he"),
|
||||
});
|
||||
assert.equal(first.delivered.length, 0);
|
||||
const second = receiver.receive({
|
||||
packetType: "data", connectionId: 9, sequence: 901, ack: 0, ackMask: 0,
|
||||
channelId: "schema", fragmentId: 1000, fragmentIndex: 1, fragmentCount: 2,
|
||||
payload: Buffer.from("llo"),
|
||||
});
|
||||
assert.equal(second.delivered.length, 1);
|
||||
assert.equal(Buffer.from(second.delivered[0]!.payload).toString(), "hello");
|
||||
});
|
||||
|
|
|
|||
Loading…
Reference in a new issue