diff --git a/AGENTS.md b/AGENTS.md index edeeab1..816a46f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -350,13 +350,18 @@ Grouped by what each one constrains. 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 unstuck. That half falls back - 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. + 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 + unstuck. That half falls back 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. What is held is capped and expiring like every other inbound store, on constants + rather than options, because a bound the application cannot raise is the point: an application that + answers nothing would otherwise grow it for the life of the link, which goal 4 forbids. A message + that falls out of the bound is one the drain stops waiting for, so `close()` can report fewer + unanswered than there were — accepted, because the alternative is holding what nothing will answer, + and both exits are logged. - **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()` @@ -400,8 +405,9 @@ Grouped by what each one constrains. `unbind()` takes through `now()`. The gate is told what happened and never reads back into the session: a collaborator that has to ask does not own its decision, which is how the first cut ended up answering the same question two different ways at admit and at release. For the same reason the - retry in `pastDrain()` asks `gate.isUp()` rather than `linkDown()`, which also reads the socket — a condition that loops on - something the gate does not gate on spins against a gate that admits it straight back. + retry in `pastDrain()` asks `gate.isUp()` rather than `linkDown()`, which also reads the socket — a + condition that loops on something the gate does not gate on spins against a gate that admits it + straight back. `LinkGate.returning` is a copy of `retrying()` taken at teardown, and stays true only because nothing stops the reconnect loop without `emitClose()` following it: `drain()` and `end()` are the only callers of `stop()`. A third caller has to shut the gate itself. diff --git a/README.md b/README.md index 80fcd3e..b036159 100644 --- a/README.md +++ b/README.md @@ -351,7 +351,8 @@ const { err, pduObj } = await session.send({ A send issued while the link is down waits for the reconnect instead of failing, and goes out once the new link is bound — up to `responseTimeout`, after which it gives up having sent nothing. A request already on the wire is the other case: the SMSC may have taken it and lost only the response, -so it fails, and `sendSms()` counts it in `unanswered`, whether the link dropped under it, the peer +so it fails, and `sendSms()` and `sms.sendDlr()` count it in `unanswered`, whether the link dropped +under it, the peer never answered in time, or you aborted it after it went out. Neither applies with `reconnect: false`, where a drop ends the session and every send after it is refused. diff --git a/src/held-messages.ts b/src/held-messages.ts index 7abdef7..5eeb99e 100644 --- a/src/held-messages.ts +++ b/src/held-messages.ts @@ -21,6 +21,7 @@ export class HeldMessages { private readonly held: ExpiringGroups; private readonly idleWaiters = new IdleWaiters(); private readonly log: SmppLog; + private readonly max: number; constructor(options: HeldMessagesOptions) { this.held = new ExpiringGroups({ @@ -29,6 +30,7 @@ export class HeldMessages { timeout: options.timeout, }); this.log = options.log; + this.max = options.max; } /** An application that answers no message at all may not grow this without end. */ @@ -37,12 +39,10 @@ export class HeldMessages { if (key === undefined) return; - if (this.held.full) { - const evicted = this.held.takeOldest(); - - if (evicted) { - this.log.warn('heldMessages - dropping the message held longest', { seqNr: evicted[0] }); - } + if (this.held.get(key)) { + this.log.warn('heldMessages - replacing a message on a re-used sequence number', { seqNr: key }); + } else if (this.held.full) { + this.dropOldest(); } this.held.set(key, pduObjs); @@ -76,6 +76,16 @@ export class HeldMessages { return this.idleWaiters.wait(() => this.held.size, timeout, signal); } + private dropOldest(): void { + const oldest = this.held.takeOldest(); + + if (!oldest) return; + + const [seqNr] = oldest; + + this.log.warn('heldMessages - buffer full, dropping the oldest message', { max: this.max, seqNr }); + } + private sweep(): void { const expired = this.held.takeExpired(); diff --git a/src/idle-waiters.ts b/src/idle-waiters.ts index 50c9311..dd29a9e 100644 --- a/src/idle-waiters.ts +++ b/src/idle-waiters.ts @@ -18,7 +18,7 @@ export class IdleWaiters { * Resolves 0 once nothing is left, or with what still is when the timeout or the signal cuts the * wait short. A timeout of 0 waits forever. */ - wait(remaining: () => number, timeout: number, signal?: AbortSignal): Promise { + wait(remaining: () => number, timeout: number, signal: AbortSignal | undefined): Promise { if (remaining() === 0) return Promise.resolve(0); if (signal?.aborted === true) return Promise.resolve(remaining()); diff --git a/src/send-window.ts b/src/send-window.ts index 46560c3..524df8d 100644 --- a/src/send-window.ts +++ b/src/send-window.ts @@ -43,7 +43,7 @@ export class SendWindow { } /** Resolves 0 once nothing is left on the wire, or with what still is. */ - idle(timeout: number, signal?: AbortSignal): Promise { + idle(timeout: number, signal: AbortSignal | undefined): Promise { return this.idleWaiters.wait(() => this.unfinished(), timeout, signal); } } diff --git a/src/sms.ts b/src/sms.ts index 8942854..5b741b4 100644 --- a/src/sms.ts +++ b/src/sms.ts @@ -137,6 +137,28 @@ function receiptTlvs(smsId: string, status: MessageState): Record[]): SendDlrResult { + const pduObjs: PduObject[] = []; + let failure: Error | undefined; + let unanswered = 0; + + for (const one of sent) { + if (one.err) { + if (one.err instanceof UnansweredError) unanswered++; + + failure ??= one.err; + } else if (one.pduObj.cmdStatus === 'ESME_ROK') { + pduObjs.push(one.pduObj); + } else { + const refusal = one.pduObj.cmdStatus ?? String(one.pduObj.cmdStatusId); + + failure ??= new Error(`deliver_sm refused by the peer: ${refusal}`); + } + } + + return failure ? { err: failure, pduObjs, unanswered } : { pduObjs, unanswered }; +} + async function sendDlr( sms: Sms, send: SmsHandlers['send'], @@ -166,19 +188,5 @@ async function sendDlr( ...(sms.session.acceptsOptionalParams() ? { tlvs: receiptTlvs(smsId, status) } : {}), }); })); - const pduObjs: PduObject[] = []; - let failure: Error | undefined; - let unanswered = 0; - - for (const one of sent) { - if (!one.err) { - pduObjs.push(one.pduObj); - } else { - if (one.err instanceof UnansweredError) unanswered++; - - failure ??= one.err; - } - } - - return failure ? { err: failure, pduObjs, unanswered } : { pduObjs, unanswered }; + return collectReceipt(sent); } diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index 0a6c6a3..3bac0f8 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -13,6 +13,8 @@ import type { Sms } from '../src/sms.ts'; import type { SmppServer } from '../src/server.ts'; import type { TestContext } from 'node:test'; import { HeldMessages } from '../src/held-messages.ts'; +import { UnansweredError } from '../src/unanswered-error.ts'; +import { createSms } from '../src/sms.ts'; import { LinkGate } from '../src/link-gate.ts'; import { Reassembler, decodeSegments } from '../src/reassembly.ts'; import { Session } from '../src/session.ts'; @@ -860,8 +862,8 @@ describe('LinkGate', () => { // Goal 4: an application that answers nothing must not grow this for the life of the link. describe('held message bounds', () => { - function message(seqNr: number): PduObject[] { - return [{ + function heldPdu(seqNr: number): PduObject { + return { cmdId: 0x00000004, cmdLength: 0, cmdName: 'submit_sm', @@ -870,7 +872,11 @@ describe('held message bounds', () => { params: { destination_addr: '46709771337', short_message: 'held', source_addr: '46701113311' }, seqNr, tlvs: {}, - }]; + }; + } + + function message(seqNr: number): PduObject[] { + return [heldPdu(seqNr)]; } test('drops the message held longest rather than holding every one', async () => { @@ -893,6 +899,37 @@ describe('held message bounds', () => { assert.equal(await held.idle(1, undefined), 1); assert.equal(await held.idle(1000, undefined), 0); }); + + // A receipt cannot be resent wholesale without duplicating the segments that landed, so sendDlr() + // names what the peer took and what it may have, the way sendSms() does. + test('a partial receipt names the segments the peer took and the ones it may have', async t => { + const session = new Session({ sock: new net.Socket() }); + + closeAfter(t, session); + + let call = 0; + const sms = createSms({ + from: '46701113311', + message: 'three segments', + pduObjs: [heldPdu(1), heldPdu(2), heldPdu(3)], + session, + to: '46709771337', + }, { + onAnswered: () => undefined, + send: () => { + call++; + + return Promise.resolve(call === 2 + ? { err: new UnansweredError(new Error('nothing came back')) } + : { pduObj: heldPdu(call) }); + }, + }); + const report = await sms.sendDlr('DELIVERED'); + + assert.ok(report.err instanceof Error); + assert.equal(report.pduObjs.length, 2); + assert.equal(report.unanswered, 1); + }); }); describe('reassembly bounds', () => { @@ -1112,6 +1149,21 @@ describe('graceful shutdown', () => { assert.deepEqual(await closed, {}); }); + // The drain refuses sends; a response was never a send, and saying so is the more useful answer. + test('names a response put through send() as the misuse it is, even mid-shutdown', async t => { + const { sent, session, sms } = await submitInFlight(t); + const closing = session.close(); + const refused = await session.send({ cmdName: 'submit_sm_resp' }); + + assert.ok(refused.err instanceof Error); + assert.match(refused.err.message, /Use sendReturn\(\)/); + + await sms.sendResp({ smsId: 'answered-after-the-misuse' }); + + assert.deepEqual((await sent).smsIds, ['answered-after-the-misuse']); + assert.deepEqual(await closing, {}); + }); + test('unbind() waits out a submit already on the wire before it unbinds', async t => { const { sent, session, sms } = await submitInFlight(t); const unbound = session.unbind();