diff --git a/AGENTS.md b/AGENTS.md index 41703ed..3ebb92d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -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`, 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 - 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.** `onDeliverSm()` answers each receipt before the group it belongs to is complete, and `teardown()` diff --git a/README.md b/README.md index 7f435b7..1df59c3 100644 --- a/README.md +++ b/README.md @@ -101,7 +101,7 @@ Every one is optional. | `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`. | | `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. | | `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. | diff --git a/src/outgoing-requests.ts b/src/outgoing-requests.ts index 4e4c490..d97ddee 100644 --- a/src/outgoing-requests.ts +++ b/src/outgoing-requests.ts @@ -84,8 +84,12 @@ export class OutgoingRequests { if (refused) return { err: refused }; - // A bind is what makes a link usable, so it cannot wait for one. - if (bindCommands.includes(input.cmdName)) return this.now(input, options); + // 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)) { + const shut = this.gate.refusal(); + + return shut ? { err: shut } : this.now(input, options); + } 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. - if (options.signal?.aborted === true) return abortedBeforeSend(); - - // 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; + return options.signal?.aborted === true ? abortedBeforeSend() : undefined; } private async attempt(input: PduObjectInput, options: SendOptions): Promise { diff --git a/src/session.ts b/src/session.ts index 24c2b10..3f232b4 100644 --- a/src/session.ts +++ b/src/session.ts @@ -304,10 +304,8 @@ export class Session extends EventEmitter { const timeout = this.options.shutdownTimeout ?? defaults.shutdownTimeout; 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. - 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); // The window empties on a teardown too, which settles everything the link was carrying. @@ -318,6 +316,19 @@ export class Session extends EventEmitter { 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. */ private end(): void { this.reconnectLoop?.stop(); diff --git a/src/sms.ts b/src/sms.ts index be341f2..5c3b6d1 100644 --- a/src/sms.ts +++ b/src/sms.ts @@ -138,11 +138,11 @@ async function sendDlr( } const total = sms.pduObjs.length; - const pduObjs: PduObject[] = []; - - for (let index = 0; index < total; index++) { + // 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) => { const smsId = segmentId(sms.smsId, index, total); - const sent = await send({ + + return send({ cmdName: 'deliver_sm', params: { destination_addr: sms.from, @@ -152,10 +152,13 @@ async function sendDlr( }, ...(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 }; diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index 963ce05..9587025 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -1041,6 +1041,7 @@ describe('graceful shutdown', () => { t: TestContext, options: Parameters[0] = {}, serverOptions: Parameters[0] = {}, + message = 'answer me', ) { const smpp = await startServer(t, serverOptions); const incoming = once(resolve => { @@ -1050,7 +1051,7 @@ describe('graceful shutdown', () => { 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 }; } @@ -1113,20 +1114,29 @@ describe('graceful shutdown', () => { 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 => { - const { sent, session, smpp, sms } = await submitInFlight(t); - const receipt = once(resolve => { session.on('dlr', resolve); }); + const { sent, session, smpp, sms } = await submitInFlight(t, {}, {}, 'x'.repeat(400)); + const received: Dlr[] = []; + const receipts = once(resolve => { + session.on('dlr', dlr => { + received.push(dlr); + + if (received.length === 3) resolve(received); + }); + }); const closing = peerOf(smpp).close(); await sms.sendResp({ smsId: 'held-through-the-drain' }); 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((await receipt).smsId, 'held-through-the-drain'); + assert.deepEqual((await receipts).map(dlr => dlr.smsId), ids); 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 => { diff --git a/todo.md b/todo.md index f4e8443..8c08242 100644 --- a/todo.md +++ b/todo.md @@ -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, 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.** `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