test: cover abort signal cleanup paths

Author: ThomasK33

src/host/hostMain.ts

@@ -53,6 +53,7 @@ import {
   addAbortListener,
   makeAbortError,
   throwIfAborted,
+  waitForScopedOperation,
 } from '../util/abort.js';
 import { invariant } from '../util/assert.js';
 import { ResourceScope } from '../util/resourceScope.js';
@@ -106,69 +107,6 @@ function rethrowAsync(error: unknown): void {
   });
 }
 
-async function waitForScopedOperation<T>(params: {
-  readonly operationName: string;
-  readonly operation: Promise<T>;
-  readonly scope: ResourceScope;
-  readonly signal: AbortSignal;
-  readonly timeoutMs?: number;
-  readonly timeoutResult?: () => T;
-}): Promise<T> {
-  const { operationName, operation, scope, signal, timeoutMs, timeoutResult } =
-    params;
-  invariant(operationName.length > 0, 'operationName must not be empty');
-  if (signal.aborted) {
-    await scope.close();
-    throw makeAbortError(signal);
-  }
-
-  const { promise, reject, resolve } = Promise.withResolvers<T>();
-  let settled = false;
-
-  const rejectWithError = (error: unknown): void => {
-    if (settled) {
-      return;
-    }
-
-    settled = true;
-    void scope.close().then(() => {
-      reject(error instanceof Error ? error : new Error(String(error)));
-    }, reject);
-  };
-
-  const resolveWithValue = (value: T): void => {
-    if (settled) {
-      return;
-    }
-
-    settled = true;
-    void scope.close().then(() => {
-      resolve(value);
-    }, reject);
-  };
-
-  if (timeoutMs !== undefined) {
-    invariant(
-      timeoutResult !== undefined,
-      'timeoutResult must be provided when timeoutMs is set',
-    );
-    const timeoutHandle = setTimeout(() => {
-      resolveWithValue(timeoutResult());
-    }, timeoutMs);
-    scope.add(`${operationName} timeout`, () => {
-      clearTimeout(timeoutHandle);
-    });
-  }
-
-  addAbortListener(scope, `${operationName} abort listener`, signal, () => {
-    rejectWithError(makeAbortError(signal));
-  });
-
-  void operation.then(resolveWithValue, rejectWithError);
-
-  return await promise;
-}
-
 function resolveHostRendererName(input: string | undefined): RendererName {
   try {
     return resolveRendererName(

src/host/rpcClient.ts

@@ -15,6 +15,7 @@ import {
 } from '../protocol/messages.js';
 import {
   addAbortListener,
+  createResourceScopedSettlers,
   makeAbortError,
   throwIfAborted,
 } from '../util/abort.js';
@@ -128,19 +129,12 @@ export async function sendRpc(
     scope.add('rpc client socket', () => {
       socket.destroy();
     });
-    let settled = false;
+    const settlers = createResourceScopedSettlers(scope, resolve, reject);
     let responseHandled = false;
     let buffer = '';
 
     const rejectWithCliError = (error: CliError): void => {
-      if (settled) {
-        return;
-      }
-
-      settled = true;
-      void scope.close().then(() => {
-        reject(error);
-      }, reject);
+      settlers.reject(error);
     };
 
     const rejectWithTransportError = (error: unknown): void => {
@@ -150,26 +144,12 @@ export async function sendRpc(
     };
 
     const resolveWithResult = (result: unknown): void => {
-      if (settled) {
-        return;
-      }
-
-      settled = true;
-      void scope.close().then(() => {
-        resolve(result);
-      }, reject);
+      settlers.resolve(result);
     };
 
     if (signal !== undefined) {
       addAbortListener(scope, 'rpc client abort listener', signal, () => {
-        if (settled) {
-          return;
-        }
-
-        settled = true;
-        void scope.close().then(() => {
-          reject(makeAbortError(signal));
-        }, reject);
+        settlers.reject(makeAbortError(signal));
       });
     }
 
@@ -306,7 +286,7 @@ export async function sendRpc(
     });
 
     socket.on('end', () => {
-      if (settled || responseHandled) {
+      if (settlers.isSettled() || responseHandled) {
         return;
       }
 

src/host/rpcServer.ts

@@ -10,7 +10,7 @@ import {
   type RpcMethod,
   type RpcResponse,
 } from '../protocol/messages.js';
-import { makeAbortError } from '../util/abort.js';
+import { createResourceScopedSettlers, makeAbortError } from '../util/abort.js';
 import { invariant } from '../util/assert.js';
 import { ResourceScope } from '../util/resourceScope.js';
 
@@ -58,51 +58,31 @@ async function probeSocketLiveness(socketPath: string): Promise<boolean> {
   return await new Promise<boolean>((resolve, reject) => {
     const scope = new ResourceScope();
     const probe = net.connect({ path: socketPath });
-    let settled = false;
-
-    const resolveWithValue = (value: boolean): void => {
-      if (settled) {
-        return;
-      }
-
-      settled = true;
-      void scope.close().then(() => {
-        resolve(value);
-      }, reject);
-    };
-
-    const rejectWithError = (error: unknown): void => {
-      if (settled) {
-        return;
-      }
-
-      settled = true;
-      void scope.close().then(() => {
-        reject(error instanceof Error ? error : new Error(String(error)));
-      }, reject);
-    };
+    const settlers = createResourceScopedSettlers(scope, resolve, reject);
 
     scope.add('rpc liveness probe socket', () => {
       probe.destroy();
     });
     const timeoutHandle = setTimeout(() => {
-      rejectWithError(new Error(`Timed out probing RPC socket: ${socketPath}`));
+      // If connect neither succeeds nor fails promptly, treat the socket path as
+      // stale rather than blocking host startup indefinitely.
+      settlers.resolve(false);
     }, SOCKET_LIVENESS_PROBE_TIMEOUT_MS);
     scope.add('rpc liveness probe timeout', () => {
       clearTimeout(timeoutHandle);
     });
 
     probe.once('connect', () => {
-      resolveWithValue(true);
+      settlers.resolve(true);
     });
 
     probe.once('error', (error: NodeJS.ErrnoException) => {
       if (error.code === 'ECONNREFUSED' || error.code === 'ENOENT') {
-        resolveWithValue(false);
+        settlers.resolve(false);
         return;
       }
 
-      rejectWithError(error);
+      settlers.reject(error);
     });
   });
 }

src/host/runCompletionCoordinator.ts

@@ -8,7 +8,7 @@ import {
   RunCompletionSentinelScanner,
   type SentinelPiece,
 } from './runCompletionSentinel.js';
-import { addAbortListener, makeAbortError } from '../util/abort.js';
+import { waitForScopedOperation } from '../util/abort.js';
 import { invariant } from '../util/assert.js';
 import { ResourceScope } from '../util/resourceScope.js';
 
@@ -280,62 +280,25 @@ export class RunCompletionCoordinator {
       'timeoutMs must be a positive integer',
     );
 
-    const { signal } = options;
-    if (signal?.aborted === true) {
+    const forgetWaiter = (): void => {
+      // Match timeout behavior: stop waiting for a client response but keep
+      // sentinel/postamble registrations active so eventual completion bytes
+      // remain hidden and replayable.
       this.#runCompletionWaiters.delete(marker);
-      throw makeAbortError(signal);
-    }
-
-    const scope = new ResourceScope();
-    const { promise, reject, resolve } =
-      Promise.withResolvers<TimedRunCompletionWaitResult>();
-    let resolved = false;
-
-    const rejectWithError = (error: unknown): void => {
-      if (resolved) {
-        return;
-      }
-
-      resolved = true;
-      void scope.close().then(() => {
-        reject(error instanceof Error ? error : new Error(String(error)));
-      }, reject);
-    };
-
-    const resolveWithResult = (result: TimedRunCompletionWaitResult): void => {
-      if (resolved) {
-        return;
-      }
-
-      resolved = true;
-      void scope.close().then(() => {
-        resolve(result);
-      }, reject);
     };
 
-    const timeoutHandle = setTimeout(() => {
-      // Keep sentinel/postamble registrations active after timeout so the
-      // eventual internal completion bytes are still hidden from artifacts.
-      this.#runCompletionWaiters.delete(marker);
-      resolveWithResult({ kind: 'timeout' });
-    }, timeoutMs);
-    scope.add('run completion timeout', () => {
-      clearTimeout(timeoutHandle);
+    return await waitForScopedOperation<TimedRunCompletionWaitResult>({
+      operationName: 'run completion',
+      operation: completionPromise,
+      scope: new ResourceScope(),
+      ...(options.signal === undefined ? {} : { signal: options.signal }),
+      timeoutMs,
+      timeoutResult: () => {
+        forgetWaiter();
+        return { kind: 'timeout' };
+      },
+      onAbort: forgetWaiter,
     });
-
-    if (signal !== undefined) {
-      addAbortListener(scope, 'run completion abort listener', signal, () => {
-        // Match timeout behavior: stop waiting for a client response but keep
-        // sentinel/postamble registrations active so eventual completion bytes
-        // remain hidden and replayable.
-        this.#runCompletionWaiters.delete(marker);
-        rejectWithError(makeAbortError(signal));
-      });
-    }
-
-    void completionPromise.then(resolveWithResult, rejectWithError);
-
-    return await promise;
   }
 
   async #appendOutput(data: string): Promise<void> {