Tear down at once on a peer unbind and stop reporting a reconnect after close

AGENTS.md

@@ -264,9 +264,12 @@ exactly 140.
   `unbind()` then waits `responseTimeout` for its own response, and sends that PDU through
   `request()` past both the window and the drain gate because it must go out either way. A stream
   the framer or the codec cannot read takes `end()` instead, and so does `close({ signal })` on an
-  aborted signal — nothing on a dead link can answer, and an abort means stop now, so draining
+  aborted signal and a peer's own `unbind` — nothing on a dead link can answer, an abort means stop
-  either would only hold a socket open for the timeout. `shutdownTimeout` stays a session option
+  now, and a peer that has declared itself finished will not answer what it still owes us, so
-  rather than a `close()` argument: `server()` builds sessions on the caller's behalf, so the option
+  draining any of the three would only hold a socket open for the timeout. `SmppServer.close()`
+  reports each session's unfinished drain through `serverError`, because its own result says
+  nothing but that the listener stopped. `shutdownTimeout` stays a session option rather than a
+  `close()` argument: `server()` builds sessions on the caller's behalf, so the option
   is the only composition point, and `close({ signal })` already covers a hard deadline.
 
 - **The TLS tests build their own self-signed certificate in DER** (`test/tls.test.ts`) instead of

README.md

@@ -309,13 +309,13 @@ TypeScript users can import `SmppLog` to have the compiler check one.
 
 ### Methods
 
-`sendSms()`, `send()`, `sendReturn()`, `unbind()` and `close()`. `close()` and `unbind()` both
+`sendSms()`, `send()`, `sendReturn()`, `unbind()` and `close()`. Both `close()` and `unbind()`
-refuse further sends, wait out the requests this end already sent — up to `shutdownTimeout`, or
+refuse further sends, wait out the requests this end already sent for up to `shutdownTimeout`, and
-until an `AbortSignal` given as `close({ signal })` says to stop now — and then tear down whatever
+then tear down whatever is left, resolving to an `err` that says what was lost. A request the peer
-is left, resolving to an `err` that says what was lost. A request the peer sent *us* is answered
+sent *us* is answered through `sendReturn()` and is not waited for. `close({ signal })` takes an
-through `sendReturn()` and is not waited for, and `unbind()` waits a further `responseTimeout` for
+`AbortSignal` that cuts the wait short; `unbind()` takes none, and waits a further
-its own response. `send()` reaches any of the 33 SMPP commands the codec knows, not just the four
+`responseTimeout` for its own response. `send()` reaches any of the 33 SMPP commands the codec
-the session handles natively:
+knows, not just the four the session handles natively:
 
 ```javascript
 const { err, pduObj } = await session.send({
@@ -353,8 +353,9 @@ The spec tables are exported both individually (`cmds`, `consts`, `encodings`, `
 ## Migrating from larvitsmpp 0.4.0
 
 - **The package is now `@larvit/smpp`** and is ESM only. `require()` no longer works.
-- **Callbacks are gone.** `client`, `server`, `sendSms`, `sendResp` and `sendDlr` are all promises
+- **Callbacks are gone.** `client`, `server`, `sendSms`, `sendResp`, `sendDlr`, `unbind` and
-  resolving to a result object with an optional `err`. Nothing rejects.
+  `session.close` are all promises resolving to a result object with an optional `err`. Nothing
+  rejects. Await `close()` or the socket outlives the call.
 - **`server()` resolves once, when it is listening**, and gives you a handle with `close()`, `port`
   and a `session` event. It no longer calls your callback once per incoming connection.
 - **The id a message is answered with goes to `sendResp({ smsId })`**, and `sms.smsId` is read-only:

src/incoming-requests.ts

