diff --git a/AGENTS.md b/AGENTS.md index dabf376..837eb4a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -264,9 +264,12 @@ exactly 140. `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 + 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()` + 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 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 diff --git a/README.md b/README.md index 9ff772d..38dc3cd 100644 --- a/README.md +++ b/README.md @@ -309,13 +309,13 @@ 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 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: +`sendSms()`, `send()`, `sendReturn()`, `unbind()` and `close()`. Both `close()` and `unbind()` +refuse further sends, wait out the requests this end already sent for up to `shutdownTimeout`, 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. `close({ signal })` takes an +`AbortSignal` that cuts the wait short; `unbind()` takes none, and 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({ @@ -353,8 +353,9 @@ The spec tables are exported both individually (`cmds`, `consts`, `encodings`, ` ## Migrating from larvitsmpp 0.4.0 - **The package is now `@larvit/smpp`** and is ESM only. `require()` no longer works. -- **Callbacks are gone.** `client`, `server`, `sendSms`, `sendResp` and `sendDlr` are all promises - resolving to a result object with an optional `err`. Nothing rejects. +- **Callbacks are gone.** `client`, `server`, `sendSms`, `sendResp`, `sendDlr`, `unbind` and + `session.close` are all promises resolving to a result object with an optional `err`. Nothing + rejects. Await `close()` or the socket outlives the call. - **`server()` resolves once, when it is listening**, and gives you a handle with `close()`, `port` and a `session` event. It no longer calls your callback once per incoming connection. - **The id a message is answered with goes to `sendResp({ smsId })`**, and `sms.smsId` is read-only: diff --git a/src/incoming-requests.ts b/src/incoming-requests.ts index d121108..4bca232 100644 --- a/src/incoming-requests.ts +++ b/src/incoming-requests.ts @@ -70,7 +70,8 @@ export class IncomingRequests { break; case 'unbind': await this.session.sendReturn(pduObj); - await this.session.close(); + // A peer that has said it is finished will not answer what we still have outstanding. + await this.session.close({ signal: AbortSignal.abort() }); break; default: await this.unhandled(pduObj); diff --git a/src/server.ts b/src/server.ts index fab068b..6708575 100644 --- a/src/server.ts +++ b/src/server.ts @@ -118,7 +118,6 @@ export class SmppServer extends EventEmitter { /** 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]; diff --git a/src/session.ts b/src/session.ts index 249b8d2..7833aa6 100644 --- a/src/session.ts +++ b/src/session.ts @@ -273,6 +273,13 @@ export class Session extends EventEmitter { return { err: bound.err }; } + // close() can land while the rebind is in flight, and it has nothing left to tear down. + if (this.reconnectLoop?.isStopped() === true) { + this.teardown(); + + return { err: new Error('Session closed while it was coming back up') }; + } + this.resetTimers(); this.log.info('session - reconnected'); this.emit('reconnected'); @@ -327,10 +334,7 @@ export class Session extends EventEmitter { return response; } - /** - * 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. - */ + /** Stops new sends and waits out the ones already issued. */ private async drain(signal: AbortSignal | undefined): Promise { this.reconnectLoop?.stop(); this.draining = true; @@ -340,8 +344,8 @@ export class Session extends EventEmitter { const timeout = this.options.shutdownTimeout ?? defaults.shutdownTimeout; const unfinished = await this.window.idle(timeout, signal); - // 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') }; + // The window empties on a teardown too, which settles everything the link was carrying. + if (this.isClosed()) return { err: new Error('The session closed before the drain finished') }; if (unfinished === 0) return {}; diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index 85122d5..f43a67c 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -8,15 +8,16 @@ import type { MessageDlr } from '../src/session.ts'; import type { PduObject, PduObjectInput } from '../src/pdu.ts'; import type { Result } from '../src/result.ts'; import type { SendSmsResult } from '../src/send-sms.ts'; -import type { Session } from '../src/session.ts'; import type { SmppLog } from '../src/log.ts'; import type { Sms } from '../src/sms.ts'; import type { SmppServer } from '../src/server.ts'; import { Reassembler, decodeSegments } from '../src/reassembly.ts'; +import { Session } from '../src/session.ts'; import { DlrMerger } from '../src/dlr-merger.ts'; import { client } from '../src/client.ts'; import { consts } from '../src/defs/constants.ts'; import { errors } from '../src/defs/errors.ts'; +import { objToPdu } from '../src/pdu.ts'; import { server } from '../src/server.ts'; import { silentLog } from '../src/log.ts'; import { submitSms } from '../src/send-sms.ts'; @@ -44,6 +45,20 @@ function once(register: (resolve: (value: T) => void) => void): Promise { }); } +function delay(ms: number): Promise { + return new Promise(resolve => { setTimeout(resolve, ms); }); +} + +type Gate = { open: () => void; passed: Promise }; + +/** A promise the test opens by hand, guarded by once() against waiting on one it never does. */ +function gate(): Gate { + const opener: { open?: () => void } = {}; + const passed = once(resolve => { opener.open = () => { resolve(true); }; }); + + return { open: () => opener.open?.(), passed }; +} + describe('merged delivery reports', () => { // 0.4.0 allocated a longSmsDlrs store to do exactly this and then never used it. test('reports once on a whole multipart message', async () => { @@ -668,7 +683,7 @@ describe('graceful shutdown', () => { const closed = await closing; assert.ok(closed.err instanceof Error); - assert.match(closed.err.message, /dropped/); + assert.match(closed.err.message, /closed before the drain finished/); assert.ok((await sent).err instanceof Error); await smpp.close(); @@ -742,4 +757,82 @@ describe('graceful shutdown', () => { assert.ok((await unanswered).err instanceof Error); }); + + // Waiting on a peer that has just declared itself finished is dead time, unbounded at 0. + test('tears down at once when the peer unbinds rather than draining for it', async () => { + const smpp = await startServer({ shutdownTimeout: 0 }); + const arrived = once(resolve => { smpp.on('session', resolve); }); + const peer = net.connect({ port: smpp.port }); + + peer.resume(); + + const bound = await arrived; + const ended = once(resolve => { bound.on('close', () => { resolve(true); }); }); + const unanswered = bound.send({ cmdName: 'enquire_link' }); + const { buffer } = objToPdu({ cmdName: 'unbind', seqNr: 1 }); + + assert.ok(buffer); + peer.write(buffer); + + assert.ok(await ended); + assert.ok((await unanswered).err instanceof Error); + + peer.destroy(); + await smpp.close(); + }); + + test('does not report a reconnect on a session closed while it was coming back up', async () => { + const listener = net.createServer(sock => { sock.resume(); }); + + await new Promise(resolve => { listener.listen(0, resolve); }); + + const address = listener.address(); + const port = typeof address === 'object' && address !== null ? address.port : 0; + const opened: net.Socket[] = []; + const open = (): Promise> => new Promise(resolve => { + const sock = net.connect({ port }, () => { resolve({ sock }); }); + + opened.push(sock); + }); + const first = await open(); + + assert.ok(first.sock); + + const rebinding = gate(); + const release = gate(); + const session = new Session({ + reconnect: { + connect: open, + maxDelay: 20, + minDelay: 10, + onConnected: async () => { + rebinding.open(); + await release.passed; + + return {}; + }, + }, + sock: first.sock, + }); + const reported: string[] = []; + + session.on('reconnected', () => reported.push('reconnected')); + first.sock.destroy(); + + assert.ok(await rebinding.passed); + + const closing = session.close(); + + release.open(); + await closing; + await delay(50); + + assert.deepEqual(reported, []); + + for (const sock of opened) { + sock.destroy(); + } + + await new Promise(resolve => { listener.close(() => { resolve(); }); }); + }); }); diff --git a/todo.md b/todo.md index 2928f3a..48f14a7 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**: 246 tests, lint and typecheck clean, verified on Node +The rewrite is **feature complete and green**: 248 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 @@ -124,7 +124,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. -- [ ] **`session.ts` is 465 lines.** The one seam left in it is a socket-to-PDU transport, which +- [ ] **`session.ts` is 468 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