fix(webapp): propagate abort signal through realtime proxy fetch (#3442)

## Summary

Fixes an RSS-only memory leak in the three realtime proxy routes
(`/realtime/v1/runs`, `/realtime/v1/runs/:id`,
`/realtime/v1/batches/:id`). Client disconnects during an in-flight
long-poll would leave the upstream fetch to Electric running with no way
to abort it, so undici kept the socket open and buffered response chunks
that would never be consumed.

## Root cause

All three routes flow through
`RealtimeClient.streamRun/streamRuns/streamBatch` → `#streamRunsWhere` →
`#performElectricRequest` → `longPollingFetch(url, { signal })`. The
chain was already signal-aware, but `#streamRunsWhere` hardcoded
`signal=undefined` when calling `#performElectricRequest`, so no signal
ever reached `longPollingFetch`.

When a downstream client aborts a long-poll mid-flight:
1. Express tears down the downstream response socket.
2. The `longPollingFetch` promise has already resolved (it returns as
soon as upstream headers arrive) and handed back `new
Response(upstream.body, {...})`.
3. `undici` keeps the upstream socket open and continues buffering
chunks into the `ReadableStream` that nothing will ever read from.
4. The upstream connection is eventually closed by Electric's own poll
timeout (~20s). During that window the per-request buffers stay in
native memory.

These buffers live below V8's accounting — no `heapUsed` or `external`
growth, no sign in heap snapshots, only RSS. An isolated standalone
reproducer (`fetch` against a slow-streaming upstream, discard the
`Response` before consuming its body) measures **~44 KB retained per
leaked request** after GC. That's consistent with the undici socket +
receive buffer + HTTP parser state for a long-lived chunked response.
The pattern is the shape documented in
[nodejs/undici#1108](nodejs/undici#1108) and
[#2143](nodejs/undici#2143).

## What changed

- **`realtimeClient.server.ts`** — add optional `signal` parameter to
`streamRun`, `streamRuns`, `streamBatch`, and the shared
`#streamRunsWhere`; thread it through to `#performElectricRequest`
instead of hardcoding `undefined`.
- **`realtime.v1.runs.$runId.ts`, `realtime.v1.runs.ts`,
`realtime.v1.batches.$batchId.ts`** — pass `getRequestAbortSignal()`
(from `httpAsyncStorage.server.ts`) at the call site. This is the signal
wired to `res.on('close')` and fires reliably on downstream disconnect.
- **`longPollingFetch.ts`** — belt-and-suspenders: cancel the upstream
body explicitly in the error path, and treat `AbortError` as a clean
`499` instead of a `500`. This both releases undici's buffers
deterministically on error and avoids spurious 500s in request logs when
a client legitimately walks away.

## Verification

Standalone reproducer: slow upstream server streams 32 KB chunks every
100 ms for 5 seconds per request. The proxy does `fetch(url)` with
varying signal/cancel strategies, creates `new Response(upstream.body,
...)`, and discards it without consuming the body (simulating the leak
path).

Results from 1 000 parallel fetches per variant, measured post-GC:

| variant | Δ heap | Δ external | Δ RSS |
| --- | --- | --- | --- |
| A. no signal, body never consumed (the bug) | +0.3 MB | 0 MB | **+59.4
MB** |
| B. signal propagated, aborted after headers (this fix) | −0.1 MB | 0
MB | +15.4 MB |
| C. no signal, explicit `res.body.cancel()` | 0 MB | 0 MB | −25.4 MB |

10-round sustained test of variant B to distinguish accumulating
retention from one-time allocator overhead:
```
round  1/10  Δ=+3.2 MB     round  6/10  Δ=-12.5 MB
round  2/10  Δ=-7.6 MB     round  7/10  Δ=-11.9 MB
round  3/10  Δ=-11.7 MB    round  8/10  Δ=-2.6 MB
round  4/10  Δ=+3.2 MB     round  9/10  Δ=-8.0 MB
round  5/10  Δ=-1.2 MB     round 10/10  Δ=-12.6 MB
```
RSS oscillates in a 49-65 MB band with no upward trend — signal
propagation fully releases the buffers.

## Risk

- Behavior change only on aborted long-polls: the upstream fetch now
cancels promptly instead of running to its natural timeout. This saves
both memory and outbound traffic to Electric.
- `AbortError` now surfaces as `499` rather than `500`. Any dashboard or
alert that counts 500s in request logs will see slightly fewer of them;
this is the intended behavior.
- Signal-aware parameter is optional on
`RealtimeClient.streamRun/streamRuns/streamBatch`, so callers that don't
opt in get the previous behavior.

## Test plan

- [ ] Existing realtime integration tests pass
- [ ] Dashboard realtime views (runs list, batch details) continue
working normally across tab open/close cycles
- [ ] Under a burst of aborted long-polls, server RSS returns to
baseline rather than climbing

Author: ericallam

.server-changes/fix-realtime-fetch-signal-leak.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Fix RSS memory leak in the realtime proxy routes. `/realtime/v1/runs`, `/realtime/v1/runs/:id`, and `/realtime/v1/batches/:id` called `fetch()` into Electric with no abort signal, so when a client disconnected mid long-poll, undici kept the upstream socket open and buffered response chunks that would never be consumed — retained only in RSS, invisible to V8 heap tooling. Thread `getRequestAbortSignal()` through `RealtimeClient.streamRun/streamRuns/streamBatch` to `longPollingFetch` and cancel the upstream body in the error path. Isolated reproducer showed ~44 KB retained per leaked request; signal propagation releases it cleanly.

apps/webapp/app/routes/realtime.v1.batches.$batchId.ts

@@ -1,5 +1,6 @@
 import { z } from "zod";
 import { $replica } from "~/db.server";
+import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { realtimeClient } from "~/services/realtimeClientGlobal.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 
@@ -33,7 +34,8 @@ export const loader = createLoaderApiRoute(
       batchRun.id,
       apiVersion,
       authentication.realtime,
-      request.headers.get("x-trigger-electric-version") ?? undefined
+      request.headers.get("x-trigger-electric-version") ?? undefined,
+      getRequestAbortSignal()
     );
   }
 );

apps/webapp/app/routes/realtime.v1.runs.$runId.ts

@@ -1,6 +1,7 @@
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { $replica } from "~/db.server";
+import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { realtimeClient } from "~/services/realtimeClientGlobal.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 
@@ -46,7 +47,12 @@ export const loader = createLoaderApiRoute(
       run.id,
       apiVersion,
       authentication.realtime,
-      request.headers.get("x-trigger-electric-version") ?? undefined
+      request.headers.get("x-trigger-electric-version") ?? undefined,
+      // Propagate abort on client disconnect so the upstream Electric long-poll
+      // fetch is cancelled too. Without this, undici buffers from the unconsumed
+      // upstream response body accumulate until Electric's poll timeout, causing
+      // steady RSS growth on api (see docs/runbooks for the H1 isolation test).
+      getRequestAbortSignal()
     );
   }
 );

