Put a receipt's segments on the wire together, like a message's

This commit is contained in:
2026-09-01 17:03:33 +02:00
parent 8becbc0ab6
commit 1e7e113bd7
7 changed files with 56 additions and 23 deletions
+2 -1
View File
@@ -354,7 +354,8 @@ Grouped by what each one constrains.
half: waiting forever is safe for the peer, whose every request is bounded by `responseTimeout`, half: waiting forever is safe for the peer, whose every request is bounded by `responseTimeout`,
and unsafe for the application, which nothing bounds — `close()` is what you reach for when the and unsafe for the application, which nothing bounds — `close()` is what you reach for when the
application is stuck, so it may not block on the application coming unstuck. That half falls back application is stuck, so it may not block on the application coming unstuck. That half falls back
to `responseTimeout`, the same answer the link gate's hold already takes. to `responseTimeout`, the same answer the link gate's hold already takes — and to that option's
default where it is 0 as well, since neither option is an answer about the application.
- **A reconnect keeps the delivery-receipt merges; everything else the link held is dropped.** - **A reconnect keeps the delivery-receipt merges; everything else the link held is dropped.**
`onDeliverSm()` answers each receipt before the group it belongs to is complete, and `teardown()` `onDeliverSm()` answers each receipt before the group it belongs to is complete, and `teardown()`
+1 -1
View File
@@ -101,7 +101,7 @@ Every one is optional.
| `enquireLinkInterval` | `20000` | How often to send `enquire_link` on a quiet link. | | `enquireLinkInterval` | `20000` | How often to send `enquire_link` on a quiet link. |
| `idleTimeout` | `2 × enquireLinkInterval` | Give up on a link the peer has stopped answering, and re-bind unless `reconnect` is `false`. | | `idleTimeout` | `2 × enquireLinkInterval` | Give up on a link the peer has stopped answering, and re-bind unless `reconnect` is `false`. |
| `responseTimeout` | `30000` | How long to wait for a response before giving up on it, and how long a send with no link waits for the next one; `0` waits forever. | | `responseTimeout` | `30000` | How long to wait for a response before giving up on it, and how long a send with no link waits for the next one; `0` waits forever. |
| `shutdownTimeout` | `5000` | How long `close()` and `unbind()` wait for the requests this end already sent and the messages the application has not answered. `0` waits forever for the requests, which the peer answers or times out; the messages fall back to `responseTimeout`, since nothing but the application ends that wait. | | `shutdownTimeout` | `5000` | How long `close()` and `unbind()` wait for the requests this end already sent and the messages the application has not answered. `0` waits forever for the requests, which the peer answers or times out; the messages fall back to `responseTimeout`, or to its default where that is 0 too, since nothing but the application ends that wait. |
| `maxOutstanding` | `10` | Requests allowed on the wire at once; further sends queue. | | `maxOutstanding` | `10` | Requests allowed on the wire at once; further sends queue. |
| `smsIdFormat` | — | The notation the SMSC writes message ids in, per place it writes them: `{ receipt: 'decimal', submitResp: 'hex' }`. Only needed where the two disagree. | | `smsIdFormat` | — | The notation the SMSC writes message ids in, per place it writes them: `{ receipt: 'decimal', submitResp: 'hex' }`. Only needed where the two disagree. |
| `reconnect` | on | Re-binds after a drop, an idle timeout, or a stream the library cannot read, backing off from `minDelay` 1 s to `maxDelay` 30 s and starting over at `minDelay` once a link has lasted `maxDelay`. `{ minDelay, maxDelay }` retunes it; `false` turns it off, so a drop ends the session. | | `reconnect` | on | Re-binds after a drop, an idle timeout, or a stream the library cannot read, backing off from `minDelay` 1 s to `maxDelay` 30 s and starting over at `minDelay` once a link has lasted `maxDelay`. `{ minDelay, maxDelay }` retunes it; `false` turns it off, so a drop ends the session. |
+7 -6
View File
@@ -84,8 +84,12 @@ export class OutgoingRequests {
if (refused) return { err: refused }; if (refused) return { err: refused };
// A bind is what makes a link usable, so it cannot wait for one. // A bind is what makes a link usable, so it cannot wait for one: it takes the gate's answer now.
if (bindCommands.includes(input.cmdName)) return this.now(input, options); if (bindCommands.includes(input.cmdName)) {
const shut = this.gate.refusal();
return shut ? { err: shut } : this.now(input, options);
}
const waitForLink = this.gate.hold(options.signal); const waitForLink = this.gate.hold(options.signal);
@@ -131,10 +135,7 @@ export class OutgoingRequests {
} }
// 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(); return options.signal?.aborted === true ? abortedBeforeSend() : 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;
} }
private async attempt(input: PduObjectInput, options: SendOptions): Promise<Attempt> { private async attempt(input: PduObjectInput, options: SendOptions): Promise<Attempt> {
+14 -3
View File
@@ -304,10 +304,8 @@ export class Session extends EventEmitter<SessionEvents> {
const timeout = this.options.shutdownTimeout ?? defaults.shutdownTimeout; const timeout = this.options.shutdownTimeout ?? defaults.shutdownTimeout;
const deadline = timeout > 0 ? Date.now() + timeout : 0; const deadline = timeout > 0 ? Date.now() + timeout : 0;
// Only the application answers a held message, so that half falls back rather than wait forever.
const answering = timeout > 0 ? timeout : (this.options.responseTimeout ?? defaults.responseTimeout);
// Answering a message can put a receipt on the wire; nothing on the wire produces a message. // Answering a message can put a receipt on the wire; nothing on the wire produces a message.
const messages = await this.incoming.drain(answering, signal); const messages = await this.incoming.drain(this.answering(timeout), signal);
const requests = await this.outgoing.drain(leftOf(deadline), signal); const requests = await this.outgoing.drain(leftOf(deadline), signal);
// The window empties on a teardown too, which settles everything the link was carrying. // The window empties on a teardown too, which settles everything the link was carrying.
@@ -318,6 +316,19 @@ export class Session extends EventEmitter<SessionEvents> {
return messages.err ? messages : requests; return messages.err ? messages : requests;
} }
/**
* How long the drain waits for the application, which is the only thing that can end that wait.
* Neither timeout may hand it "forever": both are answers about a peer, and a peer is not what
* this half is waiting for.
*/
private answering(timeout: number): number {
if (timeout > 0) return timeout;
const responseTimeout = this.options.responseTimeout ?? defaults.responseTimeout;
return responseTimeout > 0 ? responseTimeout : defaults.responseTimeout;
}
/** The session is over now, drained or not. Nothing brings it back. */ /** The session is over now, drained or not. Nothing brings it back. */
private end(): void { private end(): void {
this.reconnectLoop?.stop(); this.reconnectLoop?.stop();
+9 -6
View File
@@ -138,11 +138,11 @@ async function sendDlr(
} }
const total = sms.pduObjs.length; const total = sms.pduObjs.length;
const pduObjs: PduObject[] = []; // Together, not one after a response: a drain waiting for this message must see the whole receipt.
const sent = await Promise.all(sms.pduObjs.map((_segment, index) => {
for (let index = 0; index < total; index++) {
const smsId = segmentId(sms.smsId, index, total); const smsId = segmentId(sms.smsId, index, total);
const sent = await send({
return send({
cmdName: 'deliver_sm', cmdName: 'deliver_sm',
params: { params: {
destination_addr: sms.from, destination_addr: sms.from,
@@ -152,10 +152,13 @@ async function sendDlr(
}, },
...(sms.session.acceptsOptionalParams() ? { tlvs: receiptTlvs(smsId, status) } : {}), ...(sms.session.acceptsOptionalParams() ? { tlvs: receiptTlvs(smsId, status) } : {}),
}); });
}));
const pduObjs: PduObject[] = [];
if (sent.err) return { err: sent.err }; for (const one of sent) {
if (one.err) return { err: one.err };
pduObjs.push(sent.pduObj); pduObjs.push(one.pduObj);
} }
return { pduObjs }; return { pduObjs };
+16 -6
View File
@@ -1041,6 +1041,7 @@ describe('graceful shutdown', () => {
t: TestContext, t: TestContext,
options: Parameters<typeof client>[0] = {}, options: Parameters<typeof client>[0] = {},
serverOptions: Parameters<typeof server>[0] = {}, serverOptions: Parameters<typeof server>[0] = {},
message = 'answer me',
) { ) {
const smpp = await startServer(t, serverOptions); const smpp = await startServer(t, serverOptions);
const incoming = once<Sms>(resolve => { const incoming = once<Sms>(resolve => {
@@ -1050,7 +1051,7 @@ describe('graceful shutdown', () => {
assert.ok(session); assert.ok(session);
const sent = session.sendSms({ from: '46701113311', message: 'answer me', to: '46709771337' }); const sent = session.sendSms({ from: '46701113311', message, to: '46709771337' });
return { sent, session, sms: await incoming, smpp }; return { sent, session, sms: await incoming, smpp };
} }
@@ -1113,20 +1114,29 @@ describe('graceful shutdown', () => {
assert.ok(Date.now() - started < 2000); assert.ok(Date.now() - started < 2000);
}); });
// The README's own listener answers and then sends its receipt, one turn later. // The README's own listener answers and then sends its receipt, one turn later. Multipart, because
// a receipt sent one-after-a-response outruns that turn on every segment past the first.
test('a receipt sent right after the response still goes out mid-drain', async t => { test('a receipt sent right after the response still goes out mid-drain', async t => {
const { sent, session, smpp, sms } = await submitInFlight(t); const { sent, session, smpp, sms } = await submitInFlight(t, {}, {}, 'x'.repeat(400));
const receipt = once<Dlr>(resolve => { session.on('dlr', resolve); }); const received: Dlr[] = [];
const receipts = once<Dlr[]>(resolve => {
session.on('dlr', dlr => {
received.push(dlr);
if (received.length === 3) resolve(received);
});
});
const closing = peerOf(smpp).close(); const closing = peerOf(smpp).close();
await sms.sendResp({ smsId: 'held-through-the-drain' }); await sms.sendResp({ smsId: 'held-through-the-drain' });
const receiptSent = await sms.sendDlr('DELIVERED'); const receiptSent = await sms.sendDlr('DELIVERED');
const ids = ['held-through-the-drain-1', 'held-through-the-drain-2', 'held-through-the-drain-3'];
assert.equal(receiptSent.err, undefined); assert.equal(receiptSent.err, undefined);
assert.equal((await receipt).smsId, 'held-through-the-drain'); assert.deepEqual((await receipts).map(dlr => dlr.smsId), ids);
assert.deepEqual(await closing, {}); assert.deepEqual(await closing, {});
assert.deepEqual((await sent).smsIds, ['held-through-the-drain']); assert.deepEqual((await sent).smsIds, ids);
}); });
test('a message no listener took does not hold the shutdown up', async t => { test('a message no listener took does not hold the shutdown up', async t => {
+7
View File
@@ -132,6 +132,13 @@ 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.
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.** - [ ] **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 `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 application still holds the `Sms`, so `sendResp()` writes the old link's sequence numbers to the