| … | … | @@ -0,0 +1,84 @@ |
| 1 | // CHARACTERIZATION — does the UDS server bound its write buffering when a |
| 2 | // client pipelines many large-response requests without reading? |
| 3 | // |
| 4 | // The write path (_handleRequest) calls sock.write(encodeFrame(...)) and ignores |
| 5 | // the return value. Each response frame is capped at 16 MiB, but nothing bounds |
| 6 | // the SUM across many pipelined/unread responses. This test measures the |
| 7 | // server-side socket's buffered bytes (writableLength) after a non-reading client |
| 8 | // fires N large-response requests, to see whether buffering grows unbounded. |
| 9 | // |
| 10 | // This is a probe, not a pass/fail regression yet — it prints the numbers so we |
| 11 | // can decide whether a fix (serialized write queue awaiting 'drain') is warranted |
| 12 | // and whether bare-net exposes the hooks it would need. |
| 13 | |
| 14 | import test from 'brittle' |
| 15 | import process from '#node:process' |
| 16 | import net from '#node:net' |
| 17 | import { mkdtempSync } from '#node:fs' |
| 18 | import { join } from '#node:path' |
| 19 | import { tmpdir } from '#node:os' |
| 20 | import { UdsServer } from '../lib/transports/uds.js' |
| 21 | import { encodeFrame } from '../lib/frame.js' |
| 22 | |
| 23 | function tmpSockPath () { |
| 24 | return join(mkdtempSync(join(tmpdir(), 'hyper-svc-bp-')), 'svc.sock') |
| 25 | } |
| 26 | async function waitFor (pred, ms = 5000) { |
| 27 | const t0 = Date.now() |
| 28 | while (Date.now() - t0 < ms) { if (pred()) return true; await new Promise((r) => setTimeout(r, 20)) } |
| 29 | return false |
| 30 | } |
| 31 | |
| 32 | test('probe: server-side write buffering under a non-reading pipelining client', async (t) => { |
| 33 | const RESP_SIZE = 256 * 1024 // 256 KiB per response |
| 34 | const N = 100 |
| 35 | let dispatched = 0 |
| 36 | |
| 37 | const server = new UdsServer({ |
| 38 | path: tmpSockPath(), |
| 39 | dispatch: async () => { dispatched++; return { blob: 'x'.repeat(RESP_SIZE) } } |
| 40 | }) |
| 41 | await server.listen() |
| 42 | t.teardown(async () => { try { await server.close() } catch {} }) |
| 43 | |
| 44 | // Client that NEVER reads its responses (pause the readable side). Its OS |
| 45 | // receive buffer fills, then flow-control backpressures the server's writes. |
| 46 | const client = net.createConnection(server.path) |
| 47 | t.teardown(() => { try { client.destroy() } catch {} }) |
| 48 | await new Promise((resolve) => client.once('connect', resolve)) |
| 49 | if (typeof client.pause === 'function') client.pause() |
| 50 | |
| 51 | // Fire N pipelined requests in one burst. |
| 52 | for (let i = 0; i < N; i++) { |
| 53 | client.write(encodeFrame({ id: i, method: 'noop', params: {} })) |
| 54 | } |
| 55 | |
| 56 | const rss0 = process.memoryUsage().rss |
| 57 | |
| 58 | // Give the server time to decode + dispatch + attempt to write responses. |
| 59 | await waitFor(() => dispatched >= N, 6000) |
| 60 | await new Promise((r) => setTimeout(r, 300)) // let writes settle into buffers |
| 61 | |
| 62 | const rss1 = process.memoryUsage().rss |
| 63 | const rssGrowth = (rss1 - rss0) / 1024 / 1024 |
| 64 | const produced = N * RESP_SIZE / 1024 / 1024 |
| 65 | |
| 66 | const [serverSock] = server._connections |
| 67 | // bare-net doesn't expose writableLength; hunt for any internal buffered-bytes signal. |
| 68 | const internalKeys = serverSock ? Object.keys(serverSock).filter(k => /buffer|writ|queue|length/i.test(k)) : [] |
| 69 | |
| 70 | t.comment(`dispatched=${dispatched}/${N} (all fired from ONE decode burst → no per-request bound)`) |
| 71 | t.comment(`total produced=${produced.toFixed(1)} MiB to a client that read nothing`) |
| 72 | t.comment(`process RSS growth=${rssGrowth.toFixed(1)} MiB (unbounded ≈ tracks produced)`) |
| 73 | t.comment(`serverSock internal buffer-ish keys: ${JSON.stringify(internalKeys)}`) |
| 74 | |
| 75 | // Characterization only (no gap-assert — that would become a false regression |
| 76 | // once flow control lands). Observed 2026-07-02: RSS grew ~2x the produced |
| 77 | // bytes with ZERO flow control — a non-reading/pipelining local client can |
| 78 | // balloon server memory. Fix direction: a cooperative decode->dispatch->write |
| 79 | // pump that bounds in-flight concurrency and gates on write()'s return + 'drain' |
| 80 | // (a naive pause-on-drain is insufficient — all N requests dispatch from one |
| 81 | // synchronous decode burst, as `dispatched=N/N` above shows). Low urgency: |
| 82 | // real clients are well-behaved, and this is a local same-user UDS. |
| 83 | t.ok(dispatched === N, 'probe completed — see comments for the buffering numbers') |
| 84 | }) |