Cap the heap unanswered messages hold, and detach them from the chunk they arrived in

src/held-messages.ts

@@ -2,10 +2,12 @@ import type { PduObject } from './pdu.ts';
 import type { SmppLog } from './log.ts';
 import { ExpiringGroups } from './expiring-groups.ts';
 import { IdleWaiters } from './idle-waiters.ts';
+import { retainedOctets } from './retained-pdu.ts';
 
 export type HeldMessagesOptions = {
 	log: SmppLog;
 	max: number;
+	maxOctets: number;
 	/** Injected so expiry can be exercised without a wall clock. */
 	now?: (() => number) | undefined;
 	timeout: number;
@@ -18,12 +20,19 @@ function keyOf(pduObjs: PduObject[]): string | undefined {
 	return first ? String(first.seqNr) : undefined;
 }
 
+type Held = {
+	octets: number;
+	pduObjs: PduObject[];
+};
+
 /** The messages handed to the application that it has not answered yet, held by their segments. */
 export class HeldMessages {
-	private readonly held: ExpiringGroups<PduObject[]>;
+	private readonly held: ExpiringGroups<Held>;
 	private readonly idleWaiters = new IdleWaiters();
 	private readonly log: SmppLog;
 	private readonly max: number;
+	private readonly maxOctets: number;
+	private octets = 0;
 
 	constructor(options: HeldMessagesOptions) {
 		this.held = new ExpiringGroups({
@@ -34,6 +43,7 @@ export class HeldMessages {
 		});
 		this.log = options.log;
 		this.max = options.max;
+		this.maxOctets = options.maxOctets;
 	}
 
 	get size(): number {
@@ -48,35 +58,49 @@ export class HeldMessages {
 
 		this.sweep();
 
-		if (this.held.get(key)) {
+		const replaced = this.held.get(key);
+
+		if (replaced) {
 			this.log.warn('heldMessages - replacing a message on a re-used sequence number', { seqNr: Number(key) });
+			this.delete(key, replaced);
 		} else if (this.held.full) {
 			this.dropOldest();
 		}
 
-		this.held.set(key, pduObjs);
+		const octets = pduObjs.reduce((sum, pduObj) => sum + retainedOctets(pduObj), 0);
+
+		// The message just held stays even alone past the cap: the peer is still owed its answer.
+		while (this.held.size > 0 && this.octets + octets > this.maxOctets) {
+			this.dropOldest();
+		}
+
+		this.held.set(key, { octets, pduObjs });
+		this.octets += octets;
 	}
 
 	/** Whether a drain is still waiting for this message to be answered. */
 	has(pduObjs: PduObject[]): boolean {
 		const key = keyOf(pduObjs);
 
-		return key !== undefined && this.held.get(key) === pduObjs;
+		return key !== undefined && this.held.get(key)?.pduObjs === pduObjs;
 	}
 
 	release(pduObjs: PduObject[]): void {
 		const key = keyOf(pduObjs);
 
 		// Identity, not the key: a wrapped sequence number must not release someone else's message.
-		if (key === undefined || this.held.get(key) !== pduObjs) return;
+		const held = key === undefined ? undefined : this.held.get(key);
 
-		this.held.delete(key);
+		if (key === undefined || held?.pduObjs !== pduObjs) return;
+
+		this.delete(key, held);
 		this.settle();
 	}
 
 	/** Drops every message: their segments went with the link, so no answer of ours correlates now. */
 	clear(): void {
 		this.held.takeAll();
+		this.octets = 0;
 		this.idleWaiters.settle();
 	}
 
@@ -90,10 +114,13 @@ export class HeldMessages {
 
 		if (!oldest) return;
 
-		const [seqNr] = oldest;
+		const [seqNr, held] = oldest;
 
+		this.octets -= held.octets;
 		this.log.warn('heldMessages - buffer full, dropping the oldest message', {
 			max: this.max,
+			maxOctets: this.maxOctets,
+			octets: this.octets,
 			seqNr: Number(seqNr),
 		});
 	}
@@ -104,12 +131,21 @@ export class HeldMessages {
 
 		if (expired.length === 0) return;
 
+		for (const [, held] of expired) {
+			this.octets -= held.octets;
+		}
+
 		this.log.warn('heldMessages - messages the application never answered', {
 			messages: expired.length,
 		});
 		this.settle();
 	}
 
+	private delete(key: string, held: Held): void {
+		this.held.delete(key);
+		this.octets -= held.octets;
+	}
+
 	private settle(): void {
 		if (this.held.size === 0) this.idleWaiters.settle();
 	}

src/incoming-requests.ts

@@ -13,6 +13,7 @@ import { Reassembler, decodeSegments } from './reassembly.ts';
 import { bindCommands, defaults, standsInFor } from './session-options.ts';
 import { concatOf } from './concat.ts';
 import { createSms } from './sms.ts';
+import { detach } from './retained-pdu.ts';
 import { dlrFromPdu } from './dlr.ts';
 import { paramText } from './defs/types.ts';
 import { respIdParams, segmentId } from './sms-id.ts';
@@ -72,6 +73,7 @@ export class IncomingRequests {
 		this.held = new HeldMessages({
 			log: options.log,
 			max: defaults.maxHeldMessages,
+			maxOctets: defaults.maxHeldOctets,
 			timeout: defaults.heldMessageTimeout,
 		});
 		this.log = options.log;
@@ -213,7 +215,7 @@ export class IncomingRequests {
 		const concat = concatOf(pduObj);
 
 		if (!concat) {
-			this.emitSms([pduObj]);
+			this.emitSms([detach(pduObj)]);
 
 			return;
 		}

src/reassembly.ts

@@ -1,12 +1,11 @@
 import type { Concat } from './concat.ts';
-import type { ParamValue } from './defs/types.ts';
 import type { PduObject } from './pdu.ts';
 import type { SmppLog } from './log.ts';
-import type { Tlv } from './defs/tlvs.ts';
 import { ExpiringGroups } from './expiring-groups.ts';
 import { decodeMessage } from './message.ts';
+import { detach, retainedOctets } from './retained-pdu.ts';
 import { messageOctets } from './message-body.ts';
-import { detachedTlv, paramNumber, paramText, tlvOctets } from './defs/types.ts';
+import { paramNumber, paramText } from './defs/types.ts';
 import { uuidv7 } from './uuid.ts';
 
 /** A concatenated message given up on, whose segments the peer has already been answered for. */
@@ -52,54 +51,6 @@ type Group = {
 	total: number;
 };
 
-/** Wire reads hand back views, so retaining one segment would pin the whole PDU it arrived in. */
-function detach(pduObj: PduObject): PduObject {
-	const params: Record<string, ParamValue> = {};
-	const tlvs: Record<string, Tlv> = {};
-
-	for (const [name, value] of Object.entries(pduObj.params)) {
-		params[name] = Buffer.isBuffer(value) ? Buffer.from(value) : value;
-	}
-
-	for (const [name, tlv] of Object.entries(pduObj.tlvs)) {
-		tlvs[name] = { ...tlv, tagValue: detachedTlv(tlv.tagValue) };
-	}
-
-	// short_message holds the same octets wherever it was not decoded, so one copy covers both.
-	const octets = Buffer.isBuffer(params.short_message)
-		? params.short_message
-		: pduObj.shortMessageOctets && Buffer.from(pduObj.shortMessageOctets);
-
-	return { ...pduObj, params, shortMessageOctets: octets, tlvs };
-}
-
-// Measured heap beyond the octets, so a segment of empty fields or empty TLVs is not free.
-const segmentObjectOverhead = 1000;
-const tlvObjectOverhead = 300;
-
-// A cstring param arrives as a string, and source_addr alone can carry most of a 1 MiB PDU.
-function sizeOf(value: ParamValue): number {
-	if (Buffer.isBuffer(value)) return value.length;
-
-	return typeof value === 'string' ? value.length : 0;
-}
-
-function octetsOf(pduObj: PduObject): number {
-	let octets = segmentObjectOverhead;
-
-	for (const value of Object.values(pduObj.params)) {
-		octets += sizeOf(value);
-	}
-
-	for (const tlv of Object.values(pduObj.tlvs)) {
-		const listed = Array.isArray(tlv.tagValue) ? tlv.tagValue.length : 0;
-
-		octets += tlvOctets(tlv.tagValue) + (1 + listed) * tlvObjectOverhead;
-	}
-
-	return octets;
-}
-
 // NUL: the one octet a C-Octet String address cannot hold, so no sender can forge another's key.
 function groupKey(pduObj: PduObject, concat: Concat): string {
 	return [
@@ -169,7 +120,7 @@ export class Reassembler {
 		const group = existing ?? this.open(key, concat.total);
 		const replaced = group.parts.get(concat.part);
 		const segment = detach(pduObj);
-		const delta = octetsOf(segment) - (replaced === undefined ? 0 : octetsOf(replaced));
+		const delta = retainedOctets(segment) - (replaced === undefined ? 0 : retainedOctets(replaced));
 
 		group.parts.set(concat.part, segment);
 		group.octets += delta;

src/retained-pdu.ts

@@ -0,0 +1,53 @@
+import type { ParamValue } from './defs/types.ts';
+import type { PduObject } from './pdu.ts';
+import type { Tlv } from './defs/tlvs.ts';
+import { detachedTlv, tlvOctets } from './defs/types.ts';
+
+/** Wire reads hand back views, so retaining one PDU would pin the whole chunk it arrived in. */
+export function detach(pduObj: PduObject): PduObject {
+	const params: Record<string, ParamValue> = {};
+	const tlvs: Record<string, Tlv> = {};
+
+	for (const [name, value] of Object.entries(pduObj.params)) {
+		params[name] = Buffer.isBuffer(value) ? Buffer.from(value) : value;
+	}
+
+	for (const [name, tlv] of Object.entries(pduObj.tlvs)) {
+		tlvs[name] = { ...tlv, tagValue: detachedTlv(tlv.tagValue) };
+	}
+
+	// short_message holds the same octets wherever it was not decoded, so one copy covers both.
+	const octets = Buffer.isBuffer(params.short_message)
+		? params.short_message
+		: pduObj.shortMessageOctets && Buffer.from(pduObj.shortMessageOctets);
+
+	return { ...pduObj, params, shortMessageOctets: octets, tlvs };
+}
+
+// Measured heap beyond the octets, so a PDU of empty fields or empty TLVs is not free.
+const pduObjectOverhead = 1000;
+const tlvObjectOverhead = 300;
+
+// A cstring param arrives as a string, and source_addr alone can carry most of a 1 MiB PDU.
+function sizeOf(value: ParamValue): number {
+	if (Buffer.isBuffer(value)) return value.length;
+
+	return typeof value === 'string' ? value.length : 0;
+}
+
+/** Roughly the heap a detached PDU holds. */
+export function retainedOctets(pduObj: PduObject): number {
+	let octets = pduObjectOverhead;
+
+	for (const value of Object.values(pduObj.params)) {
+		octets += sizeOf(value);
+	}
+
+	for (const tlv of Object.values(pduObj.tlvs)) {
+		const listed = Array.isArray(tlv.tagValue) ? tlv.tagValue.length : 0;
+
+		octets += tlvOctets(tlv.tagValue) + (1 + listed) * tlvObjectOverhead;
+	}
+
+	return octets;
+}

src/session-options.ts

@@ -128,6 +128,7 @@ export const defaults = {
 	heldMessageTimeout: 300_000,
 	maxDlrMerges: 1000,
 	maxHeldMessages: 1000,
+	maxHeldOctets: defaultMaxOctets,
 	maxOutstanding: 10,
 	maxReassembly: 1000,
 	reassemblyTimeout: 300_000,