Keep the process alive while a send waits for a link
This commit is contained in:
@@ -365,7 +365,11 @@ exactly 140.
|
|||||||
answer to how long one request may wait. It bounds the hold and the answer separately, and the
|
answer to how long one request may wait. It bounds the hold and the answer separately, and the
|
||||||
wait for a `maxOutstanding` slot is bounded by nothing, so `responseTimeout` is not a deadline for
|
wait for a `maxOutstanding` slot is bounded by nothing, so `responseTimeout` is not a deadline for
|
||||||
the call; `SendOptions.signal` with `AbortSignal.timeout()` is, and both the gate and
|
the call; `SendOptions.signal` with `AbortSignal.timeout()` is, and both the gate and
|
||||||
`pending.wait()` honour it.
|
`pending.wait()` honour it. The hold's clock starts when the send is issued rather than when it
|
||||||
|
first finds the gate shut, so one budget covers every hold a single call makes — a send that spent
|
||||||
|
it queued behind the window is refused rather than held. The hold timer is the one timer here that
|
||||||
|
is not `unref()`'d: a held request is awaited with the socket already destroyed, so an unref'd one
|
||||||
|
lets a process whose only remaining work is that send exit without settling it.
|
||||||
|
|
||||||
- **The gate decides whether a link can carry a request, and a bind is what makes it one.**
|
- **The gate decides whether a link can carry a request, and a bind is what makes it one.**
|
||||||
Maintainer's call, 2026-09-01: `attach()` clears `closed` the moment a socket is handed over, one
|
Maintainer's call, 2026-09-01: `attach()` clears `closed` the moment a socket is handed over, one
|
||||||
|
|||||||
+27
-9
@@ -1,6 +1,8 @@
|
|||||||
|
import type { SmppLog } from './log.ts';
|
||||||
import type { VoidResult } from './result.ts';
|
import type { VoidResult } from './result.ts';
|
||||||
|
|
||||||
export type LinkGateOptions = {
|
export type LinkGateOptions = {
|
||||||
|
log: SmppLog;
|
||||||
now?: (() => number) | undefined;
|
now?: (() => number) | undefined;
|
||||||
/** How long a request may wait for a link. 0 waits for as long as one may still arrive. */
|
/** How long a request may wait for a link. 0 waits for as long as one may still arrive. */
|
||||||
timeout: number;
|
timeout: number;
|
||||||
@@ -20,11 +22,9 @@ function over(): Error {
|
|||||||
return new Error('Session is closed');
|
return new Error('Session is closed');
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/** Where a request with no link to go out on waits for the next one. */
|
||||||
* Where a request with no link to go out on waits for the next one. The owner reports what became of
|
|
||||||
* the link; nothing here reads back into the owner to find out.
|
|
||||||
*/
|
|
||||||
export class LinkGate {
|
export class LinkGate {
|
||||||
|
private readonly log: SmppLog;
|
||||||
private readonly now: () => number;
|
private readonly now: () => number;
|
||||||
private readonly timeout: number;
|
private readonly timeout: number;
|
||||||
private readonly waiting = new Set<Waiter>();
|
private readonly waiting = new Set<Waiter>();
|
||||||
@@ -32,6 +32,7 @@ export class LinkGate {
|
|||||||
private up = true;
|
private up = true;
|
||||||
|
|
||||||
constructor(options: LinkGateOptions) {
|
constructor(options: LinkGateOptions) {
|
||||||
|
this.log = options.log;
|
||||||
this.now = options.now ?? Date.now;
|
this.now = options.now ?? Date.now;
|
||||||
this.timeout = options.timeout;
|
this.timeout = options.timeout;
|
||||||
}
|
}
|
||||||
@@ -41,6 +42,11 @@ export class LinkGate {
|
|||||||
return this.up;
|
return this.up;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Why the gate will never admit a request, or undefined while one may still get through. */
|
||||||
|
refusal(): Error | undefined {
|
||||||
|
return this.up || this.returning ? undefined : over();
|
||||||
|
}
|
||||||
|
|
||||||
/** When a hold starting now has to give up. 0 never does. */
|
/** When a hold starting now has to give up. 0 never does. */
|
||||||
deadline(): number {
|
deadline(): number {
|
||||||
return this.timeout > 0 ? this.now() + this.timeout : 0;
|
return this.timeout > 0 ? this.now() + this.timeout : 0;
|
||||||
@@ -50,6 +56,11 @@ export class LinkGate {
|
|||||||
open(): void {
|
open(): void {
|
||||||
this.up = true;
|
this.up = true;
|
||||||
this.returning = false;
|
this.returning = false;
|
||||||
|
|
||||||
|
if (this.waiting.size > 0) {
|
||||||
|
this.log.verbose('linkGate - sending what was held for a link', { held: this.waiting.size });
|
||||||
|
}
|
||||||
|
|
||||||
this.release({});
|
this.release({});
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -65,7 +76,9 @@ export class LinkGate {
|
|||||||
wait(deadline: number, signal: AbortSignal | undefined): Promise<VoidResult> {
|
wait(deadline: number, signal: AbortSignal | undefined): Promise<VoidResult> {
|
||||||
if (this.up) return Promise.resolve({});
|
if (this.up) return Promise.resolve({});
|
||||||
|
|
||||||
if (!this.returning) return Promise.resolve({ err: over() });
|
const refused = this.refusal();
|
||||||
|
|
||||||
|
if (refused) return Promise.resolve({ err: refused });
|
||||||
|
|
||||||
if (signal?.aborted === true) return Promise.resolve({ err: aborted() });
|
if (signal?.aborted === true) return Promise.resolve({ err: aborted() });
|
||||||
|
|
||||||
@@ -77,6 +90,8 @@ export class LinkGate {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private hold(left: number, signal: AbortSignal | undefined): Promise<VoidResult> {
|
private hold(left: number, signal: AbortSignal | undefined): Promise<VoidResult> {
|
||||||
|
this.log.verbose('linkGate - holding a request until a link is back', { timeout: left });
|
||||||
|
|
||||||
return new Promise<VoidResult>(resolve => {
|
return new Promise<VoidResult>(resolve => {
|
||||||
let timer: NodeJS.Timeout | undefined = undefined;
|
let timer: NodeJS.Timeout | undefined = undefined;
|
||||||
const settle = (result: VoidResult): void => {
|
const settle = (result: VoidResult): void => {
|
||||||
@@ -86,15 +101,18 @@ export class LinkGate {
|
|||||||
this.waiting.delete(settle);
|
this.waiting.delete(settle);
|
||||||
resolve(result);
|
resolve(result);
|
||||||
};
|
};
|
||||||
|
const giveUp = (): void => {
|
||||||
|
this.log.warn('linkGate - no link came back in time', { timeout: left });
|
||||||
|
settle({ err: expired() });
|
||||||
|
};
|
||||||
|
|
||||||
function onAbort(): void {
|
function onAbort(): void {
|
||||||
settle({ err: aborted() });
|
settle({ err: aborted() });
|
||||||
}
|
}
|
||||||
|
|
||||||
if (left > 0) {
|
// Deliberately not unref()'d: a held request is awaited with a destroyed socket and no
|
||||||
timer = setTimeout(() => { settle({ err: expired() }); }, left);
|
// other handle, so an unref'd timer lets the process exit without ever settling it.
|
||||||
timer.unref();
|
if (left > 0) timer = setTimeout(giveUp, left);
|
||||||
}
|
|
||||||
|
|
||||||
signal?.addEventListener('abort', onAbort, { once: true });
|
signal?.addEventListener('abort', onAbort, { once: true });
|
||||||
this.waiting.add(settle);
|
this.waiting.add(settle);
|
||||||
|
|||||||
+1
-1
@@ -33,7 +33,7 @@ export type SendSmsResult = {
|
|||||||
err?: Error;
|
err?: Error;
|
||||||
pduObjs: PduObject[];
|
pduObjs: PduObject[];
|
||||||
smsIds: string[];
|
smsIds: string[];
|
||||||
/** Segments the link dropped under. The peer may have taken them, so sending again may duplicate. */
|
/** Segments that went out unanswered. The peer may have taken them, so sending again may duplicate. */
|
||||||
unanswered: number;
|
unanswered: number;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|||||||
+5
-5
@@ -121,7 +121,7 @@ export class Session extends EventEmitter<SessionEvents> {
|
|||||||
max: defaults.maxDlrMerges,
|
max: defaults.maxDlrMerges,
|
||||||
timeout: defaults.dlrMergeTimeout,
|
timeout: defaults.dlrMergeTimeout,
|
||||||
});
|
});
|
||||||
this.gate = new LinkGate({ timeout: options.responseTimeout ?? defaults.responseTimeout });
|
this.gate = new LinkGate({ log: this.log, timeout: options.responseTimeout ?? defaults.responseTimeout });
|
||||||
this.incoming = new IncomingRequests({
|
this.incoming = new IncomingRequests({
|
||||||
dlrMerger: this.dlrMerger,
|
dlrMerger: this.dlrMerger,
|
||||||
log: this.log,
|
log: this.log,
|
||||||
@@ -174,7 +174,7 @@ export class Session extends EventEmitter<SessionEvents> {
|
|||||||
|
|
||||||
if (refused) return { err: refused };
|
if (refused) return { err: refused };
|
||||||
|
|
||||||
// A bind is what makes a link usable, so it cannot wait for one. The door unbind() uses too.
|
// A bind is what makes a link usable, so it cannot wait for one.
|
||||||
if (bindCommands.includes(input.cmdName)) return (await this.attempt(input, options)).result;
|
if (bindCommands.includes(input.cmdName)) return (await this.attempt(input, options)).result;
|
||||||
|
|
||||||
const deadline = this.gate.deadline();
|
const deadline = this.gate.deadline();
|
||||||
@@ -189,7 +189,6 @@ export class Session extends EventEmitter<SessionEvents> {
|
|||||||
const attempt = await this.attempt(input, options).finally(() => { this.window.release(); });
|
const attempt = await this.attempt(input, options).finally(() => { this.window.release(); });
|
||||||
|
|
||||||
// Nothing reached the socket, so the next link carries it instead of the caller resending.
|
// Nothing reached the socket, so the next link carries it instead of the caller resending.
|
||||||
// The gate's own answer, or a loop round one that admits everything spins on a dead socket.
|
|
||||||
if (!attempt.retryOnNextLink || this.gate.isUp() || !this.retrying()) return attempt.result;
|
if (!attempt.retryOnNextLink || this.gate.isUp() || !this.retrying()) return attempt.result;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -206,7 +205,8 @@ export class Session extends EventEmitter<SessionEvents> {
|
|||||||
// Before the gate and the window, or an aborted call waits for what it will never use.
|
// Before the gate and the window, or an aborted call waits for what it will never use.
|
||||||
if (options.signal?.aborted === true) return abortedBeforeSend();
|
if (options.signal?.aborted === true) return abortedBeforeSend();
|
||||||
|
|
||||||
return undefined;
|
// A bind skips the gate below, so the answer it would have given is given here instead.
|
||||||
|
return bindCommands.includes(input.cmdName) ? this.gate.refusal() : undefined;
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Read through a method: a drop can land while a send is awaiting. */
|
/** Read through a method: a drop can land while a send is awaiting. */
|
||||||
@@ -338,9 +338,9 @@ export class Session extends EventEmitter<SessionEvents> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
this.resetTimers();
|
this.resetTimers();
|
||||||
|
this.gate.open();
|
||||||
this.log.info('session - reconnected');
|
this.log.info('session - reconnected');
|
||||||
this.emit('reconnected');
|
this.emit('reconnected');
|
||||||
this.gate.open();
|
|
||||||
|
|
||||||
return {};
|
return {};
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -674,7 +674,7 @@ describe('sends across a reconnect', () => {
|
|||||||
|
|
||||||
smpp.on('session', peer => { peer.on('sms', () => undefined); });
|
smpp.on('session', peer => { peer.on('sms', () => undefined); });
|
||||||
|
|
||||||
const { session } = await connect(t, smpp, { responseTimeout: 60 });
|
const { session } = await connect(t, smpp, { responseTimeout: 200 });
|
||||||
|
|
||||||
assert.ok(session);
|
assert.ok(session);
|
||||||
|
|
||||||
@@ -729,7 +729,7 @@ describe('sends across a reconnect', () => {
|
|||||||
const smpp = await startServer(t);
|
const smpp = await startServer(t);
|
||||||
const { session } = await connect(t, smpp, {
|
const { session } = await connect(t, smpp, {
|
||||||
reconnect: { maxDelay: 10_000, minDelay: 10_000 },
|
reconnect: { maxDelay: 10_000, minDelay: 10_000 },
|
||||||
responseTimeout: 60,
|
responseTimeout: 200,
|
||||||
});
|
});
|
||||||
|
|
||||||
assert.ok(session);
|
assert.ok(session);
|
||||||
@@ -818,7 +818,7 @@ describe('sends across a reconnect', () => {
|
|||||||
describe('LinkGate', () => {
|
describe('LinkGate', () => {
|
||||||
test('refuses a hold whose deadline has already passed', async () => {
|
test('refuses a hold whose deadline has already passed', async () => {
|
||||||
let now = 0;
|
let now = 0;
|
||||||
const gate = new LinkGate({ now: () => now, timeout: 100 });
|
const gate = new LinkGate({ log: silentLog, now: () => now, timeout: 100 });
|
||||||
const deadline = gate.deadline();
|
const deadline = gate.deadline();
|
||||||
|
|
||||||
gate.shut(true);
|
gate.shut(true);
|
||||||
@@ -829,9 +829,27 @@ describe('LinkGate', () => {
|
|||||||
assert.match(held.err?.message ?? '', /did not come back in time/);
|
assert.match(held.err?.message ?? '', /did not come back in time/);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// A held send is awaited with the socket destroyed, so an unref'd timer here lets a process whose
|
||||||
|
// only remaining work is that send exit without ever settling it, losing the message silently.
|
||||||
|
test('holds on a timer that keeps the process alive', async () => {
|
||||||
|
const gate = new LinkGate({ log: silentLog, timeout: 10_000 });
|
||||||
|
const timers = (): number => process.getActiveResourcesInfo().filter(name => name === 'Timeout').length;
|
||||||
|
|
||||||
|
gate.shut(true);
|
||||||
|
|
||||||
|
const before = timers();
|
||||||
|
const held = gate.wait(gate.deadline(), undefined);
|
||||||
|
|
||||||
|
assert.equal(timers(), before + 1, 'an unref\'d timer is not counted here, which is the point');
|
||||||
|
|
||||||
|
gate.open();
|
||||||
|
|
||||||
|
assert.deepEqual(await held, {});
|
||||||
|
});
|
||||||
|
|
||||||
// addEventListener never fires for a signal that already aborted, so it would wait out the timeout.
|
// addEventListener never fires for a signal that already aborted, so it would wait out the timeout.
|
||||||
test('gives up at once on a signal that was already aborted', async () => {
|
test('gives up at once on a signal that was already aborted', async () => {
|
||||||
const gate = new LinkGate({ timeout: 100 });
|
const gate = new LinkGate({ log: silentLog, timeout: 100 });
|
||||||
|
|
||||||
gate.shut(true);
|
gate.shut(true);
|
||||||
|
|
||||||
@@ -1073,7 +1091,7 @@ describe('graceful shutdown', () => {
|
|||||||
const result = await sent;
|
const result = await sent;
|
||||||
|
|
||||||
assert.ok(result.err instanceof Error);
|
assert.ok(result.err instanceof Error);
|
||||||
assert.match(result.err.message, /may have accepted/);
|
assert.match(result.err.message, /may have accepted.*Session closed before a response arrived/);
|
||||||
assert.equal(result.unanswered, 1);
|
assert.equal(result.unanswered, 1);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
@@ -236,7 +236,7 @@ describe('bind', () => {
|
|||||||
const sent = await session.send({ cmdName: 'enquire_link' });
|
const sent = await session.send({ cmdName: 'enquire_link' });
|
||||||
|
|
||||||
assert.ok(sent.err instanceof Error);
|
assert.ok(sent.err instanceof Error);
|
||||||
assert.match(sent.err.message, /may have accepted/);
|
assert.match(sent.err.message, /may have accepted.*Session closed before a response arrived/);
|
||||||
});
|
});
|
||||||
|
|
||||||
test('reports an unbind the peer left unanswered on a link that stays up', async t => {
|
test('reports an unbind the peer left unanswered on a link that stays up', async t => {
|
||||||
|
|||||||
@@ -137,8 +137,9 @@ session message is a change to every call site.
|
|||||||
bind already need, with `Session` calling it when a link comes up or goes down instead of
|
bind already need, with `Session` calling it when a link comes up or goes down instead of
|
||||||
spreading `linkDown()` and `retrying()` across both sides. It would also close two smaller
|
spreading `linkDown()` and `retrying()` across both sides. It would also close two smaller
|
||||||
things — `LinkGate.deadline()` and `wait(deadline)` are a two-call protocol whose only failure
|
things — `LinkGate.deadline()` and `wait(deadline)` are a two-call protocol whose only failure
|
||||||
mode is calling `deadline()` inside the loop, which nothing catches; and `Session.linkDown()`
|
mode is calling `deadline()` inside the loop, which nothing catches; `Session.linkDown()` is
|
||||||
is read from both sides of the seam. Every one of these is unpublished, so it is a two-way
|
read from both sides of the seam; and `UnansweredError` sits in `pending-requests.ts`, which
|
||||||
|
never uses it, for the sole edge that makes `send-sms` import that module at all. Every one of these is unpublished, so it is a two-way
|
||||||
door and belongs after 1.0.0. Raised by review, 2026-09-01.
|
door and belongs after 1.0.0. 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
|
||||||
|
|||||||
Reference in New Issue
Block a user