Tie the refusal to the link, not to the hold, so a released message is refused too
This commit is contained in:
@@ -307,6 +307,14 @@ Grouped by what each one constrains.
|
|||||||
shutdown emits `close`. A retry that opens a socket and then loses it clears `closed` through
|
shutdown emits `close`. A retry that opens a socket and then loses it clears `closed` through
|
||||||
`attach()`, which is why a second drop emits again.
|
`attach()`, which is why a second drop emits again.
|
||||||
|
|
||||||
|
- **An answer belongs to the link the message arrived on; a receipt does not.** Maintainer's call,
|
||||||
|
2026-09-01: `sendResp()` carries the peer's own sequence number, which the next link knows nothing
|
||||||
|
about, so on a message whose link went it writes nothing and returns an `err`. Writing it to the
|
||||||
|
new socket instead succeeds and reports `{}` for a response that correlates with nothing, which is
|
||||||
|
goal 2's wrong answer. `sendDlr()` still goes out, because a receipt is a request of its own
|
||||||
|
correlated by `receipted_message_id`; the accepted cost is that one sent after a refused response
|
||||||
|
names an id the peer has no record of.
|
||||||
|
|
||||||
- **`reconnect` takes `{ minDelay, maxDelay }` to retune and `false` to turn off**, so absent means
|
- **`reconnect` takes `{ minDelay, maxDelay }` to retune and `false` to turn off**, so absent means
|
||||||
on and there is one spelling for each. Only `client()` reconnects — a `server()` session is a
|
on and there is one spelling for each. Only `client()` reconnects — a `server()` session is a
|
||||||
connection the peer opened, and nothing at this end can reopen it. The retry timer is `unref()`'d,
|
connection the peer opened, and nothing at this end can reopen it. The retry timer is `unref()`'d,
|
||||||
|
|||||||
@@ -398,7 +398,7 @@ The spec tables are exported both individually (`cmds`, `consts`, `encodings`, `
|
|||||||
- **`server()` resolves once, when it is listening**, and gives you a handle with `close()`, `port`
|
- **`server()` resolves once, when it is listening**, and gives you a handle with `close()`, `port`
|
||||||
and a `session` event. It no longer calls your callback once per incoming connection.
|
and a `session` event. It no longer calls your callback once per incoming connection.
|
||||||
- **The id a message is answered with goes to `sendResp({ smsId })`**, and `sms.smsId` is read-only:
|
- **The id a message is answered with goes to `sendResp({ smsId })`**, and `sms.smsId` is read-only:
|
||||||
it reports what the response actually carried. Delete any `sms.smsId = …` line — assigning to it
|
it reports the id `sendResp()` was given, or the UUID v7 generated instead. Delete any `sms.smsId = …` line — assigning to it
|
||||||
throws a `TypeError`, since modules are always strict mode — and pass the id to `sendResp()`.
|
throws a `TypeError`, since modules are always strict mode — and pass the id to `sendResp()`.
|
||||||
- **`checkuserpass` is now `authenticate`**, takes `{ password, session, systemId, systemType }` and
|
- **`checkuserpass` is now `authenticate`**, takes `{ password, session, systemId, systemType }` and
|
||||||
returns `false` or `{ userData }`.
|
returns `false` or `{ userData }`.
|
||||||
|
|||||||
+2
-2
@@ -120,8 +120,8 @@ export class DlrMerger {
|
|||||||
}
|
}
|
||||||
|
|
||||||
clear(): void {
|
clear(): void {
|
||||||
this.groups.takeAll();
|
this.groups.clear();
|
||||||
this.spent.takeAll();
|
this.spent.clear();
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Drops every group past its deadline. Runs before each collect and on its own timer. */
|
/** Drops every group past its deadline. Runs before each collect and on its own timer. */
|
||||||
|
|||||||
@@ -58,14 +58,9 @@ export class ExpiringGroups<T> {
|
|||||||
this.idle();
|
this.idle();
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Removes every group and hands them over. */
|
clear(): void {
|
||||||
takeAll(): [string, T][] {
|
|
||||||
const taken: [string, T][] = [...this.entries].map(([key, entry]) => [key, entry.group]);
|
|
||||||
|
|
||||||
this.entries.clear();
|
this.entries.clear();
|
||||||
this.idle();
|
this.idle();
|
||||||
|
|
||||||
return taken;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Removes every group past its deadline and hands them over. */
|
/** Removes every group past its deadline and hands them over. */
|
||||||
|
|||||||
+1
-10
@@ -20,7 +20,6 @@ function keyOf(pduObjs: PduObject[]): string | undefined {
|
|||||||
|
|
||||||
/** The messages handed to the application that it has not answered yet, held by their segments. */
|
/** The messages handed to the application that it has not answered yet, held by their segments. */
|
||||||
export class HeldMessages {
|
export class HeldMessages {
|
||||||
private readonly droppedWithLink = new WeakSet<PduObject[]>();
|
|
||||||
private readonly held: ExpiringGroups<PduObject[]>;
|
private readonly held: ExpiringGroups<PduObject[]>;
|
||||||
private readonly idleWaiters = new IdleWaiters();
|
private readonly idleWaiters = new IdleWaiters();
|
||||||
private readonly log: SmppLog;
|
private readonly log: SmppLog;
|
||||||
@@ -65,11 +64,6 @@ export class HeldMessages {
|
|||||||
return key !== undefined && this.held.get(key) === pduObjs;
|
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 {
|
release(pduObjs: PduObject[]): void {
|
||||||
const key = keyOf(pduObjs);
|
const key = keyOf(pduObjs);
|
||||||
|
|
||||||
@@ -82,10 +76,7 @@ export class HeldMessages {
|
|||||||
|
|
||||||
/** Drops every message: their segments went with the link, so no answer of ours correlates now. */
|
/** Drops every message: their segments went with the link, so no answer of ours correlates now. */
|
||||||
clear(): void {
|
clear(): void {
|
||||||
for (const [, pduObjs] of this.held.takeAll()) {
|
this.held.clear();
|
||||||
this.droppedWithLink.add(pduObjs);
|
|
||||||
}
|
|
||||||
|
|
||||||
this.idleWaiters.settle();
|
this.idleWaiters.settle();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -39,6 +39,8 @@ export class IncomingRequests {
|
|||||||
private readonly session: Session;
|
private readonly session: Session;
|
||||||
private readonly smsIdFormat: SmsIdFormat;
|
private readonly smsIdFormat: SmsIdFormat;
|
||||||
private readonly systemId: string;
|
private readonly systemId: string;
|
||||||
|
/** Bumped when a link goes, so a message from an earlier one knows its answer cannot correlate. */
|
||||||
|
private linkGeneration = 0;
|
||||||
|
|
||||||
constructor(options: IncomingRequestsOptions) {
|
constructor(options: IncomingRequestsOptions) {
|
||||||
this.dlrMerger = options.dlrMerger;
|
this.dlrMerger = options.dlrMerger;
|
||||||
@@ -96,6 +98,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.held.clear();
|
this.held.clear();
|
||||||
this.reassembler.clear();
|
this.reassembler.clear();
|
||||||
}
|
}
|
||||||
@@ -163,6 +166,8 @@ export class IncomingRequests {
|
|||||||
|
|
||||||
if (!first) return;
|
if (!first) return;
|
||||||
|
|
||||||
|
const generation = this.linkGeneration;
|
||||||
|
|
||||||
const sms = createSms({
|
const sms = createSms({
|
||||||
from: paramText(first.params.source_addr),
|
from: paramText(first.params.source_addr),
|
||||||
message: decodeSegments(pduObjs),
|
message: decodeSegments(pduObjs),
|
||||||
@@ -170,7 +175,7 @@ export class IncomingRequests {
|
|||||||
session: this.session,
|
session: this.session,
|
||||||
to: paramText(first.params.destination_addr),
|
to: paramText(first.params.destination_addr),
|
||||||
}, {
|
}, {
|
||||||
lostLink: () => this.held.lostLink(pduObjs),
|
lostLink: () => this.linkGeneration !== generation,
|
||||||
// A turn later, so a listener sending its receipt straight after the response still holds.
|
// A turn later, so a listener sending its receipt straight after the response still holds.
|
||||||
onAnswered: () => { setImmediate(() => { this.held.release(pduObjs); }); },
|
onAnswered: () => { setImmediate(() => { this.held.release(pduObjs); }); },
|
||||||
// Past the refusal only while a drain is still waiting for this message; an ordinary send after.
|
// Past the refusal only while a drain is still waiting for this message; an ordinary send after.
|
||||||
|
|||||||
+1
-1
@@ -163,7 +163,7 @@ export class Reassembler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
clear(): void {
|
clear(): void {
|
||||||
this.groups.takeAll();
|
this.groups.clear();
|
||||||
this.octets = 0;
|
this.octets = 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+2
-3
@@ -54,7 +54,6 @@ export type SmsInput = {
|
|||||||
|
|
||||||
/** What the session's incoming side gives a message so it can be answered and accounted for. */
|
/** What the session's incoming side gives a message so it can be answered and accounted for. */
|
||||||
export type SmsHandlers = {
|
export type SmsHandlers = {
|
||||||
/** Whether the link this message arrived on is gone, so no answer of ours correlates. */
|
|
||||||
lostLink: () => boolean;
|
lostLink: () => boolean;
|
||||||
onAnswered: () => void;
|
onAnswered: () => void;
|
||||||
send: (input: PduObjectInput) => Promise<Result<{ pduObj: PduObject }>>;
|
send: (input: PduObjectInput) => Promise<Result<{ pduObj: PduObject }>>;
|
||||||
@@ -107,13 +106,13 @@ async function sendResp(
|
|||||||
return { err: new Error('smsId must not be empty') };
|
return { err: new Error('smsId must not be empty') };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (options.smsId !== undefined) answered.smsId = options.smsId;
|
||||||
|
|
||||||
// A response carries the sequence number it was asked on, which the next link knows nothing about.
|
// A response carries the sequence number it was asked on, which the next link knows nothing about.
|
||||||
if (lostLink()) {
|
if (lostLink()) {
|
||||||
return { err: new Error('The link this message arrived on is gone, so nothing would correlate the response') };
|
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(
|
const results = await Promise.all(sms.pduObjs.map((pduObj, index) => sms.session.sendReturn(
|
||||||
pduObj,
|
pduObj,
|
||||||
options.status ?? 'ESME_ROK',
|
options.status ?? 'ESME_ROK',
|
||||||
|
|||||||
+28
-24
@@ -557,38 +557,56 @@ describe('reconnect', () => {
|
|||||||
assert.equal(report.segments.length, 3);
|
assert.equal(report.segments.length, 3);
|
||||||
});
|
});
|
||||||
|
|
||||||
test('refuses to answer a message whose link went, rather than reporting it delivered', async t => {
|
test('refuses to answer a message whose link went, held or already answered', async t => {
|
||||||
const smpp = await startServer(t);
|
const smpp = await startServer(t);
|
||||||
const { session } = await connect(t, smpp, { reconnect: { maxDelay: 100, minDelay: 20 } });
|
const { session } = await connect(t, smpp, { reconnect: { maxDelay: 100, minDelay: 20 } });
|
||||||
|
|
||||||
assert.ok(session);
|
assert.ok(session);
|
||||||
|
|
||||||
const arrived = once<Sms>(resolve => { session.on('sms', resolve); });
|
const arrived: Sms[] = [];
|
||||||
|
const both = once<true>(resolve => {
|
||||||
|
session.on('sms', sms => {
|
||||||
|
arrived.push(sms);
|
||||||
|
|
||||||
|
if (arrived.length === 2) resolve(true);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
for (const text of ['answered before the drop', 'never answered']) {
|
||||||
void peerOf(smpp).send({
|
void peerOf(smpp).send({
|
||||||
cmdName: 'deliver_sm',
|
cmdName: 'deliver_sm',
|
||||||
params: {
|
params: {
|
||||||
destination_addr: '46701113311',
|
destination_addr: '46701113311',
|
||||||
short_message: 'answered too late',
|
short_message: text,
|
||||||
source_addr: '46709771337',
|
source_addr: '46709771337',
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
await both;
|
||||||
|
|
||||||
|
const [answered, held] = arrived;
|
||||||
|
|
||||||
|
assert.ok(answered);
|
||||||
|
assert.ok(held);
|
||||||
|
assert.equal((await answered.sendResp()).err, undefined);
|
||||||
|
|
||||||
const sms = await arrived;
|
|
||||||
const reconnected = once<true>(resolve => { session.on('reconnected', () => { resolve(true); }); });
|
const reconnected = once<true>(resolve => { session.on('reconnected', () => { resolve(true); }); });
|
||||||
|
|
||||||
await peerOf(smpp).close({ signal: AbortSignal.abort() });
|
await peerOf(smpp).close({ signal: AbortSignal.abort() });
|
||||||
await reconnected;
|
await reconnected;
|
||||||
|
|
||||||
const answered = await sms.sendResp();
|
const taken: string[] = [];
|
||||||
|
|
||||||
assert.match(answered.err?.message ?? '', /link this message arrived on is gone/);
|
peerOf(smpp).on('incomingPduObj', pduObj => { taken.push(pduObj.cmdName); });
|
||||||
|
|
||||||
|
// The answered one is out of the hold and the held one is not, so neither may reach the answer.
|
||||||
|
assert.match((await answered.sendResp()).err?.message ?? '', /link this message arrived on is gone/);
|
||||||
|
assert.match((await held.sendResp()).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.
|
// 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((await held.sendDlr('DELIVERED')).err, undefined);
|
||||||
|
assert.deepEqual(taken, ['deliver_sm'], 'a response reached the new link');
|
||||||
assert.equal(report.err, undefined);
|
|
||||||
assert.equal(report.pduObjs.length, 1);
|
|
||||||
});
|
});
|
||||||
|
|
||||||
test('does not reconnect after an explicit close', async t => {
|
test('does not reconnect after an explicit close', async t => {
|
||||||
@@ -947,20 +965,6 @@ describe('held message bounds', () => {
|
|||||||
held.clear();
|
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.
|
// 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 () => {
|
test('wakes a waiting drain when the last message expires', async () => {
|
||||||
let now = 0;
|
let now = 0;
|
||||||
|
|||||||
@@ -141,6 +141,11 @@ session message is a change to every call site.
|
|||||||
until the message expires. Same cost as the `onRequest`-answers-nothing case that was declined,
|
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.
|
but reached by a bug rather than a policy. Raised by review, 2026-09-01.
|
||||||
|
|
||||||
|
- [ ] **A refused `sendResp()` releases the hold anyway.** Every early return runs
|
||||||
|
`.finally(onAnswered)`, so `sendResp({ smsId: '' })` refuses and stops the drain waiting for a
|
||||||
|
message the peer was never answered, which `close()` then reports as answered. Raised by
|
||||||
|
review, 2026-09-01.
|
||||||
|
|
||||||
- [ ] **Does an intermediate delivery notification deserve to be a `dlr`?** `esm_class` message type
|
- [ ] **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
|
`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
|
non-final states with it hands the application a raw `id:… stat:ENROUTE` text as an inbound
|
||||||
|
|||||||
Reference in New Issue
Block a user