Drain the requests already on the wire before close() and unbind() tear down

src/client.ts

@@ -27,6 +27,7 @@ export type ClientOptions = {
 	port?: number;
 	reconnect?: { maxDelay?: number; minDelay?: number };
 	responseTimeout?: number;
+	shutdownTimeout?: number;
 	signal?: AbortSignal;
 	systemType?: string;
 	tls?: ConnectionOptions | boolean;
@@ -145,6 +146,7 @@ function createSession(options: ClientOptions, log: SmppLog, sock: Socket): Sess
 		log,
 		maxOutstanding: options.maxOutstanding,
 		responseTimeout: options.responseTimeout,
+		shutdownTimeout: options.shutdownTimeout,
 		sock,
 		...(options.reconnect
 			? {
@@ -194,18 +196,18 @@ export async function client(options: ClientOptions = {}): Promise<Result<{ sess
 	const signal = options.signal;
 
 	if (signal?.aborted === true) {
-		session.close();
+		void session.close();
 
 		return { err: new Error('Aborted before binding') };
 	}
 
 	// Registered before the bind: an abort landing while it is in flight has to close the session.
-	signal?.addEventListener('abort', () => { session.close(); }, { once: true });
+	signal?.addEventListener('abort', () => { void session.close(); }, { once: true });
 
 	const bound = await bind(session, options);
 
 	if (bound.err) {
-		session.close();
+		void session.close();
 
 		return { err: bound.err };
 	}

src/incoming-requests.ts

@@ -70,7 +70,7 @@ export class IncomingRequests {
 				break;
 			case 'unbind':
 				await this.session.sendReturn(pduObj);
-				this.session.close();
+				await this.session.close();
 				break;
 			default:
 				await this.unhandled(pduObj);

src/send-window.ts

@@ -2,6 +2,7 @@
 export class SendWindow {
 	private readonly limit: number;
 	private readonly waiting: (() => void)[] = [];
+	private readonly waitingForIdle: (() => void)[] = [];
 	private inFlight = 0;
 
 	constructor(limit: number) {
@@ -28,5 +29,35 @@ export class SendWindow {
 		}
 
 		this.inFlight--;
+
+		if (this.inFlight > 0) return;
+
+		for (const resolve of this.waitingForIdle.splice(0)) {
+			resolve();
+		}
+	}
+
+	/** Resolves once nothing is in flight, or on the timeout with how many still are. 0 never times out. */
+	idle(timeout: number): Promise<number> {
+		if (this.inFlight === 0) return Promise.resolve(0);
+
+		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);
+
+				resolve(this.inFlight);
+			};
+
+			if (timeout > 0) {
+				timer = setTimeout(done, timeout);
+				timer.unref();
+			}
+
+			this.waitingForIdle.push(done);
+		});
 	}
 }

src/server.ts

@@ -34,6 +34,7 @@ export type ServerOptions = {
 	port?: number;
 	reassemblyTimeout?: number;
 	responseTimeout?: number;
+	shutdownTimeout?: number;
 	signal?: AbortSignal;
 	systemId?: string;
 	tls?: TlsOptions | boolean;
@@ -114,20 +115,19 @@ export class SmppServer extends EventEmitter<ServerEvents> {
 		if (event !== 'serverError') this.emit('serverError', error);
 	}
 
-	/** Stops listening and closes every live session. */
-	close(): Promise<void> {
-		return new Promise(resolve => {
-			for (const session of this.sessions) {
-				try {
-					session.close();
-				} catch (thrown: unknown) {
-					this.emit('serverError', errorFrom(thrown));
-				}
-			}
+	/** Stops listening and closes every live session, draining each one first. */
+	async close(): Promise<void> {
+		const live = [...this.sessions];
 
-			this.sessions.clear();
-			this.server.close(() => { resolve(); });
-		});
+		this.sessions.clear();
+
+		await Promise.all(live.map(async session => {
+			const closed = await session.close().catch((thrown: unknown) => ({ err: errorFrom(thrown) }));
+
+			if (closed.err) this.emit('serverError', closed.err);
+		}));
+
+		return new Promise(resolve => { this.server.close(() => { resolve(); }); });
 	}
 }
 
@@ -233,6 +233,7 @@ function onConnection(sock: Socket, options: ServerOptions, server: SmppServer):
 		onRequest: (bound, pduObj) => onRequest(bound, pduObj, options),
 		reassemblyTimeout: options.reassemblyTimeout,
 		responseTimeout: options.responseTimeout,
+		shutdownTimeout: options.shutdownTimeout,
 		sock,
 		systemId: options.systemId ?? defaults.systemId,
 	});

src/session-options.ts

