diff --git a/AGENTS.md b/AGENTS.md index 86266bf..ce2ac3b 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 @@ -300,6 +301,8 @@ the file. request that handling did not answer. - The drain waits on the messages the application holds, and `sendResp()` is what says it is done with one. +- The drain's wait on the application ignores `shutdownTimeout: 0`. +- What the application holds unanswered is capped on constants. - A reconnect keeps the delivery-receipt merges; everything else the link held is dropped. - A message id base is merged at most once. - A send that never reached the socket waits for the next link; one that did is counted, not resent. diff --git a/CHANGELOG.md b/CHANGELOG.md index ae4e243..193be3f 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. The library used to + hold up to 1000 messages of any size for an application that answered none of them. - `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/docs/decisions.md b/docs/decisions.md index 5034230..72acb16 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -664,32 +664,25 @@ rule and an index of the titles below. down while the application was still answering a `submit_sm`, so the peer timed out and re-sent — the duplicate goal 2 forbids, in the direction the window already covers. No completion signal was added to the `sms` event: `sendResp()` is what an application already calls when it is done with a - message, so it is the one the drain waits for. Counting every inbound request until `sendReturn()` answered it was rejected — - an `onRequest` that deliberately answers nothing would then cost a full `shutdownTimeout` on every - close — and a message no listener took is released at once, since nothing is going to answer it. - A listener that failed before answering gives it up the same way, but only once every listener has: - a throw stops `emit()` where it stands, while a rejection leaves the others running, so the release - waits for the last of them rather than answering on their behalf. What ends the wait is the response - reaching the wire, not the call — a `sendResp()` the library refused, or one the socket would not - carry, leaves the message held, so `close()` still reports the one the peer is owed. Where the - segments were answered as they arrived there is no response left to write, so the call itself ends - the wait, an argument the library refuses excepted. `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, 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` 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 default where it is 0 as well, since neither option is an answer about - the application. What is held is capped and expiring like every other inbound store, on constants - rather than options, because a bound the application cannot raise is the point: an application that - answers nothing would otherwise grow it for the life of the link, which goal 4 forbids. A message - that falls out of the bound is one the drain stops waiting for, so `close()` can report fewer - unanswered than there were — accepted, because the alternative is holding what nothing will answer, - and both exits are logged. + message, so it is the one the drain waits for. Counting every inbound request until `sendReturn()` + answered it was rejected: an `onRequest` that deliberately answers nothing would then cost a full + `shutdownTimeout` on every close. The response reaching the wire ends the wait, so a `sendResp()` + the library refused or the socket would not carry leaves `close()` still reporting the message the + peer is owed. + +- **The drain's wait on the application ignores `shutdownTimeout: 0`.** 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 default where it is 0 as well, since neither option is an answer about the application. + +- **What the application holds unanswered is capped on constants.** A bound the application cannot + raise is the point: an application that answers nothing would otherwise grow it for the life of + the link, which goal 4 forbids. Reassembly's `maxOctets` is an option because it bounds what the + peer sends; this bounds what the application leaves unanswered. A message that falls out of a bound + is one the drain stops waiting for, so `close()` can report fewer unanswered than there were — + accepted, because the alternative is holding what nothing will answer. - **A reconnect keeps the delivery-receipt merges; everything else the link held is dropped.** `onDelivery()` answers each receipt before the group it belongs to is complete, and `teardown()` 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..0e2ad3b 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: 64 * 1024 * 1024, maxOutstanding: 10, maxReassembly: 1000, reassemblyTimeout: 300_000, diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index d4b4314..f210725 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -1483,7 +1483,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 +1501,88 @@ 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', () => { + let now = 0; + const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 2100, now: () => now, 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); + + // A message that leaves any other way gives its octets back, so two still fit afterwards. + const answered = message(4); + + held.hold(answered); + held.release(answered); + held.hold(message(5)); + assert.equal(held.size, 2, 'after a release'); + + now = 20_000; + held.sweep(); + now = 0; + held.hold(message(6)); + held.hold(message(7)); + assert.equal(held.size, 2, 'after a sweep'); + + held.clear(); + held.hold(message(8)); + held.hold(message(9)); + assert.equal(held.size, 2, 'after a clear'); + + 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', async t => { + const session = new Session({ sock: new net.Socket() }); + + closeAfter(t, session); + session.boundAs = 'transceiver'; + + const incoming = new IncomingRequests({ + dlrMerger: new DlrMerger({ log: silentLog, max: 10, timeout: 10_000 }), + log: silentLog, + sendPastDrain: () => Promise.resolve({ err: new Error('never sent') }), + session, + }); + const chunk = Buffer.alloc(64 * 1024); + const carried = submitPdu(1); + let received: Sms | undefined; + + session.on('sms', sms => { received = sms; }); + await incoming.handle({ ...carried, params: { ...carried.params, short_message: chunk.subarray(16, 20) } }); + + const retained = received?.pduObjs[0]?.params.short_message; + + assert.ok(Buffer.isBuffer(retained)); + assert.notEqual(retained.buffer, chunk.buffer); + incoming.clear(); + }); + 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 +1598,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..1cac818 100644 --- a/todo.md +++ b/todo.md @@ -8,11 +8,14 @@ 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. +- [ ] **Bound what a peer can make the application hold, not only the library.** An `sms` the + application is still answering pins its PDUs through `Sms.pduObjs` and its `sendResp`/`sendDlr` + closures, so the held-message caps free nothing while it works, and nothing slows the peer: a + peer faster than an application answering asynchronously (a DB write per message) grows the + heap without bound. Each eviction by the caps also stops the drain waiting for a message still + being answered, so `close()` can cut it off and the peer re-sends it. Flow control is a wire + change under goal 4 and the maintainer's call: answer `ESME_RTHROTTLED`, or stop reading the + socket, once the held count or octets are at the cap. From the stability review of #28. ## Status