Eric Allam
5693b62cfb
## 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](https://github.com/nodejs/undici/issues/1108) and
[#2143](https://github.com/nodejs/undici/issues/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
74 lines
2.5 KiB
TypeScript
74 lines
2.5 KiB
TypeScript
// When proxying long-polling requests, content-encoding & content-length are added
|
|
// erroneously (saying the body is gzipped when it's not) so we'll just remove
|
|
// them to avoid content decoding errors in the browser.
|
|
//
|
|
|
|
import { logger } from "~/services/logger.server";
|
|
|
|
// Similar-ish problem to https://github.com/wintercg/fetch/issues/23
|
|
export async function longPollingFetch(
|
|
url: string,
|
|
options?: RequestInit,
|
|
rewriteResponseHeaders?: Record<string, string>
|
|
) {
|
|
let upstream: Response | undefined;
|
|
try {
|
|
upstream = await fetch(url, options);
|
|
let response = upstream;
|
|
|
|
if (response.headers.get("content-encoding")) {
|
|
const headers = new Headers(response.headers);
|
|
headers.delete("content-encoding");
|
|
headers.delete("content-length");
|
|
|
|
response = new Response(response.body, {
|
|
headers,
|
|
status: response.status,
|
|
statusText: response.statusText,
|
|
});
|
|
}
|
|
|
|
if (rewriteResponseHeaders) {
|
|
const headers = new Headers(response.headers);
|
|
|
|
for (const [fromKey, toKey] of Object.entries(rewriteResponseHeaders)) {
|
|
const value = headers.get(fromKey);
|
|
if (value) {
|
|
headers.set(toKey, value);
|
|
headers.delete(fromKey);
|
|
}
|
|
}
|
|
|
|
response = new Response(response.body, {
|
|
headers,
|
|
status: response.status,
|
|
statusText: response.statusText,
|
|
});
|
|
}
|
|
|
|
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) {
|
|
logger.error("Network error:", { error: error.message });
|
|
throw new Response("Network error occurred", { status: 503 });
|
|
} else if (error instanceof Error) {
|
|
logger.error("Fetch error:", { error: error.message });
|
|
throw new Response(error.message, { status: 500 });
|
|
} else {
|
|
logger.error("Unknown error occurred during fetch");
|
|
throw new Response("An unknown error occurred", { status: 500 });
|
|
}
|
|
}
|
|
}
|