diff --git a/AGENTS.md b/AGENTS.md index a3af124..5fc6cb2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -169,13 +169,12 @@ naming the behaviour. | `ESME_RINVBCASTCHANIND` typo | Defined as `0x011`, three hex digits; the spec value is `0x0112` | | Every response carries a message id | `session.js` builds `params = {'message_id': …}` for every response it sends, `deliver_sm_resp` included; SMPP 3.4 4.6.2 makes that field unused and NULL, and Jasmin closes the connection on one | -## Multipart sends and the send window +## Multipart sends `sendSms` puts every segment of a message on the wire together instead of waiting for each response in turn, so a long message costs one round trip rather than one per segment. Nothing on the receiving side forces the order either way: this library answers each inbound segment as it arrives, -so a peer that dispatches one request at a time is never left waiting on us, and a message with more -segments than `maxOutstanding` goes out a slot at a time and still completes. +so a peer that dispatches one request at a time is never left waiting on us. ## GSM 7-bit is sent unpacked @@ -639,6 +638,23 @@ Grouped by what each one constrains. is awaited with the socket already destroyed, so an unref'd one lets a process whose only remaining work is that send exit without settling it. +- **A send queued for a send-window slot is bounded by the caller's `signal`, and by nothing else.** + Maintainer's call, 2026-09-06, from a review of PR #71: the hold above observes the signal and the + `acquire()` on the next line did not, so a caller that aborted while the window was full waited for + a slot it no longer wanted — at `responseTimeout: 0` for as long as the peer stayed quiet, which is + the deadline the README sends the caller to that signal for. Goal 4 is not re-opened by an + unbounded wait here: the queue is the application's own backlog, unbounded in depth as well as in + time because capping it would refuse a send the application asked for, and nothing in it keeps the + peer waiting — which is what separates it from the inbound stores capped on constants. Rejected: + having `release()` skip a waiter whose signal already fired, which leaves the departed waiter in + the queue where `unfinished()` still counts it and the drain waits on it; the waiter leaves as it + settles instead. Rejected: bounding this wait by `responseTimeout` as the hold is bounded — a full + window is this end's own concurrency draining as the peer answers rather than a link going nowhere, + and that bound would fail a message with more segments than `maxOutstanding` partway through + against a slow peer. The failure is a plain `Error` rather than `UnansweredError`, the same answer + an abort at the gate already gives. The drain half needs nothing: `close({ signal })` already hands the signal to + `window.idle()`, and `unbind()` taking none is the shape README states. + - **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 @@ -666,6 +682,13 @@ Grouped by what each one constrains. handlers normalise through `errorFrom()` rather than inline — a route out of the handler would land on a bare `process.nextTick` with nothing to catch it. +- **The four-line abort dance is copied across `LinkGate`, `IdleWaiters`, `PendingRequests` and + `SendWindow` rather than extracted.** Architecture review, 2026-09-06: pre-check `aborted`, attach + `{ once: true }`, detach on settle, leave the registry. What differs at each site is the registry + and what settling means — a FIFO handing over a slot, a set released together, a map keyed by + sequence number, a count recomputed at settle — so a shared `Waiters` fits two of the four and + is a shallower module than the copies. Extract it once a fifth appears. + - **`SmppLog` is a five-method contract this library declares, not a dependency.** `debug`, `error`, `info`, `verbose` and `warn` are what the code actually calls, so an application can satisfy it with an object literal. `@larvit/log` implements it structurally and stays a devDependency, where diff --git a/README.md b/README.md index 66e8556..0d5ecff 100644 --- a/README.md +++ b/README.md @@ -456,8 +456,11 @@ reconnect follows, its `sms.sendDlr()` still goes out on the new link, since a r of its own, correlated by the id it names. `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. +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: it cuts all three waits short, and a send it +stops before anything reached the socket adds nothing to `unanswered`. A message with more segments +than there are slots goes out a slot at a time and still completes, so a deadline tight enough to +expire mid-message is how you produce the partial failure above. `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/outgoing-requests.ts b/src/outgoing-requests.ts index fe661e4..7641aab 100644 --- a/src/outgoing-requests.ts +++ b/src/outgoing-requests.ts @@ -48,7 +48,7 @@ export class OutgoingRequests { this.pending = new PendingRequests(options.log); this.responseTimeout = options.responseTimeout; this.transport = options.transport; - this.window = new SendWindow(options.maxOutstanding); + this.window = new SendWindow({ limit: options.maxOutstanding, log: options.log }); } /** Read through a method: a drop can land while a request is awaiting. */ @@ -115,7 +115,9 @@ export class OutgoingRequests { if (held.err) return { err: held.err }; - await this.window.acquire(); + const slot = await this.window.acquire(options.signal); + + if (slot.err) return { err: slot.err }; const attempt = await this.attempt(input, options).finally(() => { this.window.release(); }); diff --git a/src/send-window.ts b/src/send-window.ts index 524df8d..e13d67d 100644 --- a/src/send-window.ts +++ b/src/send-window.ts @@ -1,31 +1,50 @@ +import type { SmppLog } from './log.ts'; +import type { VoidResult } from './result.ts'; import { IdleWaiters } from './idle-waiters.ts'; +export type SendWindowOptions = { + limit: number; + log: SmppLog; +}; + +type Waiter = (result: VoidResult) => void; + +function aborted(): Error { + return new Error('Aborted while waiting for a send window slot'); +} + /** Caps how many requests are on the wire at once; anything past the limit waits its turn. */ export class SendWindow { private readonly idleWaiters = new IdleWaiters(); private readonly limit: number; - private readonly waiting: (() => void)[] = []; + private readonly log: SmppLog; + private readonly waiting = new Set(); private inFlight = 0; - constructor(limit: number) { - this.limit = limit; + constructor(options: SendWindowOptions) { + this.limit = options.limit; + this.log = options.log; } - acquire(): Promise { + /** Resolves once a slot is the caller's, or with the reason it stopped waiting for one. */ + acquire(signal: AbortSignal | undefined): Promise { if (this.inFlight < this.limit) { this.inFlight++; - return Promise.resolve(); + return Promise.resolve({}); } - return new Promise(resolve => this.waiting.push(resolve)); + if (signal?.aborted === true) return Promise.resolve({ err: aborted() }); + + return this.queue(signal); } release(): void { - const next = this.waiting.shift(); + const next = this.waiting.values().next().value; if (next) { - next(); + this.waiting.delete(next); + next({}); return; } @@ -39,11 +58,34 @@ export class SendWindow { /** Everything the caller is still owed: on the wire, plus queued behind a full window. */ unfinished(): number { - return this.inFlight + this.waiting.length; + return this.inFlight + this.waiting.size; } /** Resolves 0 once nothing is left on the wire, or with what still is. */ idle(timeout: number, signal: AbortSignal | undefined): Promise { return this.idleWaiters.wait(() => this.unfinished(), timeout, signal); } + + /** A waiter leaves the queue as it settles, so release() can only hand a slot to one still in it. */ + private queue(signal: AbortSignal | undefined): Promise { + this.log.verbose('sendWindow - queueing a request behind a full window', { + limit: this.limit, + queued: this.waiting.size + 1, + }); + + return new Promise(resolve => { + const settle = (result: VoidResult): void => { + this.waiting.delete(settle); + signal?.removeEventListener('abort', onAbort); + resolve(result); + }; + + function onAbort(): void { + settle({ err: aborted() }); + } + + signal?.addEventListener('abort', onAbort, { once: true }); + this.waiting.add(settle); + }); + } } diff --git a/src/session-options.ts b/src/session-options.ts index 8c077df..c6d5ea7 100644 --- a/src/session-options.ts +++ b/src/session-options.ts @@ -136,7 +136,7 @@ export const defaults = { /** * A count below 1 does not fail loudly anywhere downstream: `maxOutstanding: 0` leaves every send - * queued behind a slot that is never freed, so the call never settles at all. + * queued behind a slot that is never freed, so a send with no `signal` never settles at all. */ export function checkSessionOptions(options: CheckableOptions): VoidResult { if (options.fromStart !== undefined) { diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index 98cee81..2e3ed84 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -18,6 +18,7 @@ import { IncomingRequests, refusedSegmentStatus } from '../src/incoming-requests import { UnansweredError } from '../src/unanswered-error.ts'; import { createSms } from '../src/sms.ts'; import { LinkGate } from '../src/link-gate.ts'; +import { SendWindow } from '../src/send-window.ts'; import { Reassembler, decodeSegments } from '../src/reassembly.ts'; import { Session } from '../src/session.ts'; import { DlrMerger } from '../src/dlr-merger.ts'; @@ -1304,6 +1305,50 @@ describe('LinkGate', () => { }); }); +describe('SendWindow', () => { + test('gives up a queued acquire the moment its signal fires', async () => { + const window = new SendWindow({ limit: 1, log: silentLog }); + const controller = new AbortController(); + + assert.deepEqual(await window.acquire(undefined), {}); + + const queued = window.acquire(controller.signal); + + controller.abort(); + + assert.match((await queued).err?.message ?? '', /Aborted while waiting for a send window slot/); + assert.equal(window.unfinished(), 1, 'a waiter that gave up is owed nothing'); + }); + + // release() hands the slot straight to the next waiter, so one nobody awaits loses it for good. + test('never hands a freed slot to a waiter that gave up', async () => { + const window = new SendWindow({ limit: 1, log: silentLog }); + const controller = new AbortController(); + + await window.acquire(undefined); + + const abandoned = window.acquire(controller.signal); + + controller.abort(); + await abandoned; + window.release(); + + assert.equal(window.unfinished(), 0, 'the slot is free, not stranded on the waiter that left'); + assert.deepEqual(await window.acquire(undefined), {}, 'so the next send takes it at once'); + }); + + test('takes no slot for a signal that was already aborted', async () => { + const window = new SendWindow({ limit: 1, log: silentLog }); + + await window.acquire(undefined); + + const refused = await window.acquire(AbortSignal.abort()); + + assert.match(refused.err?.message ?? '', /Aborted while waiting for a send window slot/); + assert.equal(window.unfinished(), 1); + }); +}); + // Goal 4: an application that answers nothing must not grow this for the life of the link. describe('held message bounds', () => { function message(seqNr: number): PduObject[] { @@ -2035,6 +2080,109 @@ describe('AbortSignal on a send', () => { assert.ok(sent.err instanceof Error); }); + + /** The peer answers nothing, so the single slot stays taken for the life of the test. */ + async function oneSlotHeld( + t: TestContext, + options: Parameters[0] = {}, + ): Promise { + const smpp = await startServer(t); + const onWire = once(resolve => { + smpp.on('session', bound => bound.on('incomingPduObj', resolve)); + }); + + smpp.on('session', bound => bound.on('sms', () => undefined)); + + const { session } = await connect(t, smpp, { maxOutstanding: 1, ...options }); + + assert.ok(session); + void session.sendSms({ from: '46701113311', message: 'holds the only slot', to: '46709771337' }); + + await onWire; + + return session; + } + + test('gives up on a send still queued behind a full window', async t => { + const session = await oneSlotHeld(t, { responseTimeout: 10_000 }); + const controller = new AbortController(); + const queued = session.sendSms( + { from: '46701113311', message: 'queued behind the held slot', to: '46709771337' }, + { signal: controller.signal }, + ); + + await delay(20); + controller.abort(); + + const sent = await within(500, queued); + + assert.ok(sent, 'an abort must not wait out a slot the caller no longer wants'); + assert.match(sent.err?.message ?? '', /Aborted while waiting for a send window slot/); + assert.equal(sent.unanswered, 0, 'it never reached the socket, so the peer cannot have taken it'); + }); + + test('gives up on a queued send where responseTimeout: 0 never would', async t => { + const session = await oneSlotHeld(t, { responseTimeout: 0 }); + const controller = new AbortController(); + const queued = session.sendSms( + { from: '46701113311', message: 'queued with nothing else to end the wait', to: '46709771337' }, + { signal: controller.signal }, + ); + + await delay(20); + controller.abort(); + + const sent = await within(500, queued); + + assert.ok(sent, 'the signal is the only bound this wait has'); + assert.match(sent.err?.message ?? '', /Aborted while waiting for a send window slot/); + assert.equal(sent.unanswered, 0); + }); + + test('leaves the freed slot to the next send rather than to the waiter that gave up', async t => { + const smpp = await startServer(t); + const holding = once(resolve => { smpp.on('session', bound => bound.on('sms', resolve)); }); + let firstTaken = false; + + smpp.on('session', bound => { + bound.on('sms', async sms => { + if (firstTaken) await sms.sendResp(); + + firstTaken = true; + }); + }); + + const { session } = await connect(t, smpp, { maxOutstanding: 1, responseTimeout: 10_000 }); + + assert.ok(session); + void session.sendSms({ from: '46701113311', message: 'holds the only slot', to: '46709771337' }); + + const held = await holding; + const controller = new AbortController(); + const abandoned = session.sendSms( + { from: '46701113311', message: 'abandoned in the queue', to: '46709771337' }, + { signal: controller.signal }, + ); + + await delay(20); + controller.abort(); + + const gaveUp = await within(500, abandoned); + + assert.ok(gaveUp, 'the waiter that gave up must settle before the slot it left is freed'); + // Any other error means it never reached the queue, so there was no waiter to strand. + assert.match(gaveUp.err?.message ?? '', /Aborted while waiting for a send window slot/); + await held.sendResp(); + + const following = await within(1000, session.sendSms({ + from: '46701113311', + message: 'takes the freed slot', + to: '46709771337', + })); + + assert.ok(following, 'a slot handed to a waiter that left is one the window never gets back'); + assert.equal(following.err, undefined); + }); }); describe('graceful shutdown', () => { diff --git a/test/session.test.ts b/test/session.test.ts index 35c5da1..ca4e4f2 100644 --- a/test/session.test.ts +++ b/test/session.test.ts @@ -1315,6 +1315,14 @@ describe('robustness', () => { to: '46709771337', }))); + const long = await session.sendSms({ + from: '46701113311', + message: 'x'.repeat(500), + to: '46709771337', + }); + + assert.equal(long.err, undefined, 'more segments than slots still completes, a slot at a time'); + assert.equal(long.smsIds.length, 4); assert.ok(peak <= 2, `peak was ${String(peak)}`); });