Bound the messages held for a shutdown, and report a partial receipt

src/held-messages.ts

@@ -1,24 +1,68 @@
 import type { PduObject } from './pdu.ts';
+import type { SmppLog } from './log.ts';
+import { ExpiringGroups } from './expiring-groups.ts';
 import { IdleWaiters } from './idle-waiters.ts';
 
+export type HeldMessagesOptions = {
+	log: SmppLog;
+	max: number;
+	timeout: number;
+};
+
+/** The peer's own sequence number, which is what our answer to this message will carry. */
+function keyOf(pduObjs: PduObject[]): string | undefined {
+	const first = pduObjs[0];
+
+	return first ? String(first.seqNr) : undefined;
+}
+
 /** 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 held: ExpiringGroups<PduObject[]>;
 	private readonly idleWaiters = new IdleWaiters();
+	private readonly log: SmppLog;
 
+	constructor(options: HeldMessagesOptions) {
+		this.held = new ExpiringGroups({
+			max: options.max,
+			onSweep: () => { this.sweep(); },
+			timeout: options.timeout,
+		});
+		this.log = options.log;
+	}
+
+	/** An application that answers no message at all may not grow this without end. */
 	hold(pduObjs: PduObject[]): void {
-		this.held.add(pduObjs);
+		const key = keyOf(pduObjs);
+
+		if (key === undefined) return;
+
+		if (this.held.full) {
+			const evicted = this.held.takeOldest();
+
+			if (evicted) {
+				this.log.warn('heldMessages - dropping the message held longest', { seqNr: evicted[0] });
+			}
+		}
+
+		this.held.set(key, pduObjs);
 	}
 
 	/** Whether a drain is still waiting for this message to be answered. */
 	has(pduObjs: PduObject[]): boolean {
-		return this.held.has(pduObjs);
+		const key = keyOf(pduObjs);
+
+		return key !== undefined && this.held.get(key) === pduObjs;
 	}
 
 	release(pduObjs: PduObject[]): void {
-		if (!this.held.delete(pduObjs)) return;
+		const key = keyOf(pduObjs);
 
-		if (this.held.size === 0) this.idleWaiters.settle();
+		// Identity, not the key: a wrapped sequence number must not release someone else's message.
+		if (key === undefined || this.held.get(key) !== pduObjs) return;
+
+		this.held.delete(key);
+		this.settle();
 	}
 
 	/** Drops every message: their segments went with the link, so no answer of ours correlates now. */
@@ -28,7 +72,22 @@ export class HeldMessages {
 	}
 
 	/** Resolves 0 once every message has been answered, or with how many have not. */
-	idle(timeout: number, signal?: AbortSignal): Promise<number> {
+	idle(timeout: number, signal: AbortSignal | undefined): Promise<number> {
 		return this.idleWaiters.wait(() => this.held.size, timeout, signal);
 	}
+
+	private sweep(): void {
+		const expired = this.held.takeExpired();
+
+		if (expired.length === 0) return;
+
+		this.log.warn('heldMessages - messages the application never answered', {
+			messages: expired.length,
+		});
+		this.settle();
+	}
+
+	private settle(): void {
+		if (this.held.size === 0) this.idleWaiters.settle();
+	}
 }

src/incoming-requests.ts

@@ -31,7 +31,7 @@ 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 held: HeldMessages;
 	private readonly log: SmppLog;
 	private readonly onRequest: OnRequest | undefined;
 	private readonly reassembler: Reassembler;
