From c3ebd6d6bc2accb26776e26343480643e837a431 Mon Sep 17 00:00:00 2001 From: Lilleman auf Larv Date: Fri, 25 Sep 2026 19:22:31 +0200 Subject: [PATCH] Cap the heap unanswered messages hold, and detach them from the chunk they arrived in --- AGENTS.md | 1 + CHANGELOG.md | 3 ++ README.md | 5 ++-- src/held-messages.ts | 50 ++++++++++++++++++++++++++++----- src/incoming-requests.ts | 4 ++- src/reassembly.ts | 55 ++----------------------------------- src/retained-pdu.ts | 53 +++++++++++++++++++++++++++++++++++ src/session-options.ts | 1 + test/session-extras.test.ts | 48 ++++++++++++++++++++++++++++++-- todo.md | 8 ------ 10 files changed, 155 insertions(+), 73 deletions(-) create mode 100644 src/retained-pdu.ts diff --git a/AGENTS.md b/AGENTS.md index 86266bf..553c09a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -70,6 +70,7 @@ src/ reassembly.ts Reassembler: capped, expiring multipart groups reconnect-loop.ts ReconnectLoop: backoff, retry timer, stopped-ness result.ts Result — the shape every fallible call returns + retained-pdu.ts A PDU copied off the wire so holding it pins nothing else, and what holding it costs send-sms.ts submitSms composition and the submitSmParams builder send-window.ts SendWindow: the maxOutstanding semaphore session-options.ts SessionOptions, ReconnectOptions, bind direction and the session defaults diff --git a/CHANGELOG.md b/CHANGELOG.md index ae4e243..d8c15a9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,9 @@ count as next to nothing, so a peer could hold far more than the cap. **Raise a `maxOctets` you tuned low**: it now holds several times fewer segments, and an incomplete message evicted over the cap is lost, since its segments were already answered. +- Unanswered `sms` messages are capped at 64 MiB per session by the `maxOctets` charge, beside the + 1000-message cap: the oldest is dropped with a warning, as over the count. A peer faster than an + application answering asynchronously could hold gigabytes per session. - `server()` refuses a `maxOctets` below 1 or not a whole number, `Infinity` included, like its other limits. `server({ maxOctets: 0 })` used to start and then refuse every multipart message. - `callback_num`, `callback_num_atag`, `callback_num_pres_ind`, `broadcast_area_identifier` and diff --git a/README.md b/README.md index ab71fd2..e6f1987 100644 --- a/README.md +++ b/README.md @@ -383,8 +383,9 @@ holds for `session.send()`. message has failed. Answering through `sendReturn()` instead leaves the wait running. 3. Tear down what is left, resolving to an `err` that says what was lost. -At most 1000 unanswered messages are held, for five minutes each; what falls out of either bound is -dropped with a warning on the log and waited for no longer. Neither bound is an option. +At most 1000 unanswered messages, and 64 MiB of them by the `maxOctets` charge, are held for five +minutes each; what falls out of a bound is dropped with a warning on the log and waited for no +longer. None of the bounds is an option. `close({ signal })` cuts the wait short. `unbind()` takes no signal, and waits a further `responseTimeout` for its own response. diff --git a/src/held-messages.ts b/src/held-messages.ts index 6a317fd..dcf3dca 100644 --- a/src/held-messages.ts +++ b/src/held-messages.ts @@ -2,10 +2,12 @@ import type { PduObject } from './pdu.ts'; import type { SmppLog } from './log.ts'; import { ExpiringGroups } from './expiring-groups.ts'; import { IdleWaiters } from './idle-waiters.ts'; +import { retainedOctets } from './retained-pdu.ts'; export type HeldMessagesOptions = { log: SmppLog; max: number; + maxOctets: number; /** Injected so expiry can be exercised without a wall clock. */ now?: (() => number) | undefined; timeout: number; @@ -18,12 +20,19 @@ function keyOf(pduObjs: PduObject[]): string | undefined { return first ? String(first.seqNr) : undefined; } +type Held = { + octets: number; + pduObjs: PduObject[]; +}; + /** The messages handed to the application that it has not answered yet, held by their segments. */ export class HeldMessages { - private readonly held: ExpiringGroups; + private readonly held: ExpiringGroups; private readonly idleWaiters = new IdleWaiters(); private readonly log: SmppLog; private readonly max: number; + private readonly maxOctets: number; + private octets = 0; constructor(options: HeldMessagesOptions) { this.held = new ExpiringGroups({ @@ -34,6 +43,7 @@ export class HeldMessages { }); this.log = options.log; this.max = options.max; + this.maxOctets = options.maxOctets; } get size(): number { @@ -48,35 +58,49 @@ export class HeldMessages { this.sweep(); - if (this.held.get(key)) { + const replaced = this.held.get(key); + + if (replaced) { this.log.warn('heldMessages - replacing a message on a re-used sequence number', { seqNr: Number(key) }); + this.delete(key, replaced); } else if (this.held.full) { this.dropOldest(); } - this.held.set(key, pduObjs); + const octets = pduObjs.reduce((sum, pduObj) => sum + retainedOctets(pduObj), 0); + + // The message just held stays even alone past the cap: the peer is still owed its answer. + while (this.held.size > 0 && this.octets + octets > this.maxOctets) { + this.dropOldest(); + } + + this.held.set(key, { octets, pduObjs }); + this.octets += octets; } /** Whether a drain is still waiting for this message to be answered. */ has(pduObjs: PduObject[]): boolean { const key = keyOf(pduObjs); - return key !== undefined && this.held.get(key) === pduObjs; + return key !== undefined && this.held.get(key)?.pduObjs === pduObjs; } release(pduObjs: PduObject[]): void { const key = keyOf(pduObjs); // Identity, not the key: a wrapped sequence number must not release someone else's message. - if (key === undefined || this.held.get(key) !== pduObjs) return; + const held = key === undefined ? undefined : this.held.get(key); - this.held.delete(key); + if (key === undefined || held?.pduObjs !== pduObjs) return; + + this.delete(key, held); this.settle(); } /** Drops every message: their segments went with the link, so no answer of ours correlates now. */ clear(): void { this.held.takeAll(); + this.octets = 0; this.idleWaiters.settle(); } @@ -90,10 +114,13 @@ export class HeldMessages { if (!oldest) return; - const [seqNr] = oldest; + const [seqNr, held] = oldest; + this.octets -= held.octets; this.log.warn('heldMessages - buffer full, dropping the oldest message', { max: this.max, + maxOctets: this.maxOctets, + octets: this.octets, seqNr: Number(seqNr), }); } @@ -104,12 +131,21 @@ export class HeldMessages { if (expired.length === 0) return; + for (const [, held] of expired) { + this.octets -= held.octets; + } + this.log.warn('heldMessages - messages the application never answered', { messages: expired.length, }); this.settle(); } + private delete(key: string, held: Held): void { + this.held.delete(key); + this.octets -= held.octets; + } + private settle(): void { if (this.held.size === 0) this.idleWaiters.settle(); } diff --git a/src/incoming-requests.ts b/src/incoming-requests.ts index e29ac52..244f897 100644 --- a/src/incoming-requests.ts +++ b/src/incoming-requests.ts @@ -13,6 +13,7 @@ import { Reassembler, decodeSegments } from './reassembly.ts'; import { bindCommands, defaults, standsInFor } from './session-options.ts'; import { concatOf } from './concat.ts'; import { createSms } from './sms.ts'; +import { detach } from './retained-pdu.ts'; import { dlrFromPdu } from './dlr.ts'; import { paramText } from './defs/types.ts'; import { respIdParams, segmentId } from './sms-id.ts'; @@ -72,6 +73,7 @@ export class IncomingRequests { this.held = new HeldMessages({ log: options.log, max: defaults.maxHeldMessages, + maxOctets: defaults.maxHeldOctets, timeout: defaults.heldMessageTimeout, }); this.log = options.log; @@ -213,7 +215,7 @@ export class IncomingRequests { const concat = concatOf(pduObj); if (!concat) { - this.emitSms([pduObj]); + this.emitSms([detach(pduObj)]); return; } diff --git a/src/reassembly.ts b/src/reassembly.ts index c53d993..1c2d6bd 100644 --- a/src/reassembly.ts +++ b/src/reassembly.ts @@ -1,12 +1,11 @@ import type { Concat } from './concat.ts'; -import type { ParamValue } from './defs/types.ts'; import type { PduObject } from './pdu.ts'; import type { SmppLog } from './log.ts'; -import type { Tlv } from './defs/tlvs.ts'; import { ExpiringGroups } from './expiring-groups.ts'; import { decodeMessage } from './message.ts'; +import { detach, retainedOctets } from './retained-pdu.ts'; import { messageOctets } from './message-body.ts'; -import { detachedTlv, paramNumber, paramText, tlvOctets } from './defs/types.ts'; +import { paramNumber, paramText } from './defs/types.ts'; import { uuidv7 } from './uuid.ts'; /** A concatenated message given up on, whose segments the peer has already been answered for. */ @@ -52,54 +51,6 @@ type Group = { total: number; }; -/** Wire reads hand back views, so retaining one segment would pin the whole PDU it arrived in. */ -function detach(pduObj: PduObject): PduObject { - const params: Record = {}; - const tlvs: Record = {}; - - for (const [name, value] of Object.entries(pduObj.params)) { - params[name] = Buffer.isBuffer(value) ? Buffer.from(value) : value; - } - - for (const [name, tlv] of Object.entries(pduObj.tlvs)) { - tlvs[name] = { ...tlv, tagValue: detachedTlv(tlv.tagValue) }; - } - - // short_message holds the same octets wherever it was not decoded, so one copy covers both. - const octets = Buffer.isBuffer(params.short_message) - ? params.short_message - : pduObj.shortMessageOctets && Buffer.from(pduObj.shortMessageOctets); - - return { ...pduObj, params, shortMessageOctets: octets, tlvs }; -} - -// Measured heap beyond the octets, so a segment of empty fields or empty TLVs is not free. -const segmentObjectOverhead = 1000; -const tlvObjectOverhead = 300; - -// A cstring param arrives as a string, and source_addr alone can carry most of a 1 MiB PDU. -function sizeOf(value: ParamValue): number { - if (Buffer.isBuffer(value)) return value.length; - - return typeof value === 'string' ? value.length : 0; -} - -function octetsOf(pduObj: PduObject): number { - let octets = segmentObjectOverhead; - - for (const value of Object.values(pduObj.params)) { - octets += sizeOf(value); - } - - for (const tlv of Object.values(pduObj.tlvs)) { - const listed = Array.isArray(tlv.tagValue) ? tlv.tagValue.length : 0; - - octets += tlvOctets(tlv.tagValue) + (1 + listed) * tlvObjectOverhead; - } - - return octets; -} - // NUL: the one octet a C-Octet String address cannot hold, so no sender can forge another's key. function groupKey(pduObj: PduObject, concat: Concat): string { return [ @@ -169,7 +120,7 @@ export class Reassembler { const group = existing ?? this.open(key, concat.total); const replaced = group.parts.get(concat.part); const segment = detach(pduObj); - const delta = octetsOf(segment) - (replaced === undefined ? 0 : octetsOf(replaced)); + const delta = retainedOctets(segment) - (replaced === undefined ? 0 : retainedOctets(replaced)); group.parts.set(concat.part, segment); group.octets += delta; diff --git a/src/retained-pdu.ts b/src/retained-pdu.ts new file mode 100644 index 0000000..641731a --- /dev/null +++ b/src/retained-pdu.ts @@ -0,0 +1,53 @@ +import type { ParamValue } from './defs/types.ts'; +import type { PduObject } from './pdu.ts'; +import type { Tlv } from './defs/tlvs.ts'; +import { detachedTlv, tlvOctets } from './defs/types.ts'; + +/** Wire reads hand back views, so retaining one PDU would pin the whole chunk it arrived in. */ +export function detach(pduObj: PduObject): PduObject { + const params: Record = {}; + const tlvs: Record = {}; + + for (const [name, value] of Object.entries(pduObj.params)) { + params[name] = Buffer.isBuffer(value) ? Buffer.from(value) : value; + } + + for (const [name, tlv] of Object.entries(pduObj.tlvs)) { + tlvs[name] = { ...tlv, tagValue: detachedTlv(tlv.tagValue) }; + } + + // short_message holds the same octets wherever it was not decoded, so one copy covers both. + const octets = Buffer.isBuffer(params.short_message) + ? params.short_message + : pduObj.shortMessageOctets && Buffer.from(pduObj.shortMessageOctets); + + return { ...pduObj, params, shortMessageOctets: octets, tlvs }; +} + +// Measured heap beyond the octets, so a PDU of empty fields or empty TLVs is not free. +const pduObjectOverhead = 1000; +const tlvObjectOverhead = 300; + +// A cstring param arrives as a string, and source_addr alone can carry most of a 1 MiB PDU. +function sizeOf(value: ParamValue): number { + if (Buffer.isBuffer(value)) return value.length; + + return typeof value === 'string' ? value.length : 0; +} + +/** Roughly the heap a detached PDU holds. */ +export function retainedOctets(pduObj: PduObject): number { + let octets = pduObjectOverhead; + + for (const value of Object.values(pduObj.params)) { + octets += sizeOf(value); + } + + for (const tlv of Object.values(pduObj.tlvs)) { + const listed = Array.isArray(tlv.tagValue) ? tlv.tagValue.length : 0; + + octets += tlvOctets(tlv.tagValue) + (1 + listed) * tlvObjectOverhead; + } + + return octets; +} diff --git a/src/session-options.ts b/src/session-options.ts index ea1dbf0..3351b92 100644 --- a/src/session-options.ts +++ b/src/session-options.ts @@ -128,6 +128,7 @@ export const defaults = { heldMessageTimeout: 300_000, maxDlrMerges: 1000, maxHeldMessages: 1000, + maxHeldOctets: defaultMaxOctets, maxOutstanding: 10, maxReassembly: 1000, reassemblyTimeout: 300_000, diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index d4b4314..7c8f8ad 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -29,6 +29,7 @@ import { client } from '../src/client.ts'; import { closeAfter, closeListenerAfter } from './teardown.ts'; import { concatOf } from '../src/concat.ts'; import { consts } from '../src/defs/constants.ts'; +import { detach } from '../src/retained-pdu.ts'; import { errors } from '../src/defs/errors.ts'; import { paramNumber, paramText } from '../src/defs/types.ts'; import { server } from '../src/server.ts'; @@ -1483,7 +1484,7 @@ describe('held message bounds', () => { } test('drops the message held longest rather than holding every one', () => { - const held = new HeldMessages({ log: silentLog, max: 2, timeout: 10_000 }); + const held = new HeldMessages({ log: silentLog, max: 2, maxOctets: 1_000_000, timeout: 10_000 }); const oldest = message(1); held.hold(oldest); @@ -1501,9 +1502,50 @@ describe('held message bounds', () => { held.clear(); }); + // submitPdu() holds 1026 octets by the maxOctets charge: its object, and the three text fields. + test('drops the message held longest once the octets held pass the cap', () => { + const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 2100, timeout: 10_000 }); + const oldest = message(1); + + held.hold(oldest); + held.hold(message(2)); + + assert.equal(held.size, 2); + + held.hold(message(3)); + + assert.equal(held.size, 2); + assert.equal(held.has(oldest), false); + + held.clear(); + }); + + // Dropping it would leave the drain blind to a message the peer is still owed an answer for. + test('keeps a message larger than the cap on its own', () => { + const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 1000, timeout: 10_000 }); + const large = message(2); + + held.hold(message(1)); + held.hold(large); + + assert.equal(held.size, 1); + assert.equal(held.has(large), true); + + held.clear(); + }); + + test('holds a message detached from the chunk it was read from', () => { + const chunk = Buffer.alloc(64 * 1024); + const carried = submitPdu(1); + const retained = detach({ ...carried, params: { ...carried.params, short_message: chunk.subarray(16, 20) } }); + + assert.ok(Buffer.isBuffer(retained.params.short_message)); + assert.notEqual(retained.params.short_message.buffer, chunk.buffer); + }); + test('gives up on a message the application never answers', () => { let now = 0; - const held = new HeldMessages({ log: silentLog, max: 10, now: () => now, timeout: 60 }); + const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 1_000_000, now: () => now, timeout: 60 }); held.hold(message(1)); now = 61; @@ -1519,7 +1561,7 @@ describe('held message bounds', () => { // Without this the drain sits out its whole budget before returning what a sweep already settled. test('wakes a waiting drain when the last message expires', async () => { let now = 0; - const held = new HeldMessages({ log: silentLog, max: 10, now: () => now, timeout: 60 }); + const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 1_000_000, now: () => now, timeout: 60 }); held.hold(message(1)); diff --git a/todo.md b/todo.md index 2650bab..843a501 100644 --- a/todo.md +++ b/todo.md @@ -6,14 +6,6 @@ hard rules first — they constrain every item below. This is a working file that sets its own rules. The documentation conventions in AGENTS.md do not govern it, and nothing here is a source anything else may cite. -## Security - -- [ ] **Bound the heap `HeldMessages` keeps, not only its count.** It holds up to 1000 messages the - application has not answered for up to 300 s, each as the PDUs it arrived in, undetached and - uncharged: one 200 KB PDU of 50,000 empty unknown TLVs is 15 MB of heap, and a completed - reassembly is up to `maxOctets`. A peer faster than an application answering asynchronously - holds gigabytes per session. From the stability review of #27. - ## Status The rewrite is **feature complete and green**: the suite, lint and typecheck are clean on Node 18