Keep a hold's release behind MessageHold, and require its listener count

src/held-messages.ts

@@ -25,26 +25,29 @@ type Held = {
 	pduObjs: PduObject[];
 };
 
+type HoldEntry = {
+	isHeld: () => boolean;
+	release: () => void;
+};
+
 /** One message offered to the application, held until it is answered or every listener gives up. */
 export class MessageHold {
-	private readonly held: HeldMessages;
-	private readonly pduObjs: PduObject[];
+	private readonly entry: HoldEntry;
 	private working: number;
 
-	constructor(held: HeldMessages, pduObjs: PduObject[], listeners: number) {
-		this.held = held;
-		this.pduObjs = pduObjs;
+	constructor(entry: HoldEntry, listeners: number) {
+		this.entry = entry;
 		this.working = listeners;
 	}
 
 	/** Whether a drain is still waiting for this message to be answered. */
 	isHeld(): boolean {
-		return this.held.has(this.pduObjs);
+		return this.entry.isHeld();
 	}
 
 	/** A turn later, so a listener sending its receipt straight after the response still holds. */
 	answered(): void {
-		setImmediate(() => { this.held.release(this.pduObjs); });
+		setImmediate(() => { this.entry.release(); });
 	}
 
 	/** A rejection leaves the other listeners running, so only the last one to fail gives the message up. */
@@ -54,9 +57,9 @@ export class MessageHold {
 		if (this.working <= 0) this.answered();
 	}
 
-	/** A message nobody took is not work a shutdown can wait for. */
-	untaken(): void {
-		this.held.release(this.pduObjs);
+	/** At once, for a message nobody took: that is not work a shutdown can wait for. */
+	release(): void {
+		this.entry.release();
 	}
 }
 
@@ -94,9 +97,11 @@ export class HeldMessages {
 		return this.held.full || this.octets >= this.maxOctets;
 	}
 
-	/** Holds a message about to be offered to `listeners` listeners. */
-	hold(pduObjs: PduObject[], listeners = 1): MessageHold {
-		const hold = new MessageHold(this, pduObjs, listeners);
+	hold(pduObjs: PduObject[], listeners: number): MessageHold {
+		const hold = new MessageHold({
+			isHeld: () => this.has(pduObjs),
+			release: () => { this.release(pduObjs); },
+		}, listeners);
 		const key = keyOf(pduObjs);
 
 		if (key === undefined) return hold;
@@ -118,14 +123,13 @@ export class HeldMessages {
 		return hold;
 	}
 
-	/** Whether a drain is still waiting for this message to be answered. */
-	has(pduObjs: PduObject[]): boolean {
+	private has(pduObjs: PduObject[]): boolean {
 		const key = keyOf(pduObjs);
 
 		return key !== undefined && this.held.get(key)?.pduObjs === pduObjs;
 	}
 
-	release(pduObjs: PduObject[]): void {
+	private release(pduObjs: PduObject[]): void {
 		const key = keyOf(pduObjs);
 
 		// Identity, not the key: a wrapped sequence number must not release someone else's message.

src/incoming-requests.ts

@@ -324,12 +324,11 @@ export class IncomingRequests {
 			bindAllows: cmdName => this.deps.bindAllows(cmdName),
 			lostLink: () => this.linkGeneration !== generation,
 			onAnswered: () => { hold.answered(); },
-			// Past the refusal only while a drain is still waiting for this message; an ordinary send after.
 			send: input => (hold.isHeld() ? this.deps.sendPastDrain(input) : this.deps.send(input)),
 		});
 
 		this.holds.set(sms, hold);
 
-		if (!this.deps.offerSms(sms)) hold.untaken();
+		if (!this.deps.offerSms(sms)) hold.release();
 	}
 }

test/session-extras.test.ts

@@ -1502,15 +1502,14 @@ describe('held message bounds', () => {
 
 	test('is full at its count, and a re-used sequence number replaces rather than adding', () => {
 		const held = new HeldMessages({ log: silentLog, max: 2, maxOctets: 1_000_000, timeout: 10_000 });
-		const first = message(1);
+		const first = held.hold(message(1), 1);
 
-		held.hold(first);
-		held.hold(message(2));
-		held.hold(message(2));
+		held.hold(message(2), 1);
+		held.hold(message(2), 1);
 
 		assert.equal(held.size, 2);
 		assert.equal(held.full(), true);
-		assert.equal(held.has(first), true);
+		assert.equal(first.isHeld(), true);
 
 		held.clear();
 	});
@@ -1519,23 +1518,22 @@ describe('held message bounds', () => {
 	test('is full at its octet cap, until a message leaves by any way out', () => {
 		let now = 0;
 		const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 2000, now: () => now, timeout: 10_000 });
-		const answered = message(1);
+		const answered = held.hold(message(1), 1);
 
-		held.hold(answered);
 		assert.equal(held.full(), false);
-		held.hold(message(2));
+		held.hold(message(2), 1);
 		assert.equal(held.full(), true);
 
-		held.release(answered);
+		answered.release();
 		assert.equal(held.full(), false, 'after a release');
-		held.hold(message(3));
+		held.hold(message(3), 1);
 
 		now = 20_000;
 		held.sweep();
 		now = 0;
 		assert.equal(held.full(), false, 'after a sweep');
-		held.hold(message(4));
-		held.hold(message(5));
+		held.hold(message(4), 1);
+		held.hold(message(5), 1);
 
 		held.clear();
 		assert.equal(held.full(), false, 'after a clear');
@@ -1624,11 +1622,11 @@ describe('held message bounds', () => {
 		let now = 0;
 		const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 1_000_000, now: () => now, timeout: 60 });
 
-		held.hold(message(1));
+		held.hold(message(1), 1);
 		now = 61;
 
 		// The next message sweeps the one that expired, so only the new one is still waited for.
-		held.hold(message(2));
+		held.hold(message(2), 1);
 
 		assert.equal(held.size, 1);
 
@@ -1640,7 +1638,7 @@ describe('held message bounds', () => {
 		let now = 0;
 		const held = new HeldMessages({ log: silentLog, max: 10, maxOctets: 1_000_000, now: () => now, timeout: 60 });
 
-		held.hold(message(1));
+		held.hold(message(1), 1);
 
 		const waiting = held.idle(1000, undefined);