@@ -42,6 +42,11 @@ export class IncomingRequests {
 
 	constructor(options: IncomingRequestsOptions) {
 		this.dlrMerger = options.dlrMerger;
+		this.held = new HeldMessages({
+			log: options.log,
+			max: defaults.maxHeldMessages,
+			timeout: defaults.heldMessageTimeout,
+		});
 		this.log = options.log;
 		this.onRequest = options.onRequest;
 		this.reassembler = new Reassembler({
@@ -96,7 +101,7 @@ export class IncomingRequests {
 	}
 
 	/** Waits out the messages the application still holds, and says how many it never answered. */
-	async drain(timeout: number, signal?: AbortSignal): Promise<VoidResult> {
+	async drain(timeout: number, signal: AbortSignal | undefined): Promise<VoidResult> {
 		const unanswered = await this.held.idle(timeout, signal);
 
 		if (unanswered === 0) return {};

src/index.ts

@@ -35,7 +35,7 @@ export { uuidv7 } from './uuid.ts';
 
 export type { BindType, ClientOptions } from './client.ts';
 export type { Dlr, Receipt } from './dlr.ts';
-export type { SendRespOptions, Sms, SmsInput } from './sms.ts';
+export type { SendDlrResult, SendRespOptions, Sms, SmsInput } from './sms.ts';
 export type { ConcatInfo } from './udh.ts';
 export type { Result, VoidResult } from './result.ts';
 export type { SmppLog } from './log.ts';

src/outgoing-requests.ts

@@ -24,6 +24,13 @@ function abortedBeforeSend(): Error {
 	return new Error('Aborted before the request was sent');
 }
 
+/** A response carries the request's sequence number, which only sendReturn() has. */
+function misuse(input: PduObjectInput): Error | undefined {
+	return input.cmdName.endsWith('_resp')
+		? new Error(`Use sendReturn() for responses, not send(): ${input.cmdName}`)
+		: undefined;
+}
+
 /** Everything this end asks of the peer: which link carries it, how many at once, and the answer. */
 export class OutgoingRequests {
 	private readonly gate: LinkGate;
@@ -67,6 +74,11 @@ export class OutgoingRequests {
 
 	/** Sends a request and resolves with the peer's response. */
 	request(input: PduObjectInput, options: SendOptions): Promise<Result<{ pduObj: PduObject }>> {
+		// Ahead of the drain, so a misuse is named as one rather than blamed on the shutdown.
+		const wrong = misuse(input);
+
+		if (wrong) return Promise.resolve({ err: wrong });
+
 		// 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') });
@@ -123,19 +135,15 @@ export class OutgoingRequests {
 
 		if (unfinished === 0) return {};
 
-		this.log.warn('session - shutting down with requests unfinished', { timeout, unfinished });
+		this.log.warn('outgoingRequests - shutting down with requests unfinished', { timeout, unfinished });
 
 		return { err: new Error(`Shut down with ${String(unfinished)} request(s) unfinished`) };
 	}
 
 	/** Why a request cannot go out at all, as opposed to not yet. */
 	private refuse(input: PduObjectInput, options: SendOptions): Error | undefined {
-		if (input.cmdName.endsWith('_resp')) {
-			return new Error(`Use sendReturn() for responses, not send(): ${input.cmdName}`);
-		}
-
 		// Before the gate and the window, or an aborted call waits for what it will never use.
-		return options.signal?.aborted === true ? abortedBeforeSend() : undefined;
+		return misuse(input) ?? (options.signal?.aborted === true ? abortedBeforeSend() : undefined);
 	}
 
 	private async attempt(input: PduObjectInput, options: SendOptions): Promise<Attempt> {

src/session-options.ts

@@ -101,8 +101,11 @@ export const undeclaredInterfaceVersion = 0x00;
 export const defaults = {
 	/** Receipts of a multipart message can be a working day apart, so the cap does the bounding. */
 	dlrMergeTimeout: 86_400_000,
+	/** The peer gave up on an unanswered message long before this; the bound is against growth. */
+	heldMessageTimeout: 300_000,
 	maxDelay: 30_000,
 	maxDlrMerges: 1000,
+	maxHeldMessages: 1000,
 	maxOutstanding: 10,
 	maxReassembly: 1000,
 	minDelay: 1000,

src/session.ts

@@ -313,14 +313,14 @@ export class Session extends EventEmitter<SessionEvents> {
 			return { err: new Error('The session closed before the drain finished') };
 		}
 
-		return messages.err ? messages : requests;
+		if (!messages.err) return requests;
+
+		if (!requests.err) return messages;
+
+		return { err: new Error(`${messages.err.message}; ${requests.err.message}`) };
 	}
 
-	/**
-	 * How long the drain waits for the application, which is the only thing that can end that wait.
-	 * Neither timeout may hand it "forever": both are answers about a peer, and a peer is not what
-	 * this half is waiting for.
-	 */
+	/** The application half's budget, which may never be "forever": nothing else ends that wait. */
 	private answering(timeout: number): number {
 		if (timeout > 0) return timeout;
 

src/sms.ts

@@ -3,11 +3,20 @@ import type { MessageState } from './defs/constants.ts';
 import type { PduObject, PduObjectInput, TlvInput } from './pdu.ts';
 import type { Result, VoidResult } from './result.ts';
 import type { Session } from './session.ts';
+import { UnansweredError } from './unanswered-error.ts';
 import { consts } from './defs/constants.ts';
 import { receiptCodes } from './dlr.ts';
 import { smppDate } from './message.ts';
 import { uuidv7 } from './uuid.ts';
 
+/** Both fields hold what the peer took, so a partial failure names what is already receipted. */
+export type SendDlrResult = {
+	err?: Error;
+	pduObjs: PduObject[];
+	/** Segments that went out unanswered. The peer may have taken them, so sending again may duplicate. */
+	unanswered: number;
+};
+
 export type SendRespOptions = {
 	/** The id the peer correlates a later delivery receipt by. Defaults to a generated UUID v7. */
 	smsId?: string;
@@ -25,7 +34,7 @@ export type Sms = {
 	message: string;
 	pduObjs: PduObject[];
 	/** Sends a delivery report back to the sender. Defaults to DELIVERED. */
-	sendDlr: (status?: MessageState) => Promise<Result<{ pduObjs: PduObject[] }>>;
+	sendDlr: (status?: MessageState) => Promise<SendDlrResult>;
 	/** Answers every segment. Part of the protocol, not optional. Defaults to ESME_ROK. */
 	sendResp: (options?: SendRespOptions) => Promise<VoidResult>;
 	session: Session;
@@ -132,9 +141,13 @@ async function sendDlr(
 	sms: Sms,
 	send: SmsHandlers['send'],
 	status: MessageState = 'DELIVERED',
-): Promise<Result<{ pduObjs: PduObject[] }>> {
+): Promise<SendDlrResult> {
 	if (!sms.session.bindAllows('deliver_sm')) {
-		return { err: new Error('A transmitter-bound session does not carry deliver_sm') };
+		return {
+			err: new Error('A transmitter-bound session does not carry deliver_sm'),
+			pduObjs: [],
+			unanswered: 0,
+		};
 	}
 
 	const total = sms.pduObjs.length;
@@ -154,12 +167,18 @@ async function sendDlr(
 		});
 	}));
 	const pduObjs: PduObject[] = [];
+	let failure: Error | undefined;
+	let unanswered = 0;
 
 	for (const one of sent) {
-		if (one.err) return { err: one.err };
+		if (!one.err) {
+			pduObjs.push(one.pduObj);
+		} else {
+			if (one.err instanceof UnansweredError) unanswered++;
 
-		pduObjs.push(one.pduObj);
+			failure ??= one.err;
+		}
 	}
 
-	return { pduObjs };
+	return failure ? { err: failure, pduObjs, unanswered } : { pduObjs, unanswered };
 }