fix(backend): cancel demo-run polling when client disconnects (#531)

backend/src/lib/waitForActivityTxHash.js

@@ -7,19 +7,48 @@
  * @param {{ maxWaitMs: number, initialDelayMs: number, maxDelayMs: number }} options
  * @param {(entry: { txHash?: string }) => boolean} [matchesEntry]
  * @param {(ms: number) => Promise<void>} [sleep]
+ * @param {AbortSignal} [signal] — when aborted, the loop breaks and returns ''.
  * @returns {Promise<string>}
  */
+function sleepWithAbort(ms, signal) {
+  return new Promise((resolve) => {
+    let timer;
+
+    const onAbort = () => {
+      clearTimeout(timer);
+      signal.removeEventListener('abort', onAbort);
+      resolve();
+    };
+
+    timer = setTimeout(() => {
+      signal?.removeEventListener('abort', onAbort);
+      resolve();
+    }, ms);
+
+    if (signal) {
+      if (signal.aborted) {
+        onAbort();
+      } else {
+        signal.addEventListener('abort', onAbort, { once: true });
+      }
+    }
+  });
+}
+
 export async function waitForActivityTxHash(
   getFeed,
   activityCountBefore,
   { maxWaitMs, initialDelayMs, maxDelayMs },
   matchesEntry,
-  sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)),
+  sleep = (ms, signal) => sleepWithAbort(ms, signal),
+  signal,
 ) {
   let elapsedMs = 0;
   let currentDelay = initialDelayMs;
 
   while (true) {
+    // Early exit when the client disconnected mid-request (see demo.js).
+    if (signal?.aborted) break;
     const feed = getFeed();
     const addedCount = Math.max(feed.length - activityCountBefore, 0);
     if (addedCount > 0) {
@@ -39,7 +68,8 @@ export async function waitForActivityTxHash(
       break;
     }
 
-    await sleep(delay);
+    await sleep(delay, signal);
+    if (signal?.aborted) break;
     elapsedMs += delay;
     currentDelay = Math.min(currentDelay * 2, maxDelayMs);
   }

backend/src/lib/waitForActivityTxHash.test.js

@@ -97,6 +97,34 @@ describe('waitForActivityTxHash', () => {
     expect(delays.length).toBeGreaterThan(0);
   });
 
+  it('aborts an in-flight default sleep and cleans up its timer', async () => {
+    vi.useFakeTimers();
+    try {
+      const controller = new AbortController();
+      const getFeed = vi.fn(() => []);
+      const resultPromise = waitForActivityTxHash(
+        getFeed,
+        0,
+        { maxWaitMs: 1000, initialDelayMs: 100, maxDelayMs: 100 },
+        undefined,
+        undefined,
+        controller.signal,
+      );
+
+      // The poll sleep is scheduled: exactly one timer is active.
+      expect(vi.getTimerCount()).toBe(1);
+
+      controller.abort();
+
+      // The abort wakes the sleep: the promise resolves and the timer is gone.
+      await expect(resultPromise).resolves.toBe('');
+      expect(vi.getTimerCount()).toBe(0);
+      expect(getFeed).toHaveBeenCalledTimes(1);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it('ignores unrelated new entries when a matcher is provided', async () => {
     const { sleep, delays } = makeSleepRecorder();
     const myId = 'request-a';

backend/src/routes/demo.js

@@ -46,6 +46,17 @@ function buildHttpClient() {
 }
 
 router.post('/demo-run', async (req, res) => {
+  // NOTE: use res.on('close'), not req.on('close'). In Node >=22, req 'close'
+  // fires once the request body is fully consumed — not on client disconnect —
+  // so keeping it registered would abort every normal request (HTTP 499).
+  // res 'close' + !writableEnded distinguishes a real mid-request disconnect
+  // from normal completion. Verified on Node 22.23.1 (CI) and 24.15.0.
+  const abortController = new AbortController();
+  const onClose = () => {
+    if (!res.writableEnded) abortController.abort();
+  };
+  res.on('close', onClose);
+
   try {
     const { serviceId, category } = req.body;
 
@@ -83,55 +94,58 @@ router.post('/demo-run', async (req, res) => {
     const httpClient = buildHttpClient();
     const activityCountBefore = getActivityFeed().length;
 
-    const abortController = new AbortController();
-    const onClose = () => abortController.abort();
-    req.on('close', onClose);
-
-    const { response, txHash: fetchedTxHash } = await httpClient.fetchWithTx(finalEndpointUrl, { signal: abortController.signal });
-    req.removeListener('close', onClose);
-
-    if (!response.ok) {
-      throw new Error(`Service responded with ${response.status}`);
-    }
-
-    const data = await response.json();
-
-    // Evaluate data quality: the response must be a non-null object (or a
-    // non-empty array) and must not carry a top-level `error` field.
-    const dataValid =
-      data !== null &&
-      typeof data === 'object' &&
-      !('error' in data) &&
-      (Array.isArray(data) ? data.length > 0 : Object.keys(data).length > 0);
-
-    if (!dataValid) {
-      logger.warn({ serviceId, category }, 'Demo run returned empty or error payload — marking data invalid');
-    }
-
-    const txHash = fetchedTxHash || (await waitForActivityTxHash(
-      getActivityFeed,
-      activityCountBefore,
-      {
-        maxWaitMs: config.demoRun.pollMaxWaitMs,
-        initialDelayMs: config.demoRun.pollInitialDelayMs,
-        maxDelayMs: config.demoRun.pollMaxDelayMs,
-      },
-      (entry) => entry.demoRunId === demoRunId,
-    ));
-    if (!txHash) {
-      logger.warn({ serviceId, category, maxWaitMs: config.demoRun.pollMaxWaitMs }, 'Activity txHash not found before poll timeout');
+    try {
+      const { response, txHash: fetchedTxHash } = await httpClient.fetchWithTx(finalEndpointUrl, { signal: abortController.signal });
+
+      if (!response.ok) {
+        throw new Error(`Service responded with ${response.status}`);
+      }
+
+      const data = await response.json();
+
+      // Evaluate data quality: the response must be a non-null object (or a
+      // non-empty array) and must not carry a top-level `error` field.
+      const dataValid =
+        data !== null &&
+        typeof data === 'object' &&
+        !('error' in data) &&
+        (Array.isArray(data) ? data.length > 0 : Object.keys(data).length > 0);
+
+      if (!dataValid) {
+        logger.warn({ serviceId, category }, 'Demo run returned empty or error payload — marking data invalid');
+      }
+
+      const txHash = fetchedTxHash || (await waitForActivityTxHash(
+        getActivityFeed,
+        activityCountBefore,
+        {
+          maxWaitMs: config.demoRun.pollMaxWaitMs,
+          initialDelayMs: config.demoRun.pollInitialDelayMs,
+          maxDelayMs: config.demoRun.pollMaxDelayMs,
+        },
+        (entry) => entry.demoRunId === demoRunId,
+        undefined,
+        abortController.signal,
+      ));
+      if (!txHash) {
+        logger.warn({ serviceId, category, maxWaitMs: config.demoRun.pollMaxWaitMs }, 'Activity txHash not found before poll timeout');
+      }
+
+      recordActivity({
+        timestamp: new Date().toISOString(),
+        agent: config.server.address,
+        service: service.name,
+        amount: service.price_usdc,
+        txHash,
+      });
+
+      logger.info({ serviceId, category, txHash, dataValid }, 'Demo run complete');
+      if (!abortController.signal.aborted && !res.writableEnded) {
+        res.json({ data, txHash, dataValid });
+      }
+    } finally {
+      res.removeListener('close', onClose);
     }
-
-    recordActivity({
-      timestamp: new Date().toISOString(),
-      agent: config.server.address,
-      service: service.name,
-      amount: service.price_usdc,
-      txHash,
-    });
-
-    logger.info({ serviceId, category, txHash, dataValid }, 'Demo run complete');
-    res.json({ data, txHash, dataValid });
   } catch (err) {
     if (err.name === 'AbortError') {
       logger.info({ serviceId: req.body?.serviceId, category: req.body?.category }, 'Demo run aborted by client');