Emit exactly one close, and read the socket through a PduTransport

Ending a session while the link was down emitted nothing: the retried
drop had already torn it down, so close() found nothing left to do. The
terminal event is now guarded by its own flag rather than by `closed`.

The socket, the framer and the parse move to PduTransport, which is what
takes session.ts back under its line cap; `session.sock` reads through a
getter. The backoff also refuses an inverted pair and an unknown key.
This commit is contained in:
2026-08-31 21:53:11 +02:00
parent 880454b4f4
commit b2d121b4f3
7 changed files with 179 additions and 85 deletions
+8
View File
@@ -50,6 +50,7 @@ src/
message.ts Encoding detection, splitting, bit counting, SMPP date formatting message.ts Encoding detection, splitting, bit counting, SMPP date formatting
pdu.ts pduToObj / objToPdu / pduReturn — synchronous, result-returning pdu.ts pduToObj / objToPdu / pduReturn — synchronous, result-returning
pdu-framer.ts PduFramer: a byte stream cut into complete PDUs 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 pending-requests.ts PendingRequests: sequence numbers, correlation, timeout, abort
reassembly.ts Reassembler: capped, expiring multipart groups reassembly.ts Reassembler: capped, expiring multipart groups
reconnect-loop.ts ReconnectLoop: backoff, retry timer, stopped-ness 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 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. 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: - **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 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 application that never read the options table got none of it. `reconnect` takes
+1 -1
View File
@@ -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. | | `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`. | | `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 `<base>-<n>`, 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. | | `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 `<base>-<n>`, 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. | | `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. | | `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. | | `sessionError` | Something failed on a live session, including a hook or listener that threw or, if it was `async`, rejected. |
+91
View File
@@ -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);
}
}
}
+16 -5
View File
@@ -143,6 +143,8 @@ function checkLimits(limits: [string, number, number][]): VoidResult {
return {}; return {};
} }
const backoffDelays: readonly string[] = ['maxDelay', 'minDelay'];
function checkReconnect(reconnect: unknown): VoidResult { function checkReconnect(reconnect: unknown): VoidResult {
if (reconnect === undefined || reconnect === false) return {}; 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') }; 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. for (const key of Object.keys(reconnect)) {
return checkLimits([ if (!backoffDelays.includes(key)) {
['maxDelay', delayOr(reconnect.maxDelay, defaults.maxDelay), 1], return { err: new Error(`reconnect has no ${key}, name ${backoffDelays.join(' or ')}`) };
['minDelay', delayOr(reconnect.minDelay, defaults.minDelay), 1], }
]); }
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<string, unknown> { function isRecord(value: unknown): value is Record<string, unknown> {
+40 -70
View File
@@ -11,14 +11,14 @@ import { DlrMerger } from './dlr-merger.ts';
import { EventEmitter } from 'node:events'; import { EventEmitter } from 'node:events';
import { IncomingRequests } from './incoming-requests.ts'; import { IncomingRequests } from './incoming-requests.ts';
import { LinkTimers } from './link-timers.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 { PendingRequests } from './pending-requests.ts';
import { ReconnectLoop } from './reconnect-loop.ts'; import { ReconnectLoop } from './reconnect-loop.ts';
import { SendWindow } from './send-window.ts'; import { SendWindow } from './send-window.ts';
import { errorFrom } from './error-from.ts'; import { errorFrom } from './error-from.ts';
import { optionalParamsMinVersion } from './defs/constants.ts'; import { optionalParamsMinVersion } from './defs/constants.ts';
import { bindCarries, bindCommands, defaultSystemId, defaults } from './session-options.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 { silentLog } from './log.ts';
import { submitSms } from './send-sms.ts'; import { submitSms } from './send-sms.ts';
@@ -47,8 +47,6 @@ export class Session extends EventEmitter<SessionEvents> {
declare prependOnceListener: <K extends keyof SessionEvents>(event: K, listener: SessionListener<K>) => this; declare prependOnceListener: <K extends keyof SessionEvents>(event: K, listener: SessionListener<K>) => this;
declare removeListener: <K extends keyof SessionEvents>(event: K, listener: SessionListener<K>) => this; declare removeListener: <K extends keyof SessionEvents>(event: K, listener: SessionListener<K>) => this;
/** Replaced on reconnect, so hold the session rather than this. */
sock: Socket;
readonly log: SmppLog; readonly log: SmppLog;
/** The role the ESME bound with, whichever end of the link this is. Undefined before any bind. */ /** 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<SessionEvents> {
private readonly pending: PendingRequests; private readonly pending: PendingRequests;
private readonly reconnectLoop: ReconnectLoop | undefined; private readonly reconnectLoop: ReconnectLoop | undefined;
private readonly timers: LinkTimers; private readonly timers: LinkTimers;
private readonly transport: PduTransport;
private readonly window: SendWindow; private readonly window: SendWindow;
private closed = false; private closed = false;
private concatReference = 0; private concatReference = 0;
private draining = false; 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. */ /** A listener that throws is the application's bug; it must not become ours. Hard rule 1. */
override emit<K extends keyof SessionEvents>( override emit<K extends keyof SessionEvents>(
@@ -126,7 +125,6 @@ export class Session extends EventEmitter<SessionEvents> {
}); });
this.pending = new PendingRequests(this.log); this.pending = new PendingRequests(this.log);
this.reconnectLoop = this.loopFor(options.reconnect); this.reconnectLoop = this.loopFor(options.reconnect);
this.sock = options.sock;
this.timers = new LinkTimers({ this.timers = new LinkTimers({
enquireLinkInterval: options.enquireLinkInterval, enquireLinkInterval: options.enquireLinkInterval,
idleTimeout: options.idleTimeout, idleTimeout: options.idleTimeout,
@@ -135,12 +133,18 @@ export class Session extends EventEmitter<SessionEvents> {
// Not close(): a link that went quiet is a drop, and a drop is what reconnect is for. // Not close(): a link that went quiet is a drop, and a drop is what reconnect is for.
onIdle: () => { this.teardown(); }, onIdle: () => { this.teardown(); },
}); });
this.transport = this.transportFor(options.sock);
this.window = new SendWindow(options.maxOutstanding ?? defaults.maxOutstanding); this.window = new SendWindow(options.maxOutstanding ?? defaults.maxOutstanding);
this.attach(options.sock); this.attach(options.sock);
this.resetTimers(); 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. */ /** Whether this session's bind direction carries a command. Consulted by the library's senders. */
bindAllows(cmdName: string): boolean { bindAllows(cmdName: string): boolean {
return bindCarries(this.boundAs, cmdName); return bindCarries(this.boundAs, cmdName);
@@ -187,7 +191,7 @@ export class Session extends EventEmitter<SessionEvents> {
tlvs?: Record<string, TlvInput>, tlvs?: Record<string, TlvInput>,
): Promise<VoidResult> { ): Promise<VoidResult> {
const built = pduReturn(pdu, status, params, tlvs); 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. // A peer that unbinds and drops the link takes our response with it; that is not a failure.
if (sent.err && !this.closed) { if (sent.err && !this.closed) {
@@ -249,6 +253,21 @@ export class Session extends EventEmitter<SessionEvents> {
return drained; 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 { private loopFor(reconnect: ReconnectOptions | undefined): ReconnectLoop | undefined {
if (!reconnect) return undefined; if (!reconnect) return undefined;
@@ -289,22 +308,9 @@ export class Session extends EventEmitter<SessionEvents> {
return {}; return {};
} }
/** Wires a freshly opened socket into this session, replacing any previous one. */
private attach(sock: Socket): void { private attach(sock: Socket): void {
// The socket being replaced is already dead, and its three handlers still point here. this.transport.attach(sock);
if (this.sock !== sock) this.sock.removeAllListeners();
this.sock = sock;
this.framer = new PduFramer();
this.closed = false; 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( private async request(
@@ -325,7 +331,7 @@ export class Session extends EventEmitter<SessionEvents> {
signal: options.signal, signal: options.signal,
timeout: this.options.responseTimeout ?? defaults.responseTimeout, timeout: this.options.responseTimeout ?? defaults.responseTimeout,
}); });
const written = this.write(built.buffer); const written = this.transport.write(built.buffer);
if (written.err) { if (written.err) {
this.pending.settle(seqNr, { err: written.err }); this.pending.settle(seqNr, { err: written.err });
@@ -366,6 +372,15 @@ export class Session extends EventEmitter<SessionEvents> {
this.reconnectLoop?.stop(); this.reconnectLoop?.stop();
this.teardown(); this.teardown();
this.dlrMerger.clear(); 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 { private teardown(): void {
@@ -376,10 +391,11 @@ export class Session extends EventEmitter<SessionEvents> {
this.pending.settleAll(new Error('Session closed before a response arrived')); this.pending.settleAll(new Error('Session closed before a response arrived'));
this.incoming.clear(); this.incoming.clear();
this.sock.destroy(); 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 { private retrying(): boolean {
return this.reconnectLoop !== undefined && !this.reconnectLoop.isStopped(); return this.reconnectLoop !== undefined && !this.reconnectLoop.isStopped();
} }
@@ -390,55 +406,9 @@ export class Session extends EventEmitter<SessionEvents> {
return this.concatReference; 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 { private onData(chunk: Buffer): void {
this.emit('data', chunk); this.emit('data', chunk);
this.resetTimers(); 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 { private dispatch(pduObj: PduObject): void {
+23 -6
View File
@@ -365,22 +365,28 @@ describe('reconnect', () => {
assert.deepEqual(events, ['disconnected', 'close']); 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 smpp = await startServer(t);
const { session } = await connect(t, smpp, { reconnect: false }); const { session } = await connect(t, smpp);
assert.ok(session); assert.ok(session);
const events: string[] = []; const events: string[] = [];
session.on('close', () => { events.push('close'); });
session.on('disconnected', () => { events.push('disconnected'); }); session.on('disconnected', () => { events.push('disconnected'); });
const closed = once<true>(resolve => { session.on('close', () => { resolve(true); }); }); const down = once<true>(resolve => { session.on('disconnected', () => { resolve(true); }); });
await peerOf(smpp).close(); 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', () => { test('refuses a backoff that would retry without pausing', () => {
@@ -391,11 +397,17 @@ describe('reconnect', () => {
/false/, /false/,
'off is spelled false, so nothing else may stand in for it', '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: false }).err, undefined);
assert.equal(checkSessionOptions({ reconnect: { maxDelay: 60_000, minDelay: 500 } }).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 smpp = await startServer(t);
const noop = (): void => undefined; const noop = (): void => undefined;
const infos: string[] = []; const infos: string[] = [];
@@ -410,11 +422,16 @@ describe('reconnect', () => {
assert.ok(session); assert.ok(session);
let disconnects = 0;
session.on('disconnected', () => { disconnects++; });
const closed = once<true>(resolve => { session.on('close', () => { resolve(true); }); }); const closed = once<true>(resolve => { session.on('close', () => { resolve(true); }); });
await peerOf(smpp).close(); await peerOf(smpp).close();
await closed; await closed;
assert.equal(disconnects, 0);
assert.ok(!infos.includes('reconnect - retrying after a drop')); assert.ok(!infos.includes('reconnect - retrying after a drop'));
}); });
-3
View File
@@ -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 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 answered. Surviving one means exposing the merge state for the application to persist and hand
back, which is a public-surface decision. 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 - [ ] **Group the session's collaborators under `src/session/`.** Only `session.ts` imports
`reassembly`, `dlr-merger`, `send-window`, `link-timers`, `reconnect-loop`, `pending-requests` `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 and `send-sms`, so the directory would make that boundary visible. Do it on the next