From 1cb988a41ae845d049c783a6c794db214df6bae0 Mon Sep 17 00:00:00 2001 From: jorge rios Date: Sun, 2 Aug 2026 23:01:33 -0300 Subject: [PATCH 1/5] fix(backend): cancel demo-run polling when client disconnects Move the disconnect listener to res.on('close') guarded by !res.writableEnded so it stays active for the whole handler, and pass the AbortSignal into waitForActivityTxHash() so the poll loop breaks early when the client disconnects mid-request. req.on('close') fires when the request body is fully consumed, not on disconnect (verified on Node 22/24), so keeping it registered would abort every normal request. --- backend/src/lib/waitForActivityTxHash.js | 3 + backend/src/routes/demo.js | 103 ++++++++++++----------- 2 files changed, 58 insertions(+), 48 deletions(-) diff --git a/backend/src/lib/waitForActivityTxHash.js b/backend/src/lib/waitForActivityTxHash.js index d219673a..cf18e032 100644 --- a/backend/src/lib/waitForActivityTxHash.js +++ b/backend/src/lib/waitForActivityTxHash.js @@ -7,6 +7,7 @@ * @param {{ maxWaitMs: number, initialDelayMs: number, maxDelayMs: number }} options * @param {(entry: { txHash?: string }) => boolean} [matchesEntry] * @param {(ms: number) => Promise} [sleep] + * @param {AbortSignal} [signal] * @returns {Promise} */ export async function waitForActivityTxHash( @@ -15,11 +16,13 @@ export async function waitForActivityTxHash( { maxWaitMs, initialDelayMs, maxDelayMs }, matchesEntry, sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)), + signal, ) { let elapsedMs = 0; let currentDelay = initialDelayMs; while (true) { + if (signal?.aborted) break; const feed = getFeed(); const addedCount = Math.max(feed.length - activityCountBefore, 0); if (addedCount > 0) { diff --git a/backend/src/routes/demo.js b/backend/src/routes/demo.js index 58114fc4..724d9ded 100644 --- a/backend/src/routes/demo.js +++ b/backend/src/routes/demo.js @@ -46,6 +46,12 @@ function buildHttpClient() { } router.post('/demo-run', async (req, res) => { + const abortController = new AbortController(); + const onClose = () => { + if (!res.writableEnded) abortController.abort(); + }; + res.on('close', onClose); + try { const { serviceId, category } = req.body; @@ -83,55 +89,56 @@ 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'); + 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'); From be5462ee9422d14733af6c5b48a13e065093079d Mon Sep 17 00:00:00 2001 From: jorge rios Date: Sun, 2 Aug 2026 23:15:04 -0300 Subject: [PATCH 2/5] docs(backend): explain why res.on('close') is used for disconnect detection Adds code comments documenting the empirical finding: req.on('close') fires when the request body is fully consumed (not on client disconnect) in Node >=22, so keeping it registered would abort every normal request. res.on('close') + !writableEnded distinguishes real disconnects from normal completion. Verified on Node 22.23.1 (CI) and 24.15.0. --- backend/src/lib/waitForActivityTxHash.js | 3 ++- backend/src/routes/demo.js | 5 +++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/backend/src/lib/waitForActivityTxHash.js b/backend/src/lib/waitForActivityTxHash.js index cf18e032..ce1d57ac 100644 --- a/backend/src/lib/waitForActivityTxHash.js +++ b/backend/src/lib/waitForActivityTxHash.js @@ -7,7 +7,7 @@ * @param {{ maxWaitMs: number, initialDelayMs: number, maxDelayMs: number }} options * @param {(entry: { txHash?: string }) => boolean} [matchesEntry] * @param {(ms: number) => Promise} [sleep] - * @param {AbortSignal} [signal] + * @param {AbortSignal} [signal] — when aborted, the loop breaks and returns ''. * @returns {Promise} */ export async function waitForActivityTxHash( @@ -22,6 +22,7 @@ export async function waitForActivityTxHash( 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); diff --git a/backend/src/routes/demo.js b/backend/src/routes/demo.js index 724d9ded..776c7260 100644 --- a/backend/src/routes/demo.js +++ b/backend/src/routes/demo.js @@ -46,6 +46,11 @@ 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(); From e9a33cf16145fa1b56e83c15274703cab22bc8c0 Mon Sep 17 00:00:00 2001 From: jorge rios Date: Sun, 2 Aug 2026 23:40:01 -0300 Subject: [PATCH 3/5] test(backend): add separate integration test for disconnect during polling New standalone file (backend/test/demo-disconnect.test.js) that covers issue #531's real scenario: client disconnect DURING the polling phase. The existing suite cannot see this bug because waitForActivityTxHash is mocked to resolve instantly, so the handler completes before the 'close' event fires. This suite runs the real poll with a real HTTP socket: - unit: abort signal in waitForActivityTxHash (already-aborted, mid-wait) - integration: normal request completes the full poll; disconnect mid-request cancels the poll early (no full 8s budget waste) Lives separately so the maintainer can adopt or discard it freely. --- backend/test/demo-disconnect.test.js | 212 +++++++++++++++++++++++++++ 1 file changed, 212 insertions(+) create mode 100644 backend/test/demo-disconnect.test.js diff --git a/backend/test/demo-disconnect.test.js b/backend/test/demo-disconnect.test.js new file mode 100644 index 00000000..793fd918 --- /dev/null +++ b/backend/test/demo-disconnect.test.js @@ -0,0 +1,212 @@ +/** + * Issue #531 — client disconnect during /api/demo-run polling. + * + * This test lives OUTSIDE the repo's existing suites on purpose: it covers + * the scenario that the current tests structurally cannot see (disconnect + * DURING the real polling phase). The existing suite mocks + * waitForActivityTxHash to resolve instantly, so the handler completes + * before the 'close' event fires and the abort never matters. + * + * Here we run the REAL waitForActivityTxHash (not mocked) and destroy the + * client socket mid-poll, exactly like a real browser tab being closed. + * + * Expected behavior with the fix: + * - normal request -> 200, poll runs (not aborted) + * - disconnect mid-poll -> poll cancels early (no full budget waste) + */ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import express from 'express'; +import http from 'node:http'; +import demoRouter from '../src/routes/demo.js'; +import * as contract from '../src/lib/contract.js'; + +vi.mock('../src/lib/contract.js', () => ({ + getService: vi.fn(), +})); + +vi.mock('../src/routes/demoValidate.js', () => ({ + validateDemoEndpoint: vi.fn().mockReturnValue('http://localhost:9999/demo'), +})); + +vi.mock('../src/routes/services.js', () => ({ + recordActivity: vi.fn(), + getActivityFeed: vi.fn(() => []), +})); + +vi.mock('@x402/core/client', () => ({ + x402Client: vi.fn().mockImplementation(() => ({ + register: vi.fn().mockReturnThis(), + })), + x402HTTPClient: vi.fn().mockImplementation(() => ({ + fetchWithTx: vi.fn().mockResolvedValue({ + response: { + ok: true, + status: 200, + json: async () => ({ weather: 'sunny' }), + headers: { get: () => null }, + }, + txHash: '', + }), + })), +})); + +vi.mock('@x402/stellar', () => ({ + createEd25519Signer: vi.fn(), +})); + +vi.mock('@x402/stellar/exact/client', () => ({ + ExactStellarScheme: vi.fn(), +})); + +import * as services from '../src/routes/services.js'; + +// buildHttpClient() overrides fetchWithTx with the real implementation, which +// calls global fetch(). Stub fetch so the route's "payment" succeeds without +// touching the network: a 200 response means "no payment required" and the +// flow proceeds straight into the polling phase. +const fetchMock = vi.fn(async () => + new Response(JSON.stringify({ weather: 'sunny' }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }), +); +vi.stubGlobal('fetch', fetchMock); + +// ── helpers ───────────────────────────────────────────────────────────────── + +function makeSleepRecorder() { + const delays = []; + const sleep = vi.fn(async (ms) => { + delays.push(ms); + }); + return { sleep, delays }; +} + +const defaultOptions = { maxWaitMs: 8000, initialDelayMs: 250, maxDelayMs: 2000 }; + +async function startServer(app) { + const server = http.createServer(app); + await new Promise((resolve) => server.listen(0, resolve)); + return server; +} + +function sendDemoRun(port, onResponse) { + const req = http.request( + { + hostname: '127.0.0.1', + port, + path: '/api/demo-run', + method: 'POST', + headers: { 'content-type': 'application/json' }, + }, + (res) => { + let data = ''; + res.on('data', (c) => (data += c)); + res.on('end', () => onResponse({ status: res.statusCode, body: data })); + }, + ); + req.on('error', () => {}); // ECONNRESET on destroy is expected + req.end(JSON.stringify({ serviceId: 1, category: 'weather' })); + return req; +} + +// ── unit tests: abort signal in waitForActivityTxHash ──────────────────────── + +import { waitForActivityTxHash } from '../src/lib/waitForActivityTxHash.js'; + +describe('waitForActivityTxHash — abort signal (issue #531)', () => { + it('breaks immediately when the signal is already aborted', async () => { + const { sleep, delays } = makeSleepRecorder(); + const controller = new AbortController(); + controller.abort(); + const getFeed = vi.fn(() => [{ txHash: 'abc123' }]); + + const result = await waitForActivityTxHash( + getFeed, + 0, + defaultOptions, + undefined, + sleep, + controller.signal, + ); + + expect(result).toBe(''); + expect(getFeed).not.toHaveBeenCalled(); + expect(delays).toEqual([]); + }); + + it('stops polling once the signal aborts mid-wait', async () => { + const { sleep, delays } = makeSleepRecorder(); + const controller = new AbortController(); + const getFeed = vi.fn(() => []); + + // Simulate a disconnect arriving during the first sleep delay. + sleep.mockImplementationOnce(async (ms) => { + delays.push(ms); + controller.abort(); + }); + + const result = await waitForActivityTxHash( + getFeed, + 0, + defaultOptions, + undefined, + sleep, + controller.signal, + ); + + expect(result).toBe(''); + expect(getFeed).toHaveBeenCalledTimes(1); + expect(delays.length).toBe(1); // one delay, then the loop stops + }); +}); + +// ── integration tests: real HTTP socket, real polling ─────────────────────── + +describe('POST /api/demo-run — client disconnect during real polling (issue #531)', () => { + beforeEach(() => { + vi.clearAllMocks(); + services.getActivityFeed.mockReturnValue([]); + services.recordActivity.mockClear(); + }); + + it('completes a normal request without aborting the poll', async () => { + contract.getService.mockResolvedValue({ name: 'Test Service', endpoint: 'test', price_usdc: '1' }); + const app = express(); + app.use(express.json()); + app.use('/api', demoRouter); + const server = await startServer(app); + + const started = Date.now(); + const result = await new Promise((resolve) => { + sendDemoRun(server.address().port, resolve); + }); + const elapsed = Date.now() - started; + + server.close(); + expect(result.status).toBe(200); + expect(elapsed).toBeGreaterThan(200); // the poll actually ran + }, 15_000); // real polling runs the full 8s budget by design + + it('cancels the poll when the client disconnects mid-request', async () => { + contract.getService.mockResolvedValue({ name: 'Test Service', endpoint: 'test', price_usdc: '1' }); + const app = express(); + app.use(express.json()); + app.use('/api', demoRouter); + const server = await startServer(app); + + const started = Date.now(); + // Real client that pays and then closes the tab ~200ms in. + await new Promise((resolve) => { + const req = sendDemoRun(server.address().port, () => resolve()); + setTimeout(() => req.destroy(), 200); + setTimeout(resolve, 400); // give the handler time to react + }); + const elapsed = Date.now() - started; + + server.close(); + // The poll budget is 8000ms; after a disconnect it must stop well before. + expect(elapsed).toBeLessThan(2000); + expect(services.recordActivity).toHaveBeenCalled(); + }); +}); From b7e9fbf0352e4b508d7dadf5fbb0d9630a48c554 Mon Sep 17 00:00:00 2001 From: jorge rios Date: Mon, 3 Aug 2026 09:43:08 -0300 Subject: [PATCH 4/5] fix(backend): abort in-flight polling sleep when client disconnects The loop's sleep now honors the AbortSignal (timers cleaned up), breaks on abort, and demo-run no longer responds once the client has gone. --- backend/src/lib/waitForActivityTxHash.js | 30 +++++++++++++++++-- backend/src/lib/waitForActivityTxHash.test.js | 19 ++++++++++++ backend/src/routes/demo.js | 4 ++- 3 files changed, 50 insertions(+), 3 deletions(-) diff --git a/backend/src/lib/waitForActivityTxHash.js b/backend/src/lib/waitForActivityTxHash.js index ce1d57ac..229a795b 100644 --- a/backend/src/lib/waitForActivityTxHash.js +++ b/backend/src/lib/waitForActivityTxHash.js @@ -10,12 +10,37 @@ * @param {AbortSignal} [signal] — when aborted, the loop breaks and returns ''. * @returns {Promise} */ +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; @@ -43,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); } diff --git a/backend/src/lib/waitForActivityTxHash.test.js b/backend/src/lib/waitForActivityTxHash.test.js index 6617522b..16baf27f 100644 --- a/backend/src/lib/waitForActivityTxHash.test.js +++ b/backend/src/lib/waitForActivityTxHash.test.js @@ -97,6 +97,25 @@ describe('waitForActivityTxHash', () => { expect(delays.length).toBeGreaterThan(0); }); + it('aborts an in-flight default sleep and cleans up its timer', async () => { + const controller = new AbortController(); + const getFeed = vi.fn(() => []); + const resultPromise = waitForActivityTxHash( + getFeed, + 0, + { maxWaitMs: 1000, initialDelayMs: 100, maxDelayMs: 100 }, + undefined, + undefined, + controller.signal, + ); + + await new Promise((resolve) => setTimeout(resolve, 10)); + controller.abort(); + + await expect(resultPromise).resolves.toBe(''); + expect(getFeed).toHaveBeenCalledTimes(1); + }); + it('ignores unrelated new entries when a matcher is provided', async () => { const { sleep, delays } = makeSleepRecorder(); const myId = 'request-a'; diff --git a/backend/src/routes/demo.js b/backend/src/routes/demo.js index 776c7260..389c8043 100644 --- a/backend/src/routes/demo.js +++ b/backend/src/routes/demo.js @@ -140,7 +140,9 @@ router.post('/demo-run', async (req, res) => { }); logger.info({ serviceId, category, txHash, dataValid }, 'Demo run complete'); - res.json({ data, txHash, dataValid }); + if (!abortController.signal.aborted && !res.writableEnded) { + res.json({ data, txHash, dataValid }); + } } finally { res.removeListener('close', onClose); } From feae59cc82a6513628550718929dace19d47221e Mon Sep 17 00:00:00 2001 From: jorge rios Date: Mon, 3 Aug 2026 10:53:35 -0300 Subject: [PATCH 5/5] test(backend): prove abort wakes polling sleep with fake timers Use fake timers to assert one timer is active before the abort and zero timers after it resolves, so a regression to a non-abortable sleep fails instead of passing by waiting longer. --- backend/src/lib/waitForActivityTxHash.test.js | 41 +++++++++++-------- 1 file changed, 25 insertions(+), 16 deletions(-) diff --git a/backend/src/lib/waitForActivityTxHash.test.js b/backend/src/lib/waitForActivityTxHash.test.js index 16baf27f..1f0b6468 100644 --- a/backend/src/lib/waitForActivityTxHash.test.js +++ b/backend/src/lib/waitForActivityTxHash.test.js @@ -98,22 +98,31 @@ describe('waitForActivityTxHash', () => { }); it('aborts an in-flight default sleep and cleans up its timer', async () => { - const controller = new AbortController(); - const getFeed = vi.fn(() => []); - const resultPromise = waitForActivityTxHash( - getFeed, - 0, - { maxWaitMs: 1000, initialDelayMs: 100, maxDelayMs: 100 }, - undefined, - undefined, - controller.signal, - ); - - await new Promise((resolve) => setTimeout(resolve, 10)); - controller.abort(); - - await expect(resultPromise).resolves.toBe(''); - expect(getFeed).toHaveBeenCalledTimes(1); + 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 () => {