Merge pull request #76 from larvit/hold-until-answered

Release a message's hold when the response goes out, or when its listener failed

AGENTS.md

@@ -34,8 +34,10 @@ one wins. They do not override the hard rules below.
    promise this library can verify. The low-level surface is a passthrough: policy binds what the
    library composes, never what the caller wrote.
 7. **Nothing that needs state wider than one session.** No throughput throttling, no persistence
-   across a restart, no coordination between processes. This is the scope floor, and it is why an
+   across a restart, no coordination between processes — and no seam handing the application state to
-   otherwise reasonable feature is declined without a fresh argument each time.
+   persist for one of those either, which commits to the same scope through the back door and
+   publishes an internal shape to do it. This is the scope floor, and it is why an otherwise
+   reasonable feature is declined without a fresh argument each time.
 8. **It builds, tests and runs the same everywhere.** Container-only toolchain, no runtime
    dependencies, the Node 18 floor verified in CI rather than asserted, every README example executed
    by the suite.
@@ -351,11 +353,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.
-  `teardown()` drops what is still held for the same reason it drops inbound segments. The release
+  A listener that failed before answering gives it up the same way, but only once every listener has:
-  is one turn late, so a listener that sends its receipt straight after the response is still
+  a throw stops `emit()` where it stands, while a rejection leaves the others running, so the release
-  holding when the drain looks; `sendDlr()` is the one send that goes out past the drain's refusal,
+  waits for the last of them rather than answering on their behalf. What ends the wait is the response
-  and only while the message is still held — past that it is an ordinary send, because the drain it
+  reaching the wire, not the call — a `sendResp()` the library refused, or one the socket would not
-  would slip past is no longer waiting for it. `shutdownTimeout: 0` does not carry over to this half:
+  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

README.md

