diff --git a/AGENTS.md b/AGENTS.md index bb7369c..a4210b2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -351,15 +351,16 @@ Grouped by what each one constrains. 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 close — and a message no listener took is released at once, since nothing is going to answer it. - 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 - 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 - would slip past is no longer waiting for it. `shutdownTimeout: 0` does not carry over to this half: + A listener that failed before answering gives it up the same way, but only once every listener has: + a throw stops `emit()` where it stands, while a rejection leaves the others running, so the release + waits for the last of them rather than answering on their behalf. What ends the wait is the response + reaching the wire, not the call — a `sendResp()` the library refused, or one the socket would not + carry, 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 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 would slip past is + no longer waiting for it. `shutdownTimeout: 0` does not carry over to this half: waiting forever is safe for the peer, whose every request is bounded by `responseTimeout` unless the caller set that to 0 as well, 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 diff --git a/README.md b/README.md index fb3d8de..9d879ae 100644 --- a/README.md +++ b/README.md @@ -332,9 +332,10 @@ TypeScript users can import `SmppLog` to have the compiler check one. `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 then tear down whatever is left, resolving to an `err` that says what was lost. They also wait for -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()` -instead leaves that wait running until it gives up. `sendDlr()` is the one send the refusal lets +every `sms` still in the application's hands, so a peer whose `submit_sm` is +being handled is answered rather than left to re-send it. That wait ends when `sendResp()` puts the +response on the wire, or when every listener that took the message has failed; answering its PDUs +through `sendReturn()` instead leaves the 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 it races the shutdown like any other send. `close({ signal })` takes an `AbortSignal` that cuts the wait short; `unbind()` takes none, and waits a further diff --git a/src/incoming-requests.ts b/src/incoming-requests.ts index fca8fa7..e4208a2 100644 --- a/src/incoming-requests.ts +++ b/src/incoming-requests.ts @@ -31,6 +31,8 @@ export type IncomingRequestsOptions = { /** Everything the peer asks of a session: messages, receipts, links and the answers to them. */ export class IncomingRequests { private readonly dlrMerger: DlrMerger; + /** The rejection handler is handed the Sms back as an `unknown`, so its hold is found by identity. */ + private readonly emitted = new WeakMap void>(); private readonly held: HeldMessages; private readonly log: SmppLog; private readonly onRequest: OnRequest | undefined; @@ -39,8 +41,6 @@ export class IncomingRequests { private readonly session: Session; private readonly smsIdFormat: SmsIdFormat; private readonly systemId: string; - /** Identity, not a type guard: the rejected event carries the Sms back as an `unknown`. */ - private readonly emitted = new WeakMap void>(); private linkGeneration = 0; constructor(options: IncomingRequestsOptions) { @@ -113,7 +113,7 @@ export class IncomingRequests { this.reassembler.clear(); } - /** A listener that rejected answered nothing and never will, so a shutdown may not wait for it. */ + /** One `sms` listener gave up on a message; the last one to do so is what releases the hold. */ listenerRejected(sms: unknown): void { if (typeof sms !== 'object' || sms === null) return; @@ -184,6 +184,8 @@ export class IncomingRequests { if (!first) return; const generation = this.linkGeneration; + // A turn later, so a listener sending its receipt straight after the response still holds. + const release = (): void => { setImmediate(() => { this.held.release(pduObjs); }); }; const sms = createSms({ from: paramText(first.params.source_addr), @@ -193,14 +195,20 @@ export class IncomingRequests { 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. - onAnswered: () => { setImmediate(() => { this.held.release(pduObjs); }); }, + onAnswered: release, // Past the refusal only while a drain is still waiting for this message; an ordinary send after. send: input => (this.held.has(pduObjs) ? this.sendPastDrain(input) : this.session.send(input)), }); + // A rejection leaves the other listeners running, so only the last one to fail gives the message up. + let working = this.session.listenerCount('sms'); + this.held.hold(pduObjs); - this.emitted.set(sms, () => { this.held.release(pduObjs); }); + this.emitted.set(sms, () => { + working--; + + if (working <= 0) release(); + }); // A message nobody took is not work a shutdown can wait for. if (!this.session.emit('sms', sms)) this.held.release(pduObjs); diff --git a/src/sms.ts b/src/sms.ts index 750ea02..9f5d898 100644 --- a/src/sms.ts +++ b/src/sms.ts @@ -93,7 +93,7 @@ async function sendResp( sms: Sms, answered: { smsId: string }, options: SendRespOptions, - handlers: SmsHandlers, + handlers: Pick, ): Promise { const total = sms.pduObjs.length; @@ -118,10 +118,11 @@ async function sendResp( { 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(); + const failure = results.find(result => result.err); - return results.find(result => result.err) ?? {}; + if (!failure) handlers.onAnswered(); + + return failure ?? {}; } /** The receipt as text, which is all of it a peer below SMPP 3.4 is allowed to be sent. */ diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index f6b9212..c3a4925 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -1012,6 +1012,33 @@ describe('held message bounds', () => { }); }); +describe('sendResp()', () => { + // A response the wire never carried leaves the peer owed one, so nothing may count it answered. + test('does not count a response that never reached the wire as an answer', async t => { + const sock = new net.Socket(); + const session = new Session({ sock }); + + closeAfter(t, session); + sock.destroy(); + + let answered = 0; + const sms = createSms({ + from: '46701113311', + message: 'never answered', + pduObjs: [submitPdu(1)], + session, + to: '46709771337', + }, { + lostLink: () => false, + onAnswered: () => { answered++; }, + send: () => Promise.resolve({ err: new Error('never sent') }), + }); + + assert.match((await sms.sendResp()).err?.message ?? '', /Socket is closed/); + assert.equal(answered, 0); + }); +}); + describe('sendDlr()', () => { // A receipt cannot be resent wholesale without duplicating the segments that landed. test('names the segments the peer took, refused, and may have taken', async t => { @@ -1429,6 +1456,34 @@ describe('graceful shutdown', () => { assert.ok((await sent).err instanceof Error); }); + // A rejection leaves the other listeners running, unlike a throw, which stops emit() where it is. + test('waits for the listener still working when another one rejected', async t => { + const smpp = await startServer(t, { shutdownTimeout: 30_000 }); + const failed = once(resolve => { + smpp.on('session', bound => { + bound.on('sessionError', resolve); + bound.on('sms', async sms => { + await delay(100); + await sms.sendResp({ smsId: 'answered-after-the-other-gave-up' }); + }); + bound.on('sms', () => Promise.reject(new Error('the audit listener gave up'))); + }); + }); + const { session } = await connect(t, smpp); + + assert.ok(session); + + const sent = session.sendSms({ + from: '46701113311', + message: 'two listeners, one gives up', + to: '46709771337', + }); + + assert.equal((await failed).message, 'the audit listener gave up'); + assert.deepEqual(await peerOf(smpp).close(), {}); + assert.deepEqual((await sent).smsIds, ['answered-after-the-other-gave-up']); + }); + // 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 });