Merge pull request #74 from larvit/answer-without-a-link
Refuse to answer a message whose link went, instead of reporting it delivered
This commit is contained in:
@@ -307,6 +307,11 @@ 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. Rejected: answering on the new link, which succeeds and reports `{}` for a response
|
||||||
|
that correlates with nothing — goal 2's wrong answer. Accepted: a receipt 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,
|
||||||
|
|||||||
@@ -316,7 +316,7 @@ TypeScript users can import `SmppLog` to have the compiler check one.
|
|||||||
|
|
||||||
| Event | Fires when |
|
| Event | Fires when |
|
||||||
| --- | --- |
|
| --- | --- |
|
||||||
| `sms` | An SMS arrives, reassembled if it was multipart. Carries `sendResp()`, `sendDlr()` and the `smsId` it was answered with. |
|
| `sms` | An SMS arrives, reassembled if it was multipart. Carries `sendResp()`, `sendDlr()` and its `smsId`. |
|
||||||
| `dlr` | A delivery report arrives, one per segment. `smsId` is undefined when the peer marked a receipt whose body carries no readable id. `statusMsg` names `statusId` unless the peer sent a `message_state` this library cannot name — then `statusId` is that raw value and `statusMsg` is whatever the body said, or `UNKNOWN`. |
|
| `dlr` | A delivery report arrives, one per segment. `smsId` is undefined when the peer marked a receipt whose body carries no readable id. `statusMsg` names `statusId` unless the peer sent a `message_state` this library cannot name — then `statusId` is that raw value and `statusMsg` is whatever the body said, or `UNKNOWN`. |
|
||||||
| `messageDlr` | Every segment of a multipart message sent with `dlr: true` has been reported on, carrying the worst status of the segments. Merging needs the SMSC to number its segment ids `<base>-<n>`, which is this library's own server's convention — an SMSC that hands out unrelated ids per segment never fires it. A base is merged once: a later message the SMSC gives the same ids is reported on through `dlr` alone, and an earlier one still collecting loses its merged report as well. |
|
| `messageDlr` | Every segment of a multipart message sent with `dlr: true` has been reported on, carrying the worst status of the segments. Merging needs the SMSC to number its segment ids `<base>-<n>`, which is this library's own server's convention — an SMSC that hands out unrelated ids per segment never fires it. A base is merged once: a later message the SMSC gives the same ids is reported on through `dlr` alone, and an earlier one still collecting loses its merged report as well. |
|
||||||
| `close` | The session is over, because nothing will bring the link back. Fires once, whether you closed it or the link failed for good. |
|
| `close` | The session is over, because nothing will bring the link back. Fires once, whether you closed it or the link failed for good. |
|
||||||
@@ -355,6 +355,11 @@ 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`,
|
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.
|
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`. Where a
|
||||||
|
reconnect follows, its `sms.sendDlr()` still goes out on the new link, 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
|
`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.
|
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.
|
Pass `{ signal: AbortSignal.timeout(ms) }` when you need one.
|
||||||
@@ -394,7 +399,8 @@ 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 }`.
|
||||||
|
|||||||
@@ -39,6 +39,7 @@ 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;
|
||||||
|
private linkGeneration = 0;
|
||||||
|
|
||||||
constructor(options: IncomingRequestsOptions) {
|
constructor(options: IncomingRequestsOptions) {
|
||||||
this.dlrMerger = options.dlrMerger;
|
this.dlrMerger = options.dlrMerger;
|
||||||
@@ -62,8 +63,17 @@ export class IncomingRequests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async handle(pduObj: PduObject): Promise<void> {
|
async handle(pduObj: PduObject): Promise<void> {
|
||||||
|
const generation = this.linkGeneration;
|
||||||
|
|
||||||
if (this.onRequest && await this.onRequest(this.session, pduObj)) return;
|
if (this.onRequest && await this.onRequest(this.session, pduObj)) return;
|
||||||
|
|
||||||
|
// The link it arrived on went while the hook ran, so nothing we answer now correlates.
|
||||||
|
if (this.linkGeneration !== generation) {
|
||||||
|
this.log.info('session - dropping a request whose link went', { cmdName: pduObj.cmdName });
|
||||||
|
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
if (!this.session.bindAllows(pduObj.cmdName)) {
|
if (!this.session.bindAllows(pduObj.cmdName)) {
|
||||||
this.log.info('session - command the peer\'s bind direction does not carry', {
|
this.log.info('session - command the peer\'s bind direction does not carry', {
|
||||||
bindType: this.session.boundAs ?? '',
|
bindType: this.session.boundAs ?? '',
|
||||||
@@ -96,6 +106,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 +174,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,6 +183,7 @@ export class IncomingRequests {
|
|||||||
session: this.session,
|
session: this.session,
|
||||||
to: paramText(first.params.destination_addr),
|
to: paramText(first.params.destination_addr),
|
||||||
}, {
|
}, {
|
||||||
|
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.
|
||||||
|
|||||||
+10
-2
@@ -38,7 +38,7 @@ export type Sms = {
|
|||||||
/** Answers every segment. Part of the protocol, not optional. Defaults to ESME_ROK. */
|
/** Answers every segment. Part of the protocol, not optional. Defaults to ESME_ROK. */
|
||||||
sendResp: (options?: SendRespOptions) => Promise<VoidResult>;
|
sendResp: (options?: SendRespOptions) => Promise<VoidResult>;
|
||||||
session: Session;
|
session: Session;
|
||||||
/** The id answered to the peer: what sendResp() was given, or a generated UUID v7. */
|
/** The id `sendResp()` was given, or a generated UUID v7. */
|
||||||
readonly smsId: string;
|
readonly smsId: string;
|
||||||
submitTime: Date;
|
submitTime: Date;
|
||||||
to: string;
|
to: string;
|
||||||
@@ -54,6 +54,7 @@ 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 = {
|
||||||
|
lostLink: () => boolean;
|
||||||
onAnswered: () => void;
|
onAnswered: () => void;
|
||||||
send: (input: PduObjectInput) => Promise<Result<{ pduObj: PduObject }>>;
|
send: (input: PduObjectInput) => Promise<Result<{ pduObj: PduObject }>>;
|
||||||
};
|
};
|
||||||
@@ -76,7 +77,8 @@ export function createSms(input: SmsInput, handlers: SmsHandlers): Sms {
|
|||||||
message: input.message,
|
message: input.message,
|
||||||
pduObjs: input.pduObjs,
|
pduObjs: input.pduObjs,
|
||||||
sendDlr: status => sendDlr(sms, handlers.send, status),
|
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,
|
session: input.session,
|
||||||
get smsId(): string {
|
get smsId(): string {
|
||||||
return answered.smsId;
|
return answered.smsId;
|
||||||
@@ -92,6 +94,7 @@ async function sendResp(
|
|||||||
sms: Sms,
|
sms: Sms,
|
||||||
answered: { smsId: string },
|
answered: { smsId: string },
|
||||||
options: SendRespOptions,
|
options: SendRespOptions,
|
||||||
|
lostLink: () => boolean,
|
||||||
): Promise<VoidResult> {
|
): Promise<VoidResult> {
|
||||||
const total = sms.pduObjs.length;
|
const total = sms.pduObjs.length;
|
||||||
|
|
||||||
@@ -105,6 +108,11 @@ async function sendResp(
|
|||||||
|
|
||||||
if (options.smsId !== undefined) answered.smsId = options.smsId;
|
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.
|
||||||
|
if (lostLink()) {
|
||||||
|
return { err: new Error('The link this message arrived on is gone, so nothing would correlate the response') };
|
||||||
|
}
|
||||||
|
|
||||||
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',
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ import type { Sms } from '../src/sms.ts';
|
|||||||
import type { SmppServer } from '../src/server.ts';
|
import type { SmppServer } from '../src/server.ts';
|
||||||
import type { TestContext } from 'node:test';
|
import type { TestContext } from 'node:test';
|
||||||
import { HeldMessages } from '../src/held-messages.ts';
|
import { HeldMessages } from '../src/held-messages.ts';
|
||||||
|
import { IncomingRequests } from '../src/incoming-requests.ts';
|
||||||
import { UnansweredError } from '../src/unanswered-error.ts';
|
import { UnansweredError } from '../src/unanswered-error.ts';
|
||||||
import { createSms } from '../src/sms.ts';
|
import { createSms } from '../src/sms.ts';
|
||||||
import { LinkGate } from '../src/link-gate.ts';
|
import { LinkGate } from '../src/link-gate.ts';
|
||||||
@@ -557,6 +558,88 @@ describe('reconnect', () => {
|
|||||||
assert.equal(report.segments.length, 3);
|
assert.equal(report.segments.length, 3);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test('refuses to answer a message whose link went, held or already answered', async t => {
|
||||||
|
const smpp = await startServer(t);
|
||||||
|
const { session } = await connect(t, smpp, { reconnect: { maxDelay: 100, minDelay: 20 } });
|
||||||
|
|
||||||
|
assert.ok(session);
|
||||||
|
|
||||||
|
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({
|
||||||
|
cmdName: 'deliver_sm',
|
||||||
|
params: {
|
||||||
|
destination_addr: '46701113311',
|
||||||
|
short_message: text,
|
||||||
|
source_addr: '46709771337',
|
||||||
|
},
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
await both;
|
||||||
|
|
||||||
|
const [answered, held] = arrived;
|
||||||
|
|
||||||
|
assert.ok(answered);
|
||||||
|
assert.ok(held);
|
||||||
|
assert.equal(answered.message, 'answered before the drop');
|
||||||
|
assert.equal((await answered.sendResp()).err, undefined);
|
||||||
|
|
||||||
|
const reconnected = once<true>(resolve => { session.on('reconnected', () => { resolve(true); }); });
|
||||||
|
|
||||||
|
await peerOf(smpp).close({ signal: AbortSignal.abort() });
|
||||||
|
await reconnected;
|
||||||
|
|
||||||
|
let taken = 0;
|
||||||
|
|
||||||
|
// A response is dispatched before `incomingPduObj`, so only the raw event sees one arrive.
|
||||||
|
peerOf(smpp).on('incomingPdu', () => { taken++; });
|
||||||
|
|
||||||
|
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/);
|
||||||
|
|
||||||
|
assert.equal((await held.sendDlr('DELIVERED')).err, undefined);
|
||||||
|
assert.equal(taken, 1, 'a refused response reached the new link');
|
||||||
|
});
|
||||||
|
|
||||||
|
test('drops a message whose link went while onRequest was still running', 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,
|
||||||
|
onRequest: async () => { await delay(10); return false; },
|
||||||
|
sendPastDrain: () => Promise.resolve({ err: new Error('never sent') }),
|
||||||
|
session,
|
||||||
|
});
|
||||||
|
let messages = 0;
|
||||||
|
|
||||||
|
session.on('sms', () => { messages++; });
|
||||||
|
|
||||||
|
const handled = incoming.handle(submitPdu(1));
|
||||||
|
|
||||||
|
incoming.clear();
|
||||||
|
|
||||||
|
await handled;
|
||||||
|
|
||||||
|
assert.equal(messages, 0);
|
||||||
|
|
||||||
|
await incoming.handle(submitPdu(2));
|
||||||
|
|
||||||
|
assert.equal(messages, 1, 'the harness delivers a message whose link stayed');
|
||||||
|
});
|
||||||
|
|
||||||
test('does not reconnect after an explicit close', async t => {
|
test('does not reconnect after an explicit close', async t => {
|
||||||
const smpp = await startServer(t);
|
const smpp = await startServer(t);
|
||||||
const { session } = await connect(t, smpp, { reconnect: { maxDelay: 50, minDelay: 10 } });
|
const { session } = await connect(t, smpp, { reconnect: { maxDelay: 50, minDelay: 10 } });
|
||||||
@@ -944,6 +1027,7 @@ describe('sendDlr()', () => {
|
|||||||
session,
|
session,
|
||||||
to: '46709771337',
|
to: '46709771337',
|
||||||
}, {
|
}, {
|
||||||
|
lostLink: () => false,
|
||||||
onAnswered: () => undefined,
|
onAnswered: () => undefined,
|
||||||
send: () => {
|
send: () => {
|
||||||
call++;
|
call++;
|
||||||
|
|||||||
@@ -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` |
|
| `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` |
|
| 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 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` |
|
| Every runnable README example | `test/readme.test.ts` |
|
||||||
| Receipt-versus-message classification by `esm_class` | `test/dlr.test.ts`, `test/session.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` |
|
| 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,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 message the reconnect dropped is answered into the void, and reported as delivered.**
|
- [ ] **A refused `sendResp()` releases the hold anyway.** Every early return runs
|
||||||
`teardown()` clears the held messages along with the inbound segments, which is right — but the
|
`.finally(onAnswered)`, so `sendResp({ smsId: '' })` refuses and stops the drain waiting for a
|
||||||
application still holds the `Sms`, so `sendResp()` writes the old link's sequence numbers to the
|
message the peer was never answered, which `close()` then reports as answered. Raised by
|
||||||
new socket, succeeds, and returns `{}` for a response that correlates with nothing at the peer.
|
review, 2026-09-01.
|
||||||
`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
|
- [ ] **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