Close the stability findings and settle the sendResp, sendSms and unbind contracts

src/reassembly.ts

@@ -2,23 +2,60 @@ import type { ConcatInfo } from './udh.ts';
 import type { LogInt } from '@larvit/log';
 import type { ParamValue } from './defs/types.ts';
 import type { PduObject } from './pdu.ts';
+import type { Tlv } from './defs/tlvs.ts';
+import { ExpiringGroups } from './expiring-groups.ts';
 import { decodeMessage } from './message.ts';
 import { paramText } from './defs/types.ts';
 
 export type ReassemblerOptions = {
 	log: LogInt;
 	max: number;
+	maxOctets?: number | undefined;
 	/** Injected so expiry can be exercised without a wall clock. */
 	now?: (() => number) | undefined;
 	timeout: number;
 };
 
+const defaultMaxOctets = 64 * 1024 * 1024;
+
 type Group = {
-	deadline: number;
+	octets: number;
 	parts: Map<number, PduObject>;
 	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] = Buffer.isBuffer(tlv.tagValue)
+			? { ...tlv, tagValue: Buffer.from(tlv.tagValue) }
+			: tlv;
+	}
+
+	return { ...pduObj, params, tlvs };
+}
+
+function octetsOf(pduObj: PduObject): number {
+	let octets = 0;
+
+	for (const value of Object.values(pduObj.params)) {
+		if (Buffer.isBuffer(value)) octets += value.length;
+	}
+
+	for (const tlv of Object.values(pduObj.tlvs)) {
+		if (Buffer.isBuffer(tlv.tagValue)) octets += tlv.tagValue.length;
+	}
+
+	return octets;
+}
+
 function groupKey(pduObj: PduObject, reference: number): string {
 	return [
 		paramText(pduObj.params.source_addr),
@@ -52,18 +89,22 @@ export function decodeSegments(pduObjs: PduObject[]): string {
 
 /** Holds the segments of incomplete multipart messages until they are whole, capped and expiring. */
 export class Reassembler {
-	private readonly groups = new Map<string, Group>();
+	private readonly groups: ExpiringGroups<Group>;
 	private readonly log: LogInt;
 	private readonly max: number;
-	private readonly now: () => number;
-	private readonly timeout: number;
-	private sweeper: NodeJS.Timeout | undefined;
+	private readonly maxOctets: number;
+	private octets = 0;
 
 	constructor(options: ReassemblerOptions) {
+		this.groups = new ExpiringGroups<Group>({
+			max: options.max,
+			now: options.now,
+			onSweep: () => { this.sweep(); },
+			timeout: options.timeout,
+		});
 		this.log = options.log;
 		this.max = options.max;
-		this.now = options.now ?? Date.now;
-		this.timeout = options.timeout;
+		this.maxOctets = options.maxOctets ?? defaultMaxOctets;
 	}
 
 	get size(): number {
@@ -76,64 +117,68 @@ export class Reassembler {
 
 		const key = groupKey(pduObj, concat.reference);
 		const group = this.groups.get(key) ?? 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));
 
-		group.parts.set(concat.part, pduObj);
+		group.parts.set(concat.part, segment);
+		group.octets += delta;
+		this.octets += delta;
 
-		if (group.parts.size < group.total) return undefined;
+		if (group.parts.size < group.total) {
+			this.trim();
+
+			return undefined;
+		}
 
 		this.groups.delete(key);
-		this.idle();
+		this.octets -= group.octets;
 
 		return [...group.parts.entries()].sort(([a], [b]) => a - b).map(([, part]) => part);
 	}
 
 	clear(): void {
 		this.groups.clear();
-		this.idle();
+		this.octets = 0;
 	}
 
 	/** Drops every group past its deadline. Runs before each collect and on its own timer. */
 	sweep(): void {
-		const now = this.now();
-
-		for (const [key, group] of this.groups) {
-			if (group.deadline > now) continue;
-
+		for (const [key, group] of this.groups.takeExpired()) {
 			this.log.info('reassembler - incomplete message expired', { key, total: group.total });
-			this.groups.delete(key);
+			this.octets -= group.octets;
 		}
-
-		this.idle();
 	}
 
 	private open(key: string, total: number): Group {
-		if (this.groups.size >= this.max) this.dropOldest();
+		if (this.groups.full) this.dropOldest();
 
-		const group: Group = { deadline: this.now() + this.timeout, parts: new Map(), total };
+		const group: Group = { octets: 0, parts: new Map(), total };
 
 		this.groups.set(key, group);
 
-		if (!this.sweeper) {
-			this.sweeper = setInterval(() => { this.sweep(); }, this.timeout);
-			this.sweeper.unref();
-		}
-
 		return group;
 	}
 
-	private dropOldest(): void {
-		const oldest = this.groups.keys().next();
-
-		if (oldest.done) return;
-
-		this.log.warn('reassembler - buffer full, dropping the oldest message', { max: this.max });
-		this.groups.delete(oldest.value);
+	/** Drops the oldest groups until the retained payload is back under the octet cap. */
+	private trim(): void {
+		while (this.octets > this.maxOctets && this.groups.size > 0) {
+			this.dropOldest();
+		}
 	}
 
-	private idle(): void {
-		if (!this.sweeper || this.groups.size > 0) return;
+	private dropOldest(): void {
+		const oldest = this.groups.takeOldest();
 
-		clearInterval(this.sweeper);
-		this.sweeper = undefined;
+		if (!oldest) return;
+
+		const [, group] = oldest;
+
+		this.log.warn('reassembler - buffer full, dropping the oldest message', {
+			max: this.max,
+			maxOctets: this.maxOctets,
+			octets: this.octets,
+		});
+		this.octets -= group.octets;
 	}
 }