AGENTS.md

@@ -70,6 +70,7 @@ src/
 	reassembly.ts        Reassembler: capped, expiring multipart groups
 	reconnect-loop.ts    ReconnectLoop: backoff, retry timer, stopped-ness
 	result.ts            Result<T> — the shape every fallible call returns
+	retained-pdu.ts      A PDU copied off the wire so holding it pins nothing else, and what holding it costs
 	send-sms.ts          submitSms composition and the submitSmParams builder
 	send-window.ts       SendWindow: the maxOutstanding semaphore
 	session-options.ts   SessionOptions, ReconnectOptions, bind direction and the session defaults
@@ -300,6 +301,8 @@ the file.
   request that handling did not answer.
 - The drain waits on the messages the application holds, and `sendResp()` is what says it is done
   with one.
+- The drain's wait on the application ignores `shutdownTimeout: 0`.
+- What the application holds unanswered is capped on constants.
 - A reconnect keeps the delivery-receipt merges; everything else the link held is dropped.
 - A message id base is merged at most once.
 - A send that never reached the socket waits for the next link; one that did is counted, not resent.

CHANGELOG.md

@@ -40,6 +40,9 @@
   count as next to nothing, so a peer could hold far more than the cap. **Raise a `maxOctets` you
   tuned low**: it now holds several times fewer segments, and an incomplete message evicted over
   the cap is lost, since its segments were already answered.
+- Unanswered `sms` messages are capped at 64 MiB per session by the `maxOctets` charge, beside the
+  1000-message cap: the oldest is dropped with a warning, as over the count. The library used to
+  hold up to 1000 messages of any size for an application that answered none of them.
 - `server()` refuses a `maxOctets` below 1 or not a whole number, `Infinity` included, like its
   other limits. `server({ maxOctets: 0 })` used to start and then refuse every multipart message.
 - `callback_num`, `callback_num_atag`, `callback_num_pres_ind`, `broadcast_area_identifier` and

README.md

@@ -383,8 +383,9 @@ holds for `session.send()`.
    message has failed. Answering through `sendReturn()` instead leaves the wait running.
 3. Tear down what is left, resolving to an `err` that says what was lost.
 
-At most 1000 unanswered messages are held, for five minutes each; what falls out of either bound is
-dropped with a warning on the log and waited for no longer. Neither bound is an option.
+At most 1000 unanswered messages, and 64 MiB of them by the `maxOctets` charge, are held for five
+minutes each; what falls out of a bound is dropped with a warning on the log and waited for no
+longer. None of the bounds is an option.
 `close({ signal })` cuts the wait short. `unbind()` takes no signal, and waits a further
 `responseTimeout` for its own response.
 

docs/decisions.md

@@ -664,32 +664,25 @@ rule and an index of the titles below.
   down while the application was still answering a `submit_sm`, so the peer timed out and re-sent —
   the duplicate goal 2 forbids, in the direction the window already covers. No completion signal was
   added to the `sms` event: `sendResp()` is what an application already calls when it is done with a
