fix: honour the web fetch timeout budget end to end (#249)

src/fetch/client.test.ts

@@ -19,4 +19,21 @@ describe('webFetch', () => {
     expect(createImpit).toHaveBeenNthCalledWith(1, expect.objectContaining({ browser: 'chrome' }));
     expect(createImpit).toHaveBeenNthCalledWith(2, expect.objectContaining({ browser: 'firefox' }));
   });
+  it('closes the safe proxy even when the ladder throws', async () => {
+    const close = vi.fn().mockResolvedValue(undefined);
+    await expect(webFetch({ url: 'https://example.com', timeoutSeconds: 5, maxChars: 0, allowPrivate: false }, {
+      plainFetch: vi.fn().mockRejectedValue(new Error('boom')),
+      createImpit: vi.fn(),
+      createSafeProxy: async () => ({ url: 'http://proxy', close }),
+    })).rejects.toThrow('boom');
+    expect(close).toHaveBeenCalledOnce();
+  });
+  it('reports an aborted fetch as a structured timeout', async () => {
+    const abort = Object.assign(new Error('The operation was aborted'), { name: 'TimeoutError' });
+    await expect(webFetch({ url: 'https://example.com', timeoutSeconds: 5, maxChars: 0, allowPrivate: false }, {
+      plainFetch: vi.fn().mockRejectedValue(abort),
+      createImpit: vi.fn(),
+      createSafeProxy: async () => ({ url: 'http://proxy', close: async () => {} }),
+    })).rejects.toMatchObject({ code: 'TIMEOUT', message: 'web fetch timed out after 5s' });
+  });
 });

src/fetch/client.ts

