diff --git a/AGENTS.md b/AGENTS.md index 03f2087..a86b4cf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -351,12 +351,32 @@ exactly 140. 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. + that has no link, 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. Once a request has been written, every way it can + fail — the link dropping under it, `responseTimeout` expiring, the caller's own abort — means the + peer may have taken it, so `attempt()` wraps all three in `UnansweredError` and `collectSent()` + counts them into `SendSmsResult.unanswered`. Counting only the dropped-link case, as the first cut + did, would have called the commonest one safe to resend. A count rather than a boolean because + `sendSms()` aggregates segments into one `err` slot, and required rather than optional so every + construction site answers. `UnansweredError` stays unexported: `unanswered` is the one spelling on + the public surface, and a `send()` error that is neither a build failure nor a pre-write abort + means the same thing. + The hold is bounded by `responseTimeout` rather than an option of its own — that is already the + answer to how long one request may wait. It bounds the hold and the answer separately, and the + wait for a `maxOutstanding` slot is bounded by nothing, so `responseTimeout` is not a deadline for + the call; `SendOptions.signal` with `AbortSignal.timeout()` is, and both the gate and + `pending.wait()` honour it. + +- **The gate decides whether a link can carry a request, and a bind is what makes it one.** + Maintainer's call, 2026-09-01: `attach()` clears `closed` the moment a socket is handed over, one + round trip before the bind is answered, so gating on `closed` let a send arriving in that window + go out unbound and come back `ESME_RINVBNDSTS` while a send that arrived a millisecond earlier was + held correctly. `LinkGate` owns the answer instead — `shut(returning)` on every teardown, + `open()` only once `comeBackUp()` has a bound link — and `Session.linkDown()` reads it rather than + `closed`. The bind itself cannot wait for what it creates, so `send()` lets the three bind + commands past the gate and the window, the same door `unbind()` takes through `attempt()`. That + keeps the exemption a predicate on the command, like the `_resp` guard beside it, rather than a + second `send()` on the public surface or a changed `ReconnectOptions.onConnected`. + The gate is told what happened and never reads back into the session: a collaborator that has to + ask does not own its decision, which is how the first cut ended up answering the same question two + different ways at admit and at release. diff --git a/README.md b/README.md index fdd0510..92778c0 100644 --- a/README.md +++ b/README.md @@ -141,11 +141,11 @@ const { err, pduObjs, smsIds, unanswered } = await session.sendSms({ from, messa `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. `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: +failed message. `unanswered` counts the segments that went out and were never answered: 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. ### Receiving @@ -344,11 +344,16 @@ 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. +A send issued while the link is down waits for the reconnect instead of failing, and goes out once +the new link is bound — up to `responseTimeout`, after which it gives up having sent nothing. A +request already on the wire is the other case: the SMSC may have taken it and lost only the response, +so it fails, and `sendSms()` counts it in `unanswered`, whether the link dropped under it, the peer +never answered in time, or you aborted it after it went out. Neither applies with `reconnect: false`, +where a drop ends the session and every send after it is refused. + +`responseTimeout` bounds the wait for a link and the wait for an answer separately, and a send also +queues for a `maxOutstanding` slot, which nothing bounds — so it is not a deadline for the call. +Pass `{ signal: AbortSignal.timeout(ms) }` when you need one. `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 diff --git a/src/link-gate.ts b/src/link-gate.ts index d97293e..6e3bc9f 100644 --- a/src/link-gate.ts +++ b/src/link-gate.ts @@ -1,57 +1,81 @@ import type { VoidResult } from './result.ts'; export type LinkGateOptions = { - /** Whether the link can carry nothing right now. */ - isDown: () => boolean; + now?: (() => number) | undefined; /** 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 aborted(): Error { + return new Error('Aborted while waiting for a link'); +} + 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. */ +function over(): Error { + return new Error('Session is closed'); +} + +/** + * Where a request with no link to go out on waits for the next one. The owner reports what became of + * the link; nothing here reads back into the owner to find out. + */ export class LinkGate { - private readonly options: LinkGateOptions; + private readonly now: () => number; + private readonly timeout: number; private readonly waiting = new Set(); + private returning = false; + private up = true; constructor(options: LinkGateOptions) { - this.options = options; + this.now = options.now ?? Date.now; + this.timeout = options.timeout; + } + + /** Whether a request can go out right now. A link that is attached but not yet bound cannot. */ + isUp(): boolean { + return this.up; } /** 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; + return this.timeout > 0 ? this.now() + this.timeout : 0; + } + + /** A link is up and bound: everything held goes out on it. */ + open(): void { + this.up = true; + this.returning = false; + this.release({}); + } + + /** The link is gone. `returning` says whether another one is on its way. */ + shut(returning: boolean): void { + this.up = false; + this.returning = returning; + + if (!returning) this.release({ err: over() }); } /** 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.up) return Promise.resolve({}); - if (!this.options.willReturn()) return Promise.resolve({ err: new Error('Session is closed') }); + if (!this.returning) return Promise.resolve({ err: over() }); - const left = deadline === 0 ? 0 : deadline - Date.now(); + if (signal?.aborted === true) return Promise.resolve({ err: aborted() }); + + const left = deadline === 0 ? 0 : deadline - this.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; @@ -64,7 +88,7 @@ export class LinkGate { }; function onAbort(): void { - settle({ err: new Error('Aborted while waiting for a link') }); + settle({ err: aborted() }); } if (left > 0) { diff --git a/src/pending-requests.ts b/src/pending-requests.ts index 515a335..5fdbb1e 100644 --- a/src/pending-requests.ts +++ b/src/pending-requests.ts @@ -12,10 +12,10 @@ 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. */ +/** The request went out and no answer came back: 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'); + constructor(cause: Error) { + super(`No answer came back, so the peer may have accepted it: ${cause.message}`, { cause }); this.name = 'UnansweredError'; } } @@ -77,10 +77,9 @@ export class PendingRequests { this.pending.get(seqNr)?.settle(result); } - /** Everything still on the wire when the link died, each one possibly accepted by the peer. */ - settleAll(): void { + settleAll(err: Error): void { for (const [seqNr] of this.pending) { - this.settle(seqNr, { err: new UnansweredError() }); + this.settle(seqNr, { err }); } } diff --git a/src/session.ts b/src/session.ts index a9f4f80..6f6d65d 100644 --- a/src/session.ts +++ b/src/session.ts @@ -13,7 +13,7 @@ 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'; +import { PendingRequests, UnansweredError } from './pending-requests.ts'; import { ReconnectLoop } from './reconnect-loop.ts'; import { SendWindow } from './send-window.ts'; import { errorFrom } from './error-from.ts'; @@ -39,8 +39,12 @@ 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 }; +function abortedBeforeSend(): Error { + return new Error('Aborted before the request was sent'); +} + +/** `retryOnNextLink`: the write failed, so nothing reached the socket and another link may carry it. */ +type Attempt = { result: Result<{ pduObj: PduObject }>; retryOnNextLink: boolean }; export class Session extends EventEmitter { declare addListener: (event: K, listener: SessionListener) => this; @@ -117,11 +121,7 @@ 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.gate = new LinkGate({ timeout: options.responseTimeout ?? defaults.responseTimeout }); this.incoming = new IncomingRequests({ dlrMerger: this.dlrMerger, log: this.log, @@ -174,6 +174,9 @@ export class Session extends EventEmitter { if (refused) return { err: refused }; + // A bind is what makes a link usable, so it cannot wait for one. The door unbind() uses too. + if (bindCommands.includes(input.cmdName)) return (await this.attempt(input, options)).result; + const deadline = this.gate.deadline(); for (;;) { @@ -186,7 +189,7 @@ export class Session extends EventEmitter { 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; + if (!attempt.retryOnNextLink || !this.linkDown() || !this.retrying()) return attempt.result; } } @@ -199,15 +202,15 @@ export class Session extends EventEmitter { // 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 a link it will not use. - if (options.signal?.aborted === true) return new Error('Aborted before the request was sent'); + // Before the gate and the window, or an aborted call waits for what it will never use. + if (options.signal?.aborted === true) return abortedBeforeSend(); return undefined; } /** Read through a method: a drop can land while a send is awaiting. */ private linkDown(): boolean { - return this.closed || this.sock.destroyed; + return !this.gate.isUp() || this.sock.destroyed; } /** Answers a request the peer sent us. Responses are never waited on. */ @@ -349,13 +352,13 @@ export class Session extends EventEmitter { 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 { result: { err: new Error('Aborted before the request was sent') }, unsent: false }; + return { result: { err: abortedBeforeSend() }, retryOnNextLink: false }; } const seqNr = this.pending.nextSeqNr(); const built = objToPdu({ ...input, seqNr }); - if (built.err) return { result: { err: built.err }, unsent: false }; + if (built.err) return { result: { err: built.err }, retryOnNextLink: false }; const response = this.pending.wait(seqNr, { signal: options.signal, @@ -366,10 +369,13 @@ export class Session extends EventEmitter { if (written.err) { this.pending.settle(seqNr, { err: written.err }); - return { result: { err: written.err }, unsent: true }; + return { result: { err: written.err }, retryOnNextLink: true }; } - return { result: await response, unsent: false }; + const answered = await response; + + // It went out, so a failure now means the peer may have taken it and the answer was the loss. + return { result: answered.err ? { err: new UnansweredError(answered.err) } : answered, retryOnNextLink: false }; } /** Stops new sends and waits out the ones already issued. */ @@ -404,7 +410,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.gate.shut(false); this.emit('close'); } @@ -412,8 +418,9 @@ export class Session extends EventEmitter { if (this.closed) return; this.closed = true; + this.gate.shut(this.retrying()); this.timers.clear(); - this.pending.settleAll(); + this.pending.settleAll(new Error('Session closed before a response arrived')); this.incoming.clear(); this.sock.destroy(); diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index 9030830..96af9b0 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -12,6 +12,7 @@ import type { SmppLog } from '../src/log.ts'; import type { Sms } from '../src/sms.ts'; import type { SmppServer } from '../src/server.ts'; import type { TestContext } from 'node:test'; +import { LinkGate } from '../src/link-gate.ts'; import { Reassembler, decodeSegments } from '../src/reassembly.ts'; import { Session } from '../src/session.ts'; import { DlrMerger } from '../src/dlr-merger.ts'; @@ -100,10 +101,10 @@ async function sendReceipt(peer: Session, smsId: string, tlvSmsId = smsId): Prom assert.equal(sent.err, undefined); } -type Gate = { open: () => void; passed: Promise }; +type Latch = { open: () => void; passed: Promise }; /** A promise the test opens by hand, guarded by once() against waiting on one it never does. */ -function gate(): Gate { +function latch(): Latch { const opener: { open?: () => void } = {}; const passed = once(resolve => { opener.open = () => { resolve(true); }; }); @@ -559,8 +560,8 @@ 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(); + function answerAfterTheFirst(smpp: SmppServer, arrived: string[]): Latch { + const first = latch(); smpp.on('session', peer => { peer.on('sms', async sms => { @@ -624,6 +625,87 @@ describe('sends across a reconnect', () => { assert.deepEqual(arrived, ['first', 'second']); }); + test('holds a send issued while the rebind is still binding', async t => { + const binding = latch(); + const release = latch(); + let binds = 0; + const smpp = await startServer(t, { + authenticate: async () => { + binds++; + + if (binds > 1) { + binding.open(); + await release.passed; + } + + return true; + }, + }); + + smpp.on('session', peer => { peer.on('sms', async sms => { await sms.sendResp(); }); }); + + const { session } = await connect(t, smpp, { reconnect: { maxDelay: 100, minDelay: 20 } }); + + assert.ok(session); + + peerOf(smpp).sock.destroy(); + + // The fresh socket is attached and the bind is in flight, so the link exists but carries nothing. + await binding.passed; + + const sending = session.sendSms({ from: '46701113311', message: 'mid-bind', to: '46709771337' }); + let settled = false; + + void sending.then(() => { settled = true; }); + await delay(30); + + assert.equal(settled, false, 'an unbound link must not take a submit_sm that would be refused'); + + release.open(); + + const sent = await sending; + + assert.equal(sent.err, undefined); + assert.equal(sent.smsIds.length, 1); + }); + + test('counts a segment the peer never answered in time as unanswered', async t => { + const smpp = await startServer(t); + + smpp.on('session', peer => { peer.on('sms', () => undefined); }); + + const { session } = await connect(t, smpp, { responseTimeout: 60 }); + + assert.ok(session); + + const sent = await session.sendSms({ from: '46701113311', message: 'no answer', to: '46709771337' }); + + assert.match(sent.err?.message ?? '', /may have accepted/); + assert.equal(sent.unanswered, 1, 'a slow SMSC may still have taken it'); + }); + + test('counts a segment aborted after it went out as unanswered', 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); + + assert.ok(session); + + const controller = new AbortController(); + const sending = session.sendSms( + { from: '46701113311', message: 'aborted mid-flight', to: '46709771337' }, + { signal: controller.signal }, + ); + + await arrived; + controller.abort(); + + const sent = await sending; + + assert.match(sent.err?.message ?? '', /may have accepted/); + assert.equal(sent.unanswered, 1, 'the abort is ours; the peer still holds the request'); + }); + 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)); }); @@ -733,6 +815,32 @@ describe('sends across a reconnect', () => { }); }); +describe('LinkGate', () => { + test('refuses a hold whose deadline has already passed', async () => { + let now = 0; + const gate = new LinkGate({ now: () => now, timeout: 100 }); + const deadline = gate.deadline(); + + gate.shut(true); + now = 101; + + const held = await gate.wait(deadline, undefined); + + assert.match(held.err?.message ?? '', /did not come back in time/); + }); + + // addEventListener never fires for a signal that already aborted, so it would wait out the timeout. + test('gives up at once on a signal that was already aborted', async () => { + const gate = new LinkGate({ timeout: 100 }); + + gate.shut(true); + + const held = await gate.wait(gate.deadline(), AbortSignal.abort()); + + assert.match(held.err?.message ?? '', /Aborted while waiting for a link/); + }); +}); + describe('reassembly bounds', () => { function segment(reference: number, part: number, total: number): PduObject { const udh = Buffer.from([0x05, 0x00, 0x03, reference, total, part]); @@ -1090,8 +1198,8 @@ describe('graceful shutdown', () => { assert.ok(first.sock); - const rebinding = gate(); - const release = gate(); + const rebinding = latch(); + const release = latch(); const session = new Session({ reconnect: { connect: open, diff --git a/todo.md b/todo.md index bfe0147..a8370ae 100644 --- a/todo.md +++ b/todo.md @@ -127,9 +127,19 @@ session message is a change to every call site. `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. + 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 + 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 + things — `LinkGate.deadline()` and `wait(deadline)` are a two-call protocol whose only failure + mode is calling `deadline()` inside the loop, which nothing catches; and `Session.linkDown()` + is read from both sides of the seam. Every one of these is unpublished, so it is a two-way + door and belongs after 1.0.0. Raised by review, 2026-09-01. - [ ] **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