-  message, so it is the one the drain waits for. Counting every inbound request until `sendReturn()` answered it was rejected —
-  an `onRequest` that deliberately answers nothing would then cost a full `shutdownTimeout` on every
-  close — and a message no listener took is released at once, since nothing is going to answer it.
-  A listener that failed before answering gives it up the same way, but only once every listener has:
-  a throw stops `emit()` where it stands, while a rejection leaves the others running, so the release
-  waits for the last of them rather than answering on their behalf. What ends the wait is the response
-  reaching the wire, not the call — a `sendResp()` the library refused, or one the socket would not
-  carry, leaves the message held, so `close()` still reports the one the peer is owed. Where the
-  segments were answered as they arrived there is no response left to write, so the call itself ends
-  the wait, an argument the library refuses excepted. `teardown()`
-  drops what is still held for the same reason it drops inbound segments. The release is one turn
-  late, so a listener that sends its receipt straight after the response is still holding when the
-  drain looks; `sendDlr()` is the one send that goes out past the drain's refusal, and only while the
-  message is still held — past that it is an ordinary send, because the drain it would slip past is
-  no longer waiting for it. `shutdownTimeout: 0` does not carry over to this half:
-  waiting forever is safe for the peer, whose every request is bounded by `responseTimeout` unless the
-  caller set that to 0 as well, and unsafe for the application, which nothing bounds — `close()` is
-  what you reach for when the application is stuck, so it may not block on the application coming
-  unstuck. That half falls back to `responseTimeout`, the same answer the link gate's hold already
-  takes — and to that option's default where it is 0 as well, since neither option is an answer about
-  the application. What is held is capped and expiring like every other inbound store, on constants
-  rather than options, because a bound the application cannot raise is the point: an application that
-  answers nothing would otherwise grow it for the life of the link, which goal 4 forbids. A message
-  that falls out of the bound is one the drain stops waiting for, so `close()` can report fewer
-  unanswered than there were — accepted, because the alternative is holding what nothing will answer,
-  and both exits are logged.
+  message, so it is the one the drain waits for. Counting every inbound request until `sendReturn()`
+  answered it was rejected: an `onRequest` that deliberately answers nothing would then cost a full
+  `shutdownTimeout` on every close. The response reaching the wire ends the wait, so a `sendResp()`
+  the library refused or the socket would not carry leaves `close()` still reporting the message the
+  peer is owed.
+
+- **The drain's wait on the application ignores `shutdownTimeout: 0`.** Waiting forever is safe for
+  the peer, whose every request is bounded by `responseTimeout` unless the caller set that to 0 as
+  well, and unsafe for the application, which nothing bounds — `close()` is what you reach for when
+  the application is stuck, so it may not block on the application coming unstuck. That half falls
+  back to `responseTimeout`, the same answer the link gate's hold already takes — and to that
+  option's default where it is 0 as well, since neither option is an answer about the application.
+
+- **What the application holds unanswered is capped on constants.** A bound the application cannot
+  raise is the point: an application that answers nothing would otherwise grow it for the life of
+  the link, which goal 4 forbids. Reassembly's `maxOctets` is an option because it bounds what the
+  peer sends; this bounds what the application leaves unanswered. A message that falls out of a bound
+  is one the drain stops waiting for, so `close()` can report fewer unanswered than there were —
+  accepted, because the alternative is holding what nothing will answer.
 
 - **A reconnect keeps the delivery-receipt merges; everything else the link held is dropped.**
   `onDelivery()` answers each receipt before the group it belongs to is complete, and `teardown()`

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: 64 * 1024 * 1024,
 	maxOutstanding: 10,
 	maxReassembly: 1000,
 	reassemblyTimeout: 300_000,

test/session-extras.test.ts

