hyper-svc / test / uds-backpressure.test.js

// CHARACTERIZATION — does the UDS server bound its write buffering when a
// client pipelines many large-response requests without reading?
//
// The write path (_handleRequest) calls sock.write(encodeFrame(...)) and ignores
// the return value. Each response frame is capped at 16 MiB, but nothing bounds
// the SUM across many pipelined/unread responses. This test measures the
// server-side socket's buffered bytes (writableLength) after a non-reading client
// fires N large-response requests, to see whether buffering grows unbounded.
//
// This is a probe, not a pass/fail regression yet — it prints the numbers so we
// can decide whether a fix (serialized write queue awaiting 'drain') is warranted
// and whether bare-net exposes the hooks it would need.

import test from 'brittle'
import process from '#node:process'
import net from '#node:net'
import { mkdtempSync } from '#node:fs'
import { join } from '#node:path'
import { tmpdir } from '#node:os'
import { UdsServer } from '../lib/transports/uds.js'
import { encodeFrame } from '../lib/frame.js'

function tmpSockPath () {
  return join(mkdtempSync(join(tmpdir(), 'hyper-svc-bp-')), 'svc.sock')
}
async function waitFor (pred, ms = 5000) {
  const t0 = Date.now()
  while (Date.now() - t0 < ms) { if (pred()) return true; await new Promise((r) => setTimeout(r, 20)) }
  return false
}

test('probe: server-side write buffering under a non-reading pipelining client', async (t) => {
  const RESP_SIZE = 256 * 1024 // 256 KiB per response
  const N = 100
  let dispatched = 0

  const server = new UdsServer({
    path: tmpSockPath(),
    dispatch: async () => { dispatched++; return { blob: 'x'.repeat(RESP_SIZE) } }
  })
  await server.listen()
  t.teardown(async () => { try { await server.close() } catch {} })

  // Client that NEVER reads its responses (pause the readable side). Its OS
  // receive buffer fills, then flow-control backpressures the server's writes.
  const client = net.createConnection(server.path)
  t.teardown(() => { try { client.destroy() } catch {} })
  await new Promise((resolve) => client.once('connect', resolve))
  if (typeof client.pause === 'function') client.pause()

  // Fire N pipelined requests in one burst.
  for (let i = 0; i < N; i++) {
    client.write(encodeFrame({ id: i, method: 'noop', params: {} }))
  }

  const rss0 = process.memoryUsage().rss

  // Give the server time to decode + dispatch + attempt to write responses.
  await waitFor(() => dispatched >= N, 6000)
  await new Promise((r) => setTimeout(r, 300)) // let writes settle into buffers

  const rss1 = process.memoryUsage().rss
  const rssGrowth = (rss1 - rss0) / 1024 / 1024
  const produced = N * RESP_SIZE / 1024 / 1024

  const [serverSock] = server._connections
  // bare-net doesn't expose writableLength; hunt for any internal buffered-bytes signal.
  const internalKeys = serverSock ? Object.keys(serverSock).filter(k => /buffer|writ|queue|length/i.test(k)) : []

  t.comment(`dispatched=${dispatched}/${N} (all fired from ONE decode burst → no per-request bound)`)
  t.comment(`total produced=${produced.toFixed(1)} MiB to a client that read nothing`)
  t.comment(`process RSS growth=${rssGrowth.toFixed(1)} MiB (unbounded ≈ tracks produced)`)
  t.comment(`serverSock internal buffer-ish keys: ${JSON.stringify(internalKeys)}`)

  // Characterization only (no gap-assert — that would become a false regression
  // once flow control lands). Observed 2026-07-02: RSS grew ~2x the produced
  // bytes with ZERO flow control — a non-reading/pipelining local client can
  // balloon server memory. Fix direction: a cooperative decode->dispatch->write
  // pump that bounds in-flight concurrency and gates on write()'s return + 'drain'
  // (a naive pause-on-drain is insufficient — all N requests dispatch from one
  // synchronous decode burst, as `dispatched=N/N` above shows). Low urgency:
  // real clients are well-behaved, and this is a local same-user UDS.
  t.ok(dispatched === N, 'probe completed — see comments for the buffering numbers')
})