diff --git a/AGENTS.md b/AGENTS.md index d837b4f..ce8c175 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -50,6 +50,7 @@ src/ message.ts Encoding detection, splitting, bit counting, SMPP date formatting pdu.ts pduToObj / objToPdu / pduReturn — synchronous, result-returning pdu-framer.ts PduFramer: a byte stream cut into complete PDUs + pdu-transport.ts PduTransport: the socket a session reads complete PDUs off pending-requests.ts PendingRequests: sequence numbers, correlation, timeout, abort reassembly.ts Reassembler: capped, expiring multipart groups reconnect-loop.ts ReconnectLoop: backoff, retry timer, stopped-ness @@ -295,6 +296,13 @@ exactly 140. that loop before tearing down, so every deliberate shutdown emits `close`. Without the split an application that opens a replacement client on `close` ends up holding two binds on one account. +- **`session.sock` is a getter over `PduTransport`, and stays public.** Maintainer's call, 2026-08-31: + `session.ts` had reached its line cap, so the socket-to-PDU seam todo.md named was opened — + `PduTransport` owns the socket, the framer and the parse, and hands the session raw bytes, framed + PDUs, parsed ones and an unreadable stream. Reading `session.sock` is unchanged; assigning it no + longer compiles, which never rewired the handlers and so never worked. The transport stays + unpublished like the other collaborators. + - **A client re-binds after a drop unless it is told not to.** Maintainer's call, 2026-08-31: surviving a dropped link is most of what the session layer is for, and behind an opt-in an application that never read the options table got none of it. `reconnect` takes diff --git a/README.md b/README.md index 52e2148..8e88f4f 100644 --- a/README.md +++ b/README.md @@ -316,7 +316,7 @@ TypeScript users can import `SmppLog` to have the compiler check one. | `sms` | An SMS arrives, reassembled if it was multipart. Carries `sendResp()`, `sendDlr()` and the `smsId` it was answered with. | | `dlr` | A delivery report arrives, one per segment. `smsId` is undefined when the peer marked a receipt whose body carries no readable id. `statusMsg` names `statusId` unless the peer sent a `message_state` this library cannot name — then `statusId` is that raw value and `statusMsg` is whatever the body said, or `UNKNOWN`. | | `messageDlr` | Every segment of a multipart message sent with `dlr: true` has been reported on, carrying the worst status of the segments. Merging needs the SMSC to number its segment ids `-`, which is this library's own server's convention — an SMSC that hands out unrelated ids per segment never fires it. A base is merged once: a later message the SMSC gives the same ids is reported on through `dlr` alone, and an earlier one still collecting loses its merged report as well. | -| `close` | The session is over: you closed it, or the link dropped with `reconnect: false`. Nothing brings it back. | +| `close` | The session is over, because nothing will bring the link back. Fires once, whether you closed it or the link failed for good. | | `disconnected` | The link dropped and the reconnect loop will retry it. Do not open a replacement client here — the session you hold comes back on its own, and `reconnected` says when. | | `reconnected` | The client re-bound after a drop. | | `sessionError` | Something failed on a live session, including a hook or listener that threw or, if it was `async`, rejected. | diff --git a/src/pdu-transport.ts b/src/pdu-transport.ts new file mode 100644 index 0000000..6ce41be --- /dev/null +++ b/src/pdu-transport.ts @@ -0,0 +1,91 @@ +import type { PduObject } from './pdu.ts'; +import type { SmppLog } from './log.ts'; +import type { Socket } from 'node:net'; +import type { VoidResult } from './result.ts'; +import { PduFramer } from './pdu-framer.ts'; +import { pduToObj } from './pdu.ts'; + +export type PduTransportOptions = { + log: SmppLog; + onClose: () => void; + /** Raw bytes, before framing. */ + onData: (chunk: Buffer) => void; + onError: (err: Error) => void; + /** A complete PDU, before it is parsed. */ + onFramed: (pdu: Buffer) => void; + onPdu: (pduObj: PduObject) => void; + /** Nothing further can be read off this stream, whatever the socket does next. */ + onUnreadable: (err: Error) => void; +}; + +/** A socket read as a stream of complete PDUs. A reconnect attaches a new socket in its place. */ +export class PduTransport { + private readonly options: PduTransportOptions; + private framer = new PduFramer(); + private socket: Socket; + + constructor(options: PduTransportOptions, sock: Socket) { + this.options = options; + this.socket = sock; + } + + get sock(): Socket { + return this.socket; + } + + /** Wires a freshly opened socket in, replacing any previous one. */ + attach(sock: Socket): void { + // The socket being replaced is already dead, and its three handlers still point here. + if (this.socket !== sock) this.socket.removeAllListeners(); + + this.socket = sock; + this.framer = new PduFramer(); + + sock.on('data', chunk => { this.read(chunk); }); + sock.on('close', () => { this.options.onClose(); }); + sock.on('error', err => { + this.options.log.warn('transport - socket error', { message: err.message }); + this.options.onError(err); + this.options.onClose(); + }); + } + + write(pdu: Buffer): VoidResult { + if (this.socket.destroyed) return { err: new Error('Socket is closed') }; + + this.socket.write(pdu); + + return {}; + } + + private read(chunk: Buffer): void { + this.options.onData(chunk); + this.framer.push(chunk); + + const framed = this.framer.next(); + + if (framed.err) { + this.options.log.warn('transport - unusable stream, closing', { message: framed.err.message }); + this.options.onUnreadable(framed.err); + + return; + } + + for (const pdu of framed.pdus) { + this.options.onFramed(pdu); + + const parsed = pduToObj(pdu); + + if (parsed.err) { + this.options.log.warn('transport - could not parse an incoming PDU, closing', { + message: parsed.err.message, + }); + this.options.onUnreadable(parsed.err); + + return; + } + + this.options.onPdu(parsed.pduObj); + } + } +} diff --git a/src/session-options.ts b/src/session-options.ts index 9c47b9b..adc28e2 100644 --- a/src/session-options.ts +++ b/src/session-options.ts @@ -143,6 +143,8 @@ function checkLimits(limits: [string, number, number][]): VoidResult { return {}; } +const backoffDelays: readonly string[] = ['maxDelay', 'minDelay']; + function checkReconnect(reconnect: unknown): VoidResult { if (reconnect === undefined || reconnect === false) return {}; @@ -150,11 +152,20 @@ function checkReconnect(reconnect: unknown): VoidResult { return { err: new Error('reconnect takes { maxDelay, minDelay }, or false to turn it off') }; } - // A minDelay of 0 never doubles, so the backoff never starts and every retry lands at once. - return checkLimits([ - ['maxDelay', delayOr(reconnect.maxDelay, defaults.maxDelay), 1], - ['minDelay', delayOr(reconnect.minDelay, defaults.minDelay), 1], - ]); + for (const key of Object.keys(reconnect)) { + if (!backoffDelays.includes(key)) { + return { err: new Error(`reconnect has no ${key}, name ${backoffDelays.join(' or ')}`) }; + } + } + + const maxDelay = delayOr(reconnect.maxDelay, defaults.maxDelay); + const minDelay = delayOr(reconnect.minDelay, defaults.minDelay); + // A delay of 0 never doubles, so the backoff never starts and every retry lands at once. + const checked = checkLimits([['maxDelay', maxDelay, 1], ['minDelay', minDelay, 1]]); + + if (checked.err || maxDelay >= minDelay) return checked; + + return { err: new Error(`maxDelay must be minDelay or more, got ${String(maxDelay)}`) }; } function isRecord(value: unknown): value is Record { diff --git a/src/session.ts b/src/session.ts index 1dbb84a..63768bc 100644 --- a/src/session.ts +++ b/src/session.ts @@ -11,14 +11,14 @@ import { DlrMerger } from './dlr-merger.ts'; import { EventEmitter } from 'node:events'; import { IncomingRequests } from './incoming-requests.ts'; import { LinkTimers } from './link-timers.ts'; -import { PduFramer } from './pdu-framer.ts'; +import { PduTransport } from './pdu-transport.ts'; import { PendingRequests } from './pending-requests.ts'; import { ReconnectLoop } from './reconnect-loop.ts'; import { SendWindow } from './send-window.ts'; import { errorFrom } from './error-from.ts'; import { optionalParamsMinVersion } from './defs/constants.ts'; import { bindCarries, bindCommands, defaultSystemId, defaults } from './session-options.ts'; -import { isResp, objToPdu, pduReturn, pduToObj } from './pdu.ts'; +import { isResp, objToPdu, pduReturn } from './pdu.ts'; import { silentLog } from './log.ts'; import { submitSms } from './send-sms.ts'; @@ -47,8 +47,6 @@ export class Session extends EventEmitter { declare prependOnceListener: (event: K, listener: SessionListener) => this; declare removeListener: (event: K, listener: SessionListener) => this; - /** Replaced on reconnect, so hold the session rather than this. */ - sock: Socket; readonly log: SmppLog; /** The role the ESME bound with, whichever end of the link this is. Undefined before any bind. */ @@ -64,12 +62,13 @@ export class Session extends EventEmitter { private readonly pending: PendingRequests; private readonly reconnectLoop: ReconnectLoop | undefined; private readonly timers: LinkTimers; + private readonly transport: PduTransport; private readonly window: SendWindow; private closed = false; private concatReference = 0; private draining = false; - private framer = new PduFramer(); + private ended = false; /** A listener that throws is the application's bug; it must not become ours. Hard rule 1. */ override emit( @@ -126,7 +125,6 @@ export class Session extends EventEmitter { }); this.pending = new PendingRequests(this.log); this.reconnectLoop = this.loopFor(options.reconnect); - this.sock = options.sock; this.timers = new LinkTimers({ enquireLinkInterval: options.enquireLinkInterval, idleTimeout: options.idleTimeout, @@ -135,12 +133,18 @@ export class Session extends EventEmitter { // Not close(): a link that went quiet is a drop, and a drop is what reconnect is for. onIdle: () => { this.teardown(); }, }); + this.transport = this.transportFor(options.sock); this.window = new SendWindow(options.maxOutstanding ?? defaults.maxOutstanding); this.attach(options.sock); this.resetTimers(); } + /** Replaced on reconnect, so hold the session rather than this. */ + get sock(): Socket { + return this.transport.sock; + } + /** Whether this session's bind direction carries a command. Consulted by the library's senders. */ bindAllows(cmdName: string): boolean { return bindCarries(this.boundAs, cmdName); @@ -187,7 +191,7 @@ export class Session extends EventEmitter { tlvs?: Record, ): Promise { const built = pduReturn(pdu, status, params, tlvs); - const sent = built.err ? { err: built.err } : this.write(built.buffer); + const sent = built.err ? { err: built.err } : this.transport.write(built.buffer); // A peer that unbinds and drops the link takes our response with it; that is not a failure. if (sent.err && !this.closed) { @@ -249,6 +253,21 @@ export class Session extends EventEmitter { return drained; } + private transportFor(sock: Socket): PduTransport { + return new PduTransport({ + log: this.log, + onClose: () => { this.onClose(); }, + onData: chunk => { this.onData(chunk); }, + onError: err => { this.emit('sessionError', err); }, + onFramed: pdu => { this.emit('incomingPdu', pdu); }, + onPdu: pduObj => { this.dispatch(pduObj); }, + onUnreadable: err => { + this.emit('sessionError', err); + this.end(); + }, + }, sock); + } + private loopFor(reconnect: ReconnectOptions | undefined): ReconnectLoop | undefined { if (!reconnect) return undefined; @@ -289,22 +308,9 @@ export class Session extends EventEmitter { return {}; } - /** Wires a freshly opened socket into this session, replacing any previous one. */ private attach(sock: Socket): void { - // The socket being replaced is already dead, and its three handlers still point here. - if (this.sock !== sock) this.sock.removeAllListeners(); - - this.sock = sock; - this.framer = new PduFramer(); + this.transport.attach(sock); this.closed = false; - - sock.on('data', chunk => { this.onData(chunk); }); - sock.on('close', () => { this.onClose(); }); - sock.on('error', err => { - this.log.warn('session - socket error', { message: err.message }); - this.emit('sessionError', err); - this.onClose(); - }); } private async request( @@ -325,7 +331,7 @@ export class Session extends EventEmitter { signal: options.signal, timeout: this.options.responseTimeout ?? defaults.responseTimeout, }); - const written = this.write(built.buffer); + const written = this.transport.write(built.buffer); if (written.err) { this.pending.settle(seqNr, { err: written.err }); @@ -366,6 +372,15 @@ export class Session extends EventEmitter { this.reconnectLoop?.stop(); this.teardown(); this.dlrMerger.clear(); + this.emitClose(); + } + + /** A session torn down by a drop the loop was retrying reaches here with nothing left to tear down. */ + private emitClose(): void { + if (this.ended) return; + + this.ended = true; + this.emit('close'); } private teardown(): void { @@ -376,10 +391,11 @@ export class Session extends EventEmitter { this.pending.settleAll(new Error('Session closed before a response arrived')); this.incoming.clear(); this.sock.destroy(); - this.emit(this.retrying() ? 'disconnected' : 'close'); + + if (this.retrying()) this.emit('disconnected'); + else this.emitClose(); } - /** A drop the loop will bring the session back from is a disconnect, not the end of it. */ private retrying(): boolean { return this.reconnectLoop !== undefined && !this.reconnectLoop.isStopped(); } @@ -390,55 +406,9 @@ export class Session extends EventEmitter { return this.concatReference; } - private write(pdu: Buffer): VoidResult { - if (this.sock.destroyed) { - return { err: new Error('Socket is closed') }; - } - - this.sock.write(pdu); - - return {}; - } - private onData(chunk: Buffer): void { this.emit('data', chunk); this.resetTimers(); - this.framer.push(chunk); - - const framed = this.framer.next(); - - if (framed.err) { - this.log.warn('session - unusable stream, closing', { message: framed.err.message }); - this.emit('sessionError', framed.err); - this.end(); - - return; - } - - for (const pdu of framed.pdus) { - if (!this.receive(pdu)) return; - } - } - - /** False means the PDU could not be read and the session has been closed. */ - private receive(pdu: Buffer): boolean { - this.emit('incomingPdu', pdu); - - const parsed = pduToObj(pdu); - - if (parsed.err) { - this.log.warn('session - could not parse an incoming PDU, closing', { - message: parsed.err.message, - }); - this.emit('sessionError', parsed.err); - this.end(); - - return false; - } - - this.dispatch(parsed.pduObj); - - return true; } private dispatch(pduObj: PduObject): void { diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index 611024e..72c10b3 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -365,22 +365,28 @@ describe('reconnect', () => { assert.deepEqual(events, ['disconnected', 'close']); }); - test('reports a drop as close when nothing will retry it', async t => { + test('emits close when the session ends while the link is still down', async t => { const smpp = await startServer(t); - const { session } = await connect(t, smpp, { reconnect: false }); + const { session } = await connect(t, smpp); assert.ok(session); const events: string[] = []; + session.on('close', () => { events.push('close'); }); session.on('disconnected', () => { events.push('disconnected'); }); - const closed = once(resolve => { session.on('close', () => { resolve(true); }); }); + const down = once(resolve => { session.on('disconnected', () => { resolve(true); }); }); await peerOf(smpp).close(); - await closed; + await down; + await session.close(); - assert.deepEqual(events, []); + assert.deepEqual(events, ['disconnected', 'close']); + + await session.close(); + + assert.deepEqual(events, ['disconnected', 'close'], 'closing twice is still one close'); }); test('refuses a backoff that would retry without pausing', () => { @@ -391,11 +397,17 @@ describe('reconnect', () => { /false/, 'off is spelled false, so nothing else may stand in for it', ); + assert.match( + checkSessionOptions({ reconnect: { maxDelay: 1000, minDelay: 30_000 } }).err?.message ?? '', + /maxDelay/, + 'a transposed pair asks to never retry faster than 30 s and gets one every second', + ); + assert.match(checkSessionOptions({ reconnect: { minDelayMs: 20 } }).err?.message ?? '', /minDelayMs/); assert.equal(checkSessionOptions({ reconnect: false }).err, undefined); assert.equal(checkSessionOptions({ reconnect: { maxDelay: 60_000, minDelay: 500 } }).err, undefined); }); - test('schedules nothing after a drop when reconnect is false', async t => { + test('reports a drop as close, and schedules nothing, when reconnect is false', async t => { const smpp = await startServer(t); const noop = (): void => undefined; const infos: string[] = []; @@ -410,11 +422,16 @@ describe('reconnect', () => { assert.ok(session); + let disconnects = 0; + + session.on('disconnected', () => { disconnects++; }); + const closed = once(resolve => { session.on('close', () => { resolve(true); }); }); await peerOf(smpp).close(); await closed; + assert.equal(disconnects, 0); assert.ok(!infos.includes('reconnect - retrying after a drop')); }); diff --git a/todo.md b/todo.md index 2f0d7d9..5049a92 100644 --- a/todo.md +++ b/todo.md @@ -125,9 +125,6 @@ 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` has one seam left in it**, 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 `reassembly`, `dlr-merger`, `send-window`, `link-timers`, `reconnect-loop`, `pending-requests` and `send-sms`, so the directory would make that boundary visible. Do it on the next