From 5c618e0061f325f800159cc05a0756fce9a2ccc5 Mon Sep 17 00:00:00 2001 From: Lilleman auf Larv Date: Sun, 30 Aug 2026 20:27:38 +0200 Subject: [PATCH] Stop accepting before the drain and report the link that dropped during it --- AGENTS.md | 24 ++++++---- README.md | 16 ++++--- src/client.ts | 6 +-- src/index.ts | 1 + src/send-window.ts | 18 ++++++-- src/server.ts | 16 ++++--- src/session-options.ts | 3 ++ src/session.ts | 58 +++++++++++++----------- test/session-extras.test.ts | 89 ++++++++++++++++++++++++++++++++++++- todo.md | 9 +++- 10 files changed, 184 insertions(+), 56 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 9bcc2e3..dabf376 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -252,14 +252,22 @@ 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. +- **A deliberate shutdown drains; an unusable link and an abort do not.** `close()` and `unbind()` + refuse further sends and wait on the send window, 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. The + window counts slots, never outcomes, so it says when to stop waiting and nothing about what + happened: it empties on a drop too, where `teardown()` settles everything the link was carrying, + which is why `drain()` reads `closed` before it reads the count. The wait covers what this end + sent — a request the peer sent us is answered through `sendReturn()`, which never enters the + window, so a server session waits for none of its inbound work; that half is in todo.md. + `shutdownTimeout` bounds the drain alone, and `0` waits forever like every other timeout here; + `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 + either would only hold a socket open for the timeout. `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 adding a devDependency or shelling out to openssl. Maintainer's call, 2026-08-26: the dev image diff --git a/README.md b/README.md index f854117..9ff772d 100644 --- a/README.md +++ b/README.md @@ -21,7 +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. | +| **Graceful shutdown** | `close()` and `unbind()` wait out the requests this end already sent, 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 @@ -101,7 +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. | +| `shutdownTimeout` | `5000` | How long `close()` and `unbind()` wait for the requests this end already sent; `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). | @@ -208,7 +208,7 @@ smpp.on('session', session => { }); console.log(smpp.port); // the port actually bound, useful when 0 was requested -await smpp.close(); // drain and close every live session, then stop listening +await smpp.close(); // stop listening, then drain and close every live session ``` `sendDlr` accepts `SCHEDULED`, `ENROUTE`, `DELIVERED`, `EXPIRED`, `DELETED`, `UNDELIVERABLE`, @@ -310,10 +310,12 @@ TypeScript users can import `SmppLog` to have the compiler check one. ### Methods `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: +refuse further sends, wait out the requests this end already sent — up to `shutdownTimeout`, or +until an `AbortSignal` given as `close({ signal })` says to stop now — and then tear down whatever +is left, resolving to an `err` that says what was lost. A request the peer sent *us* is answered +through `sendReturn()` and is not waited for, and `unbind()` waits a further `responseTimeout` for +its own response. `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 f3ddd48..58ac4ca 100644 --- a/src/client.ts +++ b/src/client.ts @@ -196,18 +196,18 @@ export async function client(options: ClientOptions = {}): Promise { void session.close(); }, { once: true }); + signal?.addEventListener('abort', () => { void session.close({ signal }); }, { once: true }); const bound = await bind(session, options); if (bound.err) { - void session.close(); + void session.close({ signal }); return { err: bound.err }; } diff --git a/src/index.ts b/src/index.ts index 42f597b..e0f724f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -46,6 +46,7 @@ export type { ServerOptions, } from './server.ts'; export type { + CloseOptions, MessageDlr, ReconnectOptions, SendOptions, diff --git a/src/send-window.ts b/src/send-window.ts index 00877b4..e1492ca 100644 --- a/src/send-window.ts +++ b/src/send-window.ts @@ -37,10 +37,20 @@ export class SendWindow { } } - /** Resolves once nothing is in flight, or on the timeout with how many still are. 0 never times out. */ - idle(timeout: number): Promise { + /** Everything the caller is still owed: on the wire, plus queued behind a full window. */ + unfinished(): number { + 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. + */ + idle(timeout: number, signal?: AbortSignal): Promise { if (this.inFlight === 0) return Promise.resolve(0); + if (signal?.aborted === true) return Promise.resolve(this.unfinished()); + return new Promise(resolve => { let timer: NodeJS.Timeout | undefined = undefined; const done = (): void => { @@ -49,7 +59,8 @@ export class SendWindow { if (timer) clearTimeout(timer); if (index !== -1) this.waitingForIdle.splice(index, 1); - resolve(this.inFlight); + signal?.removeEventListener('abort', done); + resolve(this.unfinished()); }; if (timeout > 0) { @@ -57,6 +68,7 @@ export class SendWindow { timer.unref(); } + signal?.addEventListener('abort', done, { once: true }); this.waitingForIdle.push(done); }); } diff --git a/src/server.ts b/src/server.ts index 9182602..fab068b 100644 --- a/src/server.ts +++ b/src/server.ts @@ -1,3 +1,4 @@ +import type { CloseOptions } from './session-options.ts'; import type { PduObject, TlvInput } from './pdu.ts'; import type { Result, VoidResult } from './result.ts'; import type { Server as NetServer, Socket } from 'node:net'; @@ -115,19 +116,22 @@ export class SmppServer extends EventEmitter { if (event !== 'serverError') this.emit('serverError', error); } - /** Stops listening and closes every live session, draining each one first. */ - async close(): Promise { + /** Stops listening, then drains and closes every session that was live when it stopped. */ + async close(options: CloseOptions = {}): Promise { + // Before the drain, or the listener keeps accepting connections nothing will ever close. + const stopped = new Promise(resolve => { this.server.close(() => { resolve(); }); }); const live = [...this.sessions]; this.sessions.clear(); await Promise.all(live.map(async session => { - const closed = await session.close().catch((thrown: unknown) => ({ err: errorFrom(thrown) })); + const closed = await session.close(options) + .catch((thrown: unknown) => ({ err: errorFrom(thrown) })); if (closed.err) this.emit('serverError', closed.err); })); - return new Promise(resolve => { this.server.close(() => { resolve(); }); }); + return stopped; } } @@ -311,7 +315,9 @@ function onListening(listener: NetServer, smpp: SmppServer, options: ServerOptio log.info('server - listening', { host: options.host ?? '*', port: smpp.port }); - options.signal?.addEventListener('abort', () => { void smpp.close(); }, { once: true }); + const signal = options.signal; + + signal?.addEventListener('abort', () => { void smpp.close({ signal }); }, { once: true }); } /** Starts listening for SMPP connections. Resolves once the socket is bound. */ diff --git a/src/session-options.ts b/src/session-options.ts index 080f0bf..8347353 100644 --- a/src/session-options.ts +++ b/src/session-options.ts @@ -49,6 +49,9 @@ export function bindCarries(bindType: BindType | undefined, cmdName: string): bo export type SendOptions = { signal?: AbortSignal | undefined }; +/** An already-aborted signal skips the drain; one that fires during it cuts the wait short. */ +export type CloseOptions = { signal?: AbortSignal | undefined }; + /** * First refusal on every incoming request. Returning true means the hook answered it and the * built-in handling is skipped — this is how the server owns bind without the session also diff --git a/src/session.ts b/src/session.ts index b5e8b78..45b20b0 100644 --- a/src/session.ts +++ b/src/session.ts @@ -2,7 +2,7 @@ import type { ErrorName } from './defs/errors.ts'; import type { MessageDlr } from './dlr-merger.ts'; import type { ParamValue } from './defs/types.ts'; import type { PduObject, PduObjectInput, TlvInput } from './pdu.ts'; -import type { BindType, ReconnectOptions, SendOptions, SessionEvents, SessionOptions } from './session-options.ts'; +import type { BindType, CloseOptions, ReconnectOptions, SendOptions, SessionEvents, SessionOptions } from './session-options.ts'; import type { Result, VoidResult } from './result.ts'; import type { SendSmsOptions, SendSmsResult } from './send-sms.ts'; import type { SmppLog } from './log.ts'; @@ -23,6 +23,7 @@ import { silentLog } from './log.ts'; import { submitSms } from './send-sms.ts'; export type { + CloseOptions, MessageDlr, ReconnectOptions, SendOptions, @@ -221,17 +222,15 @@ export class Session extends EventEmitter { * 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 drained = await this.drain(undefined); const wasOpen = !this.closed; - // request(), not send(): the drain gate would refuse it, and the window is empty by now. + // request(), not send(): the drain gate refuses a send, and the unbind goes out either way. const sent = wasOpen ? await this.request({ cmdName: 'unbind' }, {}) : { err: new Error('Session is closed') }; const closedOnUnbind = wasOpen && this.closed; - this.shutdown(); + this.end(); return sent.err && !closedOnUnbind ? { err: sent.err } : drained; } @@ -240,12 +239,10 @@ export class Session extends EventEmitter { * 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(); + async close(options: CloseOptions = {}): Promise { + const drained = await this.drain(options.signal); - const drained = await this.drain(); - - this.shutdown(); + this.end(); return drained; } @@ -330,34 +327,41 @@ 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 { + /** + * 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 { + this.reconnectLoop?.stop(); 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); + const unfinished = await this.window.idle(timeout, signal); - if (inFlight === 0) return {}; + // The window empties on a drop too: teardown() settles everything the link was carrying. + if (this.isClosed()) return { err: new Error('The link dropped before the drain finished') }; - this.log.warn('session - shutting down with requests still in flight', { inFlight, timeout }); + if (unfinished === 0) return {}; - return { err: new Error(`Shut down with ${String(inFlight)} request(s) still in flight`) }; + this.log.warn('session - shutting down with requests unfinished', { timeout, unfinished }); + + return { err: new Error(`Shut down with ${String(unfinished)} request(s) unfinished`) }; } - private shutdown(): void { + /** Read through a method: teardown() can land while the drain is awaiting. */ + private isClosed(): boolean { + return this.closed; + } + + /** The session is over now, drained or not. Nothing brings it back. */ + private end(): void { + this.reconnectLoop?.stop(); 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; @@ -395,7 +399,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.abort(); + this.end(); return; } @@ -416,7 +420,7 @@ export class Session extends EventEmitter { message: parsed.err.message, }); this.emit('sessionError', parsed.err); - this.abort(); + this.end(); return false; } diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index a876e0b..85122d5 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -646,7 +646,7 @@ describe('graceful shutdown', () => { const closed = await session.close(); assert.ok(closed.err instanceof Error); - assert.match(closed.err.message, /still in flight/); + assert.match(closed.err.message, /unfinished/); const result = await sent; @@ -655,4 +655,91 @@ describe('graceful shutdown', () => { await smpp.close(); }); + + // The window empties on a drop as well as on an answer, so it cannot be what the result reads. + test('reports a link that dropped mid-drain rather than calling it a clean shutdown', async () => { + const { sent, session, smpp } = await submitInFlight(); + const closing = session.close(); + + for (const bound of smpp.sessions) { + bound.sock.destroy(); + } + + const closed = await closing; + + assert.ok(closed.err instanceof Error); + assert.match(closed.err.message, /dropped/); + assert.ok((await sent).err instanceof Error); + + await smpp.close(); + }); + + // The queued segments are the whole reason the drain waits on the window and not on the pending map. + test('counts the segments still queued behind a full window', async () => { + const smpp = await startServer(); + const onWire = once(resolve => { + smpp.on('session', bound => bound.on('incomingPduObj', resolve)); + }); + const { session } = await client({ maxOutstanding: 1, port: smpp.port, shutdownTimeout: 50 }); + + assert.ok(session); + + const sent = session.sendSms({ + from: '46701113311', + message: 'x'.repeat(400), + to: '46709771337', + }); + + await onWire; + + const closed = await session.close(); + + assert.ok(closed.err instanceof Error); + assert.match(closed.err.message, /3 request\(s\)/); + assert.ok((await sent).err instanceof Error); + + await smpp.close(); + }); + + test('an aborted close tears down at once instead of waiting out the drain', async () => { + const { sent, session, smpp } = await submitInFlight({ shutdownTimeout: 30_000 }); + const controller = new AbortController(); + const started = Date.now(); + + controller.abort(); + + const closed = await session.close({ signal: controller.signal }); + + assert.ok(Date.now() - started < 1000); + assert.ok(closed.err instanceof Error); + assert.ok(session.sock.destroyed); + assert.ok((await sent).err instanceof Error); + + await smpp.close(); + }); + + test('stops accepting the moment close() is called, not when the drain ends', async () => { + const smpp = await startServer({ shutdownTimeout: 30_000 }); + const arrived = once(resolve => { smpp.on('session', resolve); }); + const silent = net.connect({ port: smpp.port }); + + silent.resume(); + + const bound = await arrived; + const unanswered = bound.send({ cmdName: 'enquire_link' }); + const closing = smpp.close(); + const late = await new Promise(resolve => { + const sock = net.connect({ port: smpp.port }); + + sock.on('connect', () => { sock.destroy(); resolve(true); }); + sock.on('error', () => { resolve(false); }); + }); + + assert.equal(late, false); + + silent.destroy(); + await closing; + + assert.ok((await unanswered).err instanceof Error); + }); }); diff --git a/todo.md b/todo.md index 5a811fe..2928f3a 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**: 242 tests, lint and typecheck clean, verified on Node +The rewrite is **feature complete and green**: 246 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,7 +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` | +| A draining `close()` and `unbind()`, bounded by `shutdownTimeout` or an abort | `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` | @@ -115,6 +115,11 @@ session message is a change to every call site. - [ ] **In-flight sends across a reconnect.** They currently fail with "Session closed before a response arrived" and the caller retries. Re-queueing them automatically would be friendlier but risks duplicate delivery, so it needs a decision before it is built. +- [ ] **The drain covers only what this end sent.** `close()` and `unbind()` wait on the send window, + which `sendReturn()` never enters, so a server session tears down without waiting for the + application to answer the messages it is holding — the duplicate-on-retry outcome again, in + the SMSC direction. A full inbound drain needs a completion signal the `sms` event does not + carry, so it is a public-surface decision. Raised by review, 2026-08-30. - [ ] **Merge state does not survive a process restart.** A drop no longer discards it, but a restart 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