@@ -70,7 +70,8 @@ export class IncomingRequests {
 				break;
 			case 'unbind':
 				await this.session.sendReturn(pduObj);
-				await this.session.close();
+				// A peer that has said it is finished will not answer what we still have outstanding.
+				await this.session.close({ signal: AbortSignal.abort() });
 				break;
 			default:
 				await this.unhandled(pduObj);

src/server.ts

@@ -118,7 +118,6 @@ export class SmppServer extends EventEmitter<ServerEvents> {
 
 	/** Stops listening, then drains and closes every session that was live when it stopped. */
 	async close(options: CloseOptions = {}): Promise<void> {
-		// Before the drain, or the listener keeps accepting connections nothing will ever close.
 		const stopped = new Promise<void>(resolve => { this.server.close(() => { resolve(); }); });
 		const live = [...this.sessions];
 

src/session.ts

@@ -273,6 +273,13 @@ export class Session extends EventEmitter<SessionEvents> {
 			return { err: bound.err };
 		}
 
+		// close() can land while the rebind is in flight, and it has nothing left to tear down.
+		if (this.reconnectLoop?.isStopped() === true) {
+			this.teardown();
+
+			return { err: new Error('Session closed while it was coming back up') };
+		}
+
 		this.resetTimers();
 		this.log.info('session - reconnected');
 		this.emit('reconnected');
@@ -327,10 +334,7 @@ export class Session extends EventEmitter<SessionEvents> {
 		return response;
 	}
 
-	/**
+	/** Stops new sends and waits out the ones already issued. */
-	 * Stops new sends and waits out the ones already issued, which only covers what this end sent:
-	 * a request the peer sent us is answered through sendReturn(), which never enters the window.
-	 */
 	private async drain(signal: AbortSignal | undefined): Promise<VoidResult> {
 		this.reconnectLoop?.stop();
 		this.draining = true;
@@ -340,8 +344,8 @@ export class Session extends EventEmitter<SessionEvents> {
 		const timeout = this.options.shutdownTimeout ?? defaults.shutdownTimeout;
 		const unfinished = await this.window.idle(timeout, signal);
 
-		// The window empties on a drop too: teardown() settles everything the link was carrying.
+		// The window empties on a teardown too, which settles everything the link was carrying.
-		if (this.isClosed()) return { err: new Error('The link dropped before the drain finished') };
+		if (this.isClosed()) return { err: new Error('The session closed before the drain finished') };
 
 		if (unfinished === 0) return {};
 

test/session-extras.test.ts

@@ -8,15 +8,16 @@ import type { MessageDlr } from '../src/session.ts';
 import type { PduObject, PduObjectInput } from '../src/pdu.ts';
 import type { Result } from '../src/result.ts';
 import type { SendSmsResult } from '../src/send-sms.ts';
-import type { Session } from '../src/session.ts';
 import type { SmppLog } from '../src/log.ts';
 import type { Sms } from '../src/sms.ts';
 import type { SmppServer } from '../src/server.ts';
 import { Reassembler, decodeSegments } from '../src/reassembly.ts';
+import { Session } from '../src/session.ts';
 import { DlrMerger } from '../src/dlr-merger.ts';
 import { client } from '../src/client.ts';
 import { consts } from '../src/defs/constants.ts';
 import { errors } from '../src/defs/errors.ts';
+import { objToPdu } from '../src/pdu.ts';
 import { server } from '../src/server.ts';
 import { silentLog } from '../src/log.ts';
 import { submitSms } from '../src/send-sms.ts';
@@ -44,6 +45,20 @@ function once<T>(register: (resolve: (value: T) => void) => void): Promise<T> {
 	});
 }
 
+function delay(ms: number): Promise<void> {
+	return new Promise(resolve => { setTimeout(resolve, ms); });
+}
+
+type Gate = { open: () => void; passed: Promise<true> };
+
+/** A promise the test opens by hand, guarded by once() against waiting on one it never does. */
+function gate(): Gate {
+	const opener: { open?: () => void } = {};
+	const passed = once<true>(resolve => { opener.open = () => { resolve(true); }; });
+
+	return { open: () => opener.open?.(), passed };
+}
+
 describe('merged delivery reports', () => {
 	// 0.4.0 allocated a longSmsDlrs store to do exactly this and then never used it.
 	test('reports once on a whole multipart message', async () => {
@@ -668,7 +683,7 @@ describe('graceful shutdown', () => {
 		const closed = await closing;
 
 		assert.ok(closed.err instanceof Error);
-		assert.match(closed.err.message, /dropped/);
+		assert.match(closed.err.message, /closed before the drain finished/);
 		assert.ok((await sent).err instanceof Error);
 
 		await smpp.close();
@@ -742,4 +757,82 @@ describe('graceful shutdown', () => {
 
 		assert.ok((await unanswered).err instanceof Error);
 	});
+
+	// Waiting on a peer that has just declared itself finished is dead time, unbounded at 0.
+	test('tears down at once when the peer unbinds rather than draining for it', async () => {
+		const smpp = await startServer({ shutdownTimeout: 0 });
+		const arrived = once<Session>(resolve => { smpp.on('session', resolve); });
+		const peer = net.connect({ port: smpp.port });
+
+		peer.resume();
+
+		const bound = await arrived;
+		const ended = once<true>(resolve => { bound.on('close', () => { resolve(true); }); });
+		const unanswered = bound.send({ cmdName: 'enquire_link' });
+		const { buffer } = objToPdu({ cmdName: 'unbind', seqNr: 1 });
+
+		assert.ok(buffer);
+		peer.write(buffer);
+
+		assert.ok(await ended);
+		assert.ok((await unanswered).err instanceof Error);
+
+		peer.destroy();
+		await smpp.close();
+	});
+
+	test('does not report a reconnect on a session closed while it was coming back up', async () => {
+		const listener = net.createServer(sock => { sock.resume(); });
+
+		await new Promise<void>(resolve => { listener.listen(0, resolve); });
+
+		const address = listener.address();
+		const port = typeof address === 'object' && address !== null ? address.port : 0;
+		const opened: net.Socket[] = [];
+		const open = (): Promise<Result<{ sock: net.Socket }>> => new Promise(resolve => {
+			const sock = net.connect({ port }, () => { resolve({ sock }); });
+
+			opened.push(sock);
+		});
+		const first = await open();
+
+		assert.ok(first.sock);
+
+		const rebinding = gate();
+		const release = gate();
+		const session = new Session({
+			reconnect: {
+				connect: open,
+				maxDelay: 20,
+				minDelay: 10,
+				onConnected: async () => {
+					rebinding.open();
+					await release.passed;
+
+					return {};
+				},
+			},
+			sock: first.sock,
+		});
+		const reported: string[] = [];
+
+		session.on('reconnected', () => reported.push('reconnected'));
+		first.sock.destroy();
+
+		assert.ok(await rebinding.passed);
+
+		const closing = session.close();
+
+		release.open();
+		await closing;
+		await delay(50);
+
+		assert.deepEqual(reported, []);
+
+		for (const sock of opened) {
+			sock.destroy();
+		}
+
+		await new Promise<void>(resolve => { listener.close(() => { resolve(); }); });
+	});
 });

todo.md

@@ -5,7 +5,7 @@ rules there constrain every item below.
 
 ## Status
 
-The rewrite is **feature complete and green**: 246 tests, lint and typecheck clean, verified on Node
+The rewrite is **feature complete and green**: 248 tests, lint and typecheck clean, verified on Node
 18, 20, 22 and 24. What is left is release work and a few things worth adding before or after 1.0.0.
 
 ```bash
@@ -124,7 +124,7 @@ session message is a change to every call site.
       loses every incomplete group, and a peer has no reason to resend a receipt it already had
       answered. Surviving one means exposing the merge state for the application to persist and hand
       back, which is a public-surface decision.
-- [ ] **`session.ts` is 465 lines.** The one seam left in it is a socket-to-PDU transport, which
+- [ ] **`session.ts` is 468 lines.** The one seam left in it is a socket-to-PDU transport, which
       would move the deliberately public `sock` field out of `Session` or turn it into a getter —
       a public-surface change, so it waits for a decision.
 - [ ] **Group the session's collaborators under `src/session/`.** Only `session.ts` imports