From 8d9656b4e0f5ad4197f66c6ea26864643f364241 Mon Sep 17 00:00:00 2001 From: Lilleman auf Larv Date: Tue, 1 Sep 2026 08:02:37 +0200 Subject: [PATCH] Hold a send with no link for the next one, and count the rest as unanswered --- AGENTS.md | 19 +++- README.md | 15 ++- src/link-gate.ts | 85 +++++++++++++++++ src/pending-requests.ts | 13 ++- src/send-sms.ts | 16 +++- src/session.ts | 93 ++++++++++++------- test/readme.test.ts | 3 +- test/session-extras.test.ts | 179 +++++++++++++++++++++++++++++++++++- test/session.test.ts | 2 +- todo.md | 13 +-- 10 files changed, 386 insertions(+), 52 deletions(-) create mode 100644 src/link-gate.ts diff --git a/AGENTS.md b/AGENTS.md index 4e6633a..03f2087 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -45,6 +45,7 @@ src/ error-from.ts errorFrom(): whatever was thrown or rejected, as an Error expiring-groups.ts ExpiringGroups: the capped, expiring store both of those share 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 log.ts SmppLog, the logger contract, and silentLog — the default message.ts Encoding detection, splitting, bit counting, SMPP date formatting @@ -178,8 +179,8 @@ exactly 140. - **The published surface is frozen at what `src/index.ts` exports today.** `Session` is exported and publicly constructible, which is why `SessionOptions` and `ReconnectOptions` are public too — that is correct, not a leak, and it has been raised twice. The collaborators `session.ts` delegates to - (`Reassembler`, `PendingRequests`, `SendWindow`, `ReconnectLoop`, `LinkTimers`, `DlrMerger`, - `submitSms`) stay unpublished so they can be reshaped. + (`Reassembler`, `PendingRequests`, `SendWindow`, `ReconnectLoop`, `LinkTimers`, `LinkGate`, + `DlrMerger`, `submitSms`) stay unpublished so they can be reshaped. - **The sub-3.4 optional-parameter rule is a predicate, not a chokepoint.** `acceptsOptionalParams()` is consulted by the library's own senders; `session.send({ tlvs })` is passed through as written, because silently stripping a caller's explicit TLVs off a deliberately public low-level surface @@ -345,3 +346,17 @@ exactly 140. as no number and so reaches `expect()` and `collect()` unchanged. Normalising the base instead would break that pair. The option is on `client()` only — a `server()` session generates its own ids and writes its own receipts, so both places are already one notation. + +- **A send waits for the next link only if it never reached the socket; one that did is counted, not + resent.** Maintainer's call, 2026-09-01: re-queueing everything unanswered would resend a + `submit_sm` the SMSC accepted and answered into a dead socket, which is delivered and billed + twice, while a request that never left this process can be lost for free. `LinkGate` holds a send + that has no link and `comeBackUp()` opens it once the rebind is bound, so a send issued between + links and a segment still queued behind a full window when the drop hit both go out on the new + one. The hold is bounded by `responseTimeout` rather than an option of its own — that is already + the answer to how long one request may take — so the worst case is twice it: the hold, then the + answer. It happens before `window.acquire()` rather than inside it, because the rebind's own + `bind()` goes through `session.send()` and would deadlock behind slots held by waiting sends. + `teardown()` settles what was on the wire with `UnansweredError`, which `collectSent()` counts + into `SendSmsResult.unanswered`: `smsIds` alone cannot tell a message that never left from one + whose every segment the peer took and answered into a socket that was already gone. diff --git a/README.md b/README.md index 6f0ba2d..fdd0510 100644 --- a/README.md +++ b/README.md @@ -100,7 +100,7 @@ Every one is optional. | `tls` | `false` | `true` for defaults, or a `tls.ConnectionOptions` object for a private CA or a client certificate. | | `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; `0` waits forever. | +| `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. | | `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. | @@ -135,13 +135,16 @@ Messages too long for one SMS are split automatically and sent as a concatenated one id per segment: ```javascript -const { err, pduObjs, smsIds } = await session.sendSms({ from, message, to }); +const { err, pduObjs, smsIds, unanswered } = await session.sendSms({ from, message, to }); ``` `err` is set when the SMSC refuses a segment, and it names the status it refused with. Because every segment goes on the wire together, `pduObjs` and `smsIds` then hold what the SMSC did accept — enough to reconcile against a later receipt, not enough to resend the rest, so treat a partial failure as a -failed message. A message needing more than 255 segments is refused before anything is sent, since +failed message. `unanswered` counts the segments the link dropped under: the SMSC may have taken +each of them and lost only the response, so a message with `unanswered` above zero cannot be sent +again without risking a duplicate, however empty `smsIds` is. A message needing more than 255 +segments is refused before anything is sent, since the concatenation header numbers segments in a single octet. `maxSegments` lowers that ceiling: most handsets and SMSCs stop well short of 255, and refusing beats a message only half delivered. @@ -341,6 +344,12 @@ const { err, pduObj } = await session.send({ }); ``` +A send issued while the link is down waits for the reconnect instead of failing, and goes out on the +new link once it is bound — up to `responseTimeout`, after which it gives up having sent nothing. A +request already on the wire when the link drops is the other case: the SMSC may have taken it and +lost only the response, so it fails, and `sendSms()` counts it in `unanswered`. Neither happens with +`reconnect: false`, where a drop ends the session and every send after it is refused. + `acceptsOptionalParams()` answers whether the peer declared SMPP 3.4 or later, which is the version at and above which the spec allows optional parameters to be sent to it; `peerInterfaceVersion` is the version it declared, `0x00` if it declared none. The library's own senders consult the first before attaching a TLV — a diff --git a/src/link-gate.ts b/src/link-gate.ts new file mode 100644 index 0000000..d97293e --- /dev/null +++ b/src/link-gate.ts @@ -0,0 +1,85 @@ +import type { VoidResult } from './result.ts'; + +export type LinkGateOptions = { + /** Whether the link can carry nothing right now. */ + isDown: () => boolean; + /** How long a request may wait for a link. 0 waits for as long as one may still arrive. */ + timeout: number; + /** Whether a link that is down will be brought back. */ + willReturn: () => boolean; +}; + +type Waiter = (result: VoidResult) => void; + +function expired(): Error { + return new Error('The link did not come back in time'); +} + +/** Where a request with no link to go out on waits for the next one. */ +export class LinkGate { + private readonly options: LinkGateOptions; + private readonly waiting = new Set(); + + constructor(options: LinkGateOptions) { + this.options = options; + } + + /** When a hold starting now has to give up. 0 never does. */ + deadline(): number { + return this.options.timeout > 0 ? Date.now() + this.options.timeout : 0; + } + + /** Resolves once a link can carry the request, or with the reason none ever will. */ + wait(deadline: number, signal: AbortSignal | undefined): Promise { + if (!this.options.isDown()) return Promise.resolve({}); + + if (!this.options.willReturn()) return Promise.resolve({ err: new Error('Session is closed') }); + + const left = deadline === 0 ? 0 : deadline - Date.now(); + + if (deadline !== 0 && left <= 0) return Promise.resolve({ err: expired() }); + + return this.hold(left, signal); + } + + /** A link is up: everything held goes out on it. */ + open(): void { + this.release({}); + } + + /** No link is coming, and this is why. */ + shut(err: Error): void { + this.release({ err }); + } + + private hold(left: number, signal: AbortSignal | undefined): Promise { + return new Promise(resolve => { + let timer: NodeJS.Timeout | undefined = undefined; + const settle = (result: VoidResult): void => { + if (timer) clearTimeout(timer); + + signal?.removeEventListener('abort', onAbort); + this.waiting.delete(settle); + resolve(result); + }; + + function onAbort(): void { + settle({ err: new Error('Aborted while waiting for a link') }); + } + + if (left > 0) { + timer = setTimeout(() => { settle({ err: expired() }); }, left); + timer.unref(); + } + + signal?.addEventListener('abort', onAbort, { once: true }); + this.waiting.add(settle); + }); + } + + private release(result: VoidResult): void { + for (const settle of [...this.waiting]) { + settle(result); + } + } +} diff --git a/src/pending-requests.ts b/src/pending-requests.ts index 7949106..515a335 100644 --- a/src/pending-requests.ts +++ b/src/pending-requests.ts @@ -12,6 +12,14 @@ type Pending = { settle: (result: Result<{ pduObj: PduObject }>) => void; }; +/** The request went out and the link died before an answer: the peer may have accepted it. */ +export class UnansweredError extends Error { + constructor() { + super('The link dropped after the request went out; the peer may have accepted it'); + this.name = 'UnansweredError'; + } +} + /** Hands out sequence numbers and matches responses to the requests waiting for them. */ export class PendingRequests { private readonly log: SmppLog; @@ -69,9 +77,10 @@ export class PendingRequests { this.pending.get(seqNr)?.settle(result); } - settleAll(err: Error): void { + /** Everything still on the wire when the link died, each one possibly accepted by the peer. */ + settleAll(): void { for (const [seqNr] of this.pending) { - this.settle(seqNr, { err }); + this.settle(seqNr, { err: new UnansweredError() }); } } diff --git a/src/send-sms.ts b/src/send-sms.ts index 2fb7593..96689df 100644 --- a/src/send-sms.ts +++ b/src/send-sms.ts @@ -4,6 +4,7 @@ import type { PduObject, PduObjectInput } from './pdu.ts'; import type { Result } from './result.ts'; import type { SmppLog } from './log.ts'; import type { SmsIdNotation } from './sms-id.ts'; +import { UnansweredError } from './pending-requests.ts'; import { consts } from './defs/constants.ts'; import { detect } from './defs/encodings.ts'; import { normaliseSmsId } from './sms-id.ts'; @@ -28,7 +29,13 @@ export type SendSmsOptions = { }; /** Both arrays hold what the peer accepted, so a partial failure names what is already delivered. */ -export type SendSmsResult = { err?: Error; pduObjs: PduObject[]; smsIds: string[] }; +export type SendSmsResult = { + err?: Error; + pduObjs: PduObject[]; + smsIds: string[]; + /** Segments the link dropped under. The peer may have taken them, so sending again may duplicate. */ + unanswered: number; +}; /** What sending needs from the session: a concat reference and a way onto the wire. */ export type SendSmsDeps = { @@ -107,9 +114,12 @@ function collectSent( const pduObjs: PduObject[] = []; const smsIds: string[] = []; let failure: Error | undefined; + let unanswered = 0; for (const one of sent) { if (one.err) { + if (one.err instanceof UnansweredError) unanswered++; + failure ??= one.err; } else if (one.pduObj.cmdStatus === 'ESME_ROK') { pduObjs.push(one.pduObj); @@ -121,7 +131,7 @@ function collectSent( } } - return failure ? { err: failure, pduObjs, smsIds } : { pduObjs, smsIds }; + return failure ? { err: failure, pduObjs, smsIds, unanswered } : { pduObjs, smsIds, unanswered }; } /** Puts a message on the wire as one submit_sm per segment. */ @@ -131,7 +141,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: [] }; + if (refused) return { err: refused, pduObjs: [], smsIds: [], unanswered: 0 }; const multipart = segments.length > 1; diff --git a/src/session.ts b/src/session.ts index ab229f5..a9f4f80 100644 --- a/src/session.ts +++ b/src/session.ts @@ -10,6 +10,7 @@ import type { Socket } from 'node:net'; import { DlrMerger } from './dlr-merger.ts'; import { EventEmitter } from 'node:events'; import { IncomingRequests } from './incoming-requests.ts'; +import { LinkGate } from './link-gate.ts'; import { LinkTimers } from './link-timers.ts'; import { PduTransport } from './pdu-transport.ts'; import { PendingRequests } from './pending-requests.ts'; @@ -38,6 +39,9 @@ export { bindCommands, defaultSystemId }; /** A listener may return a promise: an `async` one that rejects is routed like one that throws. */ type SessionListener = (...args: SessionEvents[K]) => unknown; +/** `unsent` means nothing reached the socket, so the next link can still carry this request. */ +type Attempt = { result: Result<{ pduObj: PduObject }>; unsent: boolean }; + export class Session extends EventEmitter { declare addListener: (event: K, listener: SessionListener) => this; declare off: (event: K, listener: SessionListener) => this; @@ -57,6 +61,7 @@ export class Session extends EventEmitter { userData: unknown = undefined; private readonly dlrMerger: DlrMerger; + private readonly gate: LinkGate; private readonly incoming: IncomingRequests; private readonly options: SessionOptions; private readonly pending: PendingRequests; @@ -112,6 +117,11 @@ export class Session extends EventEmitter { max: defaults.maxDlrMerges, timeout: defaults.dlrMergeTimeout, }); + this.gate = new LinkGate({ + isDown: () => this.linkDown(), + timeout: options.responseTimeout ?? defaults.responseTimeout, + willReturn: () => this.retrying(), + }); this.incoming = new IncomingRequests({ dlrMerger: this.dlrMerger, log: this.log, @@ -160,26 +170,44 @@ export class Session extends EventEmitter { input: PduObjectInput, options: SendOptions = {}, ): Promise> { + const refused = this.refuseSend(input, options); + + if (refused) return { err: refused }; + + const deadline = this.gate.deadline(); + + for (;;) { + const held = await this.gate.wait(deadline, options.signal); + + if (held.err) return { err: held.err }; + + await this.window.acquire(); + + const attempt = await this.attempt(input, options).finally(() => { this.window.release(); }); + + // Nothing reached the socket, so the next link carries it instead of the caller resending. + if (!attempt.unsent || !this.linkDown() || !this.retrying()) return attempt.result; + } + } + + /** Why a request cannot go out at all, as opposed to not yet. */ + private refuseSend(input: PduObjectInput, options: SendOptions): Error | undefined { if (input.cmdName.endsWith('_resp')) { - return { err: new Error(`Use sendReturn() for responses, not send(): ${input.cmdName}`) }; + return new Error(`Use sendReturn() for responses, not send(): ${input.cmdName}`); } - if (this.closed) return { err: new Error('Session is closed') }; + // 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'); - if (this.draining) return { err: new Error('Session is shutting down') }; + // Before the gate and the window, or an aborted call waits for a link it will not use. + if (options.signal?.aborted === true) return new Error('Aborted before the request was sent'); - // Before the window, or a full window makes an aborted call wait for a slot it will not use. - if (options.signal?.aborted === true) { - return { err: new Error('Aborted before the request was sent') }; - } + return undefined; + } - await this.window.acquire(); - - try { - return await this.request(input, options); - } finally { - this.window.release(); - } + /** Read through a method: a drop can land while a send is awaiting. */ + private linkDown(): boolean { + return this.closed || this.sock.destroyed; } /** Answers a request the peer sent us. Responses are never waited on. */ @@ -207,7 +235,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: [] }; + return { + err: new Error('A receiver-bound session does not carry submit_sm'), + pduObjs: [], + smsIds: [], + unanswered: 0, + }; } const sent = await submitSms({ @@ -229,9 +262,9 @@ export class Session extends EventEmitter { async unbind(): Promise { const drained = await this.drain(undefined); const wasOpen = !this.closed; - // request(), not send(): the drain gate refuses a send, and the unbind goes out either way. + // attempt(), not send(): the drain gate refuses a send, and the unbind goes out either way. const sent = wasOpen - ? await this.request({ cmdName: 'unbind' }, {}) + ? (await this.attempt({ cmdName: 'unbind' }, {})).result : { err: new Error('Session is closed') }; const closedOnUnbind = wasOpen && this.closed; @@ -303,6 +336,7 @@ export class Session extends EventEmitter { this.resetTimers(); this.log.info('session - reconnected'); this.emit('reconnected'); + this.gate.open(); return {}; } @@ -312,19 +346,16 @@ export class Session extends EventEmitter { this.closed = false; } - private async request( - input: PduObjectInput, - options: SendOptions, - ): Promise> { + private async attempt(input: PduObjectInput, options: SendOptions): Promise { // pending.wait() alone settles the caller while the request still goes out to the peer. if (options.signal?.aborted === true) { - return { err: new Error('Aborted before the request was sent') }; + return { result: { err: new Error('Aborted before the request was sent') }, unsent: false }; } const seqNr = this.pending.nextSeqNr(); const built = objToPdu({ ...input, seqNr }); - if (built.err) return { err: built.err }; + if (built.err) return { result: { err: built.err }, unsent: false }; const response = this.pending.wait(seqNr, { signal: options.signal, @@ -335,10 +366,10 @@ export class Session extends EventEmitter { if (written.err) { this.pending.settle(seqNr, { err: written.err }); - return { err: written.err }; + return { result: { err: written.err }, unsent: true }; } - return response; + return { result: await response, unsent: false }; } /** Stops new sends and waits out the ones already issued. */ @@ -346,13 +377,13 @@ export class Session extends EventEmitter { this.reconnectLoop?.stop(); this.draining = true; - if (this.closed || this.sock.destroyed) return {}; + if (this.linkDown()) return {}; const timeout = this.options.shutdownTimeout ?? defaults.shutdownTimeout; const unfinished = await this.window.idle(timeout, signal); // 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 (this.linkDown()) return { err: new Error('The session closed before the drain finished') }; if (unfinished === 0) return {}; @@ -361,11 +392,6 @@ export class Session extends EventEmitter { return { err: new Error(`Shut down with ${String(unfinished)} request(s) unfinished`) }; } - /** Read through a method: teardown() can land while the drain is awaiting. */ - private isClosed(): boolean { - return this.closed; - } - /** The session is over now, drained or not. Nothing brings it back. */ private end(): void { this.reconnectLoop?.stop(); @@ -378,6 +404,7 @@ export class Session extends EventEmitter { if (this.ended) return; this.ended = true; + this.gate.shut(new Error('Session closed before the link came back')); this.emit('close'); } @@ -386,7 +413,7 @@ export class Session extends EventEmitter { this.closed = true; this.timers.clear(); - this.pending.settleAll(new Error('Session closed before a response arrived')); + this.pending.settleAll(); this.incoming.clear(); this.sock.destroy(); diff --git a/test/readme.test.ts b/test/readme.test.ts index 44794cb..d68ce28 100644 --- a/test/readme.test.ts +++ b/test/readme.test.ts @@ -83,7 +83,7 @@ describe('README: Client', () => { closeAfter(t, session); const reported = once(resolve => { session.on('dlr', resolve); }); - const { err: sendErr, smsIds } = await session.sendSms({ + const { err: sendErr, smsIds, unanswered } = await session.sendSms({ dlr: true, from: '46701113311', message: '«baff»', @@ -92,6 +92,7 @@ describe('README: Client', () => { assert.equal(sendErr, undefined); assert.equal(smsIds.length, 1); + assert.equal(unanswered, 0); assert.equal((await reported).smsId, smsIds[0]); }); diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index 3f0dae6..9030830 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -557,6 +557,182 @@ describe('reconnect', () => { }); }); +describe('sends across a reconnect', () => { + /** Answers every message after the first, which is left to hold the send window open. */ + function answerAfterTheFirst(smpp: SmppServer, arrived: string[]): Gate { + const first = gate(); + + smpp.on('session', peer => { + peer.on('sms', async sms => { + arrived.push(sms.message); + + if (arrived.length === 1) first.open(); + else await sms.sendResp(); + }); + }); + + return first; + } + + test('holds a send issued while the link is down and puts it on the new link', async t => { + const smpp = await startServer(t); + const arrived: string[] = []; + + smpp.on('session', peer => { peer.on('sms', async sms => { arrived.push(sms.message); await sms.sendResp(); }); }); + + const { session } = await connect(t, smpp, { reconnect: { maxDelay: 100, minDelay: 20 } }); + + assert.ok(session); + + const down = once(resolve => { session.on('disconnected', () => { resolve(true); }); }); + + await peerOf(smpp).close(); + await down; + + const sent = await session.sendSms({ from: '46701113311', message: 'held', to: '46709771337' }); + + assert.equal(sent.err, undefined); + assert.equal(sent.smsIds.length, 1); + assert.equal(sent.unanswered, 0); + assert.deepEqual(arrived, ['held']); + }); + + test('puts a segment still queued behind a full window on the new link', async t => { + const smpp = await startServer(t); + const arrived: string[] = []; + const first = answerAfterTheFirst(smpp, arrived); + const { session } = await connect(t, smpp, { + maxOutstanding: 1, + reconnect: { maxDelay: 100, minDelay: 20 }, + }); + + assert.ok(session); + + const holding = session.sendSms({ from: '46701113311', message: 'first', to: '46709771337' }); + + await first.passed; + + const queued = session.sendSms({ from: '46701113311', message: 'second', to: '46709771337' }); + + peerOf(smpp).sock.destroy(); + + const [dropped, resent] = await Promise.all([holding, queued]); + + assert.equal(dropped.unanswered, 1); + assert.equal(resent.err, undefined, 'a request that never reached the socket is not lost with it'); + assert.equal(resent.smsIds.length, 1); + assert.deepEqual(arrived, ['first', 'second']); + }); + + test('reports a segment the link dropped under as unanswered, not as never sent', async t => { + const smpp = await startServer(t); + const arrived = once(resolve => { smpp.on('session', peer => peer.on('sms', resolve)); }); + const { session } = await connect(t, smpp, { reconnect: false }); + + assert.ok(session); + + const sending = session.sendSms({ from: '46701113311', message: 'in flight', to: '46709771337' }); + + await arrived; + peerOf(smpp).sock.destroy(); + + const sent = await sending; + + assert.match(sent.err?.message ?? '', /may have accepted/); + assert.equal(sent.unanswered, 1, 'the peer may have accepted it, so sending it again would duplicate'); + assert.deepEqual(sent.smsIds, []); + }); + + test('gives up a held send after responseTimeout, with nothing put on the wire', async t => { + const smpp = await startServer(t); + const { session } = await connect(t, smpp, { + reconnect: { maxDelay: 10_000, minDelay: 10_000 }, + responseTimeout: 60, + }); + + assert.ok(session); + + const down = once(resolve => { session.on('disconnected', () => { resolve(true); }); }); + + await peerOf(smpp).close(); + await down; + + const sent = await session.sendSms({ from: '46701113311', message: 'held', to: '46709771337' }); + + assert.match(sent.err?.message ?? '', /did not come back/); + assert.equal(sent.unanswered, 0, 'nothing reached the peer, so the message can be sent again'); + }); + + test('fails a held send when the session closes rather than leaving it waiting', async t => { + const smpp = await startServer(t); + const { session } = await connect(t, smpp, { + reconnect: { maxDelay: 10_000, minDelay: 10_000 }, + }); + + assert.ok(session); + + const down = once(resolve => { session.on('disconnected', () => { resolve(true); }); }); + + await peerOf(smpp).close(); + await down; + + const sending = session.sendSms({ from: '46701113311', message: 'held', to: '46709771337' }); + let settled = false; + + void sending.then(() => { settled = true; }); + await delay(30); + + assert.equal(settled, false, 'the send waits for a link rather than failing on the spot'); + + await session.close(); + + const sent = await sending; + + assert.match(sent.err?.message ?? '', /closed/); + assert.equal(sent.unanswered, 0); + }); + + test('aborts a held send instead of making it wait out the link', async t => { + const smpp = await startServer(t); + const { session } = await connect(t, smpp, { + reconnect: { maxDelay: 10_000, minDelay: 10_000 }, + }); + + assert.ok(session); + + const down = once(resolve => { session.on('disconnected', () => { resolve(true); }); }); + + await peerOf(smpp).close(); + await down; + + const controller = new AbortController(); + const sending = session.sendSms( + { from: '46701113311', message: 'held', to: '46709771337' }, + { signal: controller.signal }, + ); + + controller.abort(); + + const sent = await sending; + + assert.match(sent.err?.message ?? '', /Aborted while waiting for a link/); + assert.equal(sent.unanswered, 0); + }); + + test('refuses a send outright once the session is over', async t => { + const smpp = await startServer(t); + const { session } = await connect(t, smpp, { reconnect: false }); + + assert.ok(session); + await session.close(); + + const sent = await session.sendSms({ from: '46701113311', message: 'too late', to: '46709771337' }); + + assert.match(sent.err?.message ?? '', /closed/); + assert.equal(sent.unanswered, 0); + }); +}); + describe('reassembly bounds', () => { function segment(reference: number, part: number, total: number): PduObject { const udh = Buffer.from([0x05, 0x00, 0x03, reference, total, part]); @@ -789,7 +965,8 @@ describe('graceful shutdown', () => { const result = await sent; assert.ok(result.err instanceof Error); - assert.equal(result.err.message, 'Session closed before a response arrived'); + assert.match(result.err.message, /may have accepted/); + assert.equal(result.unanswered, 1); }); // The window empties on a drop as well as on an answer, so it cannot be what the result reads. diff --git a/test/session.test.ts b/test/session.test.ts index 6f1084b..a8e382b 100644 --- a/test/session.test.ts +++ b/test/session.test.ts @@ -236,7 +236,7 @@ describe('bind', () => { const sent = await session.send({ cmdName: 'enquire_link' }); assert.ok(sent.err instanceof Error); - assert.equal(sent.err.message, 'Session closed before a response arrived'); + assert.match(sent.err.message, /may have accepted/); }); test('reports an unbind the peer left unanswered on a link that stays up', async t => { diff --git a/todo.md b/todo.md index b83bb50..bfe0147 100644 --- a/todo.md +++ b/todo.md @@ -57,6 +57,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 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` | | A listener that throws, or rejects, reaching `sessionError`/`serverError` rather than the process | `test/session.test.ts`, `test/error-from.test.ts` | @@ -113,9 +114,6 @@ session message is a change to every call site. ## Worth doing, not blocking -- [ ] **In-flight sends across a reconnect.** They currently fail with "Session closed before a - response arrived" and the caller retries. Re-queueing them automatically would be friendlier - but risks duplicate delivery, so it needs a decision before it is built. - [ ] **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 @@ -126,9 +124,12 @@ session message is a change to every call site. answered. Surviving one means exposing the merge state for the application to persist and hand 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. `pdu-transport` joined them - on 2026-08-31 without the move being made, so it is a move of its own now. + `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. `session.ts` sits within a couple of lines of the + 350-line cap again; the outgoing request path — `send()`, `refuseSend()`, `attempt()` and the + gate and window they drive — is the next thing that would come out of it. - [ ] **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