diff --git a/AGENTS.md b/AGENTS.md index ad670c6..41703ed 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -66,7 +66,7 @@ src/ index.ts Public surface. Named exports only, no default export. client.ts client() -> { err, session } server.ts server() -> { err, server }, server owns the listener + close() - session.ts Session: dispatch, events, and the collaborators below + session.ts Session: the socket's life, dispatch, events, and the collaborators below sms.ts The live handle emitted as the 'sms' event (sendResp/sendDlr) dlr.ts Delivery receipts: text and TLV parsing, receipt status codes dlr-merger.ts DlrMerger: per-segment receipts counted into one MessageDlr @@ -79,6 +79,7 @@ src/ 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 + outgoing-requests.ts OutgoingRequests: the gate, the window, the pending map and the retry pdu.ts pduToObj / objToPdu / pduReturn — synchronous, result-returning pdu-framer.ts PduFramer: a byte stream cut into complete PDUs pdu-transport.ts PduTransport: the socket a session reads complete PDUs off @@ -91,6 +92,7 @@ src/ 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, and their reference + unanswered-error.ts UnansweredError: it went out and no answer came back uuid.ts uuidv7() — the ids the library generates for messages defs/ commands.ts The 33 commands, their ids and ordered parameter lists @@ -252,7 +254,10 @@ Grouped by what each one constrains. `super.on()` call needs one. The cost is that a subclass can no longer reach those seven through `super` — re-declaring them the same way is its way out. `unknown` rather than `void | Promise` because a listener may return anything: `session.on('close', () => - set.delete(session))` returns a boolean. + set.delete(session))` returns a boolean. This also settles what the drain can wait on: a listener's + own promise would be the better completion signal, and reaching it needs `listeners()`, which + cannot be re-declared the same way — Node types it invariantly enough that widening `void` to + `unknown` is `TS2416`. Re-probed 2026-09-01; `sendResp()` stays the signal. ### The wire @@ -344,7 +349,12 @@ Grouped by what each one constrains. `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. + and only while the message is still held — past that it is an ordinary send, because the drain it + would slip past is no longer waiting for it. `shutdownTimeout: 0` does not carry over to this + half: waiting forever is safe for the peer, whose every request is bounded by `responseTimeout`, + and unsafe for the application, which nothing bounds — `close()` is what you reach for when the + application is stuck, so it may not block on the application coming unstuck. That half falls back + to `responseTimeout`, the same answer the link gate's hold already takes. - **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()` diff --git a/README.md b/README.md index 559b05c..7f435b7 100644 --- a/README.md +++ b/README.md @@ -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 and the messages the application has not answered; `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 for the requests, which the peer answers or times out; the messages fall back to `responseTimeout`, since nothing but the application ends that wait. | | `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,10 +331,12 @@ 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. 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 +then tear down whatever is left, resolving to an `err` that says what was lost. They also wait for +every `sms` the application has not called `sendResp()` on, so a peer whose `submit_sm` is still +being handled is answered rather than left to re-send it — answering its PDUs through `sendReturn()` +instead leaves that wait running until it gives up. `sendDlr()` is the one send the refusal lets +past, and it catches the wait when issued straight after `sendResp()`; await anything in between and +it races the shutdown like any other send. `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 index 3a12be4..4e1b9b2 100644 --- a/src/held-messages.ts +++ b/src/held-messages.ts @@ -10,6 +10,11 @@ export class HeldMessages { this.held.add(pduObjs); } + /** Whether a drain is still waiting for this message to be answered. */ + has(pduObjs: PduObject[]): boolean { + return this.held.has(pduObjs); + } + release(pduObjs: PduObject[]): void { if (!this.held.delete(pduObjs)) return; diff --git a/src/incoming-requests.ts b/src/incoming-requests.ts index 2f28d1d..e560e80 100644 --- a/src/incoming-requests.ts +++ b/src/incoming-requests.ts @@ -21,8 +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>; + /** Past a drain's refusal, for a receipt the drain is itself waiting for. */ + sendPastDrain: (input: PduObjectInput) => Promise>; session: Session; smsIdFormat?: SmsIdFormat | undefined; systemId?: string | undefined; @@ -35,7 +35,7 @@ export class IncomingRequests { private readonly log: SmppLog; private readonly onRequest: OnRequest | undefined; private readonly reassembler: Reassembler; - private readonly sendHeld: IncomingRequestsOptions['sendHeld']; + private readonly sendPastDrain: IncomingRequestsOptions['sendPastDrain']; private readonly session: Session; private readonly smsIdFormat: SmsIdFormat; private readonly systemId: string; @@ -50,7 +50,7 @@ export class IncomingRequests { maxOctets: options.maxOctets, timeout: options.reassemblyTimeout ?? defaults.reassemblyTimeout, }); - this.sendHeld = options.sendHeld; + this.sendPastDrain = options.sendPastDrain; this.session = options.session; this.smsIdFormat = options.smsIdFormat ?? {}; this.systemId = options.systemId ?? defaults.systemId; @@ -167,7 +167,8 @@ export class IncomingRequests { }, { // A turn later, so a listener sending its receipt straight after the response still holds. onAnswered: () => { setImmediate(() => { this.held.release(pduObjs); }); }, - send: this.sendHeld, + // Past the refusal only while a drain is still waiting for this message; an ordinary send after. + send: input => (this.held.has(pduObjs) ? this.sendPastDrain(input) : this.session.send(input)), }); this.held.hold(pduObjs); diff --git a/src/link-gate.ts b/src/link-gate.ts index a6a51b2..6711790 100644 --- a/src/link-gate.ts +++ b/src/link-gate.ts @@ -47,9 +47,11 @@ export class LinkGate { return this.up || this.returning ? undefined : over(); } - /** When a hold starting now has to give up. 0 never does. */ - deadline(): number { - return this.timeout > 0 ? this.now() + this.timeout : 0; + /** One budget for a request, however many links it waits through. 0 never gives up. */ + hold(signal: AbortSignal | undefined): () => Promise { + const deadline = this.timeout > 0 ? this.now() + this.timeout : 0; + + return () => this.wait(deadline, signal); } /** A link is up and bound: everything held goes out on it. */ @@ -73,7 +75,7 @@ export class LinkGate { } /** Resolves once a link can carry the request, or with the reason none ever will. */ - wait(deadline: number, signal: AbortSignal | undefined): Promise { + private wait(deadline: number, signal: AbortSignal | undefined): Promise { if (this.up) return Promise.resolve({}); const refused = this.refusal(); @@ -86,10 +88,10 @@ export class LinkGate { if (deadline !== 0 && left <= 0) return Promise.resolve({ err: expired() }); - return this.hold(left, signal); + return this.waitForLink(left, signal); } - private hold(left: number, signal: AbortSignal | undefined): Promise { + private waitForLink(left: number, signal: AbortSignal | undefined): Promise { this.log.verbose('linkGate - holding a request until a link is back', { timeout: left }); return new Promise(resolve => { diff --git a/src/outgoing-requests.ts b/src/outgoing-requests.ts new file mode 100644 index 0000000..4e4c490 --- /dev/null +++ b/src/outgoing-requests.ts @@ -0,0 +1,168 @@ +import type { PduObject, PduObjectInput } from './pdu.ts'; +import type { PduTransport } from './pdu-transport.ts'; +import type { Result, VoidResult } from './result.ts'; +import type { SendOptions } from './session-options.ts'; +import type { SmppLog } from './log.ts'; +import { LinkGate } from './link-gate.ts'; +import { PendingRequests } from './pending-requests.ts'; +import { SendWindow } from './send-window.ts'; +import { UnansweredError } from './unanswered-error.ts'; +import { bindCommands } from './session-options.ts'; +import { objToPdu } from './pdu.ts'; + +export type OutgoingRequestsOptions = { + log: SmppLog; + maxOutstanding: number; + responseTimeout: number; + transport: PduTransport; +}; + +/** `retryOnNextLink`: the write failed, so nothing reached the socket and another link may carry it. */ +type Attempt = { result: Result<{ pduObj: PduObject }>; retryOnNextLink: boolean }; + +function abortedBeforeSend(): Error { + return new Error('Aborted before the request was sent'); +} + +/** Everything this end asks of the peer: which link carries it, how many at once, and the answer. */ +export class OutgoingRequests { + private readonly gate: LinkGate; + private readonly log: SmppLog; + private readonly pending: PendingRequests; + private readonly responseTimeout: number; + private readonly transport: PduTransport; + private readonly window: SendWindow; + + private draining = false; + + constructor(options: OutgoingRequestsOptions) { + this.gate = new LinkGate({ log: options.log, timeout: options.responseTimeout }); + this.log = options.log; + this.pending = new PendingRequests(options.log); + this.responseTimeout = options.responseTimeout; + this.transport = options.transport; + this.window = new SendWindow(options.maxOutstanding); + } + + /** Read through a method: a drop can land while a request is awaiting. */ + linkDown(): boolean { + return !this.gate.isUp() || this.transport.sock.destroyed; + } + + /** A link is up and bound, so everything held for one goes out on it. */ + linkUp(): void { + this.gate.open(); + } + + /** The link is gone; `returning` says whether another one is on its way. */ + linkLost(returning: boolean): void { + this.gate.shut(returning); + this.pending.settleAll(new Error('Session closed before a response arrived')); + } + + /** Hands a response to the request waiting for it. False means nothing was. */ + deliver(pduObj: PduObject): boolean { + return this.pending.deliver(pduObj); + } + + /** Sends a request and resolves with the peer's response. */ + request(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.pastDrain(input, options); + } + + /** The same path without that refusal, which a receipt for a held message has to take. */ + async pastDrain( + input: PduObjectInput, + options: SendOptions, + ): Promise> { + const refused = this.refuse(input, options); + + if (refused) return { err: refused }; + + // A bind is what makes a link usable, so it cannot wait for one. + if (bindCommands.includes(input.cmdName)) return this.now(input, options); + + const waitForLink = this.gate.hold(options.signal); + + for (;;) { + const held = await waitForLink(); + + 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.retryOnNextLink || this.gate.isUp() || this.gate.refusal()) return attempt.result; + } + } + + /** Past the gate, the window and a drain, for what has to go out either way. */ + async now(input: PduObjectInput, options: SendOptions = {}): Promise> { + return (await this.attempt(input, options)).result; + } + + /** Refuses every request from here on, on a link that is already down as much as a live one. */ + stopAccepting(): void { + this.draining = true; + } + + /** Waits out the requests already on the wire, and says how many never finished. */ + async drain(timeout: number, signal: AbortSignal | undefined): Promise { + const unfinished = await this.window.idle(timeout, signal); + + if (unfinished === 0) return {}; + + this.log.warn('session - shutting down with requests unfinished', { timeout, unfinished }); + + return { err: new Error(`Shut down with ${String(unfinished)} request(s) unfinished`) }; + } + + /** Why a request cannot go out at all, as opposed to not yet. */ + private refuse(input: PduObjectInput, options: SendOptions): Error | undefined { + if (input.cmdName.endsWith('_resp')) { + return new Error(`Use sendReturn() for responses, not send(): ${input.cmdName}`); + } + + // Before the gate and the window, or an aborted call waits for what it will never use. + if (options.signal?.aborted === true) return abortedBeforeSend(); + + // A bind skips the gate below, so the answer it would have given is given here instead. + return bindCommands.includes(input.cmdName) ? this.gate.refusal() : undefined; + } + + 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: abortedBeforeSend() }, retryOnNextLink: false }; + } + + const seqNr = this.pending.nextSeqNr(); + const built = objToPdu({ ...input, seqNr }); + + if (built.err) return { result: { err: built.err }, retryOnNextLink: false }; + + const response = this.pending.wait(seqNr, { + signal: options.signal, + timeout: this.responseTimeout, + }); + const written = this.transport.write(built.buffer); + + if (written.err) { + this.pending.settle(seqNr, { err: written.err }); + + return { result: { err: written.err }, retryOnNextLink: true }; + } + + 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 }; + } +} diff --git a/src/pending-requests.ts b/src/pending-requests.ts index 5fdbb1e..7949106 100644 --- a/src/pending-requests.ts +++ b/src/pending-requests.ts @@ -12,14 +12,6 @@ type Pending = { settle: (result: Result<{ pduObj: PduObject }>) => void; }; -/** The request went out and no answer came back: the peer may have accepted it. */ -export class UnansweredError extends Error { - constructor(cause: Error) { - super(`No answer came back, so the peer may have accepted it: ${cause.message}`, { cause }); - this.name = 'UnansweredError'; - } -} - /** Hands out sequence numbers and matches responses to the requests waiting for them. */ export class PendingRequests { private readonly log: SmppLog; diff --git a/src/send-sms.ts b/src/send-sms.ts index 3277330..8b018b4 100644 --- a/src/send-sms.ts +++ b/src/send-sms.ts @@ -4,7 +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 { UnansweredError } from './unanswered-error.ts'; import { consts } from './defs/constants.ts'; import { detect } from './defs/encodings.ts'; import { normaliseSmsId } from './sms-id.ts'; diff --git a/src/session.ts b/src/session.ts index b297233..24c2b10 100644 --- a/src/session.ts +++ b/src/session.ts @@ -10,17 +10,15 @@ 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 { OutgoingRequests } from './outgoing-requests.ts'; 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 { isResp, pduReturn } from './pdu.ts'; import { silentLog } from './log.ts'; import { submitSms, unsent } from './send-sms.ts'; import { ConcatReference } from './udh.ts'; @@ -41,13 +39,6 @@ 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; -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; declare off: (event: K, listener: SessionListener) => this; @@ -68,17 +59,14 @@ export class Session extends EventEmitter { private readonly concatReference = new ConcatReference(); private readonly dlrMerger: DlrMerger; - private readonly gate: LinkGate; private readonly incoming: IncomingRequests; private readonly options: SessionOptions; - private readonly pending: PendingRequests; + private readonly outgoing: OutgoingRequests; private readonly reconnectLoop: ReconnectLoop | undefined; private readonly timers: LinkTimers; private readonly transport: PduTransport; - private readonly window: SendWindow; private closed = false; - private draining = false; private ended = false; /** A listener that throws is the application's bug; it must not become ours. Hard rule 1. */ @@ -123,7 +111,6 @@ export class Session extends EventEmitter { max: defaults.maxDlrMerges, timeout: defaults.dlrMergeTimeout, }); - this.gate = new LinkGate({ log: this.log, timeout: options.responseTimeout ?? defaults.responseTimeout }); this.incoming = new IncomingRequests({ dlrMerger: this.dlrMerger, log: this.log, @@ -131,12 +118,11 @@ export class Session extends EventEmitter { maxReassembly: options.maxReassembly, onRequest: options.onRequest, reassemblyTimeout: options.reassemblyTimeout, - sendHeld: input => this.sendThrough(input, {}), + sendPastDrain: input => this.outgoing.pastDrain(input, {}), session: this, smsIdFormat: options.smsIdFormat, systemId: options.systemId, }); - this.pending = new PendingRequests(this.log); this.reconnectLoop = this.loopFor(options.reconnect); this.timers = new LinkTimers({ enquireLinkInterval: options.enquireLinkInterval, @@ -147,7 +133,12 @@ export class Session extends EventEmitter { onIdle: () => { this.teardown(); }, }); this.transport = this.transportFor(options.sock); - this.window = new SendWindow(options.maxOutstanding ?? defaults.maxOutstanding); + this.outgoing = new OutgoingRequests({ + log: this.log, + maxOutstanding: options.maxOutstanding ?? defaults.maxOutstanding, + responseTimeout: options.responseTimeout ?? defaults.responseTimeout, + transport: this.transport, + }); this.resetTimers(); } @@ -170,56 +161,7 @@ export class Session extends EventEmitter { /** Sends a request and resolves with the peer's response. */ 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> { - const refused = this.refuseSend(input, options); - - if (refused) return { err: refused }; - - // A bind is what makes a link usable, so it cannot wait for one. - if (bindCommands.includes(input.cmdName)) return (await this.attempt(input, options)).result; - - 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.retryOnNextLink || this.gate.isUp() || !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 new Error(`Use sendReturn() for responses, not send(): ${input.cmdName}`); - } - - // Before the gate and the window, or an aborted call waits for what it will never use. - if (options.signal?.aborted === true) return abortedBeforeSend(); - - // A bind skips the gate below, so the answer it would have given is given here instead. - return bindCommands.includes(input.cmdName) ? this.gate.refusal() : undefined; - } - - /** Read through a method: a drop can land while a send is awaiting. */ - private linkDown(): boolean { - return !this.gate.isUp() || this.sock.destroyed; + return this.outgoing.request(input, options); } /** Answers a request the peer sent us. Responses are never waited on. */ @@ -269,9 +211,9 @@ export class Session extends EventEmitter { async unbind(): Promise { const drained = await this.drain(undefined); const wasOpen = !this.closed; - // attempt(), not send(): the drain gate refuses a send, and the unbind goes out either way. + // now(), not send(): a drain refuses a send, and the unbind goes out either way. const sent = wasOpen - ? (await this.attempt({ cmdName: 'unbind' }, {})).result + ? await this.outgoing.now({ cmdName: 'unbind' }) : { err: new Error('Session is closed') }; const closedOnUnbind = wasOpen && this.closed; @@ -341,7 +283,7 @@ export class Session extends EventEmitter { } this.resetTimers(); - this.gate.open(); + this.outgoing.linkUp(); this.log.info('session - reconnected'); this.emit('reconnected'); @@ -353,58 +295,27 @@ export class Session extends EventEmitter { this.closed = false; } - 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: abortedBeforeSend() }, retryOnNextLink: false }; - } - - const seqNr = this.pending.nextSeqNr(); - const built = objToPdu({ ...input, seqNr }); - - if (built.err) return { result: { err: built.err }, retryOnNextLink: false }; - - const response = this.pending.wait(seqNr, { - signal: options.signal, - timeout: this.options.responseTimeout ?? defaults.responseTimeout, - }); - const written = this.transport.write(built.buffer); - - if (written.err) { - this.pending.settle(seqNr, { err: written.err }); - - return { result: { err: written.err }, retryOnNextLink: true }; - } - - 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 messages we hold and the requests already issued. */ private async drain(signal: AbortSignal | undefined): Promise { this.reconnectLoop?.stop(); - this.draining = true; + this.outgoing.stopAccepting(); - if (this.linkDown()) return {}; + if (this.outgoing.linkDown()) return {}; const timeout = this.options.shutdownTimeout ?? defaults.shutdownTimeout; const deadline = timeout > 0 ? Date.now() + timeout : 0; + // Only the application answers a held message, so that half falls back rather than wait forever. + const answering = timeout > 0 ? timeout : (this.options.responseTimeout ?? defaults.responseTimeout); // 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); + const messages = await this.incoming.drain(answering, signal); + const requests = await this.outgoing.drain(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 (this.outgoing.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 }); - - return { err: new Error(`Shut down with ${String(unfinished)} request(s) unfinished`) }; + return messages.err ? messages : requests; } /** The session is over now, drained or not. Nothing brings it back. */ @@ -419,7 +330,7 @@ export class Session extends EventEmitter { if (this.ended) return; this.ended = true; - this.gate.shut(false); + this.outgoing.linkLost(false); this.emit('close'); } @@ -427,9 +338,8 @@ export class Session extends EventEmitter { if (this.closed) return; this.closed = true; - this.gate.shut(this.retrying()); + this.outgoing.linkLost(this.retrying()); this.timers.clear(); - this.pending.settleAll(new Error('Session closed before a response arrived')); this.incoming.clear(); this.sock.destroy(); @@ -448,7 +358,7 @@ export class Session extends EventEmitter { private dispatch(pduObj: PduObject): void { if (isResp(pduObj)) { - if (!this.pending.deliver(pduObj)) { + if (!this.outgoing.deliver(pduObj)) { this.log.debug('session - response with no matching request', { seqNr: pduObj.seqNr }); } diff --git a/src/unanswered-error.ts b/src/unanswered-error.ts new file mode 100644 index 0000000..b6f45b8 --- /dev/null +++ b/src/unanswered-error.ts @@ -0,0 +1,7 @@ +/** The request went out and no answer came back: the peer may have accepted it. */ +export class UnansweredError extends Error { + constructor(cause: Error) { + super(`No answer came back, so the peer may have accepted it: ${cause.message}`, { cause }); + this.name = 'UnansweredError'; + } +} diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index ed6eedf..963ce05 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -819,12 +819,12 @@ describe('LinkGate', () => { test('refuses a hold whose deadline has already passed', async () => { let now = 0; const gate = new LinkGate({ log: silentLog, now: () => now, timeout: 100 }); - const deadline = gate.deadline(); + const waitForLink = gate.hold(undefined); gate.shut(true); now = 101; - const held = await gate.wait(deadline, undefined); + const held = await waitForLink(); assert.match(held.err?.message ?? '', /did not come back in time/); }); @@ -836,7 +836,7 @@ describe('LinkGate', () => { gate.shut(true); const before = timers(); - const held = gate.wait(gate.deadline(), undefined); + const held = gate.hold(undefined)(); assert.equal(timers(), before + 1, 'an unref\'d timer is not counted here, which is the point'); @@ -851,7 +851,7 @@ describe('LinkGate', () => { gate.shut(true); - const held = await gate.wait(gate.deadline(), AbortSignal.abort()); + const held = await gate.hold(AbortSignal.abort())(); assert.match(held.err?.message ?? '', /Aborted while waiting for a link/); }); @@ -1102,6 +1102,17 @@ describe('graceful shutdown', () => { assert.match(closed.err.message, /1 message\(s\) unanswered/); }); + // Waiting forever is safe for the peer, which every request times out on. The application is not. + test('falls back to responseTimeout for a held message when the shutdown waits forever', async t => { + const { smpp } = await submitInFlight(t, {}, { responseTimeout: 200, shutdownTimeout: 0 }); + const started = Date.now(); + const closed = await peerOf(smpp).close(); + + assert.ok(closed.err instanceof Error); + assert.match(closed.err.message, /1 message\(s\) unanswered/); + assert.ok(Date.now() - started < 2000); + }); + // 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); diff --git a/todo.md b/todo.md index 2a62cb4..f4e8443 100644 --- a/todo.md +++ b/todo.md @@ -57,6 +57,7 @@ Rules the API follows: | `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` | +| `OutgoingRequests`: the gate, the window, the pending map and the retry under one owner, told when a link comes up or goes down | `test/session-extras.test.ts`, `test/session.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` | @@ -118,26 +119,25 @@ 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. -- [ ] **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, `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 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 - things — `LinkGate.deadline()` and `wait(deadline)` are a two-call protocol whose only failure - mode is calling `deadline()` inside the loop, which nothing catches; `Session.linkDown()` is - read from both sides of the seam; and `UnansweredError` sits in `pending-requests.ts`, which - never uses it, for the sole edge that makes `send-sms` import that module at all. Every one of - these is unpublished, so it is a two-way door and belongs after 1.0.0. Raised by review, +- [ ] **Group the session's collaborators under `src/session/`.** `session.ts` imports + `dlr-merger`, `incoming-requests`, `link-timers`, `outgoing-requests`, `pdu-transport`, + `reconnect-loop` and `send-sms`, and nothing else does, so the directory would make that + boundary visible. The `OutgoingRequests` extraction this was to be done with landed on + 2026-09-01, so it is the remaining half. Raised by review, 2026-09-01. + +- [ ] **`leftOf()` and the link gate's own budget are one concept counted twice.** + `idle-waiters.ts` reads what is left of a budget as `Math.max(1, deadline - now)`, because 0 + means "forever" there; `link-gate.ts` runs the same subtraction and calls `<= 0` expired. + Neither is reachable from the other, so nothing can disagree today, but a reader who learns one + and applies it to the other is wrong. A budget type both take would close it. Raised by review, 2026-09-01. + +- [ ] **A message the reconnect dropped is answered into the void, and reported as delivered.** + `teardown()` clears the held messages along with the inbound segments, which is right — but the + application still holds the `Sms`, so `sendResp()` writes the old link's sequence numbers to the + new socket, succeeds, and returns `{}` for a response that correlates with nothing at the peer. + `HeldMessages` now knows exactly which messages went that way, so saying so is a small + addition. Goal 2, low frequency. 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