From bebd74b42b5b5e35d5b0e94083b9fe1f6c0f6b7b Mon Sep 17 00:00:00 2001 From: Lilleman auf Larv Date: Sun, 30 Aug 2026 20:08:36 +0200 Subject: [PATCH] Drain the requests already on the wire before close() and unbind() tear down --- AGENTS.md | 9 +++++ README.md | 13 ++++-- src/client.ts | 8 ++-- src/incoming-requests.ts | 2 +- src/send-window.ts | 31 ++++++++++++++ src/server.ts | 27 +++++++------ src/session-options.ts | 5 +++ src/session.ts | 67 +++++++++++++++++++++++++----- test/interop.test.ts | 2 +- test/readme.test.ts | 6 +-- test/session-extras.test.ts | 81 ++++++++++++++++++++++++++++++++----- test/session.test.ts | 70 +++++++++++++++++--------------- todo.md | 12 ++---- 13 files changed, 248 insertions(+), 85 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 98fd71f..9bcc2e3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -252,6 +252,15 @@ exactly 140. one's. Every segment still reaches the application as a `dlr`. The rule covers the bases the merger opened — `expect()` ignores a lone id, so a single-part message never claims one. +- **A deliberate shutdown drains; an unusable link does not.** `close()` and `unbind()` refuse + further sends and wait for the send window to empty, not the pending map — the map misses a + segment still queued behind a full window, and finishing a half-sent multipart message is the + point. What `shutdownTimeout` leaves is torn down and reported as an `err`, and `0` waits forever + like every other timeout here. A stream the framer or the codec cannot read takes `abort()` + instead: nothing on that link can answer, so draining it would only hold a dead socket open for + the timeout. `unbind()` sends its own PDU past the window, since the drain already waited for + every slot. + - **The TLS tests build their own self-signed certificate in DER** (`test/tls.test.ts`) instead of adding a devDependency or shelling out to openssl. Maintainer's call, 2026-08-26: the dev image `node:24.18.0-bookworm-slim` ships no openssl binary, so a shelled-out fixture would pass in CI diff --git a/README.md b/README.md index 0329d0c..f854117 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,7 @@ by hand on top of a library; it is built in here. | **Submit window** | `maxOutstanding` holds requests in flight at 10; further sends queue instead of overrunning the SMSC. | | **Delivery receipts** | Correlated by `receipted_message_id`/`message_state` where the SMSC sends them, falling back to parsing the receipt text — what Kannel and several others send. | | **Multipart** | Long messages split on send; concatenated `deliver_sm` reassembled into one `sms`. | +| **Graceful shutdown** | `close()` and `unbind()` wait out the requests already on the wire, so a submit the SMSC accepted is not reported as a failure. | | **Never throws** | Everything fallible resolves to `{ err?, … }`, the codec included. | Throughput throttling is deliberately absent: an operator's rate limit is scoped to the account, and @@ -100,6 +101,7 @@ Every one is optional. | `enquireLinkInterval` | `20000` | How often to send `enquire_link` on a quiet link. | | `idleTimeout` | `2 × enquireLinkInterval` | Give up on a link the peer has stopped answering; with `reconnect` set, it re-binds. | | `responseTimeout` | `30000` | How long to wait for a response before giving up on it; `0` waits forever. | +| `shutdownTimeout` | `5000` | How long `close()` and `unbind()` wait for the requests already on the wire; `0` waits forever. | | `maxOutstanding` | `10` | Requests allowed on the wire at once; further sends queue. | | `reconnect` | off | `{ minDelay, maxDelay }` to re-bind automatically after a drop or an idle timeout, with exponential backoff. | | `log` | silent | Any object with `debug`, `error`, `info`, `verbose` and `warn` methods — see [Logging](#logging). | @@ -206,7 +208,7 @@ smpp.on('session', session => { }); console.log(smpp.port); // the port actually bound, useful when 0 was requested -await smpp.close(); // stop listening and close every live session +await smpp.close(); // drain and close every live session, then stop listening ``` `sendDlr` accepts `SCHEDULED`, `ENROUTE`, `DELIVERED`, `EXPIRED`, `DELETED`, `UNDELIVERABLE`, @@ -228,7 +230,7 @@ A message whose `data_coding` says 8-bit binary arrives as Latin-1, so `Buffer.f | `maxReassembly` | `1000` | Incomplete multipart messages held per session. | | `maxOctets` | `67108864` | Bytes of incomplete multipart messages held per session. | | `reassemblyTimeout` | `300000` | How long a late segment can still join an incomplete message. | -| `responseTimeout`, `maxOutstanding`, `log`, `signal` | as for the client | | +| `responseTimeout`, `shutdownTimeout`, `maxOutstanding`, `log`, `signal` | as for the client | | ### Bind direction @@ -307,8 +309,11 @@ TypeScript users can import `SmppLog` to have the compiler check one. ### Methods -`sendSms()`, `send()`, `sendReturn()`, `unbind()` and `close()`. `send()` reaches any of the 33 SMPP -commands the codec knows, not just the four the session handles natively: +`sendSms()`, `send()`, `sendReturn()`, `unbind()` and `close()`. `close()` and `unbind()` both +refuse further sends, wait out the requests already on the wire up to `shutdownTimeout`, and then +tear down whatever is left; each resolves to an `err` naming how many requests it gave up on. +`send()` reaches any of the 33 SMPP commands the codec knows, not just the four the session handles +natively: ```javascript const { err, pduObj } = await session.send({ diff --git a/src/client.ts b/src/client.ts index 6c5afb0..f3ddd48 100644 --- a/src/client.ts +++ b/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 { 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 }; } diff --git a/src/incoming-requests.ts b/src/incoming-requests.ts index 4bccb59..d121108 100644 --- a/src/incoming-requests.ts +++ b/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); diff --git a/src/send-window.ts b/src/send-window.ts index 6f2dea4..00877b4 100644 --- a/src/send-window.ts +++ b/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 { + if (this.inFlight === 0) return Promise.resolve(0); + + return new Promise(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); + }); } } diff --git a/src/server.ts b/src/server.ts index bad202e..9182602 100644 --- a/src/server.ts +++ b/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 { if (event !== 'serverError') this.emit('serverError', error); } - /** Stops listening and closes every live session. */ - close(): Promise { - 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 { + 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, }); diff --git a/src/session-options.ts b/src/session-options.ts index 4da295c..080f0bf 100644 --- a/src/session-options.ts +++ b/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; }; diff --git a/src/session.ts b/src/session.ts index 3e506cb..b5e8b78 100644 --- a/src/session.ts +++ b/src/session.ts @@ -67,6 +67,7 @@ export class Session extends EventEmitter { 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 { 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 { 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 { + 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 { 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 { return response; } + /** Stops new sends and waits out the ones already issued. A dead link has nothing to wait for. */ + private async drain(): Promise { + 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 { 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 { message: parsed.err.message, }); this.emit('sessionError', parsed.err); - this.close(); + this.abort(); return false; } diff --git a/test/interop.test.ts b/test/interop.test.ts index 10d9457..3cdd541 100644 --- a/test/interop.test.ts +++ b/test/interop.test.ts @@ -232,7 +232,7 @@ describe('a live session against the reference implementation', () => { const port = refServer.address()?.port ?? 0; const { err, session } = await client({ port }); - t.after(() => { session?.close(); }); + t.after(() => session?.close()); assert.equal(err, undefined); assert.ok(session); diff --git a/test/readme.test.ts b/test/readme.test.ts index 8c8c0f9..061f2a1 100644 --- a/test/readme.test.ts +++ b/test/readme.test.ts @@ -77,7 +77,7 @@ describe('README: Client', () => { }); if (err) throw err; - t.after(() => { session.close(); }); + t.after(() => session.close()); const reported = once(resolve => { session.on('dlr', resolve); }); const { err: sendErr, smsIds } = await session.sendSms({ @@ -100,7 +100,7 @@ describe('README: Client', () => { const { err, session } = await client(); if (err) throw err; - t.after(() => { session.close(); }); + t.after(() => session.close()); const { signal } = new AbortController(); const [sms, sent] = await Promise.all([ @@ -128,7 +128,7 @@ describe('README: Client', () => { const { err, session } = await client(); if (err) throw err; - t.after(() => { session.close(); }); + t.after(() => session.close()); const incoming = once(resolve => { session.on('sms', resolve); }); const peer = await bound; diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index e27cbde..a876e0b 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -83,7 +83,7 @@ describe('merged delivery reports', () => { assert.equal(report.statusMsg, 'DELIVERED'); assert.deepEqual(perSegment, ['merge-me-1', 'merge-me-2', 'merge-me-3']); - session.close(); + await session.close(); await smpp.close(); }); @@ -119,7 +119,7 @@ describe('merged delivery reports', () => { assert.equal(report.statusMsg, 'UNDELIVERABLE'); assert.equal(report.segments.length, 3); - session.close(); + await session.close(); await smpp.close(); }); }); @@ -203,7 +203,7 @@ describe('sendSms()', () => { assert.ok(sent.err instanceof Error); assert.match(sent.err.message, /ESME_RMSGQFUL/); - session.close(); + await session.close(); await smpp.close(); }); @@ -324,7 +324,7 @@ describe('reconnect', () => { // Drop the connection from the server's side, as a peer restart would. for (const serverSession of smpp.sessions) { - serverSession.close(); + await serverSession.close(); } await reconnected; @@ -341,7 +341,7 @@ describe('reconnect', () => { assert.equal(sent.err, undefined); assert.deepEqual(messages, ['after reconnect']); - session.close(); + await session.close(); await smpp.close(); }); @@ -373,7 +373,7 @@ describe('reconnect', () => { const reconnected = once(resolve => { session.on('reconnected', () => { resolve(true); }); }); - peerOf(smpp).close(); + await peerOf(smpp).close(); await reconnected; const merged = once(resolve => { session.on('messageDlr', resolve); }); @@ -386,7 +386,7 @@ describe('reconnect', () => { assert.equal(report.smsId, 'across-the-drop'); assert.equal(report.segments.length, 3); - session.close(); + await session.close(); await smpp.close(); }); @@ -402,7 +402,7 @@ describe('reconnect', () => { let reconnects = 0; session.on('reconnected', () => { reconnects++; }); - session.close(); + await session.close(); await new Promise(resolve => setTimeout(resolve, 150)); @@ -586,10 +586,73 @@ describe('AbortSignal on a send', () => { assert.ok(sent.err instanceof Error); - session.close(); + await session.close(); for (const sock of accepted) sock.destroy(); await new Promise(resolve => silent.close(() => { resolve(); })); }); }); + +describe('graceful shutdown', () => { + async function submitInFlight(options: Parameters[0] = {}) { + const smpp = await startServer(); + const incoming = once(resolve => { + smpp.on('session', bound => bound.on('sms', resolve)); + }); + const { session } = await client({ port: smpp.port, ...options }); + + assert.ok(session); + + const sent = session.sendSms({ from: '46701113311', message: 'answer me', to: '46709771337' }); + + return { sent, session, sms: await incoming, smpp }; + } + + test('close() waits out a submit already on the wire and refuses new ones', async () => { + const { sent, session, sms, smpp } = await submitInFlight(); + const closed = session.close(); + const refused = await session.sendSms({ + from: '46701113311', + message: 'too late', + to: '46709771337', + }); + + assert.ok(refused.err instanceof Error); + assert.equal(refused.err.message, 'Session is shutting down'); + + await sms.sendResp({ smsId: 'answered-while-draining' }); + + assert.deepEqual((await sent).smsIds, ['answered-while-draining']); + assert.deepEqual(await closed, {}); + + await smpp.close(); + }); + + test('unbind() waits out a submit already on the wire before it unbinds', async () => { + const { sent, session, sms, smpp } = await submitInFlight(); + const unbound = session.unbind(); + + await sms.sendResp({ smsId: 'answered-before-unbind' }); + + assert.deepEqual((await sent).smsIds, ['answered-before-unbind']); + assert.deepEqual(await unbound, {}); + + await smpp.close(); + }); + + test('gives up on a request that outlasts shutdownTimeout', async () => { + const { sent, session, smpp } = await submitInFlight({ shutdownTimeout: 50 }); + const closed = await session.close(); + + assert.ok(closed.err instanceof Error); + assert.match(closed.err.message, /still in flight/); + + const result = await sent; + + assert.ok(result.err instanceof Error); + assert.equal(result.err.message, 'Session closed before a response arrived'); + + await smpp.close(); + }); +}); diff --git a/test/session.test.ts b/test/session.test.ts index 1a16285..341911b 100644 --- a/test/session.test.ts +++ b/test/session.test.ts @@ -220,7 +220,7 @@ describe('bind', () => { assert.ok(sent.err instanceof Error); assert.equal(sent.err.message, 'Session closed before a response arrived'); - session.close(); + await session.close(); await peer.close(); }); @@ -268,7 +268,7 @@ describe('bind', () => { assert.deepEqual(bound.userData, { userId: 123 }); - session.close(); + await session.close(); await smpp.close(); }); @@ -404,7 +404,7 @@ describe('bind', () => { const { session } = await connect(smpp); assert.ok(session); - t.after(() => { session.close(); }); + t.after(() => session.close()); assert.equal(session.peerInterfaceVersion, 0x50); assert.ok(session.acceptsOptionalParams()); }); @@ -418,7 +418,7 @@ describe('bind', () => { const { session } = await client({ port: peer.port }); assert.ok(session); - t.after(() => { session.close(); }); + t.after(() => session.close()); assert.equal(session.peerInterfaceVersion, 0x00); assert.equal(session.acceptsOptionalParams(), false); }); @@ -440,7 +440,7 @@ describe('bind direction', () => { assert.ok(sent.pduObj); assert.equal(sent.pduObj.cmdStatus, 'ESME_RINVBNDSTS'); - session.close(); + await session.close(); await smpp.close(); }); @@ -461,7 +461,7 @@ describe('bind direction', () => { assert.deepEqual(sent.smsIds, []); assert.equal(arrived.length, 0); - session.close(); + await session.close(); await smpp.close(); }); @@ -481,7 +481,7 @@ describe('bind direction', () => { assert.ok(sent.pduObj); assert.equal(sent.pduObj.cmdStatus, 'ESME_RINVBNDSTS'); - session.close(); + await session.close(); await smpp.close(); }); @@ -506,7 +506,7 @@ describe('bind direction', () => { assert.ok(report.err instanceof Error); assert.match(report.err.message, /transmitter-bound/); - session.close(); + await session.close(); await smpp.close(); }); }); @@ -548,7 +548,7 @@ describe('sending', () => { assert.equal(submitted.params.source_addr_ton, 5); assert.equal(submitted.params.dest_addr_ton, 1); - session.close(); + await session.close(); await smpp.close(); }); @@ -577,7 +577,7 @@ describe('sending', () => { assert.equal(sent.pduObjs.length, 4); assert.deepEqual(sent.smsIds, ['long-id-1', 'long-id-2', 'long-id-3', 'long-id-4']); - session.close(); + await session.close(); await smpp.close(); }); @@ -603,7 +603,7 @@ describe('sending', () => { assert.equal(sms.message, message); assert.match(sms.smsId, /^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/); - session.close(); + await session.close(); await smpp.close(); }); @@ -630,7 +630,7 @@ describe('sending', () => { assert.equal(sms.message, 'تست'); assert.ok(sms.flash); - session.close(); + await session.close(); await smpp.close(); }); @@ -664,7 +664,7 @@ describe('sending', () => { assert.equal(params.source_addr_npi, consts.NPI.PRIVATE); assert.equal(params.source_addr_ton, consts.TON.ABBREVIATED); - session.close(); + await session.close(); await smpp.close(); }); }); @@ -679,7 +679,7 @@ describe('receiving', () => { const { session } = await connect(smpp); assert.ok(session); - t.after(() => { session.close(); }); + t.after(() => session.close()); return { peer: await bound, session }; } @@ -808,7 +808,7 @@ describe('delivery reports', () => { assert.equal(receipt.tlvs.receipted_message_id?.tagValue, 'dlr-id'); assert.equal(receipt.tlvs.message_state?.tagValue, 2); - session.close(); + await session.close(); await smpp.close(); }); @@ -845,7 +845,7 @@ describe('delivery reports', () => { // 0.4.0 wrote stat:UNDELIVERABLE, which is not the spec's seven-character field. assert.match(await raw, /stat:UNDELIV /); - session.close(); + await session.close(); await smpp.close(); }); @@ -918,7 +918,7 @@ describe('delivery reports', () => { assert.deepEqual(perSegment, ['unrequested-1', 'unrequested-2', 'unrequested-3']); assert.equal(merged, 0); - session.close(); + await session.close(); await smpp.close(); }); }); @@ -1044,7 +1044,7 @@ describe('robustness', () => { assert.ok(peak <= 2, `peak was ${String(peak)}`); - session.close(); + await session.close(); await smpp.close(); }); @@ -1062,7 +1062,7 @@ describe('robustness', () => { assert.ok(reported instanceof Error, 'a response that never reached the wire should be reported'); assert.equal(reported.message, sent.err.message); - session.close(); + await session.close(); }); test('ignores events from the socket it left behind on a reconnect', async () => { @@ -1080,7 +1080,7 @@ describe('robustness', () => { const dead = session.sock; for (const serverSession of smpp.sessions) { - serverSession.close(); + await serverSession.close(); } await reconnected; @@ -1099,7 +1099,7 @@ describe('robustness', () => { assert.equal(closes, 0); assert.equal(sent.err, undefined); - session.close(); + await session.close(); await smpp.close(); }); @@ -1129,7 +1129,7 @@ describe('robustness', () => { const { session } = await connect(smpp); assert.ok(session); - t.after(() => { session.close(); }); + t.after(() => session.close()); const peer = await bound; const controller = new AbortController(); @@ -1161,7 +1161,7 @@ describe('robustness', () => { const { session } = await connect(smpp, { maxOutstanding: 1, responseTimeout: 5000 }); assert.ok(session); - t.after(() => { session.close(); }); + t.after(() => session.close()); const held = session.sendSms({ from: '46701113311', message: 'holds the slot', to: '46709771337' }); const controller = new AbortController(); @@ -1177,7 +1177,7 @@ describe('robustness', () => { assert.notEqual(aborted, false, 'an aborted send should not wait for the window'); assert.ok(aborted !== false && aborted.err instanceof Error); - session.close(); + await session.close(); await held; }); @@ -1262,7 +1262,7 @@ describe('application hooks that throw or reject', () => { assert.ok(reported instanceof Error, 'a throwing sms listener should reach the session'); assert.equal(reported.message, 'listener exploded'); - session.close(); + await session.close(); await smpp.close(); }); @@ -1287,7 +1287,7 @@ describe('application hooks that throw or reject', () => { assert.ok(sent.err instanceof Error); - session.close(); + await session.close(); await smpp.close(); }); @@ -1319,7 +1319,7 @@ describe('application hooks that throw or reject', () => { assert.ok(reported instanceof Error, 'a rejecting sms listener should reach the session'); assert.equal(reported.message, 'null'); - session.close(); + await session.close(); await smpp.close(); }); @@ -1343,7 +1343,7 @@ describe('application hooks that throw or reject', () => { assert.ok(sent.err instanceof Error); - session.close(); + await session.close(); await smpp.close(); }); @@ -1359,7 +1359,7 @@ describe('application hooks that throw or reject', () => { assert.ok(reported instanceof Error, 'a rejecting session listener should reach the server'); assert.equal(reported.message, 'session listener rejected'); - session?.close(); + await session?.close(); await smpp.close(); }); @@ -1376,7 +1376,7 @@ describe('application hooks that throw or reject', () => { await smpp.close(); assert.equal(smpp.sessions.size, 0); - session.close(); + await session.close(); }); test('refuses a send window that can never free a slot', async () => { @@ -1387,6 +1387,12 @@ describe('application hooks that throw or reject', () => { assert.match(err.message, /maxOutstanding/); assert.equal(session, undefined); + const negative = await connect(smpp, { shutdownTimeout: -1 }); + + assert.ok(negative.err instanceof Error); + assert.match(negative.err.message, /shutdownTimeout/); + assert.equal(negative.session, undefined); + await smpp.close(); }); @@ -1492,7 +1498,7 @@ describe('link timers', () => { 'a peer that answers nothing should time the link out', ); - session.close(); + await session.close(); await peer.close(); }); @@ -1510,7 +1516,7 @@ describe('link timers', () => { assert.ok(await raceWithin(2000, back), 'a link that timed out should be reconnected'); - session.close(); + await session.close(); await peer.close(); }); }); diff --git a/todo.md b/todo.md index aa03081..5a811fe 100644 --- a/todo.md +++ b/todo.md @@ -5,7 +5,7 @@ rules there constrain every item below. ## Status -The rewrite is **feature complete and green**: 236 tests, lint and typecheck clean, verified on Node +The rewrite is **feature complete and green**: 242 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 @@ -55,6 +55,7 @@ Rules the API follows: | Delivery receipt parsing, TLV and text | `test/dlr.test.ts` | | Session, client, server: bind, auth, send, reassembly, DLRs, timeouts, abort, send window | `test/session.test.ts` | | Merged multipart DLRs including across a reconnect, reassembly bounds, per-send abort, the segment cap | `test/session-extras.test.ts` | +| A draining `close()` and `unbind()`, bounded by `shutdownTimeout` | `test/session-extras.test.ts` | | Every runnable README example | `test/readme.test.ts` | | Receipt-versus-message classification by `esm_class` | `test/dlr.test.ts`, `test/session.test.ts` | | A listener that throws, or rejects, reaching `sessionError`/`serverError` rather than the process | `test/session.test.ts`, `test/error-from.test.ts` | @@ -118,14 +119,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. -- [ ] **`close()` and `unbind()` drop in-flight requests instead of draining them.** `teardown()` - settles every pending request with "Session closed before a response arrived" and destroys the - socket in the same tick, so a submit the SMSC has already accepted is reported to the caller - as a failure — the ambiguous outcome that produces a duplicate on retry. Give both a drain: - refuse new sends, wait out the pending responses up to a `shutdownTimeout`, then tear down - whatever is left. Distinct from re-queueing across a reconnect above — this is the deliberate - shutdown path, where there is nothing to come back to. -- [ ] **`session.ts` is 386 lines.** The one seam left in it is a socket-to-PDU transport, which +- [ ] **`session.ts` is 465 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