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
8 changed files with 50 additions and 21 deletions
Showing only changes of commit b9167073d4 - Show all commits
+5 -4
View File
@@ -40,10 +40,11 @@
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.
- A message arriving while the application holds 1000 unanswered, or 64 MiB of them by the - A message arriving while the application holds 1000 unanswered, or 64 MiB of them counted the way
`maxOctets` charge, is refused with `ESME_RTHROTTLED` (`ESME_RX_T_APPN` on a `deliver_sm`), so the `maxOctets` counts segments, is refused with `ESME_RTHROTTLED` (`ESME_RX_T_APPN` on a
peer keeps it and retries. The oldest used to be dropped to make room, which freed nothing while `deliver_sm`), so the peer keeps it and retries. **Call `sendResp()` on every `sms`, multipart
the application still held it and let `close()` stop waiting for a message the peer was owed. included**: 1000 left unanswered now stop inbound traffic for up to five minutes, where the oldest
used to be dropped with a warning.
- A `submit_sm` segment the reassembly buffer has no room for is refused with `ESME_RTHROTTLED`, - A `submit_sm` segment the reassembly buffer has no room for is refused with `ESME_RTHROTTLED`,
where it was `ESME_RMSGQFUL`. 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
+11 -7
View File
@@ -101,9 +101,10 @@ session.on('sms', async sms => {
}); });
``` ```
Call `sendResp()` for every message; it is part of the protocol. Delivery receipts reach you as Call `sendResp()` for every message, multipart included: until you do, it counts toward the bound
`dlr` events, not here. A multipart message arrives reassembled and already answered segment by past which the peer's messages are refused. Delivery receipts reach you as `dlr` events, not here. A
segment, so `sendResp()` there only says you are done with it: [Receiving in depth](#receiving-in-depth). multipart message arrives reassembled and already answered segment by segment, so `sendResp()` there
puts nothing on the wire and releases it: [Receiving in depth](#receiving-in-depth).
## Run an SMPP server ## Run an SMPP server
@@ -383,10 +384,7 @@ 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.
A message arriving while 1000 unanswered messages, or 64 MiB of them by the `maxOctets` charge, are A message left unanswered for five minutes is no longer waited for.
held is refused with `ESME_RTHROTTLED` (`ESME_RX_T_APPN` on a `deliver_sm`), 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.
@@ -434,6 +432,12 @@ const { err, pduObj } = await session.send({
each is two messages. each is two messages.
- **Answered on arrival.** Each segment was answered as it landed, before you see the message: - **Answered on arrival.** Each segment was answered as it landed, before you see the message:
[Server in depth](#server-in-depth). [Server in depth](#server-in-depth).
- **Unanswered messages.** While 1000 messages you have not called `sendResp()` on, or 64 MiB of
them counted the way `maxOctets` counts segments, are held, every new message is refused with
`ESME_RTHROTTLED` (`ESME_RX_T_APPN` on a `deliver_sm`) so the peer retries it, and no `sms` fires.
Reaching the bound logs one `warn`, and falling back below it one `info`. A message left five
minutes is dropped from the count with a `warn`; a later `sendResp()` still answers it. Neither
bound is an option.
- **Where the body is.** A body in the `message_payload` TLV, SMPP's way of carrying up to 64 KB and - **Where the body is.** A body in the `message_payload` TLV, SMPP's way of carrying up to 64 KB and
the only place a `data_sm` has, reads exactly like one in `short_message`, concatenated messages the only place a `data_sm` has, reads exactly like one in `short_message`, concatenated messages
and receipts included. A PDU filling both is read from `short_message`. and receipts included. A PDU filling both is read from `short_message`.
+3 -1
View File
@@ -685,7 +685,9 @@ rule and an index of the titles below.
retry status, so the peer keeps it (goal 2). Rejected: dropping the oldest to make room, which 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 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 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 `enquire_link` on the link. Reaching the bound is a log line and nothing more (goal 8): an event
or a public count would be surface for what the application already knows, since it is the one
not answering. 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 report fewer unanswered than there were — accepted, because the alternative is holding what
nothing will answer. nothing will answer.
+1
View File
@@ -19,6 +19,7 @@ services:
image: nicolaka/netshoot:v0.16 image: nicolaka/netshoot:v0.16
<<: *log-limits <<: *log-limits
command: ["sleep", "infinity"] command: ["sleep", "infinity"]
init: true
dumbclient-w2000: dumbclient-w2000:
build: ./interop-tests/peers/dumbclient build: ./interop-tests/peers/dumbclient
+5 -6
View File
@@ -100,8 +100,8 @@ event to a listener added after it fired. Fixed by attaching every session's `cl
Three runs of `./interop-tests/run.py dumbclient`. Run 1 (the original 300,000-count soak) surfaced Three runs of `./interop-tests/run.py dumbclient`. Run 1 (the original 300,000-count soak) surfaced
the S6 harness bug above; run 2 fixed it; run 3, with the namespace owner above and the held-message the S6 harness bug above; run 2 fixed it; run 3, with the namespace owner above and the held-message
throttle in `src/`, is the one Scenarios reports. The capture figures below are run 2's. `smppload.test.ts` passed on every run it was given (three, across the investigation throttle in `src/`, is the one Scenarios reports. The capture figures below are run 2's.
above); its one scenario needs no repeat - a second run reproduces the identical corrupted PDU, `smppload.test.ts` passed on every run it was given (three, across the investigation above); its one scenario needs no repeat - a second run reproduces the identical corrupted PDU,
adding nothing. adding nothing.
``` ```
@@ -152,10 +152,9 @@ that never gets as far as a readable PDU.
- **smppload's `bind_transceiver` is corrupted on the wire** - see Setup. Not chased past `oserl`'s - **smppload's `bind_transceiver` is corrupted on the wire** - see Setup. Not chased past `oserl`'s
`pack/2` (which is correct on inspection) given the time-box. `pack/2` (which is correct on inspection) given the time-box.
- **`smpp-dumb-client` treats `ESME_RTHROTTLED` as final** - a throttled message counts as sent - **`smpp-dumb-client` treats `ESME_RTHROTTLED` as final** - a throttled message counts as sent
and is never resubmitted. Its `enquire_link` and is never resubmitted. Its `enquire_link` interval (10s once bound as an ESME) is hardcoded
interval (10s once bound as an ESME) is also hardcoded (`smpp.go`, `enquireSender(10)`), not (`smpp.go`, `enquireSender(10)`), not exposed through `config.yml` at all - the no-ping binary
exposed through `config.yml` at all - the no-ping binary built for S6 patches the call site out built for S6 patches the call site out rather than configuring it.
rather than configuring it.
- **`smpp.remote` takes a literal IP, never a hostname** (`net.ParseIP`, no DNS resolution) - see - **`smpp.remote` takes a literal IP, never a hostname** (`net.ParseIP`, no DNS resolution) - see
Setup. Setup.
+14
View File
@@ -71,6 +71,7 @@ export class IncomingRequests {
private readonly smsIdFormat: SmsIdFormat; private readonly smsIdFormat: SmsIdFormat;
private readonly systemId: string; private readonly systemId: string;
private linkGeneration = 0; private linkGeneration = 0;
private refusing = false;
constructor(options: IncomingRequestsOptions) { constructor(options: IncomingRequestsOptions) {
this.dlrMerger = options.dlrMerger; this.dlrMerger = options.dlrMerger;
@@ -148,6 +149,7 @@ export class IncomingRequests {
/** Drops the segments of every message that never became whole, and of every one still held. */ /** Drops the segments of every message that never became whole, and of every one still held. */
clear(): void { clear(): void {
this.linkGeneration++; this.linkGeneration++;
this.refusing = false;
this.held.clear(); this.held.clear();
this.reassembler.clear(); this.reassembler.clear();
} }
@@ -217,6 +219,13 @@ export class IncomingRequests {
*/ */
private async onMessage(pduObj: PduObject): Promise<void> { private async onMessage(pduObj: PduObject): Promise<void> {
if (this.held.full()) { if (this.held.full()) {
if (!this.refusing) {
this.refusing = true;
this.log.warn('session - unanswered messages at their bound, refusing new ones until the application answers', {
messages: this.held.size,
});
}
this.log.verbose('session - unanswered messages at their bound, asking the peer to retry', { this.log.verbose('session - unanswered messages at their bound, asking the peer to retry', {
cmdName: pduObj.cmdName, cmdName: pduObj.cmdName,
seqNr: pduObj.seqNr, seqNr: pduObj.seqNr,
@@ -226,6 +235,11 @@ export class IncomingRequests {
return; return;
} }
if (this.refusing) {
this.refusing = false;
this.log.info('session - unanswered messages below their bound, accepting again', { messages: this.held.size });
}
const concat = concatOf(pduObj); const concat = concatOf(pduObj);
if (!concat) { if (!concat) {
+6 -3
View File
@@ -1530,9 +1530,10 @@ describe('held message bounds', () => {
closeAfter(t, session); closeAfter(t, session);
session.boundAs = 'transceiver'; session.boundAs = 'transceiver';
const warnings: string[] = [];
const incoming = new IncomingRequests({ const incoming = new IncomingRequests({
dlrMerger: new DlrMerger({ log: silentLog, max: 10, timeout: 10_000 }), dlrMerger: new DlrMerger({ log: silentLog, max: 10, timeout: 10_000 }),
log: silentLog, log: { ...silentLog, warn: message => { warnings.push(message); } },
sendPastDrain: () => Promise.resolve({ err: new Error('never sent') }), sendPastDrain: () => Promise.resolve({ err: new Error('never sent') }),
session, session,
}); });
@@ -1557,12 +1558,14 @@ describe('held message bounds', () => {
assert.equal(received.length, defaults.maxHeldMessages); assert.equal(received.length, defaults.maxHeldMessages);
assert.deepEqual(answers, ['ESME_RTHROTTLED', 'ESME_RTHROTTLED']); assert.deepEqual(answers, ['ESME_RTHROTTLED', 'ESME_RTHROTTLED']);
assert.equal(warnings.length, 1, 'reaching the bound warns once, not per refusal');
// The refused first segment joined no group, so the second one completes nothing. // The refused first segment joined no group, so the second one is taken and completes nothing.
await received[0]?.sendResp(); await received[0]?.sendResp();
await delay(0); await new Promise(resolve => { setImmediate(resolve); });
await incoming.handle(segment(7, 2, 2)); await incoming.handle(segment(7, 2, 2));
assert.equal(answers.at(-1), 'ESME_ROK');
assert.equal(received.length, defaults.maxHeldMessages); assert.equal(received.length, defaults.maxHeldMessages);
incoming.clear(); incoming.clear();
}); });
+5
View File
@@ -405,6 +405,11 @@ and is also what the panel ranked hardest — two methods, one answer.
## Worth doing, not blocking ## Worth doing, not blocking
- [ ] **Make the dumbclient soak's memory sample evidence of no library leak again.** Its rss ends at
its maximum (298 MiB, heapUsed 81 MiB after 173,820 messages), which the harness's own per-id
`Set` and `answerOrder` explain but cannot separate from a leak in `src/`: sample the heap
after the bookkeeping is cleared. From the stability review of #29.
- [ ] **Decide whether `alert_notification` reaches the application as more than `incomingPduObj`.** - [ ] **Decide whether `alert_notification` reaches the application as more than `incomingPduObj`.**
It is the SMSC saying a handset it could not reach is reachable again (`esme_addr`, It is the SMSC saying a handset it could not reach is reachable again (`esme_addr`,
`ms_availability_status`); a client has no `onRequest`, so the raw PDU event is the only way in. `ms_availability_status`); a client has no `onRequest`, so the raw PDU event is the only way in.