Wait out the messages the application holds before shutting a session down

src/held-messages.ts

@@ -0,0 +1,29 @@
+import type { PduObject } from './pdu.ts';
+import { IdleWaiters } from './idle-waiters.ts';
+
+/** The messages handed to the application that it has not answered yet, held by their segments. */
+export class HeldMessages {
+	private readonly held = new Set<PduObject[]>();
+	private readonly idleWaiters = new IdleWaiters();
+
+	hold(pduObjs: PduObject[]): void {
+		this.held.add(pduObjs);
+	}
+
+	release(pduObjs: PduObject[]): void {
+		if (!this.held.delete(pduObjs)) return;
+
+		if (this.held.size === 0) this.idleWaiters.settle();
+	}
+
+	/** Drops every message: their segments went with the link, so no answer of ours correlates now. */
+	clear(): void {
+		this.held.clear();
+		this.idleWaiters.settle();
+	}
+
+	/** Resolves 0 once every message has been answered, or with how many have not. */
+	idle(timeout: number, signal?: AbortSignal): Promise<number> {
+		return this.idleWaiters.wait(() => this.held.size, timeout, signal);
+	}
+}

src/idle-waiters.ts

@@ -0,0 +1,47 @@
+/** What is left of a budget, in the shape a wait takes it: 0 waits forever. */
+export function leftOf(deadline: number): number {
+	return deadline === 0 ? 0 : Math.max(1, deadline - Date.now());
+}
+
+/** Everything waiting for a count to fall to zero, and how such a wait is cut short. */
+export class IdleWaiters {
+	private readonly waiting: (() => void)[] = [];
+
+	/** Wakes everything waiting, whatever the count reads now. */
+	settle(): void {
+		for (const resolve of this.waiting.splice(0)) {
+			resolve();
+		}
+	}
+
+	/**
+	 * 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> {
+		if (remaining() === 0) return Promise.resolve(0);
+
+		if (signal?.aborted === true) return Promise.resolve(remaining());
+
+		return new Promise<number>(resolve => {
+			let timer: NodeJS.Timeout | undefined = undefined;
+			const done = (): void => {
+				const index = this.waiting.indexOf(done);
+
+				if (timer) clearTimeout(timer);
+				if (index !== -1) this.waiting.splice(index, 1);
+
+				signal?.removeEventListener('abort', done);
+				resolve(remaining());
+			};
+
+			if (timeout > 0) {
+				timer = setTimeout(done, timeout);
+				timer.unref();
+			}
+
+			signal?.addEventListener('abort', done, { once: true });
+			this.waiting.push(done);
+		});
+	}
+}

src/incoming-requests.ts

@@ -1,9 +1,11 @@
 import type { DlrMerger } from './dlr-merger.ts';
 import type { OnRequest } from './session-options.ts';
-import type { PduObject } from './pdu.ts';
+import type { PduObject, PduObjectInput } from './pdu.ts';
+import type { Result, VoidResult } from './result.ts';
 import type { Session } from './session.ts';
 import type { SmppLog } from './log.ts';
 import type { SmsIdFormat } from './sms-id.ts';
+import { HeldMessages } from './held-messages.ts';
 import { Reassembler, decodeSegments } from './reassembly.ts';
 import { bindCommands, defaults } from './session-options.ts';
 import { concatInfo } from './udh.ts';
@@ -19,6 +21,8 @@ export type IncomingRequestsOptions = {
 	maxReassembly?: number | undefined;
 	onRequest?: OnRequest | undefined;
 	reassemblyTimeout?: number | undefined;
+	/** Past the drain gate: a receipt answering a message the shutdown is still waiting for. */
+	sendHeld: (input: PduObjectInput) => Promise<Result<{ pduObj: PduObject }>>;
 	session: Session;
 	smsIdFormat?: SmsIdFormat | undefined;
 	systemId?: string | undefined;
@@ -27,9 +31,11 @@ export type IncomingRequestsOptions = {
 /** Everything the peer asks of a session: messages, receipts, links and the answers to them. */
 export class IncomingRequests {
 	private readonly dlrMerger: DlrMerger;
+	private readonly held = new HeldMessages();
 	private readonly log: SmppLog;
 	private readonly onRequest: OnRequest | undefined;
 	private readonly reassembler: Reassembler;
+	private readonly sendHeld: IncomingRequestsOptions['sendHeld'];
 	private readonly session: Session;
 	private readonly smsIdFormat: SmsIdFormat;
 	private readonly systemId: string;
@@ -44,6 +50,7 @@ export class IncomingRequests {
 			maxOctets: options.maxOctets,
 			timeout: options.reassemblyTimeout ?? defaults.reassemblyTimeout,
 		});
+		this.sendHeld = options.sendHeld;
 		this.session = options.session;
 		this.smsIdFormat = options.smsIdFormat ?? {};
 		this.systemId = options.systemId ?? defaults.systemId;
@@ -82,11 +89,23 @@ export class IncomingRequests {
 		}
 	}
 
-	/** Drops the segments of every message that never became whole. */
+	/** Drops the segments of every message that never became whole, and of every one still held. */
 	clear(): void {
+		this.held.clear();
 		this.reassembler.clear();
 	}
 
+	/** Waits out the messages the application still holds, and says how many it never answered. */
+	async drain(timeout: number, signal?: AbortSignal): Promise<VoidResult> {
+		const unanswered = await this.held.idle(timeout, signal);
+
+		if (unanswered === 0) return {};
+
+		this.log.warn('session - shutting down with messages unanswered', { timeout, unanswered });
+
+		return { err: new Error(`Shut down with ${String(unanswered)} message(s) unanswered`) };
+	}
+
 	private async unhandled(pduObj: PduObject): Promise<void> {
 		if (bindCommands.includes(pduObj.cmdName)) {
 			this.log.info('session - bind on an already bound session', { cmdName: pduObj.cmdName });
@@ -139,12 +158,21 @@ export class IncomingRequests {
 
 		if (!first) return;
 
-		this.session.emit('sms', createSms({
+		const sms = createSms({
 			from: paramText(first.params.source_addr),
 			message: decodeSegments(pduObjs),
 			pduObjs,
 			session: this.session,
 			to: paramText(first.params.destination_addr),
-		}));
+		}, {
+			// A turn later, so a listener sending its receipt straight after the response still holds.
+			onAnswered: () => { setImmediate(() => { this.held.release(pduObjs); }); },
+			send: this.sendHeld,
+		});
+
+		this.held.hold(pduObjs);
+
+		// A message nobody took is not work a shutdown can wait for.
+		if (!this.session.emit('sms', sms)) this.held.release(pduObjs);
 	}
 }

src/send-sms.ts

@@ -37,6 +37,11 @@ export type SendSmsResult = {
 	unanswered: number;
 };
 
+/** A message that never went out, in the shape a caller aggregating segments still reads. */
+export function unsent(err: Error): SendSmsResult {
+	return { err, pduObjs: [], smsIds: [], unanswered: 0 };
+}
+
 /** What sending needs from the session: a concat reference and a way onto the wire. */
 export type SendSmsDeps = {
 	log: SmppLog;
@@ -141,7 +146,7 @@ export async function submitSms(deps: SendSmsDeps, sms: SendSmsOptions): Promise
 	const segments = splitMessage(sms.message, { encoding, reference: deps.reference });
 	const refused = checkSegments(allowed, segments.length);
 
-	if (refused) return { err: refused, pduObjs: [], smsIds: [], unanswered: 0 };
+	if (refused) return unsent(refused);
 
 	const multipart = segments.length > 1;
 

src/send-window.ts

@@ -1,8 +1,10 @@
+import { IdleWaiters } from './idle-waiters.ts';
+
 /** Caps how many requests are on the wire at once; anything past the limit waits its turn. */
 export class SendWindow {
+	private readonly idleWaiters = new IdleWaiters();
 	private readonly limit: number;
 	private readonly waiting: (() => void)[] = [];
-	private readonly waitingForIdle: (() => void)[] = [];
 	private inFlight = 0;
 
 	constructor(limit: number) {
@@ -32,9 +34,7 @@ export class SendWindow {
 
 		if (this.inFlight > 0) return;
 
-		for (const resolve of this.waitingForIdle.splice(0)) {
-			resolve();
-		}
+		this.idleWaiters.settle();
 	}
 
 	/** Everything the caller is still owed: on the wire, plus queued behind a full window. */
@@ -42,34 +42,8 @@ export class SendWindow {
 		return this.inFlight + this.waiting.length;
 	}
 
-	/**
-	 * 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.
-	 */
+	/** Resolves 0 once nothing is left on the wire, or with what still is. */
 	idle(timeout: number, signal?: AbortSignal): Promise<number> {
-		if (this.inFlight === 0) return Promise.resolve(0);
-
-		if (signal?.aborted === true) return Promise.resolve(this.unfinished());
-
-		return new Promise<number>(resolve => {
-			let timer: NodeJS.Timeout | undefined = undefined;
-			const done = (): void => {
-				const index = this.waitingForIdle.indexOf(done);
-
-				if (timer) clearTimeout(timer);
-				if (index !== -1) this.waitingForIdle.splice(index, 1);
-
-				signal?.removeEventListener('abort', done);
-				resolve(this.unfinished());
-			};
-
-			if (timeout > 0) {
-				timer = setTimeout(done, timeout);
-				timer.unref();
-			}
-
-			signal?.addEventListener('abort', done, { once: true });
-			this.waitingForIdle.push(done);
-		});
+		return this.idleWaiters.wait(() => this.unfinished(), timeout, signal);
 	}
 }

src/session.ts

@@ -16,12 +16,14 @@ import { PduTransport } from './pdu-transport.ts';
 import { PendingRequests, UnansweredError } from './pending-requests.ts';
 import { ReconnectLoop } from './reconnect-loop.ts';
 import { SendWindow } from './send-window.ts';
+import { leftOf } from './idle-waiters.ts';
 import { errorFrom } from './error-from.ts';
 import { optionalParamsMinVersion } from './defs/constants.ts';
 import { bindCarries, bindCommands, defaultSystemId, defaults } from './session-options.ts';
 import { isResp, objToPdu, pduReturn } from './pdu.ts';
 import { silentLog } from './log.ts';
-import { submitSms } from './send-sms.ts';
+import { submitSms, unsent } from './send-sms.ts';
+import { ConcatReference } from './udh.ts';
 
 export type {
 	CloseOptions,
@@ -64,6 +66,7 @@ export class Session extends EventEmitter<SessionEvents> {
 	peerInterfaceVersion: number | undefined = undefined;
 	userData: unknown = undefined;
 
+	private readonly concatReference = new ConcatReference();
 	private readonly dlrMerger: DlrMerger;
 	private readonly gate: LinkGate;
 	private readonly incoming: IncomingRequests;
@@ -75,7 +78,6 @@ export class Session extends EventEmitter<SessionEvents> {
 	private readonly window: SendWindow;
 
 	private closed = false;
-	private concatReference = 0;
 	private draining = false;
 	private ended = false;
 
@@ -129,6 +131,7 @@ export class Session extends EventEmitter<SessionEvents> {
 			maxReassembly: options.maxReassembly,
 			onRequest: options.onRequest,
 			reassemblyTimeout: options.reassemblyTimeout,
+			sendHeld: input => this.sendThrough(input, {}),
 			session: this,
 			smsIdFormat: options.smsIdFormat,
 			systemId: options.systemId,
@@ -166,7 +169,15 @@ export class Session extends EventEmitter<SessionEvents> {
 	}
 
 	/** Sends a request and resolves with the peer's response. */
-	async send(
+	send(input: PduObjectInput, options: SendOptions = {}): Promise<Result<{ pduObj: PduObject }>> {
+		// A drain on a live link. A link that is down is the gate's answer, which says closed instead.
+		if (this.draining && !this.linkDown()) return Promise.resolve({ err: new Error('Session is shutting down') });
+
+		return this.sendThrough(input, options);
+	}
+
+	/** The same path without that refusal, which a receipt for a held message has to take. */
+	private async sendThrough(
 		input: PduObjectInput,
 		options: SendOptions = {},
 	): Promise<Result<{ pduObj: PduObject }>> {
@@ -199,9 +210,6 @@ export class Session extends EventEmitter<SessionEvents> {
 			return new Error(`Use sendReturn() for responses, not send(): ${input.cmdName}`);
 		}
 
-		// A drain on a live link. A link that is down is the gate's answer, which says closed instead.
-		if (this.draining && !this.linkDown()) return new Error('Session is shutting down');
-
 		// Before the gate and the window, or an aborted call waits for what it will never use.
 		if (options.signal?.aborted === true) return abortedBeforeSend();
 
@@ -239,17 +247,12 @@ export class Session extends EventEmitter<SessionEvents> {
 
 	async sendSms(sms: SendSmsOptions, options: SendOptions = {}): Promise<SendSmsResult> {
 		if (!this.bindAllows('submit_sm')) {
-			return {
-				err: new Error('A receiver-bound session does not carry submit_sm'),
-				pduObjs: [],
-				smsIds: [],
-				unanswered: 0,
-			};
+			return unsent(new Error('A receiver-bound session does not carry submit_sm'));
 		}
 
 		const sent = await submitSms({
 			log: this.log,
-			reference: this.nextConcatReference(),
+			reference: this.concatReference.next(),
 			respIdNotation: this.options.smsIdFormat?.submitResp,
 			send: input => this.send(input, options),
 		}, sms);
@@ -379,7 +382,7 @@ export class Session extends EventEmitter<SessionEvents> {
 		return { result: answered.err ? { err: new UnansweredError(answered.err) } : answered, retryOnNextLink: false };
 	}
 
-	/** Stops new sends and waits out the ones already issued. */
+	/** Stops new sends and waits out the messages we hold and the requests already issued. */
 	private async drain(signal: AbortSignal | undefined): Promise<VoidResult> {
 		this.reconnectLoop?.stop();
 		this.draining = true;
@@ -387,11 +390,16 @@ export class Session extends EventEmitter<SessionEvents> {
 		if (this.linkDown()) return {};
 
 		const timeout = this.options.shutdownTimeout ?? defaults.shutdownTimeout;
-		const unfinished = await this.window.idle(timeout, signal);
+		const deadline = timeout > 0 ? Date.now() + timeout : 0;
+		// Answering a message can put a receipt on the wire; nothing on the wire produces a message.
+		const messages = await this.incoming.drain(timeout, signal);
+		const unfinished = await this.window.idle(leftOf(deadline), signal);
 
 		// The window empties on a teardown too, which settles everything the link was carrying.
 		if (this.linkDown()) return { err: new Error('The session closed before the drain finished') };
 
+		if (messages.err) return messages;
+
 		if (unfinished === 0) return {};
 
 		this.log.warn('session - shutting down with requests unfinished', { timeout, unfinished });
@@ -433,12 +441,6 @@ export class Session extends EventEmitter<SessionEvents> {
 		return this.reconnectLoop !== undefined && !this.reconnectLoop.isStopped();
 	}
 
-	private nextConcatReference(): number {
-		this.concatReference = this.concatReference >= 255 ? 1 : this.concatReference + 1;
-
-		return this.concatReference;
-	}
-
 	private onData(chunk: Buffer): void {
 		this.emit('data', chunk);
 		this.resetTimers();

src/sms.ts

@@ -1,6 +1,6 @@
 import type { ErrorName } from './defs/errors.ts';
 import type { MessageState } from './defs/constants.ts';
-import type { PduObject, TlvInput } from './pdu.ts';
+import type { PduObject, PduObjectInput, TlvInput } from './pdu.ts';
 import type { Result, VoidResult } from './result.ts';
 import type { Session } from './session.ts';
 import { consts } from './defs/constants.ts';
@@ -43,12 +43,18 @@ export type SmsInput = {
 	to: string;
 };
 
+/** What the session's incoming side gives a message so it can be answered and accounted for. */
+export type SmsHandlers = {
+	onAnswered: () => void;
+	send: (input: PduObjectInput) => Promise<Result<{ pduObj: PduObject }>>;
+};
+
 /** Each segment of a multipart message gets its own message_id, as a separate submit_sm must. */
 function segmentId(smsId: string, index: number, total: number): string {
 	return total === 1 ? smsId : `${smsId}-${String(index + 1)}`;
 }
 
-export function createSms(input: SmsInput): Sms {
+export function createSms(input: SmsInput, handlers: SmsHandlers): Sms {
 	const first = input.pduObjs[0];
 	const registered = first?.params.registered_delivery;
 	const dataCoding = first?.params.data_coding;
@@ -60,8 +66,8 @@ export function createSms(input: SmsInput): Sms {
 		from: input.from,
 		message: input.message,
 		pduObjs: input.pduObjs,
-		sendDlr: status => sendDlr(sms, status),
-		sendResp: options => sendResp(sms, answered, options ?? {}),
+		sendDlr: status => sendDlr(sms, handlers.send, status),
+		sendResp: options => sendResp(sms, answered, options ?? {}).finally(handlers.onAnswered),
 		session: input.session,
 		get smsId(): string {
 			return answered.smsId;
@@ -124,6 +130,7 @@ function receiptTlvs(smsId: string, status: MessageState): Record<string, TlvInp
 
 async function sendDlr(
 	sms: Sms,
+	send: SmsHandlers['send'],
 	status: MessageState = 'DELIVERED',
 ): Promise<Result<{ pduObjs: PduObject[] }>> {
 	if (!sms.session.bindAllows('deliver_sm')) {
@@ -135,7 +142,7 @@ async function sendDlr(
 
 	for (let index = 0; index < total; index++) {
 		const smsId = segmentId(sms.smsId, index, total);
-		const sent = await sms.session.send({
+		const sent = await send({
 			cmdName: 'deliver_sm',
 			params: {
 				destination_addr: sms.from,

src/udh.ts

@@ -1,3 +1,14 @@
+/** The 8-bit reference tying a long SMS's segments together, counted per session. */
+export class ConcatReference {
+	private current = 0;
+
+	next(): number {
+		this.current = this.current >= 255 ? 1 : this.current + 1;
+
+		return this.current;
+	}
+}
+
 export type ConcatInfo = {
 	part: number;
 	reference: number;