Let a signal abort a send queued behind a full send window (#90)

* Regression tests for aborting a send queued behind a full window

* Let a signal abort a send queued behind a full send window

* Log a queued send, tighten its regression tests and settle the decision record

* Take the send window's queue off an array so an abort cannot stall the loop

---------

Co-authored-by: Mikael Göransson <mikael.goransson@timewave.se>

src/outgoing-requests.ts

@@ -48,7 +48,7 @@ export class OutgoingRequests {
 		this.pending = new PendingRequests(options.log);
 		this.responseTimeout = options.responseTimeout;
 		this.transport = options.transport;
-		this.window = new SendWindow(options.maxOutstanding);
+		this.window = new SendWindow({ limit: options.maxOutstanding, log: options.log });
 	}
 
 	/** Read through a method: a drop can land while a request is awaiting. */
@@ -115,7 +115,9 @@ export class OutgoingRequests {
 
 			if (held.err) return { err: held.err };
 
-			await this.window.acquire();
+			const slot = await this.window.acquire(options.signal);
+
+			if (slot.err) return { err: slot.err };
 
 			const attempt = await this.attempt(input, options).finally(() => { this.window.release(); });
 

src/send-window.ts

@@ -1,31 +1,50 @@
+import type { SmppLog } from './log.ts';
+import type { VoidResult } from './result.ts';
 import { IdleWaiters } from './idle-waiters.ts';
 
+export type SendWindowOptions = {
+	limit: number;
+	log: SmppLog;
+};
+
+type Waiter = (result: VoidResult) => void;
+
+function aborted(): Error {
+	return new Error('Aborted while waiting for a send window slot');
+}
+
 /** Caps how many requests are on the wire at once; anything past the limit waits its turn. */
 export class SendWindow {
 	private readonly idleWaiters = new IdleWaiters();
 	private readonly limit: number;
-	private readonly waiting: (() => void)[] = [];
+	private readonly log: SmppLog;
+	private readonly waiting = new Set<Waiter>();
 	private inFlight = 0;
 
-	constructor(limit: number) {
-		this.limit = limit;
+	constructor(options: SendWindowOptions) {
+		this.limit = options.limit;
+		this.log = options.log;
 	}
 
-	acquire(): Promise<void> {
+	/** Resolves once a slot is the caller's, or with the reason it stopped waiting for one. */
+	acquire(signal: AbortSignal | undefined): Promise<VoidResult> {
 		if (this.inFlight < this.limit) {
 			this.inFlight++;
 
-			return Promise.resolve();
+			return Promise.resolve({});
 		}
 
-		return new Promise<void>(resolve => this.waiting.push(resolve));
+		if (signal?.aborted === true) return Promise.resolve({ err: aborted() });
+
+		return this.queue(signal);
 	}
 
 	release(): void {
-		const next = this.waiting.shift();
+		const next = this.waiting.values().next().value;
 
 		if (next) {
-			next();
+			this.waiting.delete(next);
+			next({});
 
 			return;
 		}
@@ -39,11 +58,34 @@ export class SendWindow {
 
 	/** Everything the caller is still owed: on the wire, plus queued behind a full window. */
 	unfinished(): number {
-		return this.inFlight + this.waiting.length;
+		return this.inFlight + this.waiting.size;
 	}
 
 	/** Resolves 0 once nothing is left on the wire, or with what still is. */
 	idle(timeout: number, signal: AbortSignal | undefined): Promise<number> {
 		return this.idleWaiters.wait(() => this.unfinished(), timeout, signal);
 	}
+
+	/** A waiter leaves the queue as it settles, so release() can only hand a slot to one still in it. */
+	private queue(signal: AbortSignal | undefined): Promise<VoidResult> {
+		this.log.verbose('sendWindow - queueing a request behind a full window', {
+			limit: this.limit,
+			queued: this.waiting.size + 1,
+		});
+
+		return new Promise<VoidResult>(resolve => {
+			const settle = (result: VoidResult): void => {
+				this.waiting.delete(settle);
+				signal?.removeEventListener('abort', onAbort);
+				resolve(result);
+			};
+
+			function onAbort(): void {
+				settle({ err: aborted() });
+			}
+
+			signal?.addEventListener('abort', onAbort, { once: true });
+			this.waiting.add(settle);
+		});
+	}
 }

src/session-options.ts

@@ -136,7 +136,7 @@ export const defaults = {
 
 /**
  * A count below 1 does not fail loudly anywhere downstream: `maxOutstanding: 0` leaves every send
- * queued behind a slot that is never freed, so the call never settles at all.
+ * queued behind a slot that is never freed, so a send with no `signal` never settles at all.
  */
 export function checkSessionOptions(options: CheckableOptions): VoidResult {
 	if (options.fromStart !== undefined) {