@@ -51,6 +51,9 @@ export async function webFetch(options: WebFetchOptions, dependencies: WebFetchD
     let tier: WebFetchResult['tier'] = 'plain'; let profile: WebFetchResult['profile'];
     if (isJavaScriptShell(body)) throw new CliError('FETCH_REQUIRES_BROWSER', 'This page requires browser rendering.', 'Use webcmd web fetch-browser for this URL.');
     if (isChallengeResponse(response.status, headersOf(response), body)) {
+      // ponytail: impit's timeout covers the request, not the body stream, so a
+      // trickling escalation body can outlive the budget. Race readBody against
+      // the deadline if that shows up in practice.
       for (const browser of ['chrome', 'firefox'] as const) {
         const impit = createImpit({ browser, proxyUrl: proxy.url, timeout: remaining() });
         response = await impit.fetch(options.url, { redirect: 'manual', timeout: remaining() });
@@ -63,5 +66,15 @@ export async function webFetch(options: WebFetchOptions, dependencies: WebFetchD
     const extracted = extractFetchedContent({ body, contentType: response.headers.get('content-type') ?? '', url: options.url });
     const clipped = truncate(extracted.content, options.maxChars);
     return { status: response.status, requestedUrl: options.url, finalUrl: response.url || options.url, contentType: response.headers.get('content-type') ?? '', tier, ...(profile && { profile }), title: extracted.title, extractionSource: extracted.source, truncated: clipped.truncated, content: clipped.content };
+  } catch (error) {
+    throw asFetchError(error, options.timeoutSeconds);
   } finally { await proxy.close(); }
 }
+
+/** An aborted fetch surfaces as a DOMException; agents need the structured timeout instead. */
+function asFetchError(error: unknown, timeoutSeconds: number): unknown {
+  if (error instanceof CliError) return error;
+  const name = (error as { name?: string } | null)?.name;
+  if (name === 'TimeoutError' || name === 'AbortError') return new TimeoutError('web fetch', timeoutSeconds);
+  return error;
+}

src/fetch/safe-proxy.test.ts

@@ -1,9 +1,35 @@
+import * as net from 'node:net';
 import { describe, expect, it } from 'vitest';
-import { isSafeAddress } from './safe-proxy.js';
+import { createSafeProxy, isSafeAddress } from './safe-proxy.js';
 
 describe('isSafeAddress', () => {
   it.each(['127.0.0.1', '10.0.0.1', '172.16.0.1', '192.168.1.1', '169.254.169.254', '0.0.0.0', '::1', '::', 'fe80::1', '::ffff:127.0.0.1'])('rejects private address %s', address => {
     expect(isSafeAddress(address)).toBe(false);
   });
   it('allows public IPv4 addresses', () => expect(isSafeAddress('93.184.216.34')).toBe(true));
 });
+
+describe('createSafeProxy close', () => {
+  it('does not wait for an idle CONNECT tunnel to drain', async () => {
+    // Stands in for the upstream host: accepts and then never says anything,
+    // exactly like the keep-alive tunnels impit leaves behind.
+    const upstream = net.createServer(() => {});
+    await new Promise<void>(done => upstream.listen(0, '127.0.0.1', () => done()));
+    const upstreamPort = (upstream.address() as net.AddressInfo).port;
+    const proxy = await createSafeProxy({ allowPrivate: true });
+    const proxyPort = Number(new URL(proxy.url).port);
+
+    const client = net.connect({ host: '127.0.0.1', port: proxyPort });
+    client.on('error', () => {});
+    await new Promise<void>(done => {
+      client.once('data', () => done());
+      client.write(`CONNECT 127.0.0.1:${upstreamPort} HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n`);
+    });
+
+    const started = Date.now();
+    await proxy.close();
+    expect(Date.now() - started).toBeLessThan(1000);
+    await new Promise<void>(done => client.once('close', () => done()));
+    await new Promise<void>(done => upstream.close(() => done()));
+  });
+});

src/fetch/safe-proxy.ts

@@ -1,6 +1,7 @@
 import { lookup as dnsLookup } from 'node:dns';
 import * as http from 'node:http';
 import * as net from 'node:net';
+import type { Duplex } from 'node:stream';
 
 export interface SafeProxy { url: string; close(): Promise<void>; }
 export interface SafeProxyOptions { allowPrivate?: boolean; lookup?: typeof dnsLookup; }
@@ -38,6 +39,18 @@ async function resolve(host: string, lookup: typeof dnsLookup, allowPrivate: boo
 export async function createSafeProxy(options: SafeProxyOptions = {}): Promise<SafeProxy> {
   const lookup = options.lookup ?? dnsLookup;
   const allowPrivate = options.allowPrivate === true;
+  // Sockets opened through this proxy, including the upstream halves the HTTP
+  // server never learns about. `close()` destroys them: a keep-alive CONNECT
+  // tunnel otherwise keeps `server.close()` pending until the peer or the OS
+  // gives up, which turns a bounded fetch budget into a minutes-long hang.
+  const sockets = new Set<Duplex>();
+  const track = <T extends Duplex>(socket: T): T => {
+    sockets.add(socket);
+    socket.on('close', () => sockets.delete(socket));
+    // A destroyed peer must not resurface as an unhandled 'error' event.
+    socket.on('error', () => socket.destroy());
+    return socket;
+  };
   const server = http.createServer(async (request, response) => {
     try {
       const target = new URL(request.url ?? '');
@@ -46,22 +59,32 @@ export async function createSafeProxy(options: SafeProxyOptions = {}): Promise<S
         response.writeHead(upstreamResponse.statusCode ?? 502, upstreamResponse.headers);
         upstreamResponse.pipe(response);
       });
+      upstream.on('socket', track);
       upstream.on('error', error => response.destroy(error));
       request.pipe(upstream);
     } catch (error) { response.writeHead(403).end(error instanceof Error ? error.message : 'Unsafe fetch destination'); }
   });
+  server.on('connection', track);
   server.on('connect', async (request, client, head) => {
+    track(client);
     try {
       const [host, portText] = (request.url ?? '').replace(/^\[/, '').replace(']', '').split(':');
       if (!host) throw new Error('Invalid CONNECT target');
       const address = await resolve(host, lookup, allowPrivate);
-      const upstream = net.connect({ host: address, port: Number(portText) || 443 });
+      const upstream = track(net.connect({ host: address, port: Number(portText) || 443 }));
       upstream.once('connect', () => { client.write('HTTP/1.1 200 Connection Established\r\n\r\n'); if (head.length) upstream.write(head); upstream.pipe(client); client.pipe(upstream); });
       upstream.once('error', error => client.destroy(error));
     } catch (error) { client.end(`HTTP/1.1 403 Forbidden\r\n\r\n${error instanceof Error ? error.message : ''}`); }
   });
   await new Promise<void>((resolveListen, reject) => { server.once('error', reject); server.listen(0, '127.0.0.1', () => resolveListen()); });
   const address = server.address();
   if (!address || typeof address === 'string') throw new Error('Safe proxy did not bind');
-  return { url: `http://127.0.0.1:${address.port}`, close: () => new Promise((resolveClose, reject) => server.close(error => error ? reject(error) : resolveClose())) };
+  return {
+    url: `http://127.0.0.1:${address.port}`,
+    close: () => new Promise((resolveClose, reject) => {
+      for (const socket of sockets) socket.destroy();
+      sockets.clear();
+      server.close(error => error ? reject(error) : resolveClose());
+    }),
+  };
 }