Refuse a message past the held-message bound with ESME_RTHROTTLED, and answer a full reassembly buffer the same way #29

Merged
lilleman merged 5 commits from held-throttle into main 2026-09-26 15:24:43 +02:00
9 changed files with 113 additions and 104 deletions
Showing only changes of commit 855ccbe75f - Show all commits
+4 -1
View File
@@ -302,7 +302,10 @@ the file.
- The drain waits on the messages the application holds, and `sendResp()` is what says it is done - The drain waits on the messages the application holds, and `sendResp()` is what says it is done
with one. with one.
- The drain's wait on the application ignores `shutdownTimeout: 0`. - The drain's wait on the application ignores `shutdownTimeout: 0`.
- What the application holds unanswered is capped on constants. - What the application holds unanswered is capped on constants, and a message past the cap is
refused.
- A store at its bound answers `ESME_RTHROTTLED` to a `submit_sm` and `ESME_RX_T_APPN` to a
`deliver_sm`.
- A reconnect keeps the delivery-receipt merges; everything else the link held is dropped. - A reconnect keeps the delivery-receipt merges; everything else the link held is dropped.
- A message id base is merged at most once. - 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. - A send that never reached the socket waits for the next link; one that did is counted, not resent.
+6 -3
View File
@@ -40,9 +40,12 @@
count as next to nothing, so a peer could hold far more than the cap. **Raise a `maxOctets` you 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 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. 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 - A message arriving while the application holds 1000 unanswered, or 64 MiB of them by the
1000-message cap: the oldest is dropped with a warning, as over the count. The library used to `maxOctets` charge, is refused with `ESME_RTHROTTLED` (`ESME_RX_T_APPN` on a `deliver_sm`), so the
hold up to 1000 messages of any size for an application that answered none of them. peer keeps it and retries. The oldest used to be dropped to make room, which freed nothing while
the application still held it and let `close()` stop waiting for a message the peer was owed.
- A segment the reassembly buffer has no room for is refused with `ESME_RTHROTTLED`, where it was
`ESME_RMSGQFUL`.
- `server()` refuses a `maxOctets` below 1 or not a whole number, `Infinity` included, like its - `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. 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 - `callback_num`, `callback_num_atag`, `callback_num_pres_ind`, `broadcast_area_identifier` and
+6 -5
View File
@@ -383,9 +383,10 @@ holds for `session.send()`.
message has failed. Answering through `sendReturn()` instead leaves the wait running. 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. 3. Tear down what is left, resolving to an `err` that says what was lost.
At most 1000 unanswered messages, and 64 MiB of them by the `maxOctets` charge, are held for five At most 1000 unanswered messages, and 64 MiB of them by the `maxOctets` charge, are held. Past
minutes each; what falls out of a bound is dropped with a warning on the log and waited for no either bound a new message is refused with `ESME_RTHROTTLED` (`ESME_RX_T_APPN` on a `deliver_sm`),
longer. None of the bounds is an option. so the peer retries it. One held five minutes 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 `close({ signal })` cuts the wait short. `unbind()` takes no signal, and waits a further
`responseTimeout` for its own response. `responseTimeout` for its own response.
@@ -488,8 +489,8 @@ even where the SMSC took some of its segments; their receipts still arrive as `d
**Multipart is answered on arrival.** Each segment is answered as it lands, because a relaying SMSC **Multipart is answered on arrival.** Each segment is answered as it lands, because a relaying SMSC
will not send the next until the last is answered. The answer is `ESME_ROK`, unless the segment will not send the next until the last is answered. The answer is `ESME_ROK`, unless the segment
numbers itself into no message this session can join, which refuses it, or the reassembly buffer is numbers itself into no message this session can join, which refuses it, or the reassembly buffer or
full, which asks the SMSC to keep it and try again. `sms.answeredOnArrival` says whether the message the unanswered messages are at their bound, which asks the SMSC to keep it and try again. `sms.answeredOnArrival` says whether the message
you hold was answered that way; a segment count cannot, since a peer may number a message one part you hold was answered that way; a segment count cannot, since a peer may number a message one part
of one. of one.
+20 -7
View File
@@ -603,7 +603,7 @@ rule and an index of the titles below.
untouched, and is where a caller-chosen id and a refusal live; `onRequest` is the escape hatch for untouched, and is where a caller-chosen id and a refusal live; `onRequest` is the escape hatch for
an application that must refuse a PDU the `sms` event could not have shown it yet. `collect()` an application that must refuse a PDU the `sms` event could not have shown it yet. `collect()`
answers every segment it will not carry rather than leaving it unanswered, which is the same stall answers every segment it will not carry rather than leaving it unanswered, which is the same stall
in miniature: the field that numbered it where the segment belongs to no group, `ESME_RMSGQFUL` in miniature: the field that numbered it where the segment belongs to no group, the retry status
where the segment's own arrival overran the octet cap, since a peer told that still holds it. Rejected: where the segment's own arrival overran the octet cap, since a peer told that still holds it. Rejected:
answering every segment but the one that completes the group, which leaves the peer holding some answering every segment but the one that completes the group, which leaves the peer holding some
segments accepted and one refused with nothing in SMPP to retract the rest, and still cannot honour segments accepted and one refused with nothing in SMPP to retract the rest, and still cannot honour
@@ -677,12 +677,25 @@ rule and an index of the titles below.
back to `responseTimeout`, the same answer the link gate's hold already takes — and to that 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. 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 - **What the application holds unanswered is capped on constants, and a message past the cap is
raise is the point: an application that answers nothing would otherwise grow it for the life of refused.** A bound the application cannot raise is the point: an application that answers nothing
the link, which goal 4 forbids. Reassembly's `maxOctets` is an option because it bounds what the would otherwise grow it for the life of the link, which goal 4 forbids. Reassembly's `maxOctets`
peer sends; this bounds what the application leaves unanswered. A message that falls out of a bound is an option because it bounds what the peer sends; this bounds what the application leaves
is one the drain stops waiting for, so `close()` can report fewer unanswered than there were — unanswered. Maintainer's call, 2026-09-26: a message arriving past the cap is refused with the
accepted, because the alternative is holding what nothing will answer. retry status, so the peer keeps it (goal 2). Rejected: dropping the oldest to make room, which
frees nothing while the application still holds its `Sms`, and stops the drain waiting for a
message the peer is owed. Rejected: pausing the socket, which also stalls every answer and
`enquire_link` on the link. A message held past its timeout is still dropped, so `close()` can
report fewer unanswered than there were — accepted, because the alternative is holding what
nothing will answer.
- **A store at its bound answers `ESME_RTHROTTLED` to a `submit_sm` and `ESME_RX_T_APPN` to a
`deliver_sm`.** Maintainer's call, 2026-09-26, for reassembly and held messages alike, so "keep it
and retry" has one spelling per direction. `ESME_RTHROTTLED` asks the sender to slow down, which
is what the peer outrunning us needs, and operators send it (Vonage, LINK Mobility, Route Mobile,
Jasmin), so clients built against them meet it (goal 1). Rejected: `ESME_RMSGQFUL`, which names an
exhausted queue and no rate. `ESME_RTHROTTLED` is the SMSC's to send, so an ESME answers with
SMPP 3.4's temporary receiver error, the one an SMSC retries on (goal 3).
- **A reconnect keeps the delivery-receipt merges; everything else the link held is dropped.** - **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()` `onDelivery()` answers each receipt before the group it belongs to is complete, and `teardown()`
+7 -26
View File
@@ -30,7 +30,6 @@ export class HeldMessages {
private readonly held: ExpiringGroups<Held>; private readonly held: ExpiringGroups<Held>;
private readonly idleWaiters = new IdleWaiters(); private readonly idleWaiters = new IdleWaiters();
private readonly log: SmppLog; private readonly log: SmppLog;
private readonly max: number;
private readonly maxOctets: number; private readonly maxOctets: number;
private octets = 0; private octets = 0;
@@ -42,7 +41,6 @@ export class HeldMessages {
timeout: options.timeout, timeout: options.timeout,
}); });
this.log = options.log; this.log = options.log;
this.max = options.max;
this.maxOctets = options.maxOctets; this.maxOctets = options.maxOctets;
} }
@@ -50,7 +48,13 @@ export class HeldMessages {
return this.held.size; return this.held.size;
} }
/** An application that answers no message at all may not grow this without end. */ /** Whether a message arriving now is past the bound, once the expired are swept. */
full(): boolean {
this.sweep();
return this.held.full || this.octets >= this.maxOctets;
}
hold(pduObjs: PduObject[]): void { hold(pduObjs: PduObject[]): void {
const key = keyOf(pduObjs); const key = keyOf(pduObjs);
@@ -63,17 +67,10 @@ export class HeldMessages {
if (replaced) { if (replaced) {
this.log.warn('heldMessages - replacing a message on a re-used sequence number', { seqNr: Number(key) }); this.log.warn('heldMessages - replacing a message on a re-used sequence number', { seqNr: Number(key) });
this.delete(key, replaced); this.delete(key, replaced);
} else if (this.held.full) {
this.dropOldest();
} }
const octets = pduObjs.reduce((sum, pduObj) => sum + retainedOctets(pduObj), 0); 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.held.set(key, { octets, pduObjs });
this.octets += octets; this.octets += octets;
} }
@@ -109,22 +106,6 @@ export class HeldMessages {
return this.idleWaiters.wait(() => this.held.size, timeout, signal); return this.idleWaiters.wait(() => this.held.size, timeout, signal);
} }
private dropOldest(): void {
const oldest = this.held.takeOldest();
if (!oldest) return;
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),
});
}
/** Drops every message past its deadline. Runs before each hold and on its own timer. */ /** Drops every message past its deadline. Runs before each hold and on its own timer. */
sweep(): void { sweep(): void {
const expired = this.held.takeExpired(); const expired = this.held.takeExpired();
+16 -2
View File
@@ -19,7 +19,11 @@ import { paramText } from './defs/types.ts';
import { respIdParams, segmentId } from './sms-id.ts'; import { respIdParams, segmentId } from './sms-id.ts';
import { respNameFor } from './defs/commands.ts'; import { respNameFor } from './defs/commands.ts';
/** SMPP 3.4 lists ESME_RMSGQFUL under submit_sm_resp only; 4.6.2's retryable code is another. */ /** Asks the peer to keep the message and retry: RTHROTTLED is the SMSC's to send, so an ESME's is another. */
function throttledStatus(carriedAs: string): ErrorName {
return carriedAs === 'submit_sm' ? 'ESME_RTHROTTLED' : 'ESME_RX_T_APPN';
}
export function refusedSegmentStatus( export function refusedSegmentStatus(
carriedAs: string, carriedAs: string,
refusal: Refusal, refusal: Refusal,
@@ -30,7 +34,7 @@ export function refusedSegmentStatus(
return spelling === 'sar' ? 'ESME_RINVTLVVAL' : 'ESME_RINVESMCLASS'; return spelling === 'sar' ? 'ESME_RINVTLVVAL' : 'ESME_RINVESMCLASS';
} }
return carriedAs === 'submit_sm' ? 'ESME_RMSGQFUL' : 'ESME_RX_T_APPN'; return throttledStatus(carriedAs);
} }
const lostReasons: Record<LostGroup['reason'], string> = { const lostReasons: Record<LostGroup['reason'], string> = {
@@ -212,6 +216,16 @@ export class IncomingRequests {
* one request at a time never sends the second segment until the first has been answered. * one request at a time never sends the second segment until the first has been answered.
*/ */
private async onMessage(pduObj: PduObject): Promise<void> { private async onMessage(pduObj: PduObject): Promise<void> {
if (this.held.full()) {
this.log.info('session - unanswered messages at their bound, asking the peer to retry', {
cmdName: pduObj.cmdName,
seqNr: pduObj.seqNr,
});
await this.session.sendReturn(pduObj, throttledStatus(this.carriedAs(pduObj)));
return;
}
const concat = concatOf(pduObj); const concat = concatOf(pduObj);
if (!concat) { if (!concat) {
+53 -48
View File
@@ -1482,76 +1482,82 @@ describe('held message bounds', () => {
return [submitPdu(seqNr)]; return [submitPdu(seqNr)];
} }
test('drops the message held longest rather than holding every one', () => { test('is full at its count, and a re-used sequence number replaces rather than adding', () => {
const held = new HeldMessages({ log: silentLog, max: 2, maxOctets: 1_000_000, timeout: 10_000 }); const held = new HeldMessages({ log: silentLog, max: 2, maxOctets: 1_000_000, timeout: 10_000 });
const oldest = message(1); const first = message(1);
held.hold(oldest); held.hold(first);
held.hold(message(2)); held.hold(message(2));
held.hold(message(2)); held.hold(message(2));
assert.equal(held.size, 2, 'a re-used sequence number replaces rather than evicting');
assert.equal(held.has(oldest), true);
held.hold(message(3));
assert.equal(held.size, 2); assert.equal(held.size, 2);
assert.equal(held.has(oldest), false); assert.equal(held.full(), true);
assert.equal(held.has(first), true);
held.clear(); held.clear();
}); });
// submitPdu() holds 1026 octets by the maxOctets charge: its object, and the three text fields. // 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', () => { test('is full at its octet cap, until a message leaves by any way out', () => {
let now = 0; let now = 0;
const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 2100, now: () => now, timeout: 10_000 }); const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 2000, now: () => now, timeout: 10_000 });
const oldest = message(1); const answered = 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.hold(answered);
assert.equal(held.full(), false);
held.hold(message(2));
assert.equal(held.full(), true);
held.release(answered); held.release(answered);
held.hold(message(5)); assert.equal(held.full(), false, 'after a release');
assert.equal(held.size, 2, 'after a release'); held.hold(message(3));
now = 20_000; now = 20_000;
held.sweep(); held.sweep();
now = 0; now = 0;
held.hold(message(6)); assert.equal(held.full(), false, 'after a sweep');
held.hold(message(7)); held.hold(message(4));
assert.equal(held.size, 2, 'after a sweep'); held.hold(message(5));
held.clear();
held.hold(message(8));
held.hold(message(9));
assert.equal(held.size, 2, 'after a clear');
held.clear(); held.clear();
assert.equal(held.full(), false, 'after a clear');
}); });
// Dropping it would leave the drain blind to a message the peer is still owed an answer for. // Dropping one the application still holds frees nothing, and the drain stops waiting for it.
test('keeps a message larger than the cap on its own', () => { test('refuses what arrives past the bound with a status that asks the peer to retry', async t => {
const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 1000, timeout: 10_000 }); const session = new Session({ sock: new net.Socket() });
const large = message(2);
held.hold(message(1)); closeAfter(t, session);
held.hold(large); session.boundAs = 'transceiver';
assert.equal(held.size, 1); const incoming = new IncomingRequests({
assert.equal(held.has(large), true); dlrMerger: new DlrMerger({ log: silentLog, max: 10, timeout: 10_000 }),
log: silentLog,
sendPastDrain: () => Promise.resolve({ err: new Error('never sent') }),
session,
});
const answers: (ErrorName | undefined)[] = [];
let messages = 0;
held.clear(); session.sendReturn = (_pdu, status) => {
answers.push(status);
return Promise.resolve({});
};
session.on('sms', () => { messages++; });
for (let seqNr = 1; seqNr <= 1000; seqNr++) {
await incoming.handle(submitPdu(seqNr));
}
assert.equal(messages, 1000);
await incoming.handle(submitPdu(1001));
await incoming.handle(segment(7, 1, 2));
assert.equal(messages, 1000);
assert.deepEqual(answers, ['ESME_RTHROTTLED', 'ESME_RTHROTTLED']);
incoming.clear();
}); });
test('holds a message detached from the chunk it was read from', async t => { test('holds a message detached from the chunk it was read from', async t => {
@@ -2213,16 +2219,15 @@ describe('reassembly bounds', () => {
// Jasmin dispatches one request per connector at a time: holding a group unanswered until it was // Jasmin dispatches one request per connector at a time: holding a group unanswered until it was
// whole deadlocked every multi-segment message against it (interop-tests/findings/03-jasmin.md). // whole deadlocked every multi-segment message against it (interop-tests/findings/03-jasmin.md).
describe('the status a refused segment is answered with', () => { describe('the status a refused segment is answered with', () => {
// SMPP 3.4 lists ESME_RMSGQFUL under submit_sm_resp only; 4.6.2's retryable code is another.
test('names one the command the segment arrived on defines', () => { test('names one the command the segment arrived on defines', () => {
assert.equal(refusedSegmentStatus('submit_sm', 'full', 'udh'), 'ESME_RMSGQFUL'); assert.equal(refusedSegmentStatus('submit_sm', 'full', 'udh'), 'ESME_RTHROTTLED');
assert.equal(refusedSegmentStatus('deliver_sm', 'full', 'udh'), 'ESME_RX_T_APPN'); assert.equal(refusedSegmentStatus('deliver_sm', 'full', 'udh'), 'ESME_RX_T_APPN');
assert.equal(refusedSegmentStatus('submit_sm', 'unplaceable', 'udh'), 'ESME_RINVESMCLASS'); assert.equal(refusedSegmentStatus('submit_sm', 'unplaceable', 'udh'), 'ESME_RINVESMCLASS');
assert.equal(refusedSegmentStatus('deliver_sm', 'unplaceable', 'udh'), 'ESME_RINVESMCLASS'); assert.equal(refusedSegmentStatus('deliver_sm', 'unplaceable', 'udh'), 'ESME_RINVESMCLASS');
// esm_class is 0x00 on a sar_* segment and entirely valid: the TLV values are what cannot be honoured. // esm_class is 0x00 on a sar_* segment and entirely valid: the TLV values are what cannot be honoured.
assert.equal(refusedSegmentStatus('submit_sm', 'unplaceable', 'sar'), 'ESME_RINVTLVVAL'); assert.equal(refusedSegmentStatus('submit_sm', 'unplaceable', 'sar'), 'ESME_RINVTLVVAL');
assert.equal(refusedSegmentStatus('deliver_sm', 'unplaceable', 'sar'), 'ESME_RINVTLVVAL'); assert.equal(refusedSegmentStatus('deliver_sm', 'unplaceable', 'sar'), 'ESME_RINVTLVVAL');
assert.equal(refusedSegmentStatus('submit_sm', 'full', 'sar'), 'ESME_RMSGQFUL'); assert.equal(refusedSegmentStatus('submit_sm', 'full', 'sar'), 'ESME_RTHROTTLED');
}); });
// Which command that is, for the one that travels both ways, is what the end it arrived at says. // Which command that is, for the one that travels both ways, is what the end it arrived at says.
@@ -2232,7 +2237,7 @@ describe('the status a refused segment is answered with', () => {
assert.equal(standsInFor('deliver_sm', 'esme'), 'deliver_sm'); assert.equal(standsInFor('deliver_sm', 'esme'), 'deliver_sm');
assert.equal(standsInFor('submit_sm', 'smsc'), 'submit_sm'); assert.equal(standsInFor('submit_sm', 'smsc'), 'submit_sm');
assert.equal(standsInFor('enquire_link', 'smsc'), 'enquire_link'); assert.equal(standsInFor('enquire_link', 'smsc'), 'enquire_link');
assert.equal(refusedSegmentStatus(standsInFor('data_sm', 'smsc'), 'full', 'udh'), 'ESME_RMSGQFUL'); assert.equal(refusedSegmentStatus(standsInFor('data_sm', 'smsc'), 'full', 'udh'), 'ESME_RTHROTTLED');
assert.equal(refusedSegmentStatus(standsInFor('data_sm', 'esme'), 'full', 'udh'), 'ESME_RX_T_APPN'); assert.equal(refusedSegmentStatus(standsInFor('data_sm', 'esme'), 'full', 'udh'), 'ESME_RX_T_APPN');
}); });
}); });
+1 -1
View File
@@ -932,7 +932,7 @@ describe('receiving', () => {
assert.ok(refused.pduObj); assert.ok(refused.pduObj);
assert.equal(refused.pduObj.cmdName, 'data_sm_resp'); assert.equal(refused.pduObj.cmdName, 'data_sm_resp');
assert.equal(refused.pduObj.cmdStatus, 'ESME_RMSGQFUL'); assert.equal(refused.pduObj.cmdStatus, 'ESME_RTHROTTLED');
}); });
test('reassembles a concatenated message whose segments arrived in message_payload', async t => { test('reassembles a concatenated message whose segments arrived in message_payload', async t => {
-11
View File
@@ -6,17 +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 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. govern it, and nothing here is a source anything else may cite.
## Security
- [ ] **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 ## Status
The rewrite is **feature complete and green**: the suite, lint and typecheck are clean on Node 18 The rewrite is **feature complete and green**: the suite, lint and typecheck are clean on Node 18