@@ -78,6 +78,8 @@ export type SessionOptions = {
 	reassemblyTimeout?: number | undefined;
 	reconnect?: ReconnectOptions | undefined;
 	responseTimeout?: number | undefined;
+	/** How long a drain waits for the requests already on the wire. 0 waits forever. */
+	shutdownTimeout?: number | undefined;
 	sock: Socket;
 	/** This end's own identity, answered to the peer in place of the one it sent. */
 	systemId?: string | undefined;
@@ -98,6 +100,7 @@ export const defaults = {
 	minDelay: 1000,
 	reassemblyTimeout: 300_000,
 	responseTimeout: 30_000,
+	shutdownTimeout: 5000,
 	systemId: defaultSystemId,
 };
 
@@ -112,6 +115,7 @@ export function checkSessionOptions(options: SessionCounts): VoidResult {
 		['maxReassembly', options.maxReassembly ?? defaults.maxReassembly, 1],
 		['reassemblyTimeout', options.reassemblyTimeout ?? defaults.reassemblyTimeout, 0],
 		['responseTimeout', options.responseTimeout ?? defaults.responseTimeout, 0],
+		['shutdownTimeout', options.shutdownTimeout ?? defaults.shutdownTimeout, 0],
 	];
 
 	for (const [name, value, min] of limits) {
@@ -129,4 +133,5 @@ export type SessionCounts = {
 	maxReassembly?: number | undefined;
 	reassemblyTimeout?: number | undefined;
 	responseTimeout?: number | undefined;
+	shutdownTimeout?: number | undefined;
 };

src/session.ts

@@ -67,6 +67,7 @@ export class Session extends EventEmitter<SessionEvents> {
 
 	private closed = false;
 	private concatReference = 0;
+	private draining = false;
 	private framer = new PduFramer();
 
 	/** A listener that throws is the application's bug; it must not become ours. Hard rule 1. */
@@ -160,6 +161,8 @@ export class Session extends EventEmitter<SessionEvents> {
 
 		if (this.closed) return { err: new Error('Session is closed') };
 
+		if (this.draining) return { err: new Error('Session is shutting down') };
+
 		// Before the window, or a full window makes an aborted call wait for a slot it will not use.
 		if (options.signal?.aborted === true) {
 			return { err: new Error('Aborted before the request was sent') };
@@ -213,22 +216,38 @@ export class Session extends EventEmitter<SessionEvents> {
 		return sent;
 	}
 
-	/** Unbinds politely, then closes. Many SMSCs drop the link instead of answering, which is fine. */
+	/**
+	 * Drains, unbinds politely, then closes. Many SMSCs drop the link instead of answering the
+	 * unbind, which is fine. Reports the unbind's own failure ahead of an unfinished drain.
+	 */
 	async unbind(): Promise<VoidResult> {
+		this.reconnectLoop?.stop();
+
+		const drained = await this.drain();
 		const wasOpen = !this.closed;
-		const sent = await this.send({ cmdName: 'unbind' });
+		// request(), not send(): the drain gate would refuse it, and the window is empty by now.
+		const sent = wasOpen
+			? await this.request({ cmdName: 'unbind' }, {})
+			: { err: new Error('Session is closed') };
 		const closedOnUnbind = wasOpen && this.closed;
 
-		this.close();
+		this.shutdown();
 
-		return sent.err && !closedOnUnbind ? { err: sent.err } : {};
+		return sent.err && !closedOnUnbind ? { err: sent.err } : drained;
 	}
 
-	/** Closes for good. A session closed this way never reconnects. */
-	close(): void {
+	/**
+	 * Closes for good: refuses new sends, waits out the requests already on the wire up to
+	 * `shutdownTimeout`, then tears down whatever is left. A session closed this way never reconnects.
+	 */
+	async close(): Promise<VoidResult> {
 		this.reconnectLoop?.stop();
-		this.teardown();
-		this.dlrMerger.clear();
+
+		const drained = await this.drain();
+
+		this.shutdown();
+
+		return drained;
 	}
 
 	private loopFor(reconnect: ReconnectOptions | undefined): ReconnectLoop | undefined {
@@ -311,6 +330,34 @@ export class Session extends EventEmitter<SessionEvents> {
 		return response;
 	}
 
+	/** Stops new sends and waits out the ones already issued. A dead link has nothing to wait for. */
+	private async drain(): Promise<VoidResult> {
+		this.draining = true;
+		this.timers.clear();
+
+		if (this.closed || this.sock.destroyed) return {};
+
+		const timeout = this.options.shutdownTimeout ?? defaults.shutdownTimeout;
+		const inFlight = await this.window.idle(timeout);
+
+		if (inFlight === 0) return {};
+
+		this.log.warn('session - shutting down with requests still in flight', { inFlight, timeout });
+
+		return { err: new Error(`Shut down with ${String(inFlight)} request(s) still in flight`) };
+	}
+
+	private shutdown(): void {
+		this.teardown();
+		this.dlrMerger.clear();
+	}
+
+	/** The link is unusable, so nothing can answer and there is nothing to drain. */
+	private abort(): void {
+		this.reconnectLoop?.stop();
+		this.shutdown();
+	}
+
 	private teardown(): void {
 		if (this.closed) return;
 
@@ -348,7 +395,7 @@ export class Session extends EventEmitter<SessionEvents> {
 		if (framed.err) {
 			this.log.warn('session - unusable stream, closing', { message: framed.err.message });
 			this.emit('sessionError', framed.err);
-			this.close();
+			this.abort();
 
 			return;
 		}
@@ -369,7 +416,7 @@ export class Session extends EventEmitter<SessionEvents> {
 				message: parsed.err.message,
 			});
 			this.emit('sessionError', parsed.err);
-			this.close();
+			this.abort();
 
 			return false;
 		}