From 67c0402def9995d8e1109da30aa6538e81e0ba86 Mon Sep 17 00:00:00 2001 From: Lilleman auf Larv Date: Tue, 1 Sep 2026 17:42:22 +0200 Subject: [PATCH] Bound the messages held for a shutdown, and report a partial receipt --- AGENTS.md | 19 +++++----- README.md | 2 +- src/held-messages.ts | 71 +++++++++++++++++++++++++++++++++---- src/incoming-requests.ts | 9 +++-- src/index.ts | 2 +- src/outgoing-requests.ts | 20 +++++++---- src/session-options.ts | 3 ++ src/session.ts | 12 +++---- src/sms.ts | 31 ++++++++++++---- test/session-extras.test.ts | 66 +++++++++++++++++++++++++++++++++- todo.md | 7 ++-- 11 files changed, 201 insertions(+), 41 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 3ebb92d..edeeab1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -72,7 +72,7 @@ src/ dlr-merger.ts DlrMerger: per-segment receipts counted into one MessageDlr error-from.ts errorFrom(): whatever was thrown or rejected, as an Error expiring-groups.ts ExpiringGroups: the capped, expiring store both of those share - held-messages.ts HeldMessages: the messages handed to the application and not yet answered + held-messages.ts HeldMessages: capped, expiring messages the application has not answered idle-waiters.ts IdleWaiters: waiting for a count to fall to zero, and what is left of a budget incoming-requests.ts Every request the peer sends: messages, receipts, links, unknown commands link-gate.ts LinkGate: where a request with no link to go out on waits for the next one @@ -351,7 +351,8 @@ Grouped by what each one constrains. holding when the drain looks; `sendDlr()` is the one send that goes out past the drain's refusal, 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`, + half: waiting forever is safe for the peer, whose every request is bounded by `responseTimeout` + unless the caller set that to 0 as well, 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 — and to that option's @@ -393,13 +394,13 @@ Grouped by what each one constrains. 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`. `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()`. 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. For the same reason the retry in `send()` asks - `gate.isUp()` rather than `linkDown()`, which also reads the socket — a condition that loops on + on every teardown, `open()` only once `comeBackUp()` has a bound link — and + `OutgoingRequests.linkDown()` reads it rather than `closed`. The bind itself cannot wait for what it + creates, so `pastDrain()` lets the three bind commands past the gate and the window, the same door + `unbind()` takes through `now()`. 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. For the same reason the + retry in `pastDrain()` asks `gate.isUp()` rather than `linkDown()`, which also reads the socket — a condition that loops on something the gate does not gate on spins against a gate that admits it straight back. `LinkGate.returning` is a copy of `retrying()` taken at teardown, and stays true only because nothing stops the reconnect loop without `emitClose()` following it: `drain()` and `end()` are the diff --git a/README.md b/README.md index 1df59c3..80fcd3e 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 for the requests, which the peer answers or times out; the messages fall back to `responseTimeout`, or to its default where that is 0 too, since nothing but the application ends that wait. | +| `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 end when the peer answers or `responseTimeout` expires — so setting both to `0` never ends. The messages fall back to `responseTimeout`, or to its default where that is `0` too, 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. | diff --git a/src/held-messages.ts b/src/held-messages.ts index 4e1b9b2..7abdef7 100644 --- a/src/held-messages.ts +++ b/src/held-messages.ts @@ -1,24 +1,68 @@ import type { PduObject } from './pdu.ts'; +import type { SmppLog } from './log.ts'; +import { ExpiringGroups } from './expiring-groups.ts'; import { IdleWaiters } from './idle-waiters.ts'; +export type HeldMessagesOptions = { + log: SmppLog; + max: number; + timeout: number; +}; + +/** The peer's own sequence number, which is what our answer to this message will carry. */ +function keyOf(pduObjs: PduObject[]): string | undefined { + const first = pduObjs[0]; + + return first ? String(first.seqNr) : undefined; +} + /** The messages handed to the application that it has not answered yet, held by their segments. */ export class HeldMessages { - private readonly held = new Set(); + private readonly held: ExpiringGroups; private readonly idleWaiters = new IdleWaiters(); + private readonly log: SmppLog; + constructor(options: HeldMessagesOptions) { + this.held = new ExpiringGroups({ + max: options.max, + onSweep: () => { this.sweep(); }, + timeout: options.timeout, + }); + this.log = options.log; + } + + /** An application that answers no message at all may not grow this without end. */ hold(pduObjs: PduObject[]): void { - this.held.add(pduObjs); + const key = keyOf(pduObjs); + + if (key === undefined) return; + + if (this.held.full) { + const evicted = this.held.takeOldest(); + + if (evicted) { + this.log.warn('heldMessages - dropping the message held longest', { seqNr: evicted[0] }); + } + } + + this.held.set(key, pduObjs); } /** Whether a drain is still waiting for this message to be answered. */ has(pduObjs: PduObject[]): boolean { - return this.held.has(pduObjs); + const key = keyOf(pduObjs); + + return key !== undefined && this.held.get(key) === pduObjs; } release(pduObjs: PduObject[]): void { - if (!this.held.delete(pduObjs)) return; + const key = keyOf(pduObjs); - if (this.held.size === 0) this.idleWaiters.settle(); + // Identity, not the key: a wrapped sequence number must not release someone else's message. + if (key === undefined || this.held.get(key) !== pduObjs) return; + + this.held.delete(key); + this.settle(); } /** Drops every message: their segments went with the link, so no answer of ours correlates now. */ @@ -28,7 +72,22 @@ export class HeldMessages { } /** Resolves 0 once every message has been answered, or with how many have not. */ - idle(timeout: number, signal?: AbortSignal): Promise { + idle(timeout: number, signal: AbortSignal | undefined): Promise { return this.idleWaiters.wait(() => this.held.size, timeout, signal); } + + private sweep(): void { + const expired = this.held.takeExpired(); + + if (expired.length === 0) return; + + this.log.warn('heldMessages - messages the application never answered', { + messages: expired.length, + }); + this.settle(); + } + + private settle(): void { + if (this.held.size === 0) this.idleWaiters.settle(); + } } diff --git a/src/incoming-requests.ts b/src/incoming-requests.ts index e560e80..202f3af 100644 --- a/src/incoming-requests.ts +++ b/src/incoming-requests.ts @@ -31,7 +31,7 @@ export type IncomingRequestsOptions = { /** Everything the peer asks of a session: messages, receipts, links and the answers to them. */ export class IncomingRequests { private readonly dlrMerger: DlrMerger; - private readonly held = new HeldMessages(); + private readonly held: HeldMessages; private readonly log: SmppLog; private readonly onRequest: OnRequest | undefined; private readonly reassembler: Reassembler; @@ -42,6 +42,11 @@ export class IncomingRequests { constructor(options: IncomingRequestsOptions) { this.dlrMerger = options.dlrMerger; + this.held = new HeldMessages({ + log: options.log, + max: defaults.maxHeldMessages, + timeout: defaults.heldMessageTimeout, + }); this.log = options.log; this.onRequest = options.onRequest; this.reassembler = new Reassembler({ @@ -96,7 +101,7 @@ export class IncomingRequests { } /** Waits out the messages the application still holds, and says how many it never answered. */ - async drain(timeout: number, signal?: AbortSignal): Promise { + async drain(timeout: number, signal: AbortSignal | undefined): Promise { const unanswered = await this.held.idle(timeout, signal); if (unanswered === 0) return {}; diff --git a/src/index.ts b/src/index.ts index 72a4615..3cf22e5 100644 --- a/src/index.ts +++ b/src/index.ts @@ -35,7 +35,7 @@ export { uuidv7 } from './uuid.ts'; export type { BindType, ClientOptions } from './client.ts'; export type { Dlr, Receipt } from './dlr.ts'; -export type { SendRespOptions, Sms, SmsInput } from './sms.ts'; +export type { SendDlrResult, SendRespOptions, Sms, SmsInput } from './sms.ts'; export type { ConcatInfo } from './udh.ts'; export type { Result, VoidResult } from './result.ts'; export type { SmppLog } from './log.ts'; diff --git a/src/outgoing-requests.ts b/src/outgoing-requests.ts index d97ddee..e3b4090 100644 --- a/src/outgoing-requests.ts +++ b/src/outgoing-requests.ts @@ -24,6 +24,13 @@ function abortedBeforeSend(): Error { return new Error('Aborted before the request was sent'); } +/** A response carries the request's sequence number, which only sendReturn() has. */ +function misuse(input: PduObjectInput): Error | undefined { + return input.cmdName.endsWith('_resp') + ? new Error(`Use sendReturn() for responses, not send(): ${input.cmdName}`) + : undefined; +} + /** 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; @@ -67,6 +74,11 @@ export class OutgoingRequests { /** Sends a request and resolves with the peer's response. */ request(input: PduObjectInput, options: SendOptions): Promise> { + // Ahead of the drain, so a misuse is named as one rather than blamed on the shutdown. + const wrong = misuse(input); + + if (wrong) return Promise.resolve({ err: wrong }); + // 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') }); @@ -123,19 +135,15 @@ export class OutgoingRequests { if (unfinished === 0) return {}; - this.log.warn('session - shutting down with requests unfinished', { timeout, unfinished }); + this.log.warn('outgoingRequests - 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. - return options.signal?.aborted === true ? abortedBeforeSend() : undefined; + return misuse(input) ?? (options.signal?.aborted === true ? abortedBeforeSend() : undefined); } private async attempt(input: PduObjectInput, options: SendOptions): Promise { diff --git a/src/session-options.ts b/src/session-options.ts index eb79360..514541b 100644 --- a/src/session-options.ts +++ b/src/session-options.ts @@ -101,8 +101,11 @@ export const undeclaredInterfaceVersion = 0x00; export const defaults = { /** Receipts of a multipart message can be a working day apart, so the cap does the bounding. */ dlrMergeTimeout: 86_400_000, + /** The peer gave up on an unanswered message long before this; the bound is against growth. */ + heldMessageTimeout: 300_000, maxDelay: 30_000, maxDlrMerges: 1000, + maxHeldMessages: 1000, maxOutstanding: 10, maxReassembly: 1000, minDelay: 1000, diff --git a/src/session.ts b/src/session.ts index 3f232b4..34868e9 100644 --- a/src/session.ts +++ b/src/session.ts @@ -313,14 +313,14 @@ export class Session extends EventEmitter { return { err: new Error('The session closed before the drain finished') }; } - return messages.err ? messages : requests; + if (!messages.err) return requests; + + if (!requests.err) return messages; + + return { err: new Error(`${messages.err.message}; ${requests.err.message}`) }; } - /** - * How long the drain waits for the application, which is the only thing that can end that wait. - * Neither timeout may hand it "forever": both are answers about a peer, and a peer is not what - * this half is waiting for. - */ + /** The application half's budget, which may never be "forever": nothing else ends that wait. */ private answering(timeout: number): number { if (timeout > 0) return timeout; diff --git a/src/sms.ts b/src/sms.ts index 5c3b6d1..8942854 100644 --- a/src/sms.ts +++ b/src/sms.ts @@ -3,11 +3,20 @@ import type { MessageState } from './defs/constants.ts'; import type { PduObject, PduObjectInput, TlvInput } from './pdu.ts'; import type { Result, VoidResult } from './result.ts'; import type { Session } from './session.ts'; +import { UnansweredError } from './unanswered-error.ts'; import { consts } from './defs/constants.ts'; import { receiptCodes } from './dlr.ts'; import { smppDate } from './message.ts'; import { uuidv7 } from './uuid.ts'; +/** Both fields hold what the peer took, so a partial failure names what is already receipted. */ +export type SendDlrResult = { + err?: Error; + pduObjs: PduObject[]; + /** Segments that went out unanswered. The peer may have taken them, so sending again may duplicate. */ + unanswered: number; +}; + export type SendRespOptions = { /** The id the peer correlates a later delivery receipt by. Defaults to a generated UUID v7. */ smsId?: string; @@ -25,7 +34,7 @@ export type Sms = { message: string; pduObjs: PduObject[]; /** Sends a delivery report back to the sender. Defaults to DELIVERED. */ - sendDlr: (status?: MessageState) => Promise>; + sendDlr: (status?: MessageState) => Promise; /** Answers every segment. Part of the protocol, not optional. Defaults to ESME_ROK. */ sendResp: (options?: SendRespOptions) => Promise; session: Session; @@ -132,9 +141,13 @@ async function sendDlr( sms: Sms, send: SmsHandlers['send'], status: MessageState = 'DELIVERED', -): Promise> { +): Promise { if (!sms.session.bindAllows('deliver_sm')) { - return { err: new Error('A transmitter-bound session does not carry deliver_sm') }; + return { + err: new Error('A transmitter-bound session does not carry deliver_sm'), + pduObjs: [], + unanswered: 0, + }; } const total = sms.pduObjs.length; @@ -154,12 +167,18 @@ async function sendDlr( }); })); const pduObjs: PduObject[] = []; + let failure: Error | undefined; + let unanswered = 0; for (const one of sent) { - if (one.err) return { err: one.err }; + if (!one.err) { + pduObjs.push(one.pduObj); + } else { + if (one.err instanceof UnansweredError) unanswered++; - pduObjs.push(one.pduObj); + failure ??= one.err; + } } - return { pduObjs }; + return failure ? { err: failure, pduObjs, unanswered } : { pduObjs, unanswered }; } diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index 9587025..0a6c6a3 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 { HeldMessages } from '../src/held-messages.ts'; import { LinkGate } from '../src/link-gate.ts'; import { Reassembler, decodeSegments } from '../src/reassembly.ts'; import { Session } from '../src/session.ts'; @@ -857,6 +858,43 @@ describe('LinkGate', () => { }); }); +// 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[] { + return [{ + cmdId: 0x00000004, + cmdLength: 0, + cmdName: 'submit_sm', + cmdStatus: 'ESME_ROK', + cmdStatusId: 0, + params: { destination_addr: '46709771337', short_message: 'held', source_addr: '46701113311' }, + seqNr, + tlvs: {}, + }]; + } + + test('drops the message held longest rather than holding every one', async () => { + const held = new HeldMessages({ log: silentLog, max: 2, timeout: 10_000 }); + const oldest = message(1); + + held.hold(oldest); + held.hold(message(2)); + held.hold(message(3)); + + assert.equal(held.has(oldest), false); + assert.equal(await held.idle(1, undefined), 2); + }); + + test('gives up on a message the application never answers', async () => { + const held = new HeldMessages({ log: silentLog, max: 10, timeout: 20 }); + + held.hold(message(1)); + + assert.equal(await held.idle(1, undefined), 1); + assert.equal(await held.idle(1000, undefined), 0); + }); +}); + describe('reassembly bounds', () => { function segment(reference: number, part: number, total: number): PduObject { const udh = Buffer.from([0x05, 0x00, 0x03, reference, total, part]); @@ -1108,10 +1146,36 @@ describe('graceful shutdown', () => { const { smpp } = await submitInFlight(t, {}, { responseTimeout: 200, shutdownTimeout: 0 }); const started = Date.now(); const closed = await peerOf(smpp).close(); + const waited = Date.now() - started; assert.ok(closed.err instanceof Error); assert.match(closed.err.message, /1 message\(s\) unanswered/); - assert.ok(Date.now() - started < 2000); + assert.ok(waited >= 190, `waited ${String(waited)} ms, so the fallback was not what bounded it`); + assert.ok(waited < 2000); + }); + + // leftOf() floors what is left at 1 ms: at 0 the request half would read "wait forever" instead. + test('still ends when the message half has spent the whole shutdown budget', async t => { + const { smpp } = await submitInFlight(t, {}, { shutdownTimeout: 100 }); + const bound = peerOf(smpp); + // The client listens for no 'sms', so this one is never answered and stays in the window. + const unanswered = bound.send({ + cmdName: 'submit_sm', + params: { + destination_addr: '46701113311', + short_message: 'nothing answers this', + source_addr: '46709771337', + }, + }); + const closed = await Promise.race([ + bound.close(), + new Promise<{ err?: Error }>(resolve => { + setTimeout(() => { resolve({ err: new Error('close() never returned') }); }, 2000).unref(); + }), + ]); + + assert.match(closed.err?.message ?? '', /1 message\(s\) unanswered; .*1 request\(s\) unfinished/); + assert.ok((await unanswered).err instanceof Error); }); // The README's own listener answers and then sends its receipt, one turn later. Multipart, because diff --git a/todo.md b/todo.md index 8c08242..1f1cacb 100644 --- a/todo.md +++ b/todo.md @@ -58,6 +58,7 @@ Rules the API follows: | 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` | +| Held messages capped and expiring, so an application that answers nothing cannot grow them | `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` | @@ -135,9 +136,9 @@ session message is a change to every call site. - [ ] **An `sms` listener that rejects before answering costs a whole `shutdownTimeout`.** One that *throws* is fine: `emit()` catches it, returns false, and `emitSms()` releases the hold. A rejecting `async` one reaches `sessionError` through `captureRejections`, which hands - the handler an `unknown[]` the `Sms` cannot be read out of without a cast, so nothing releases. - Same cost as the `onRequest`-answers-nothing case that was declined, but reached by a bug - rather than a policy. Raised by review, 2026-09-01. + the handler an `unknown[]` the `Sms` cannot be read out of without a cast, so nothing releases + until the message expires. Same cost as the `onRequest`-answers-nothing case that was declined, + but reached by a bug rather than a policy. 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