@@ -1483,7 +1483,7 @@ describe('held message bounds', () => {
 	}
 
 	test('drops the message held longest rather than holding every one', () => {
-		const held = new HeldMessages({ log: silentLog, max: 2, timeout: 10_000 });
+		const held = new HeldMessages({ log: silentLog, max: 2, maxOctets: 1_000_000, timeout: 10_000 });
 		const oldest = message(1);
 
 		held.hold(oldest);
@@ -1501,9 +1501,88 @@ describe('held message bounds', () => {
 		held.clear();
 	});
 
+	// submitPdu() holds 1026 octets by the maxOctets charge: its object, and the three text fields.
+	test('drops the message held longest once the octets held pass the cap', () => {
+		let now = 0;
+		const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 2100, now: () => now, timeout: 10_000 });
+		const oldest = message(1);
+
+		held.hold(oldest);
+		held.hold(message(2));
+
+		assert.equal(held.size, 2);
+
+		held.hold(message(3));
+
+		assert.equal(held.size, 2);
+		assert.equal(held.has(oldest), false);
+
+		// A message that leaves any other way gives its octets back, so two still fit afterwards.
+		const answered = message(4);
+
+		held.hold(answered);
+		held.release(answered);
+		held.hold(message(5));
+		assert.equal(held.size, 2, 'after a release');
+
+		now = 20_000;
+		held.sweep();
+		now = 0;
+		held.hold(message(6));
+		held.hold(message(7));
+		assert.equal(held.size, 2, 'after a sweep');
+
+		held.clear();
+		held.hold(message(8));
+		held.hold(message(9));
+		assert.equal(held.size, 2, 'after a clear');
+
+		held.clear();
+	});
+
+	// Dropping it would leave the drain blind to a message the peer is still owed an answer for.
+	test('keeps a message larger than the cap on its own', () => {
+		const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 1000, timeout: 10_000 });
+		const large = message(2);
+
+		held.hold(message(1));
+		held.hold(large);
+
+		assert.equal(held.size, 1);
+		assert.equal(held.has(large), true);
+
+		held.clear();
+	});
+
+	test('holds a message detached from the chunk it was read from', async t => {
+		const session = new Session({ sock: new net.Socket() });
+
+		closeAfter(t, session);
+		session.boundAs = 'transceiver';
+
+		const incoming = new IncomingRequests({
+			dlrMerger: new DlrMerger({ log: silentLog, max: 10, timeout: 10_000 }),
+			log: silentLog,
+			sendPastDrain: () => Promise.resolve({ err: new Error('never sent') }),
+			session,
+		});
+		const chunk = Buffer.alloc(64 * 1024);
+		const carried = submitPdu(1);
+		let received: Sms | undefined;
+
+		session.on('sms', sms => { received = sms; });
+		await incoming.handle({ ...carried, params: { ...carried.params, short_message: chunk.subarray(16, 20) } });
+
+		const retained = received?.pduObjs[0]?.params.short_message;
+
+		assert.ok(Buffer.isBuffer(retained));
+		assert.notEqual(retained.buffer, chunk.buffer);
+		incoming.clear();
+	});
+
 	test('gives up on a message the application never answers', () => {
 		let now = 0;
-		const held = new HeldMessages({ log: silentLog, max: 10, now: () => now, timeout: 60 });
+		const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 1_000_000, now: () => now, timeout: 60 });
 
 		held.hold(message(1));
 		now = 61;
@@ -1519,7 +1598,7 @@ describe('held message bounds', () => {
 	// Without this the drain sits out its whole budget before returning what a sweep already settled.
 	test('wakes a waiting drain when the last message expires', async () => {
 		let now = 0;
-		const held = new HeldMessages({ log: silentLog, max: 10, now: () => now, timeout: 60 });
+		const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 1_000_000, now: () => now, timeout: 60 });
 
 		held.hold(message(1));
 

todo.md

@@ -8,11 +8,14 @@ govern it, and nothing here is a source anything else may cite.
 
 ## Security
 
-- [ ] **Bound the heap `HeldMessages` keeps, not only its count.** It holds up to 1000 messages the
-      application has not answered for up to 300 s, each as the PDUs it arrived in, undetached and
-      uncharged: one 200 KB PDU of 50,000 empty unknown TLVs is 15 MB of heap, and a completed
-      reassembly is up to `maxOctets`. A peer faster than an application answering asynchronously
-      holds gigabytes per session. From the stability review of #27.
+- [ ] **Bound what a peer can make the application hold, not only the library.** An `sms` the
+      application is still answering pins its PDUs through `Sms.pduObjs` and its `sendResp`/`sendDlr`
+      closures, so the held-message caps free nothing while it works, and nothing slows the peer: a
+      peer faster than an application answering asynchronously (a DB write per message) grows the
+      heap without bound. Each eviction by the caps also stops the drain waiting for a message still
+      being answered, so `close()` can cut it off and the peer re-sends it. Flow control is a wire
+      change under goal 4 and the maintainer's call: answer `ESME_RTHROTTLED`, or stop reading the
+      socket, once the held count or octets are at the cap. From the stability review of #28.
 
 ## Status