Say what a peer did with each receipt segment, and leave no drop unlogged

AGENTS.md

@@ -350,13 +350,18 @@ Grouped by what each one constrains.
   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
+  would slip past is no longer waiting for it. `shutdownTimeout: 0` does not carry over to this half:
-  half: waiting forever is safe for the peer, whose every request is bounded by `responseTimeout`
+  waiting forever is safe for the peer, whose every request is bounded by `responseTimeout` unless the
-  unless the caller set that to 0 as well,
+  caller set that to 0 as well, and unsafe for the application, which nothing bounds — `close()` is
-  and unsafe for the application, which nothing bounds — `close()` is what you reach for when the
+  what you reach for when the application is stuck, so it may not block on the application coming
-  application is stuck, so it may not block on the application coming unstuck. That half falls back
+  unstuck. That half falls back to `responseTimeout`, the same answer the link gate's hold already
-  to `responseTimeout`, the same answer the link gate's hold already takes — and to that option's
+  takes — and to that option's default where it is 0 as well, since neither option is an answer about
-  default where it is 0 as well, since neither option is an answer about the application.
+  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.
 
 - **A reconnect keeps the delivery-receipt merges; everything else the link held is dropped.**
   `onDeliverSm()` answers each receipt before the group it belongs to is complete, and `teardown()`
@@ -400,8 +405,9 @@ Grouped by what each one constrains.
   `unbind()` takes through `now()`. The gate is told what happened and never reads back into the
   session: a collaborator that has to ask does not own its decision, which is how the first cut ended
   up answering the same question two different ways at admit and at release. For the same reason the
-  retry in `pastDrain()` asks `gate.isUp()` rather than `linkDown()`, which also reads the socket — a condition that loops on
+  retry in `pastDrain()` asks `gate.isUp()` rather than `linkDown()`, which also reads the socket — a
-  something the gate does not gate on spins against a gate that admits it straight back.
+  condition that loops on something the gate does not gate on spins against a gate that admits it
+  straight back.
   `LinkGate.returning` is a copy of `retrying()` taken at teardown, and stays true only because
   nothing stops the reconnect loop without `emitClose()` following it: `drain()` and `end()` are the
   only callers of `stop()`. A third caller has to shut the gate itself.

README.md

