Fix early Issues disconnect handling (#135)

Author: mchwang

test/issue-board.test.ts

@@ -171,6 +171,53 @@ describe('issue endpoints', { timeout: 30_000 }, () => {
     } finally { spy.mockRestore(); await app.close(); }
   });
 
+  it('stops waiting when the response closed before the Issues listener was attached', async () => {
+    root = mkdtempSync(join(tmpdir(), 'codeboost-issues-'));
+    const { gateway, calls } = heldGateway();
+    const waits: Promise<unknown>[] = [];
+    let closed = false;
+    let closedBeforeRefresh = false;
+    let disconnectNext = true;
+    const resumeHandler = deferred<void>();
+    const original = IssueBoard.prototype.refresh;
+    const spy = vi.spyOn(IssueBoard.prototype, 'refresh').mockImplementation(function (this: IssueBoard, signal) {
+      closedBeforeRefresh = closed;
+      const wait = original.call(this, signal);
+      waits.push(wait.then(() => 'settled', error => String(error)));
+      return wait;
+    });
+    const app = await startServer(createDemo(join(root, 'demo')), 0, undefined, undefined, undefined, gateway,
+      undefined, undefined, undefined, { beforeIssueRefreshWait: () => resumeHandler.promise });
+    app.server.prependListener('request', (req, res) => {
+      if (req.url !== '/api/issues' || !disconnectNext) return;
+      disconnectNext = false;
+      req.once('end', () => {
+        res.once('close', () => { closed = true; resumeHandler.resolve(); });
+        res.destroy();
+      });
+    });
+    try {
+      const leaving = call(app.url, app.token, 'POST', { action: 'refresh' }).catch(() => undefined);
+      await expect.poll(() => waits.length).toBe(1);
+      expect(closedBeforeRefresh).toBe(true);
+      const staying = call(app.url, app.token, 'POST', { action: 'refresh' });
+      await expect.poll(() => waits.length).toBe(2);
+      let settled = false;
+      void waits[0]!.then(() => { settled = true; });
+      await expect.poll(() => settled, { timeout: 500 }).toBe(true);
+      expect(await waits[0]).toContain('Client disconnected');
+      expect(calls[0]!.signal!.aborted).toBe(false);
+      calls[0]!.result.resolve(snapshot());
+      await expect(staying).resolves.toMatchObject({ status: 200, body: { state: { state: 'fresh' } } });
+      expect(calls).toHaveLength(1);
+      await leaving;
+    } finally {
+      calls[0]?.result.resolve(snapshot());
+      spy.mockRestore();
+      await app.close();
+    }
+  });
+
   it('shutdown aborts an admitted refresh and waits for the retrieval to settle', async () => {
     root = mkdtempSync(join(tmpdir(), 'codeboost-issues-'));
     const { gateway, calls } = heldGateway();

web/server.ts

@@ -50,7 +50,11 @@ export type RunnerSetup = (service: ReviewService, capability: ShutdownCapabilit
 export const RUNNER_NOT_CONFIGURED = 'The runner is not configured. Add a runner block to the review configuration and restart codeboost.';
 /** Demos never run the runner, whatever their configuration says. */
 export const RUNNER_NOT_IN_DEMO = 'Demos do not run the runner. Use a review configuration with a runner block.';
-export async function startServer(config: ReviewConfig, port = 4318, questionAgent?: QuestionAgent, mergeGateway?: MergeGateway, shutdownDrainMs = MAX_SHUTDOWN_DRAIN_MS, issueGateway?: IssueGateway, runnerDeps?: RunnerDeps, planningInput?: PlanningDeps | PlanningSetup, runnerSetup?: RunnerSetup) {
+export interface ServerTestHooks {
+  /** Test-only gate for forcing a response lifecycle transition before the Issues wait is armed. */
+  beforeIssueRefreshWait?(): Promise<void> | void;
+}
+export async function startServer(config: ReviewConfig, port = 4318, questionAgent?: QuestionAgent, mergeGateway?: MergeGateway, shutdownDrainMs = MAX_SHUTDOWN_DRAIN_MS, issueGateway?: IssueGateway, runnerDeps?: RunnerDeps, planningInput?: PlanningDeps | PlanningSetup, runnerSetup?: RunnerSetup, testHooks: ServerTestHooks = {}) {
   if (!Number.isSafeInteger(shutdownDrainMs) || shutdownDrainMs < 1 || shutdownDrainMs > MAX_SHUTDOWN_DRAIN_MS) throw new Error('Invalid shutdown drain deadline.');
   const service = new ReviewService(config), token = randomBytes(32).toString('hex');
   let questions: Questions, merges: MergeCoordinator | null, issues: IssueBoard, runner: RunnerCoordinator | null, suggestions: SuggestionCoordinator | null;
@@ -488,10 +492,13 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge
         if (stopping && input.action === 'merge') { json(503, { error: 'The review server is shutting down.' }); return; }
         if(path==='/api/issues') {
           if(input?.action!=='refresh')throw new Error('Unsupported issue action.');
+          await testHooks.beforeIssueRefreshWait?.();
           // A departing browser stops waiting; the board keeps the shared refresh for other callers.
           const departed=new AbortController();
           const depart=()=>{if(!res.writableEnded)departed.abort(new Error('Client disconnected.'));};
           res.once('close',depart);
+          // `close` may have fired while the request body was being read, before this listener existed.
+          if(res.closed||res.destroyed)depart();
           try { json(200,await issues.refresh(AbortSignal.any([requestAbort.signal,departed.signal]))); }
           finally { res.removeListener('close',depart); }
           return;