Release a message's hold when the response goes out, or when its listener failed
This commit is contained in:
@@ -351,7 +351,11 @@ Grouped by what each one constrains.
|
|||||||
drain waits for. Counting every inbound request until `sendReturn()` answered it was rejected —
|
drain waits for. Counting every inbound request until `sendReturn()` answered it was rejected —
|
||||||
an `onRequest` that deliberately answers nothing would then cost a full `shutdownTimeout` on every
|
an `onRequest` that deliberately answers nothing would then cost a full `shutdownTimeout` on every
|
||||||
close — and a message no listener took is released at once, since nothing is going to answer it.
|
close — and a message no listener took is released at once, since nothing is going to answer it.
|
||||||
`teardown()` drops what is still held for the same reason it drops inbound segments. The release
|
A listener that threw or rejected before answering releases it the same way, for the same reason,
|
||||||
|
with `sessionError` carrying the failure. What ends the wait is the response reaching the wire, not
|
||||||
|
the call: a `sendResp()` the library refused leaves the message held, so `close()` still reports the
|
||||||
|
one the peer is owed. `teardown()` drops what is still held for the same reason it drops inbound
|
||||||
|
segments. The release
|
||||||
is one turn late, so a listener that sends its receipt straight after the response is still
|
is one turn late, so a listener that sends its receipt straight after the response is still
|
||||||
holding when the drain looks; `sendDlr()` is the one send that goes out past the drain's refusal,
|
holding when the drain looks; `sendDlr()` is the one send that goes out past the drain's refusal,
|
||||||
and only while the message is still held — past that it is an ordinary send, because the drain it
|
and only while the message is still held — past that it is an ordinary send, because the drain it
|
||||||
|
|||||||
@@ -332,7 +332,7 @@ TypeScript users can import `SmppLog` to have the compiler check one.
|
|||||||
`sendSms()`, `send()`, `sendReturn()`, `unbind()` and `close()`. Both `close()` and `unbind()`
|
`sendSms()`, `send()`, `sendReturn()`, `unbind()` and `close()`. Both `close()` and `unbind()`
|
||||||
refuse further sends, wait out the requests this end already sent for up to `shutdownTimeout`, and
|
refuse further sends, wait out the requests this end already sent for up to `shutdownTimeout`, and
|
||||||
then tear down whatever is left, resolving to an `err` that says what was lost. They also wait for
|
then tear down whatever is left, resolving to an `err` that says what was lost. They also wait for
|
||||||
every `sms` the application has not called `sendResp()` on, so a peer whose `submit_sm` is still
|
every `sms` that `sendResp()` has not answered, so a peer whose `submit_sm` is still
|
||||||
being handled is answered rather than left to re-send it — answering its PDUs through `sendReturn()`
|
being handled is answered rather than left to re-send it — answering its PDUs through `sendReturn()`
|
||||||
instead leaves that wait running until it gives up. `sendDlr()` is the one send the refusal lets
|
instead leaves that wait running until it gives up. `sendDlr()` is the one send the refusal lets
|
||||||
past, and it catches the wait when issued straight after `sendResp()`; await anything in between and
|
past, and it catches the wait when issued straight after `sendResp()`; await anything in between and
|
||||||
|
|||||||
@@ -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;
|
||||||
|
/** Identity, not a type guard: the rejected event carries the Sms back as an `unknown`. */
|
||||||
|
private readonly emitted = new WeakMap<object, () => void>();
|
||||||
private linkGeneration = 0;
|
private linkGeneration = 0;
|
||||||
|
|
||||||
constructor(options: IncomingRequestsOptions) {
|
constructor(options: IncomingRequestsOptions) {
|
||||||
@@ -111,6 +113,13 @@ export class IncomingRequests {
|
|||||||
this.reassembler.clear();
|
this.reassembler.clear();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** A listener that rejected answered nothing and never will, so a shutdown may not wait for it. */
|
||||||
|
listenerRejected(sms: unknown): void {
|
||||||
|
if (typeof sms !== 'object' || sms === null) return;
|
||||||
|
|
||||||
|
this.emitted.get(sms)?.();
|
||||||
|
}
|
||||||
|
|
||||||
/** Waits out the messages the application still holds, and says how many it never answered. */
|
/** Waits out the messages the application still holds, and says how many it never answered. */
|
||||||
async drain(timeout: number, signal: AbortSignal | undefined): Promise<VoidResult> {
|
async drain(timeout: number, signal: AbortSignal | undefined): Promise<VoidResult> {
|
||||||
const unanswered = await this.held.idle(timeout, signal);
|
const unanswered = await this.held.idle(timeout, signal);
|
||||||
@@ -191,6 +200,7 @@ export class IncomingRequests {
|
|||||||
});
|
});
|
||||||
|
|
||||||
this.held.hold(pduObjs);
|
this.held.hold(pduObjs);
|
||||||
|
this.emitted.set(sms, () => { this.held.release(pduObjs); });
|
||||||
|
|
||||||
// A message nobody took is not work a shutdown can wait for.
|
// A message nobody took is not work a shutdown can wait for.
|
||||||
if (!this.session.emit('sms', sms)) this.held.release(pduObjs);
|
if (!this.session.emit('sms', sms)) this.held.release(pduObjs);
|
||||||
|
|||||||
+3
-1
@@ -93,11 +93,13 @@ export class Session extends EventEmitter<SessionEvents> {
|
|||||||
reason: unknown,
|
reason: unknown,
|
||||||
...args: [event: keyof SessionEvents, ...rest: unknown[]]
|
...args: [event: keyof SessionEvents, ...rest: unknown[]]
|
||||||
): void {
|
): void {
|
||||||
const [event] = args;
|
const [event, ...rest] = args;
|
||||||
const error = errorFrom(reason);
|
const error = errorFrom(reason);
|
||||||
|
|
||||||
this.log.error('session - a listener rejected', { event, message: error.message });
|
this.log.error('session - a listener rejected', { event, message: error.message });
|
||||||
|
|
||||||
|
if (event === 'sms') this.incoming.listenerRejected(rest[0]);
|
||||||
|
|
||||||
if (event !== 'sessionError') this.emit('sessionError', error);
|
if (event !== 'sessionError') this.emit('sessionError', error);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+6
-4
@@ -77,8 +77,7 @@ 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 ?? {}, handlers.lostLink)
|
sendResp: options => sendResp(sms, answered, options ?? {}, handlers),
|
||||||
.finally(handlers.onAnswered),
|
|
||||||
session: input.session,
|
session: input.session,
|
||||||
get smsId(): string {
|
get smsId(): string {
|
||||||
return answered.smsId;
|
return answered.smsId;
|
||||||
@@ -94,7 +93,7 @@ async function sendResp(
|
|||||||
sms: Sms,
|
sms: Sms,
|
||||||
answered: { smsId: string },
|
answered: { smsId: string },
|
||||||
options: SendRespOptions,
|
options: SendRespOptions,
|
||||||
lostLink: () => boolean,
|
handlers: SmsHandlers,
|
||||||
): Promise<VoidResult> {
|
): Promise<VoidResult> {
|
||||||
const total = sms.pduObjs.length;
|
const total = sms.pduObjs.length;
|
||||||
|
|
||||||
@@ -109,7 +108,7 @@ 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.
|
// A response carries the sequence number it was asked on, which the next link knows nothing about.
|
||||||
if (lostLink()) {
|
if (handlers.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') };
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -119,6 +118,9 @@ async function sendResp(
|
|||||||
{ message_id: segmentId(answered.smsId, index, total) },
|
{ message_id: segmentId(answered.smsId, index, total) },
|
||||||
)));
|
)));
|
||||||
|
|
||||||
|
// Only here: a refusal above put nothing on the wire, so the peer is still owed its response.
|
||||||
|
handlers.onAnswered();
|
||||||
|
|
||||||
return results.find(result => result.err) ?? {};
|
return results.find(result => result.err) ?? {};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1401,6 +1401,45 @@ describe('graceful shutdown', () => {
|
|||||||
assert.ok((await sent).err instanceof Error);
|
assert.ok((await sent).err instanceof Error);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// emit() releases the hold of a listener that throws; one that rejects may cost no more than that.
|
||||||
|
test('a listener that rejected before answering does not hold the shutdown up', async t => {
|
||||||
|
const smpp = await startServer(t, { shutdownTimeout: 30_000 });
|
||||||
|
const failed = once<Error>(resolve => {
|
||||||
|
smpp.on('session', bound => {
|
||||||
|
bound.on('sessionError', resolve);
|
||||||
|
bound.on('sms', () => Promise.reject(new Error('the listener gave up')));
|
||||||
|
});
|
||||||
|
});
|
||||||
|
const { session } = await connect(t, smpp);
|
||||||
|
|
||||||
|
assert.ok(session);
|
||||||
|
|
||||||
|
const sent = session.sendSms({
|
||||||
|
from: '46701113311',
|
||||||
|
message: 'the listener rejects',
|
||||||
|
to: '46709771337',
|
||||||
|
});
|
||||||
|
|
||||||
|
assert.equal((await failed).message, 'the listener gave up');
|
||||||
|
|
||||||
|
const started = Date.now();
|
||||||
|
|
||||||
|
assert.deepEqual(await peerOf(smpp).close(), {});
|
||||||
|
assert.ok(Date.now() - started < 1000);
|
||||||
|
assert.ok((await sent).err instanceof Error);
|
||||||
|
});
|
||||||
|
|
||||||
|
// Nothing reached the peer, so a drain counting this answered would report an outcome that never was.
|
||||||
|
test('leaves a message the library refused to answer unanswered', async t => {
|
||||||
|
const { sms, smpp } = await submitInFlight(t, {}, { shutdownTimeout: 50 });
|
||||||
|
const refused = await sms.sendResp({ smsId: '' });
|
||||||
|
const closed = await peerOf(smpp).close();
|
||||||
|
|
||||||
|
assert.match(refused.err?.message ?? '', /smsId must not be empty/);
|
||||||
|
assert.ok(closed.err instanceof Error);
|
||||||
|
assert.match(closed.err.message, /1 message\(s\) unanswered/);
|
||||||
|
});
|
||||||
|
|
||||||
test('gives up on a request that outlasts shutdownTimeout', async t => {
|
test('gives up on a request that outlasts shutdownTimeout', async t => {
|
||||||
const { sent, session } = await submitInFlight(t, { shutdownTimeout: 50 });
|
const { sent, session } = await submitInFlight(t, { shutdownTimeout: 50 });
|
||||||
const closed = await session.close();
|
const closed = await session.close();
|
||||||
|
|||||||
@@ -61,6 +61,7 @@ Rules the API follows:
|
|||||||
| 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` |
|
| A message whose link dropped refused an answer, with its receipt still allowed out | `test/session-extras.test.ts` |
|
||||||
|
| The hold released exactly when the peer was answered: a refused `sendResp()` keeps it, a listener that rejected drops it | `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` |
|
||||||
@@ -134,18 +135,6 @@ session message is a change to every call site.
|
|||||||
and applies it to the other is wrong. A budget type both take would close it. Raised by review,
|
and applies it to the other is wrong. A budget type both take would close it. Raised by review,
|
||||||
2026-09-01.
|
2026-09-01.
|
||||||
|
|
||||||
- [ ] **An `sms` listener that rejects before answering costs a whole `shutdownTimeout`.**
|
|
||||||
One that *throws* is fine: `emit()` catches it, returns false, and `emitSms()` releases the
|
|
||||||
hold. A rejecting `async` one reaches `sessionError` through `captureRejections`, which hands
|
|
||||||
the handler an `unknown[]` the `Sms` cannot be read out of without a cast, so nothing releases
|
|
||||||
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 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