apps/webapp/app/routes/realtime.v1.runs.ts

@@ -1,4 +1,5 @@
 import { z } from "zod";
+import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { realtimeClient } from "~/services/realtimeClientGlobal.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 
@@ -31,7 +32,8 @@ export const loader = createLoaderApiRoute(
       searchParams,
       apiVersion,
       authentication.realtime,
-      request.headers.get("x-trigger-electric-version") ?? undefined
+      request.headers.get("x-trigger-electric-version") ?? undefined,
+      getRequestAbortSignal()
     );
   }
 );

apps/webapp/app/services/realtimeClient.server.ts

@@ -115,15 +115,17 @@ export class RealtimeClient {
     runId: string,
     apiVersion: API_VERSIONS,
     requestOptions?: RealtimeRequestOptions,
-    clientVersion?: string
+    clientVersion?: string,
+    signal?: AbortSignal
   ) {
     return this.#streamRunsWhere(
       url,
       environment,
       `id='${runId}'`,
       apiVersion,
       requestOptions,
-      clientVersion
+      clientVersion,
+      signal
     );
   }
 
@@ -133,7 +135,8 @@ export class RealtimeClient {
     batchId: string,
     apiVersion: API_VERSIONS,
     requestOptions?: RealtimeRequestOptions,
-    clientVersion?: string
+    clientVersion?: string,
+    signal?: AbortSignal
   ) {
     const whereClauses: string[] = [
       `"runtimeEnvironmentId"='${environment.id}'`,
@@ -148,7 +151,8 @@ export class RealtimeClient {
       whereClause,
       apiVersion,
       requestOptions,
-      clientVersion
+      clientVersion,
+      signal
     );
   }
 
@@ -158,7 +162,8 @@ export class RealtimeClient {
     params: RealtimeRunsParams,
     apiVersion: API_VERSIONS,
     requestOptions?: RealtimeRequestOptions,
-    clientVersion?: string
+    clientVersion?: string,
+    signal?: AbortSignal
   ) {
     const whereClauses: string[] = [`"runtimeEnvironmentId"='${environment.id}'`];
 
@@ -180,7 +185,8 @@ export class RealtimeClient {
       whereClause,
       apiVersion,
       requestOptions,
-      clientVersion
+      clientVersion,
+      signal
     );
 
     if (createdAtFilter) {
@@ -274,7 +280,8 @@ export class RealtimeClient {
     whereClause: string,
     apiVersion: API_VERSIONS,
     requestOptions?: RealtimeRequestOptions,
-    clientVersion?: string
+    clientVersion?: string,
+    signal?: AbortSignal
   ) {
     const electricUrl = this.#constructRunsElectricUrl(
       url,
@@ -288,7 +295,7 @@ export class RealtimeClient {
       electricUrl,
       environment,
       apiVersion,
-      undefined,
+      signal,
       clientVersion
     );
   }

apps/webapp/app/utils/longPollingFetch.ts

@@ -11,8 +11,10 @@ export async function longPollingFetch(
   options?: RequestInit,
   rewriteResponseHeaders?: Record<string, string>
 ) {
+  let upstream: Response | undefined;
   try {
-    let response = await fetch(url, options);
+    upstream = await fetch(url, options);
+    let response = upstream;
 
     if (response.headers.get("content-encoding")) {
       const headers = new Headers(response.headers);
@@ -46,16 +48,24 @@ export async function longPollingFetch(
 
     return response;
   } catch (error) {
+    // Release upstream undici socket + buffers explicitly. Without this the
+    // ReadableStream stays open and undici keeps buffering chunks into memory
+    // until the upstream times out (see H1 isolation test — ~44 KB retained
+    // per unconsumed-body fetch in RSS).
+    try { await upstream?.body?.cancel(); } catch {}
+
+    // AbortError is the expected path when downstream disconnects with a
+    // propagated signal — treat as a clean client-close, not a server error.
+    if (error instanceof Error && error.name === "AbortError") {
+      throw new Response(null, { status: 499 });
+    }
     if (error instanceof TypeError) {
-      // Network error or other fetch-related errors
       logger.error("Network error:", { error: error.message });
       throw new Response("Network error occurred", { status: 503 });
     } else if (error instanceof Error) {
-      // HTTP errors or other known errors
       logger.error("Fetch error:", { error: error.message });
       throw new Response(error.message, { status: 500 });
     } else {
-      // Unknown errors
       logger.error("Unknown error occurred during fetch");
       throw new Response("An unknown error occurred", { status: 500 });
     }