From 3bcea108e3ddfe55e605992b41fd9686692724fa Mon Sep 17 00:00:00 2001 From: Lilleman auf Larv Date: Tue, 1 Sep 2026 13:58:28 +0200 Subject: [PATCH] Wait out the messages the application holds before shutting a session down --- AGENTS.md | 17 ++++++++- README.md | 10 +++--- src/held-messages.ts | 29 +++++++++++++++ src/idle-waiters.ts | 47 ++++++++++++++++++++++++ src/incoming-requests.ts | 36 ++++++++++++++++--- src/send-sms.ts | 7 +++- src/send-window.ts | 38 ++++---------------- src/session.ts | 44 ++++++++++++----------- src/sms.ts | 17 ++++++--- src/udh.ts | 11 ++++++ test/session-extras.test.ts | 71 +++++++++++++++++++++++++++++++++++-- todo.md | 15 ++++---- 12 files changed, 263 insertions(+), 79 deletions(-) create mode 100644 src/held-messages.ts create mode 100644 src/idle-waiters.ts diff --git a/AGENTS.md b/AGENTS.md index 3151286..ad670c6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -72,6 +72,8 @@ src/ dlr-merger.ts DlrMerger: per-segment receipts counted into one MessageDlr error-from.ts errorFrom(): whatever was thrown or rejected, as an Error expiring-groups.ts ExpiringGroups: the capped, expiring store both of those share + held-messages.ts HeldMessages: the messages handed to the application and not yet answered + idle-waiters.ts IdleWaiters: waiting for a count to fall to zero, and what is left of a budget incoming-requests.ts Every request the peer sends: messages, receipts, links, unknown commands link-gate.ts LinkGate: where a request with no link to go out on waits for the next one link-timers.ts LinkTimers: the enquire_link heartbeat and the idle timeout @@ -88,7 +90,7 @@ src/ send-window.ts SendWindow: the maxOutstanding semaphore session-options.ts SessionOptions, ReconnectOptions, bind direction and the session defaults sms-id.ts The notation a peer writes message ids in, normalised for comparison - udh.ts User data header: the concatenation fields of a long SMS + udh.ts User data header: the concatenation fields of a long SMS, and their reference uuid.ts uuidv7() — the ids the library generates for messages defs/ commands.ts The 33 commands, their ids and ordered parameter lists @@ -331,6 +333,19 @@ Grouped by what each one constrains. reports each session's unfinished drain through `serverError`, because its own result says nothing but that the listener stopped. +- **The drain waits on the messages the application holds, and `sendResp()` is what says it is done + with one.** Maintainer's call, 2026-09-01: waiting on the send window alone tore a server session + down while the application was still answering a `submit_sm`, so the peer timed out and re-sent — + the duplicate goal 2 forbids, in the direction the window already covers. No completion signal was + added to the `sms` event: `sendResp()` is the answer the peer is waiting for, so it is the one the + drain waits for. Counting every inbound request until `sendReturn()` answered it was rejected — + an `onRequest` that deliberately answers nothing would then cost a full `shutdownTimeout` on every + close — and a message no listener took is released at once, since nothing is going to answer it. + `teardown()` drops what is still held for the same reason it drops inbound segments. The release + is one turn late, so a listener that sends its receipt straight after the response is still + holding when the drain looks; `sendDlr()` is the one send that goes out past the drain's refusal, + being part of answering a message the drain is itself waiting for. + - **A reconnect keeps the delivery-receipt merges; everything else the link held is dropped.** `onDeliverSm()` answers each receipt before the group it belongs to is complete, and `teardown()` runs on every path — an idle timeout and a failed rebind, not only `close()` — so clearing the diff --git a/README.md b/README.md index 92778c0..559b05c 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 this end already sent, 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 and the messages the application has not answered yet, so neither end has to guess whether a message got through. | | **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, and re-bind unless `reconnect` is `false`. | | `responseTimeout` | `30000` | How long to wait for a response before giving up on it, and how long a send with no link waits for the next one; `0` waits forever. | -| `shutdownTimeout` | `5000` | How long `close()` and `unbind()` wait for the requests this end already sent; `0` waits forever. | +| `shutdownTimeout` | `5000` | How long `close()` and `unbind()` wait for the requests this end already sent and the messages the application has not answered; `0` waits forever. | | `maxOutstanding` | `10` | Requests allowed on the wire at once; further sends queue. | | `smsIdFormat` | — | The notation the SMSC writes message ids in, per place it writes them: `{ receipt: 'decimal', submitResp: 'hex' }`. Only needed where the two disagree. | | `reconnect` | on | Re-binds after a drop, an idle timeout, or a stream the library cannot read, backing off from `minDelay` 1 s to `maxDelay` 30 s and starting over at `minDelay` once a link has lasted `maxDelay`. `{ minDelay, maxDelay }` retunes it; `false` turns it off, so a drop ends the session. | @@ -331,8 +331,10 @@ TypeScript users can import `SmppLog` to have the compiler check one. `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 +then tear down whatever is left, resolving to an `err` that says what was lost. They wait on the +`sms` events the application has not answered yet too, so a peer whose `submit_sm` is still being +handled is answered rather than left to re-send it; `sendDlr()` goes out during that wait, and every +other send is refused. `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: diff --git a/src/held-messages.ts b/src/held-messages.ts new file mode 100644 index 0000000..3a12be4 --- /dev/null +++ b/src/held-messages.ts @@ -0,0 +1,29 @@ +import type { PduObject } from './pdu.ts'; +import { IdleWaiters } from './idle-waiters.ts'; + +/** The messages handed to the application that it has not answered yet, held by their segments. */ +export class HeldMessages { + private readonly held = new Set(); + private readonly idleWaiters = new IdleWaiters(); + + hold(pduObjs: PduObject[]): void { + this.held.add(pduObjs); + } + + release(pduObjs: PduObject[]): void { + if (!this.held.delete(pduObjs)) return; + + if (this.held.size === 0) this.idleWaiters.settle(); + } + + /** Drops every message: their segments went with the link, so no answer of ours correlates now. */ + clear(): void { + this.held.clear(); + this.idleWaiters.settle(); + } + + /** Resolves 0 once every message has been answered, or with how many have not. */ + idle(timeout: number, signal?: AbortSignal): Promise { + return this.idleWaiters.wait(() => this.held.size, timeout, signal); + } +} diff --git a/src/idle-waiters.ts b/src/idle-waiters.ts new file mode 100644 index 0000000..50c9311 --- /dev/null +++ b/src/idle-waiters.ts @@ -0,0 +1,47 @@ +/** What is left of a budget, in the shape a wait takes it: 0 waits forever. */ +export function leftOf(deadline: number): number { + return deadline === 0 ? 0 : Math.max(1, deadline - Date.now()); +} + +/** Everything waiting for a count to fall to zero, and how such a wait is cut short. */ +export class IdleWaiters { + private readonly waiting: (() => void)[] = []; + + /** Wakes everything waiting, whatever the count reads now. */ + settle(): void { + for (const resolve of this.waiting.splice(0)) { + resolve(); + } + } + + /** + * 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. + */ + wait(remaining: () => number, timeout: number, signal?: AbortSignal): Promise { + if (remaining() === 0) return Promise.resolve(0); + + if (signal?.aborted === true) return Promise.resolve(remaining()); + + return new Promise(resolve => { + let timer: NodeJS.Timeout | undefined = undefined; + const done = (): void => { + const index = this.waiting.indexOf(done); + + if (timer) clearTimeout(timer); + if (index !== -1) this.waiting.splice(index, 1); + + signal?.removeEventListener('abort', done); + resolve(remaining()); + }; + + if (timeout > 0) { + timer = setTimeout(done, timeout); + timer.unref(); + } + + signal?.addEventListener('abort', done, { once: true }); + this.waiting.push(done); + }); + } +} diff --git a/src/incoming-requests.ts b/src/incoming-requests.ts index e81898a..2f28d1d 100644 --- a/src/incoming-requests.ts +++ b/src/incoming-requests.ts @@ -1,9 +1,11 @@ import type { DlrMerger } from './dlr-merger.ts'; import type { OnRequest } from './session-options.ts'; -import type { PduObject } from './pdu.ts'; +import type { PduObject, PduObjectInput } from './pdu.ts'; +import type { Result, VoidResult } from './result.ts'; import type { Session } from './session.ts'; import type { SmppLog } from './log.ts'; import type { SmsIdFormat } from './sms-id.ts'; +import { HeldMessages } from './held-messages.ts'; import { Reassembler, decodeSegments } from './reassembly.ts'; import { bindCommands, defaults } from './session-options.ts'; import { concatInfo } from './udh.ts'; @@ -19,6 +21,8 @@ export type IncomingRequestsOptions = { maxReassembly?: number | undefined; onRequest?: OnRequest | undefined; reassemblyTimeout?: number | undefined; + /** Past the drain gate: a receipt answering a message the shutdown is still waiting for. */ + sendHeld: (input: PduObjectInput) => Promise>; session: Session; smsIdFormat?: SmsIdFormat | undefined; systemId?: string | undefined; @@ -27,9 +31,11 @@ export type IncomingRequestsOptions = { /** Everything the peer asks of a session: messages, receipts, links and the answers to them. */ export class IncomingRequests { private readonly dlrMerger: DlrMerger; + private readonly held = new HeldMessages(); private readonly log: SmppLog; private readonly onRequest: OnRequest | undefined; private readonly reassembler: Reassembler; + private readonly sendHeld: IncomingRequestsOptions['sendHeld']; private readonly session: Session; private readonly smsIdFormat: SmsIdFormat; private readonly systemId: string; @@ -44,6 +50,7 @@ export class IncomingRequests { maxOctets: options.maxOctets, timeout: options.reassemblyTimeout ?? defaults.reassemblyTimeout, }); + this.sendHeld = options.sendHeld; this.session = options.session; this.smsIdFormat = options.smsIdFormat ?? {}; this.systemId = options.systemId ?? defaults.systemId; @@ -82,11 +89,23 @@ export class IncomingRequests { } } - /** Drops the segments of every message that never became whole. */ + /** Drops the segments of every message that never became whole, and of every one still held. */ clear(): void { + this.held.clear(); this.reassembler.clear(); } + /** Waits out the messages the application still holds, and says how many it never answered. */ + async drain(timeout: number, signal?: AbortSignal): Promise { + const unanswered = await this.held.idle(timeout, signal); + + if (unanswered === 0) return {}; + + this.log.warn('session - shutting down with messages unanswered', { timeout, unanswered }); + + return { err: new Error(`Shut down with ${String(unanswered)} message(s) unanswered`) }; + } + private async unhandled(pduObj: PduObject): Promise { if (bindCommands.includes(pduObj.cmdName)) { this.log.info('session - bind on an already bound session', { cmdName: pduObj.cmdName }); @@ -139,12 +158,21 @@ export class IncomingRequests { if (!first) return; - this.session.emit('sms', createSms({ + const sms = createSms({ from: paramText(first.params.source_addr), message: decodeSegments(pduObjs), pduObjs, session: this.session, to: paramText(first.params.destination_addr), - })); + }, { + // A turn later, so a listener sending its receipt straight after the response still holds. + onAnswered: () => { setImmediate(() => { this.held.release(pduObjs); }); }, + send: this.sendHeld, + }); + + this.held.hold(pduObjs); + + // A message nobody took is not work a shutdown can wait for. + if (!this.session.emit('sms', sms)) this.held.release(pduObjs); } } diff --git a/src/send-sms.ts b/src/send-sms.ts index d329846..3277330 100644 --- a/src/send-sms.ts +++ b/src/send-sms.ts @@ -37,6 +37,11 @@ export type SendSmsResult = { unanswered: number; }; +/** A message that never went out, in the shape a caller aggregating segments still reads. */ +export function unsent(err: Error): SendSmsResult { + return { err, pduObjs: [], smsIds: [], unanswered: 0 }; +} + /** What sending needs from the session: a concat reference and a way onto the wire. */ export type SendSmsDeps = { log: SmppLog; @@ -141,7 +146,7 @@ export async function submitSms(deps: SendSmsDeps, sms: SendSmsOptions): Promise const segments = splitMessage(sms.message, { encoding, reference: deps.reference }); const refused = checkSegments(allowed, segments.length); - if (refused) return { err: refused, pduObjs: [], smsIds: [], unanswered: 0 }; + if (refused) return unsent(refused); const multipart = segments.length > 1; diff --git a/src/send-window.ts b/src/send-window.ts index e1492ca..46560c3 100644 --- a/src/send-window.ts +++ b/src/send-window.ts @@ -1,8 +1,10 @@ +import { IdleWaiters } from './idle-waiters.ts'; + /** Caps how many requests are on the wire at once; anything past the limit waits its turn. */ export class SendWindow { + private readonly idleWaiters = new IdleWaiters(); private readonly limit: number; private readonly waiting: (() => void)[] = []; - private readonly waitingForIdle: (() => void)[] = []; private inFlight = 0; constructor(limit: number) { @@ -32,9 +34,7 @@ export class SendWindow { if (this.inFlight > 0) return; - for (const resolve of this.waitingForIdle.splice(0)) { - resolve(); - } + this.idleWaiters.settle(); } /** Everything the caller is still owed: on the wire, plus queued behind a full window. */ @@ -42,34 +42,8 @@ export class SendWindow { 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. - */ + /** Resolves 0 once nothing is left on the wire, or with what still is. */ 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 => { - const index = this.waitingForIdle.indexOf(done); - - if (timer) clearTimeout(timer); - if (index !== -1) this.waitingForIdle.splice(index, 1); - - signal?.removeEventListener('abort', done); - resolve(this.unfinished()); - }; - - if (timeout > 0) { - timer = setTimeout(done, timeout); - timer.unref(); - } - - signal?.addEventListener('abort', done, { once: true }); - this.waitingForIdle.push(done); - }); + return this.idleWaiters.wait(() => this.unfinished(), timeout, signal); } } diff --git a/src/session.ts b/src/session.ts index 247d62f..b297233 100644 --- a/src/session.ts +++ b/src/session.ts @@ -16,12 +16,14 @@ import { PduTransport } from './pdu-transport.ts'; import { PendingRequests, UnansweredError } from './pending-requests.ts'; import { ReconnectLoop } from './reconnect-loop.ts'; import { SendWindow } from './send-window.ts'; +import { leftOf } from './idle-waiters.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 } from './pdu.ts'; import { silentLog } from './log.ts'; -import { submitSms } from './send-sms.ts'; +import { submitSms, unsent } from './send-sms.ts'; +import { ConcatReference } from './udh.ts'; export type { CloseOptions, @@ -64,6 +66,7 @@ export class Session extends EventEmitter { peerInterfaceVersion: number | undefined = undefined; userData: unknown = undefined; + private readonly concatReference = new ConcatReference(); private readonly dlrMerger: DlrMerger; private readonly gate: LinkGate; private readonly incoming: IncomingRequests; @@ -75,7 +78,6 @@ export class Session extends EventEmitter { private readonly window: SendWindow; private closed = false; - private concatReference = 0; private draining = false; private ended = false; @@ -129,6 +131,7 @@ export class Session extends EventEmitter { maxReassembly: options.maxReassembly, onRequest: options.onRequest, reassemblyTimeout: options.reassemblyTimeout, + sendHeld: input => this.sendThrough(input, {}), session: this, smsIdFormat: options.smsIdFormat, systemId: options.systemId, @@ -166,7 +169,15 @@ export class Session extends EventEmitter { } /** Sends a request and resolves with the peer's response. */ - async send( + send(input: PduObjectInput, options: SendOptions = {}): Promise> { + // A drain on a live link. A link that is down is the gate's answer, which says closed instead. + if (this.draining && !this.linkDown()) return Promise.resolve({ err: new Error('Session is shutting down') }); + + return this.sendThrough(input, options); + } + + /** The same path without that refusal, which a receipt for a held message has to take. */ + private async sendThrough( input: PduObjectInput, options: SendOptions = {}, ): Promise> { @@ -199,9 +210,6 @@ export class Session extends EventEmitter { return new Error(`Use sendReturn() for responses, not send(): ${input.cmdName}`); } - // A drain on a live link. A link that is down is the gate's answer, which says closed instead. - if (this.draining && !this.linkDown()) return new Error('Session is shutting down'); - // Before the gate and the window, or an aborted call waits for what it will never use. if (options.signal?.aborted === true) return abortedBeforeSend(); @@ -239,17 +247,12 @@ export class Session extends EventEmitter { async sendSms(sms: SendSmsOptions, options: SendOptions = {}): Promise { if (!this.bindAllows('submit_sm')) { - return { - err: new Error('A receiver-bound session does not carry submit_sm'), - pduObjs: [], - smsIds: [], - unanswered: 0, - }; + return unsent(new Error('A receiver-bound session does not carry submit_sm')); } const sent = await submitSms({ log: this.log, - reference: this.nextConcatReference(), + reference: this.concatReference.next(), respIdNotation: this.options.smsIdFormat?.submitResp, send: input => this.send(input, options), }, sms); @@ -379,7 +382,7 @@ export class Session extends EventEmitter { return { result: answered.err ? { err: new UnansweredError(answered.err) } : answered, retryOnNextLink: false }; } - /** Stops new sends and waits out the ones already issued. */ + /** Stops new sends and waits out the messages we hold and the requests already issued. */ private async drain(signal: AbortSignal | undefined): Promise { this.reconnectLoop?.stop(); this.draining = true; @@ -387,11 +390,16 @@ export class Session extends EventEmitter { if (this.linkDown()) return {}; const timeout = this.options.shutdownTimeout ?? defaults.shutdownTimeout; - const unfinished = await this.window.idle(timeout, signal); + const deadline = timeout > 0 ? Date.now() + timeout : 0; + // Answering a message can put a receipt on the wire; nothing on the wire produces a message. + const messages = await this.incoming.drain(timeout, signal); + const unfinished = await this.window.idle(leftOf(deadline), signal); // The window empties on a teardown too, which settles everything the link was carrying. if (this.linkDown()) return { err: new Error('The session closed before the drain finished') }; + if (messages.err) return messages; + if (unfinished === 0) return {}; this.log.warn('session - shutting down with requests unfinished', { timeout, unfinished }); @@ -433,12 +441,6 @@ export class Session extends EventEmitter { return this.reconnectLoop !== undefined && !this.reconnectLoop.isStopped(); } - private nextConcatReference(): number { - this.concatReference = this.concatReference >= 255 ? 1 : this.concatReference + 1; - - return this.concatReference; - } - private onData(chunk: Buffer): void { this.emit('data', chunk); this.resetTimers(); diff --git a/src/sms.ts b/src/sms.ts index cefdcf9..be341f2 100644 --- a/src/sms.ts +++ b/src/sms.ts @@ -1,6 +1,6 @@ import type { ErrorName } from './defs/errors.ts'; import type { MessageState } from './defs/constants.ts'; -import type { PduObject, TlvInput } from './pdu.ts'; +import type { PduObject, PduObjectInput, TlvInput } from './pdu.ts'; import type { Result, VoidResult } from './result.ts'; import type { Session } from './session.ts'; import { consts } from './defs/constants.ts'; @@ -43,12 +43,18 @@ export type SmsInput = { to: string; }; +/** What the session's incoming side gives a message so it can be answered and accounted for. */ +export type SmsHandlers = { + onAnswered: () => void; + send: (input: PduObjectInput) => Promise>; +}; + /** Each segment of a multipart message gets its own message_id, as a separate submit_sm must. */ function segmentId(smsId: string, index: number, total: number): string { return total === 1 ? smsId : `${smsId}-${String(index + 1)}`; } -export function createSms(input: SmsInput): Sms { +export function createSms(input: SmsInput, handlers: SmsHandlers): Sms { const first = input.pduObjs[0]; const registered = first?.params.registered_delivery; const dataCoding = first?.params.data_coding; @@ -60,8 +66,8 @@ export function createSms(input: SmsInput): Sms { from: input.from, message: input.message, pduObjs: input.pduObjs, - sendDlr: status => sendDlr(sms, status), - sendResp: options => sendResp(sms, answered, options ?? {}), + sendDlr: status => sendDlr(sms, handlers.send, status), + sendResp: options => sendResp(sms, answered, options ?? {}).finally(handlers.onAnswered), session: input.session, get smsId(): string { return answered.smsId; @@ -124,6 +130,7 @@ function receiptTlvs(smsId: string, status: MessageState): Record> { if (!sms.session.bindAllows('deliver_sm')) { @@ -135,7 +142,7 @@ async function sendDlr( for (let index = 0; index < total; index++) { const smsId = segmentId(sms.smsId, index, total); - const sent = await sms.session.send({ + const sent = await send({ cmdName: 'deliver_sm', params: { destination_addr: sms.from, diff --git a/src/udh.ts b/src/udh.ts index 37fba2d..1b6c399 100644 --- a/src/udh.ts +++ b/src/udh.ts @@ -1,3 +1,14 @@ +/** The 8-bit reference tying a long SMS's segments together, counted per session. */ +export class ConcatReference { + private current = 0; + + next(): number { + this.current = this.current >= 255 ? 1 : this.current + 1; + + return this.current; + } +} + export type ConcatInfo = { part: number; reference: number; diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index 4770767..ed6eedf 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -1037,8 +1037,12 @@ describe('AbortSignal on a send', () => { }); describe('graceful shutdown', () => { - async function submitInFlight(t: TestContext, options: Parameters[0] = {}) { - const smpp = await startServer(t); + async function submitInFlight( + t: TestContext, + options: Parameters[0] = {}, + serverOptions: Parameters[0] = {}, + ) { + const smpp = await startServer(t, serverOptions); const incoming = once(resolve => { smpp.on('session', bound => bound.on('sms', resolve)); }); @@ -1079,6 +1083,69 @@ describe('graceful shutdown', () => { assert.deepEqual(await unbound, {}); }); + test('close() waits for a message the application has not answered yet', async t => { + const { sent, smpp, sms } = await submitInFlight(t); + const closing = peerOf(smpp).close(); + + await delay(50); + await sms.sendResp({ smsId: 'answered-during-the-inbound-drain' }); + + assert.deepEqual(await closing, {}); + assert.deepEqual((await sent).smsIds, ['answered-during-the-inbound-drain']); + }); + + test('gives up on a message the application never answers', async t => { + const { smpp } = await submitInFlight(t, {}, { shutdownTimeout: 50 }); + const closed = await peerOf(smpp).close(); + + assert.ok(closed.err instanceof Error); + assert.match(closed.err.message, /1 message\(s\) unanswered/); + }); + + // The README's own listener answers and then sends its receipt, one turn later. + test('a receipt sent right after the response still goes out mid-drain', async t => { + const { sent, session, smpp, sms } = await submitInFlight(t); + const receipt = once(resolve => { session.on('dlr', resolve); }); + const closing = peerOf(smpp).close(); + + await sms.sendResp({ smsId: 'held-through-the-drain' }); + + const receiptSent = await sms.sendDlr('DELIVERED'); + + assert.equal(receiptSent.err, undefined); + assert.equal((await receipt).smsId, 'held-through-the-drain'); + assert.deepEqual(await closing, {}); + assert.deepEqual((await sent).smsIds, ['held-through-the-drain']); + }); + + test('a message no listener took does not hold the shutdown up', async t => { + const smpp = await startServer(t, { shutdownTimeout: 30_000 }); + const { session } = await connect(t, smpp); + + assert.ok(session); + + const bound = peerOf(smpp); + const arrived = once(resolve => { + bound.on('incomingPduObj', pduObj => { + if (pduObj.cmdName === 'submit_sm') resolve(pduObj); + }); + }); + const sent = session.sendSms({ + from: '46701113311', + message: 'nobody is listening', + to: '46709771337', + }); + + await arrived; + await delay(50); + + const started = Date.now(); + + assert.deepEqual(await bound.close(), {}); + assert.ok(Date.now() - started < 1000); + assert.ok((await sent).err instanceof Error); + }); + test('gives up on a request that outlasts shutdownTimeout', async t => { const { sent, session } = await submitInFlight(t, { shutdownTimeout: 50 }); const closed = await session.close(); diff --git a/todo.md b/todo.md index 8ef295f..2a62cb4 100644 --- a/todo.md +++ b/todo.md @@ -56,6 +56,7 @@ Rules the API follows: | Merged multipart DLRs including across a reconnect, reassembly bounds, per-send abort, the segment cap | `test/session-extras.test.ts` | | `smsIdFormat`: a peer's `submit_sm_resp` and receipt ids read into one notation before they are compared | `test/dlr.test.ts`, `test/session-extras.test.ts` | | A draining `close()` and `unbind()`, bounded by `shutdownTimeout` or an abort | `test/session-extras.test.ts` | +| A drain that also waits out the messages the application has not answered, with `sendDlr()` the one send that passes it | `test/session-extras.test.ts` | | A send with no link held for the next one, and one the link dropped under counted as `unanswered` | `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` | @@ -113,11 +114,6 @@ session message is a change to every call site. ## Worth doing, not blocking -- [ ] **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 @@ -125,13 +121,14 @@ session message is a change to every call site. - [ ] **Group the session's collaborators under `src/session/`.** Only `session.ts` imports `reassembly`, `dlr-merger`, `send-window`, `link-timers`, `link-gate`, `reconnect-loop`, `pending-requests` and `send-sms`, so the directory would make that boundary visible. - `pdu-transport` joined them on 2026-08-31 and `link-gate` on 2026-09-01, both without the move - being made, so it is a move of its own now. Do it together with the extraction below rather + `pdu-transport` joined them on 2026-08-31, `link-gate` and `idle-waiters` on 2026-09-01, all + without the move being made, so it is a move of its own now. Do it together with the extraction below rather than before it — three reactive splits at whatever boundary fitted under the line cap is what produced the current shape. Raised by review, 2026-09-01. - [ ] **An `OutgoingRequests` collaborator, owning `LinkGate`, `SendWindow` and `PendingRequests`.** - `session.ts` sits a few lines under its 350-line cap and every split so far has been made to - get back under it. The seam that holds: one object owning the gate, the window, the pending + `session.ts` sits three lines under its 350-line cap and every split so far has been made to + get back under it — the inbound drain on 2026-09-01 only fitted once `ConcatReference` and the + unsent-result shape moved out to `udh.ts` and `send-sms.ts`. The seam that holds: one object owning the gate, the window, the pending map and the retry loop, exposing a gated `request()` and the ungated door `unbind()` and the bind already need, with `Session` calling it when a link comes up or goes down instead of spreading `linkDown()` and `retrying()` across both sides. It would also close two smaller