Refuse to answer a message whose link went, instead of reporting it delivered

This commit is contained in:
2026-09-01 19:16:21 +02:00
parent a91c55cc64
commit 82a95593bd
9 changed files with 84 additions and 12 deletions
+4
View File
@@ -355,6 +355,10 @@ so it fails, and `sendSms()` and `sms.sendDlr()` count it in `unanswered`, wheth
under it, the peer never answered in time, or you aborted it after it went out. Neither applies with `reconnect: false`,
where a drop ends the session and every send after it is refused.
An answer cannot wait for a link that way, because it carries the sequence number the message arrived
on: `sms.sendResp()` on a message whose link dropped writes nothing and returns an `err`. Its
`sms.sendDlr()` still goes out, since a receipt is a request of its own, correlated by the id it names.
`responseTimeout` bounds the wait for a link and the wait for an answer separately, and a send also
queues for a `maxOutstanding` slot, which nothing bounds — so it is not a deadline for the call.
Pass `{ signal: AbortSignal.timeout(ms) }` when you need one.
+2 -2
View File
@@ -120,8 +120,8 @@ export class DlrMerger {
}
clear(): void {
this.groups.clear();
this.spent.clear();
this.groups.takeAll();
this.spent.takeAll();
}
/** Drops every group past its deadline. Runs before each collect and on its own timer. */
+6 -1
View File
@@ -58,9 +58,14 @@ export class ExpiringGroups<T> {
this.idle();
}
clear(): void {
/** Removes every group and hands them over. */
takeAll(): [string, T][] {
const taken: [string, T][] = [...this.entries].map(([key, entry]) => [key, entry.group]);
this.entries.clear();
this.idle();
return taken;
}
/** Removes every group past its deadline and hands them over. */
+10 -1
View File
@@ -20,6 +20,7 @@ function keyOf(pduObjs: PduObject[]): string | undefined {
/** The messages handed to the application that it has not answered yet, held by their segments. */
export class HeldMessages {
private readonly droppedWithLink = new WeakSet<PduObject[]>();
private readonly held: ExpiringGroups<PduObject[]>;
private readonly idleWaiters = new IdleWaiters();
private readonly log: SmppLog;
@@ -64,6 +65,11 @@ export class HeldMessages {
return key !== undefined && this.held.get(key) === pduObjs;
}
/** Whether this message's link went before it was answered, so no answer of ours correlates. */
lostLink(pduObjs: PduObject[]): boolean {
return this.droppedWithLink.has(pduObjs);
}
release(pduObjs: PduObject[]): void {
const key = keyOf(pduObjs);
@@ -76,7 +82,10 @@ export class HeldMessages {
/** Drops every message: their segments went with the link, so no answer of ours correlates now. */
clear(): void {
this.held.clear();
for (const [, pduObjs] of this.held.takeAll()) {
this.droppedWithLink.add(pduObjs);
}
this.idleWaiters.settle();
}
+1
View File
@@ -170,6 +170,7 @@ export class IncomingRequests {
session: this.session,
to: paramText(first.params.destination_addr),
}, {
lostLink: () => this.held.lostLink(pduObjs),
// A turn later, so a listener sending its receipt straight after the response still holds.
onAnswered: () => { setImmediate(() => { this.held.release(pduObjs); }); },
// Past the refusal only while a drain is still waiting for this message; an ordinary send after.
+1 -1
View File
@@ -163,7 +163,7 @@ export class Reassembler {
}
clear(): void {
this.groups.clear();
this.groups.takeAll();
this.octets = 0;
}
+10 -1
View File
@@ -54,6 +54,8 @@ export type SmsInput = {
/** What the session's incoming side gives a message so it can be answered and accounted for. */
export type SmsHandlers = {
/** Whether the link this message arrived on is gone, so no answer of ours correlates. */
lostLink: () => boolean;
onAnswered: () => void;
send: (input: PduObjectInput) => Promise<Result<{ pduObj: PduObject }>>;
};
@@ -76,7 +78,8 @@ export function createSms(input: SmsInput, handlers: SmsHandlers): Sms {
message: input.message,
pduObjs: input.pduObjs,
sendDlr: status => sendDlr(sms, handlers.send, status),
sendResp: options => sendResp(sms, answered, options ?? {}).finally(handlers.onAnswered),
sendResp: options => sendResp(sms, answered, options ?? {}, handlers.lostLink)
.finally(handlers.onAnswered),
session: input.session,
get smsId(): string {
return answered.smsId;
@@ -92,6 +95,7 @@ async function sendResp(
sms: Sms,
answered: { smsId: string },
options: SendRespOptions,
lostLink: () => boolean,
): Promise<VoidResult> {
const total = sms.pduObjs.length;
@@ -103,6 +107,11 @@ async function sendResp(
return { err: new Error('smsId must not be empty') };
}
// A response carries the sequence number it was asked on, which the next link knows nothing about.
if (lostLink()) {
return { err: new Error('The link this message arrived on is gone, so nothing would correlate the response') };
}
if (options.smsId !== undefined) answered.smsId = options.smsId;
const results = await Promise.all(sms.pduObjs.map((pduObj, index) => sms.session.sendReturn(
+49
View File
@@ -557,6 +557,40 @@ describe('reconnect', () => {
assert.equal(report.segments.length, 3);
});
test('refuses to answer a message whose link went, rather than reporting it delivered', async t => {
const smpp = await startServer(t);
const { session } = await connect(t, smpp, { reconnect: { maxDelay: 100, minDelay: 20 } });
assert.ok(session);
const arrived = once<Sms>(resolve => { session.on('sms', resolve); });
void peerOf(smpp).send({
cmdName: 'deliver_sm',
params: {
destination_addr: '46701113311',
short_message: 'answered too late',
source_addr: '46709771337',
},
});
const sms = await arrived;
const reconnected = once<true>(resolve => { session.on('reconnected', () => { resolve(true); }); });
await peerOf(smpp).close({ signal: AbortSignal.abort() });
await reconnected;
const answered = await sms.sendResp();
assert.match(answered.err?.message ?? '', /link this message arrived on is gone/);
// A receipt is a request of its own, correlated by its id, so the new link carries it.
const report = await sms.sendDlr('DELIVERED');
assert.equal(report.err, undefined);
assert.equal(report.pduObjs.length, 1);
});
test('does not reconnect after an explicit close', async t => {
const smpp = await startServer(t);
const { session } = await connect(t, smpp, { reconnect: { maxDelay: 50, minDelay: 10 } });
@@ -913,6 +947,20 @@ describe('held message bounds', () => {
held.clear();
});
test('remembers which messages went with the link, and which were answered', () => {
const held = new HeldMessages({ log: silentLog, max: 10, timeout: 10_000 });
const answered = message(1);
const dropped = message(2);
held.hold(answered);
held.hold(dropped);
held.release(answered);
held.clear();
assert.equal(held.lostLink(dropped), true);
assert.equal(held.lostLink(answered), false);
});
// 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;
@@ -944,6 +992,7 @@ describe('sendDlr()', () => {
session,
to: '46709771337',
}, {
lostLink: () => false,
onAnswered: () => undefined,
send: () => {
call++;
+1 -6
View File
@@ -60,6 +60,7 @@ Rules the API follows:
| `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` |
| A message whose link dropped refused an answer, with its receipt still allowed out | `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` |
| A listener that throws, or rejects, reaching `sessionError`/`serverError` rather than the process | `test/session.test.ts`, `test/error-from.test.ts` |
@@ -140,12 +141,6 @@ session message is a change to every call site.
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
application still holds the `Sms`, so `sendResp()` writes the old link's sequence numbers to the
new socket, succeeds, and returns `{}` for a response that correlates with nothing at the peer.
`HeldMessages` now knows exactly which messages went that way, so saying so is a small
addition. Goal 2, low frequency. Raised by review, 2026-09-01.
- [ ] **Does an intermediate delivery notification deserve to be a `dlr`?** `esm_class` message type
`INTERMEDIATE_DELIVERY` (0x20) is classified as a message today, so a peer that reports
non-final states with it hands the application a raw `id:… stat:ENROUTE` text as an inbound