From b1c790b9a089cdfa3abc536a3b3c73d4a4a0fc5b Mon Sep 17 00:00:00 2001 From: Lilleman auf Larv Date: Mon, 31 Aug 2026 22:27:16 +0200 Subject: [PATCH] Retry a stream we cannot read, and cover the framer reset on attach MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A framing or codec error tears the link down rather than the session, so the loop retries it on a fresh socket with a fresh framer — which is what a desynced stream needs. Removing that reset failed nothing before; the reconnect test now leaves half a PDU on the dying link, and does. `disconnected` counts failed links rather than outages, which the README now says, and the transport wires its socket as it is built. --- AGENTS.md | 22 ++++++++++++++++++---- README.md | 2 +- src/pdu-transport.ts | 15 +++++++++------ src/session-options.ts | 8 ++++++-- src/session.ts | 4 +--- test/session-extras.test.ts | 25 +++++++++++++++++++++++++ todo.md | 4 ++-- 7 files changed, 62 insertions(+), 18 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index ce8c175..a6ecdfd 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -274,10 +274,10 @@ exactly 140. `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 and a peer's own `unbind` — nothing on a dead link can answer, an abort means stop - now, and a peer that has declared itself finished will not answer what it still owes us, so - draining any of the three would only hold a socket open for the timeout. `SmppServer.close()` + the framer or the codec cannot read takes `teardown()` instead, and `close({ signal })` on an + aborted signal and a peer's own `unbind` take `end()` — nothing on a dead link can answer, an abort + means stop now, and a peer that has declared itself finished will not answer what it still owes us, + so draining any of the three would only hold a socket open for the timeout. `SmppServer.close()` reports each session's unfinished drain through `serverError`, because its own result says nothing but that the listener stopped. `shutdownTimeout` stays a session option rather than a `close()` argument: `server()` builds sessions on the caller's behalf, so the option @@ -296,6 +296,20 @@ 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. +- **A stream this library cannot read is a dead link, not a dead session.** Maintainer's call, + 2026-08-31: a framing or codec error tears the link down through `teardown()`, so the reconnect + loop retries it on a fresh socket with a fresh framer — which is what a desynced stream needs, and + the common cause. `sessionError` still carries every failure, so a peer that only ever sends + garbage is visible in the log rather than silent, and the backoff caps the retries at one per + `maxDelay`. Ending the session outright was inherited from when reconnect was opt-in, where the + distinction could not arise. + +- **`disconnected` counts failed links, not outages.** A retry that opens a socket and then loses its + bind re-enters `attach()` and so emits again, which makes it deliberately not one-to-one with + `reconnected`: each emission is a link that went down, and suppressing the later ones would leave a + failed rebind with nothing but a log line. The README says so, because the pairing is what a reader + would otherwise assume. + - **`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 diff --git a/README.md b/README.md index 8e88f4f..3c9f8ea 100644 --- a/README.md +++ b/README.md @@ -317,7 +317,7 @@ TypeScript users can import `SmppLog` to have the compiler check one. | `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, 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. Fires again for each attempt that reconnects and then fails, so it is not one-to-one with `reconnected`. | | `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. | | `data` | Raw bytes arrived on the socket. | diff --git a/src/pdu-transport.ts b/src/pdu-transport.ts index 6ce41be..d27a823 100644 --- a/src/pdu-transport.ts +++ b/src/pdu-transport.ts @@ -14,7 +14,7 @@ export type PduTransportOptions = { /** 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. */ + /** Nothing further can be read off this stream; the socket has to go. */ onUnreadable: (err: Error) => void; }; @@ -27,20 +27,23 @@ export class PduTransport { constructor(options: PduTransportOptions, sock: Socket) { this.options = options; this.socket = sock; + this.wire(sock); } get sock(): Socket { return this.socket; } - /** Wires a freshly opened socket in, replacing any previous one. */ + /** Takes over a freshly opened socket. Half a PDU left on the old one must not prefix this 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.removeAllListeners(); this.socket = sock; this.framer = new PduFramer(); + this.wire(sock); + } + private wire(sock: Socket): void { sock.on('data', chunk => { this.read(chunk); }); sock.on('close', () => { this.options.onClose(); }); sock.on('error', err => { @@ -65,7 +68,7 @@ export class PduTransport { const framed = this.framer.next(); if (framed.err) { - this.options.log.warn('transport - unusable stream, closing', { message: framed.err.message }); + this.options.log.warn('transport - unusable stream', { message: framed.err.message }); this.options.onUnreadable(framed.err); return; @@ -77,7 +80,7 @@ export class PduTransport { const parsed = pduToObj(pdu); if (parsed.err) { - this.options.log.warn('transport - could not parse an incoming PDU, closing', { + this.options.log.warn('transport - could not parse an incoming PDU', { message: parsed.err.message, }); this.options.onUnreadable(parsed.err); diff --git a/src/session-options.ts b/src/session-options.ts index adc28e2..eb79360 100644 --- a/src/session-options.ts +++ b/src/session-options.ts @@ -163,9 +163,13 @@ function checkReconnect(reconnect: unknown): VoidResult { // 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; + if (checked.err) return checked; - return { err: new Error(`maxDelay must be minDelay or more, got ${String(maxDelay)}`) }; + if (maxDelay < minDelay) { + return { err: new Error(`maxDelay must be minDelay (${String(minDelay)}) or more, got ${String(maxDelay)}`) }; + } + + return {}; } function isRecord(value: unknown): value is Record { diff --git a/src/session.ts b/src/session.ts index 63768bc..ab229f5 100644 --- a/src/session.ts +++ b/src/session.ts @@ -136,7 +136,6 @@ export class Session extends EventEmitter { this.transport = this.transportFor(options.sock); this.window = new SendWindow(options.maxOutstanding ?? defaults.maxOutstanding); - this.attach(options.sock); this.resetTimers(); } @@ -263,7 +262,7 @@ export class Session extends EventEmitter { onPdu: pduObj => { this.dispatch(pduObj); }, onUnreadable: err => { this.emit('sessionError', err); - this.end(); + this.teardown(); }, }, sock); } @@ -375,7 +374,6 @@ export class Session extends EventEmitter { 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; diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index 72c10b3..d1da3af 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -389,6 +389,26 @@ describe('reconnect', () => { assert.deepEqual(events, ['disconnected', 'close'], 'closing twice is still one close'); }); + test('re-binds after a stream it cannot read, rather than ending the session', async t => { + const smpp = await startServer(t); + const { session } = await connect(t, smpp); + + assert.ok(session); + + const events: string[] = []; + + session.on('close', () => { events.push('close'); }); + session.on('sessionError', () => { events.push('sessionError'); }); + + const reconnected = once(resolve => { session.on('reconnected', () => { resolve(true); }); }); + + // A cmd_length below the 16-octet header is a stream no framing can recover from. + peerOf(smpp).sock.write(Buffer.from([0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 0, 0, 0, 0, 1])); + await reconnected; + + assert.deepEqual(events, ['sessionError']); + }); + test('refuses a backoff that would retry without pausing', () => { assert.match(checkSessionOptions({ reconnect: { minDelay: 0 } }).err?.message ?? '', /minDelay/); assert.match(checkSessionOptions({ reconnect: { maxDelay: -1 } }).err?.message ?? '', /maxDelay/); @@ -453,6 +473,11 @@ describe('reconnect', () => { assert.ok(session); const reconnected = once(resolve => { session.on('reconnected', () => { resolve(true); }); }); + const halfPdu = once(resolve => { session.on('data', () => { resolve(true); }); }); + + // A PDU header promising 32 octets and sending 8: the next link must not continue it. + peerOf(smpp).sock.write(Buffer.from([0, 0, 0, 32, 0, 0, 0, 4])); + await halfPdu; // Drop the connection from the server's side, as a peer restart would. for (const serverSession of smpp.sessions) { diff --git a/todo.md b/todo.md index 5049a92..b83bb50 100644 --- a/todo.md +++ b/todo.md @@ -127,8 +127,8 @@ session message is a change to every call site. back, which is a public-surface 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 - extraction out of `session.ts`, not as a move of its own. + and `send-sms`, so the directory would make that boundary visible. `pdu-transport` joined them + on 2026-08-31 without the move being made, so it is a move of its own now. - [ ] **Does an intermediate delivery notification deserve to be a `dlr`?** `esm_class` message type `INTERMEDIATE_DELIVERY` (0x20) is classified as a message today, so a peer that reports non-final states with it hands the application a raw `id:… stat:ENROUTE` text as an inbound