@@ -332,12 +332,13 @@ 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` the application has not called `sendResp()` on, so a peer whose `submit_sm` is still
+every `sms` still in the application's hands, so a peer whose `submit_sm` is being handled is
-being handled is answered rather than left to re-send it — answering its PDUs through `sendReturn()`
+answered rather than left to re-send it. That wait ends when `sendResp()` puts the response on the
-instead leaves that wait running until it gives up. `sendDlr()` is the one send the refusal lets
+wire, or when every listener that took the message has failed; answering its PDUs through
-past, and it catches the wait when issued straight after `sendResp()`; await anything in between and
+`sendReturn()` instead leaves the wait running until it gives up. `sendDlr()` is the one send the
-it races the shutdown like any other send. `close({ signal })` takes an
+refusal lets past, and it catches the wait when issued straight after `sendResp()`; await anything in
-`AbortSignal` that cuts the wait short; `unbind()` takes none, and waits a further
+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
 `responseTimeout` for its own response. `send()` reaches any of the 33 SMPP commands the codec
 knows, not just the four the session handles natively:
 

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<object, () => void>();
 	private readonly held: HeldMessages;
 	private readonly log: SmppLog;
 	private readonly onRequest: OnRequest | undefined;
@@ -111,6 +113,13 @@ export class IncomingRequests {
 		this.reassembler.clear();
 	}
 
+	/** 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;
+
+		this.emitted.get(sms)?.();
+	}
+
 	/** Waits out the messages the application still holds, and says how many it never answered. */
 	async drain(timeout: number, signal: AbortSignal | undefined): Promise<VoidResult> {
 		const unanswered = await this.held.idle(timeout, signal);
@@ -175,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),
@@ -184,13 +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: release,
-			onAnswered: () => { setImmediate(() => { this.held.release(pduObjs); }); },
 			// 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, () => {
+			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);

src/session.ts

@@ -93,11 +93,13 @@ export class Session extends EventEmitter<SessionEvents> {
 		reason: unknown,
 		...args: [event: keyof SessionEvents, ...rest: unknown[]]
 	): void {
-		const [event] = args;
+		const [event, ...rest] = args;
 		const error = errorFrom(reason);
 
 		this.log.error('session - a listener rejected', { event, message: error.message });
 
+		if (event === 'sms') this.incoming.listenerRejected(rest[0]);
+
 		if (event !== 'sessionError') this.emit('sessionError', error);
 	}
 

src/sms.ts

@@ -77,8 +77,7 @@ export function createSms(input: SmsInput, handlers: SmsHandlers): Sms {
 		message: input.message,
 		pduObjs: input.pduObjs,
 		sendDlr: status => sendDlr(sms, handlers.send, status),
-		sendResp: options => sendResp(sms, answered, options ?? {}, handlers.lostLink)
+		sendResp: options => sendResp(sms, answered, options ?? {}, handlers),
-			.finally(handlers.onAnswered),
 		session: input.session,
 		get smsId(): string {
 			return answered.smsId;
@@ -94,7 +93,7 @@ async function sendResp(
 	sms: Sms,
 	answered: { smsId: string },
 	options: SendRespOptions,
-	lostLink: () => boolean,
+	handlers: Pick<SmsHandlers, 'lostLink' | 'onAnswered'>,
 ): Promise<VoidResult> {
 	const total = sms.pduObjs.length;
 
@@ -109,7 +108,7 @@ async function sendResp(
 	if (options.smsId !== undefined) answered.smsId = options.smsId;
 
 	// A response carries the sequence number it was asked on, which the next link knows nothing about.
-	if (lostLink()) {
+	if (handlers.lostLink()) {
 		return { err: new Error('The link this message arrived on is gone, so nothing would correlate the response') };
 	}
 
@@ -119,7 +118,11 @@ async function sendResp(
 		{ message_id: segmentId(answered.smsId, index, total) },
 	)));
 
-	return results.find(result => result.err) ?? {};
+	const failure = 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. */

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 => {
@@ -1401,6 +1428,72 @@ describe('graceful shutdown', () => {
 		assert.ok((await sent).err instanceof Error);
 	});
 
+	// emit() releases the hold of a listener that throws; one that rejects may cost no more than that.
+	test('a listener that rejected before answering does not hold the shutdown up', async t => {
+		const smpp = await startServer(t, { shutdownTimeout: 30_000 });
+		const failed = once<Error>(resolve => {
+			smpp.on('session', bound => {
+				bound.on('sessionError', resolve);
+				bound.on('sms', () => Promise.reject(new Error('the listener gave up')));
+			});
+		});
+		const { session } = await connect(t, smpp);
+
+		assert.ok(session);
+
+		const sent = session.sendSms({
+			from: '46701113311',
+			message: 'the listener rejects',
+			to: '46709771337',
+		});
+
+		assert.equal((await failed).message, 'the listener gave up');
+
+		const started = Date.now();
+
+		assert.deepEqual(await peerOf(smpp).close(), {});
+		assert.ok(Date.now() - started < 1000);
+		assert.ok((await sent).err instanceof Error);
+	});
+
+	test('waits for the listener still working when another one rejected', async t => {
+		const smpp = await startServer(t, { shutdownTimeout: 30_000 });
+		const failed = once<Error>(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 });
+		const refused = await sms.sendResp({ smsId: '' });
+		const closed = await peerOf(smpp).close();
+
+		assert.match(refused.err?.message ?? '', /smsId must not be empty/);
+		assert.ok(closed.err instanceof Error);
+		assert.match(closed.err.message, /1 message\(s\) unanswered/);
+	});
+
 	test('gives up on a request that outlasts shutdownTimeout', async t => {
 		const { sent, session } = await submitInFlight(t, { shutdownTimeout: 50 });
 		const closed = await session.close();

todo.md

@@ -61,6 +61,7 @@ Rules the API follows:
 | Held messages capped and expiring, so an application that answers nothing cannot grow them | `test/session-extras.test.ts` |
 | A send with no link held for the next one, and one the link dropped under counted as `unanswered` | `test/session-extras.test.ts` |
 | A message whose link dropped refused an answer, with its receipt still allowed out | `test/session-extras.test.ts` |
+| The hold released exactly when the peer was answered: a refused `sendResp()` keeps it, a listener that rejected drops it | `test/session-extras.test.ts` |
 | Every runnable README example | `test/readme.test.ts` |
 | Receipt-versus-message classification by `esm_class` | `test/dlr.test.ts`, `test/session.test.ts` |
 | A listener that throws, or rejects, reaching `sessionError`/`serverError` rather than the process | `test/session.test.ts`, `test/error-from.test.ts` |
@@ -117,10 +118,6 @@ session message is a change to every call site.
 
 ## Worth doing, not blocking
 
-- [ ] **Merge state does not survive a process restart.** A drop no longer discards it, but a restart
-      loses every incomplete group, and a peer has no reason to resend a receipt it already had
-      answered. Surviving one means exposing the merge state for the application to persist and hand
-      back, which is a public-surface decision.
 - [ ] **Group the session's collaborators under `src/session/`.** `session.ts` imports
       `dlr-merger`, `incoming-requests`, `link-timers`, `outgoing-requests`, `pdu-transport`,
       `reconnect-loop` and `send-sms`, and nothing else does, so the directory would make that
@@ -134,18 +131,6 @@ 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
-      until the message expires. 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 refused `sendResp()` releases the hold anyway.** Every early return runs
-      `.finally(onAnswered)`, so `sendResp({ smsId: '' })` refuses and stops the drain waiting for a
-      message the peer was never answered, which `close()` then reports as answered. Raised by
-      review, 2026-09-01.
-
 - [ ] **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
       non-final states with it hands the application a raw `id:… stat:ENROUTE` text as an inbound
@@ -179,6 +164,13 @@ session message is a change to every call site.
 
 ## Declined
 
+- **Merge state surviving a process restart.** Declined by AGENTS.md goal 7, maintainer's call,
+  2026-09-02. A restart loses every incomplete receipt group and a peer has no reason to resend one it
+  already had answered, so the loss is real — but surviving it means handing the application the merge
+  state to persist, which the scope floor covers as squarely as holding the state here would, and
+  which publishes the shape of `DlrMerger`'s groups against goal 6. Nothing is foreclosed: the seam
+  can still be added after 1.0.0 as a minor.
+
 - **Throughput throttling — a TPS cap, and backing off on `ESME_RTHROTTLED`.** Declined by AGENTS.md
   goal 7: an operator's rate limit is scoped to the account, while the widest thing this library owns
   is a session, so a bucket here cannot see a second process binding the same account and is wrong in