@@ -351,7 +351,8 @@ const { err, pduObj } = await session.send({
 A send issued while the link is down waits for the reconnect instead of failing, and goes out once
 the new link is bound — up to `responseTimeout`, after which it gives up having sent nothing. A
 request already on the wire is the other case: the SMSC may have taken it and lost only the response,
-so it fails, and `sendSms()` counts it in `unanswered`, whether the link dropped under it, the peer
+so it fails, and `sendSms()` and `sms.sendDlr()` count it in `unanswered`, whether the link dropped
+under it, the peer
 never answered in time, or you aborted it after it went out. Neither applies with `reconnect: false`,
 where a drop ends the session and every send after it is refused.
 

src/held-messages.ts

@@ -21,6 +21,7 @@ export class HeldMessages {
 	private readonly held: ExpiringGroups<PduObject[]>;
 	private readonly idleWaiters = new IdleWaiters();
 	private readonly log: SmppLog;
+	private readonly max: number;
 
 	constructor(options: HeldMessagesOptions) {
 		this.held = new ExpiringGroups({
@@ -29,6 +30,7 @@ export class HeldMessages {
 			timeout: options.timeout,
 		});
 		this.log = options.log;
+		this.max = options.max;
 	}
 
 	/** An application that answers no message at all may not grow this without end. */
@@ -37,12 +39,10 @@ export class HeldMessages {
 
 		if (key === undefined) return;
 
-		if (this.held.full) {
+		if (this.held.get(key)) {
-			const evicted = this.held.takeOldest();
+			this.log.warn('heldMessages - replacing a message on a re-used sequence number', { seqNr: key });
-
+		} else if (this.held.full) {
-			if (evicted) {
+			this.dropOldest();
-				this.log.warn('heldMessages - dropping the message held longest', { seqNr: evicted[0] });
-			}
 		}
 
 		this.held.set(key, pduObjs);
@@ -76,6 +76,16 @@ export class HeldMessages {
 		return this.idleWaiters.wait(() => this.held.size, timeout, signal);
 	}
 
+	private dropOldest(): void {
+		const oldest = this.held.takeOldest();
+
+		if (!oldest) return;
+
+		const [seqNr] = oldest;
+
+		this.log.warn('heldMessages - buffer full, dropping the oldest message', { max: this.max, seqNr });
+	}
+
 	private sweep(): void {
 		const expired = this.held.takeExpired();
 

src/idle-waiters.ts

@@ -18,7 +18,7 @@ export class IdleWaiters {
 	 * Resolves 0 once nothing is left, or with what still is when the timeout or the signal cuts the
 	 * wait short. A timeout of 0 waits forever.
 	 */
-	wait(remaining: () => number, timeout: number, signal?: AbortSignal): Promise<number> {
+	wait(remaining: () => number, timeout: number, signal: AbortSignal | undefined): Promise<number> {
 		if (remaining() === 0) return Promise.resolve(0);
 
 		if (signal?.aborted === true) return Promise.resolve(remaining());

src/send-window.ts

@@ -43,7 +43,7 @@ export class SendWindow {
 	}
 
 	/** Resolves 0 once nothing is left on the wire, or with what still is. */
-	idle(timeout: number, signal?: AbortSignal): Promise<number> {
+	idle(timeout: number, signal: AbortSignal | undefined): Promise<number> {
 		return this.idleWaiters.wait(() => this.unfinished(), timeout, signal);
 	}
 }

src/sms.ts

@@ -137,6 +137,28 @@ function receiptTlvs(smsId: string, status: MessageState): Record<string, TlvInp
 	};
 }
 
+function collectReceipt(sent: Result<{ pduObj: PduObject }>[]): SendDlrResult {
+	const pduObjs: PduObject[] = [];
+	let failure: Error | undefined;
+	let unanswered = 0;
+
+	for (const one of sent) {
+		if (one.err) {
+			if (one.err instanceof UnansweredError) unanswered++;
+
+			failure ??= one.err;
+		} else if (one.pduObj.cmdStatus === 'ESME_ROK') {
+			pduObjs.push(one.pduObj);
+		} else {
+			const refusal = one.pduObj.cmdStatus ?? String(one.pduObj.cmdStatusId);
+
+			failure ??= new Error(`deliver_sm refused by the peer: ${refusal}`);
+		}
+	}
+
+	return failure ? { err: failure, pduObjs, unanswered } : { pduObjs, unanswered };
+}
+
 async function sendDlr(
 	sms: Sms,
 	send: SmsHandlers['send'],
@@ -166,19 +188,5 @@ async function sendDlr(
 			...(sms.session.acceptsOptionalParams() ? { tlvs: receiptTlvs(smsId, status) } : {}),
 		});
 	}));
-	const pduObjs: PduObject[] = [];
+	return collectReceipt(sent);
-	let failure: Error | undefined;
-	let unanswered = 0;
-
-	for (const one of sent) {
-		if (!one.err) {
-			pduObjs.push(one.pduObj);
-		} else {
-			if (one.err instanceof UnansweredError) unanswered++;
-
-			failure ??= one.err;
-		}
-	}
-
-	return failure ? { err: failure, pduObjs, unanswered } : { pduObjs, unanswered };
 }

test/session-extras.test.ts

@@ -13,6 +13,8 @@ import type { Sms } from '../src/sms.ts';
 import type { SmppServer } from '../src/server.ts';
 import type { TestContext } from 'node:test';
 import { HeldMessages } from '../src/held-messages.ts';
+import { UnansweredError } from '../src/unanswered-error.ts';
+import { createSms } from '../src/sms.ts';
 import { LinkGate } from '../src/link-gate.ts';
 import { Reassembler, decodeSegments } from '../src/reassembly.ts';
 import { Session } from '../src/session.ts';
@@ -860,8 +862,8 @@ describe('LinkGate', () => {
 
 // Goal 4: an application that answers nothing must not grow this for the life of the link.
 describe('held message bounds', () => {
-	function message(seqNr: number): PduObject[] {
+	function heldPdu(seqNr: number): PduObject {
-		return [{
+		return {
 			cmdId: 0x00000004,
 			cmdLength: 0,
 			cmdName: 'submit_sm',
@@ -870,7 +872,11 @@ describe('held message bounds', () => {
 			params: { destination_addr: '46709771337', short_message: 'held', source_addr: '46701113311' },
 			seqNr,
 			tlvs: {},
-		}];
+		};
+	}
+
+	function message(seqNr: number): PduObject[] {
+		return [heldPdu(seqNr)];
 	}
 
 	test('drops the message held longest rather than holding every one', async () => {
@@ -893,6 +899,37 @@ describe('held message bounds', () => {
 		assert.equal(await held.idle(1, undefined), 1);
 		assert.equal(await held.idle(1000, undefined), 0);
 	});
+
+	// A receipt cannot be resent wholesale without duplicating the segments that landed, so sendDlr()
+	// names what the peer took and what it may have, the way sendSms() does.
+	test('a partial receipt names the segments the peer took and the ones it may have', async t => {
+		const session = new Session({ sock: new net.Socket() });
+
+		closeAfter(t, session);
+
+		let call = 0;
+		const sms = createSms({
+			from: '46701113311',
+			message: 'three segments',
+			pduObjs: [heldPdu(1), heldPdu(2), heldPdu(3)],
+			session,
+			to: '46709771337',
+		}, {
+			onAnswered: () => undefined,
+			send: () => {
+				call++;
+
+				return Promise.resolve(call === 2
+					? { err: new UnansweredError(new Error('nothing came back')) }
+					: { pduObj: heldPdu(call) });
+			},
+		});
+		const report = await sms.sendDlr('DELIVERED');
+
+		assert.ok(report.err instanceof Error);
+		assert.equal(report.pduObjs.length, 2);
+		assert.equal(report.unanswered, 1);
+	});
 });
 
 describe('reassembly bounds', () => {
@@ -1112,6 +1149,21 @@ describe('graceful shutdown', () => {
 		assert.deepEqual(await closed, {});
 	});
 
+	// The drain refuses sends; a response was never a send, and saying so is the more useful answer.
+	test('names a response put through send() as the misuse it is, even mid-shutdown', async t => {
+		const { sent, session, sms } = await submitInFlight(t);
+		const closing = session.close();
+		const refused = await session.send({ cmdName: 'submit_sm_resp' });
+
+		assert.ok(refused.err instanceof Error);
+		assert.match(refused.err.message, /Use sendReturn\(\)/);
+
+		await sms.sendResp({ smsId: 'answered-after-the-misuse' });
+
+		assert.deepEqual((await sent).smsIds, ['answered-after-the-misuse']);
+		assert.deepEqual(await closing, {});
+	});
+
 	test('unbind() waits out a submit already on the wire before it unbinds', async t => {
 		const { sent, session, sms } = await submitInFlight(t);
 		const unbound = session.unbind();