Cancel the Stream, Not the Connection, When One HTTP/2 Request Times Out

// One HTTP/2 connection, several requests, one of them is slow.
// What should the client close when the slow one hits its deadline: the stream, or the whole session?
import http2 from "node:http2";
const { NGHTTP2_CANCEL } = http2.constants;
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

const server = http2.createServer();                       // cleartext HTTP/2 is enough for this
let sessions = 0; let serverLog = [];
server.on("session", (s) => { sessions++; });
server.on("stream", (stream, headers) => {
  const delay = Number(headers[":path"].slice(1));          // GET /<milliseconds>
  stream.on("close", () => { if (stream.rstCode) serverLog.push(`server saw RST_STREAM(code ${stream.rstCode}) for the /${delay} stream`); });
  setTimeout(() => { if (!stream.destroyed) { stream.respond({ ":status": 200 }); stream.end(`done after ${delay} ms`); } }, delay);
});
await new Promise((r) => server.listen(0, r));
const url = `http://localhost:${server.address().port}`;

async function run(strategy) {
  sessions = 0; serverLog = [];
  const session = http2.connect(url);
  session.on("error", () => {});
  await new Promise((r) => session.on("connect", r));
  const t0 = Date.now();
  const request = (ms, deadline) => new Promise((resolve) => {
    const req = session.request({ ":path": `/${ms}` });
    let body = "";
    const timer = setTimeout(() => {
      if (strategy === "close-stream") req.close(NGHTTP2_CANCEL);       // RST_STREAM(CANCEL) for this stream only
      else session.destroy();                                           // tear down the TCP connection
      resolve(`TIMEOUT after ${Date.now() - t0} ms`);
    }, deadline);
    req.on("data", (d) => (body += d));
    req.on("error", () => {});
    req.on("close", () => { clearTimeout(timer); resolve(body.startsWith("done") ? `ok (${body})` : "FAILED: closed without a response"); });
  });
  // one slow request (3 s) with a 500 ms deadline, two others that need 800 ms and have 2 s deadlines
  const results = await Promise.all([request(3000, 500), request(800, 2000), request(800, 2000)]);
  console.log(`  strategy "${strategy}": slow -> ${results[0]} | other #1 -> ${results[1]} | other #2 -> ${results[2]} | TCP connections opened: ${sessions}`);
  await sleep(300);
  console.log(`      ${serverLog.length ? serverLog.sort().join("; ") : "server saw no RST_STREAM"}`);
  session.destroy(); await sleep(100);
}
console.log("timeout handling on a shared HTTP/2 connection");
await run("close-session");
await run("close-stream");
server.close();

timeout handling on a shared HTTP/2 connection
  strategy "close-session": slow -> TIMEOUT after 505 ms | other #1 -> FAILED: closed without a response | other #2 -> FAILED: closed without a response | TCP connections opened: 1
      server saw RST_STREAM(code 8) for the /800 stream
  strategy "close-stream": slow -> TIMEOUT after 501 ms | other #1 -> ok (done after 800 ms) | other #2 -> ok (done after 800 ms) | TCP connections opened: 1
      server saw RST_STREAM(code 8) for the /3000 stream

// Correlating replies on one shared connection: by position (FIFO) or by request id.
// One slow request (A), then two quick ones (B, C), 500 ms deadline each. Run: node mux_lab.mjs
import { WebSocketServer, WebSocket } from "ws";
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// --- server: one handler per connection; "ordered" replies in request order (like a pipelined protocol), "tagged" replies as soon as ready
const wss = new WebSocketServer({ port: 0 });
await new Promise((r) => wss.on("listening", r));
wss.on("connection", (ws) => {
  let chain = Promise.resolve();
  ws.on("message", (raw) => {
    const m = JSON.parse(raw);
    const work = sleep(m.delay).then(() => JSON.stringify({ id: m.id, tag: m.tag }));
    if (m.ordered) chain = chain.then(() => work).then((out) => ws.send(out));    // must wait for everything before it
    else work.then((out) => ws.send(out));
  });
});
const url = `ws://127.0.0.1:${wss.address().port}`;
const open = async () => { const c = new WebSocket(url); await new Promise((r) => c.on("open", r)); return c; };

// --- client 1: replies carry no id, the n-th reply belongs to the n-th request
async function positional() {
  const c = await open(); const waiting = []; let late = 0;
  c.on("message", (raw) => { const w = waiting.shift(); if (w) w.resolve(JSON.parse(raw).tag); else late++; });
  const call = (tag, delay, deadline = 500) => new Promise((resolve) => {
    const w = { resolve };
    waiting.push(w);
    c.send(JSON.stringify({ tag, delay, ordered: true }));
    setTimeout(() => { const i = waiting.indexOf(w); if (i >= 0) { waiting.splice(i, 1); resolve("TIMEOUT"); } }, deadline);
  });
  const first = await Promise.all([call("A", 2000), call("B", 50), call("C", 50)]);
  console.log("positional, first round :", first.join(" "));
  const d = await call("D", 10, 3000);                       // a new request, while A's reply is still on its way
  console.log(`positional, request D   : asked for D, got "${d}"`);
  await sleep(500);
  c.close();
}

// --- client 2: replies carry the request id; the timeout removes exactly that id
async function tagged() {
  const c = await open(); const pending = new Map(); let nextId = 1, orphans = 0;
  c.on("message", (raw) => { const m = JSON.parse(raw); const w = pending.get(m.id); if (w) { pending.delete(m.id); w(m.tag); } else orphans++; });
  const call = (tag, delay, deadline = 500) => new Promise((resolve) => {
    const id = nextId++;
    pending.set(id, resolve);
    c.send(JSON.stringify({ id, tag, delay }));
    setTimeout(() => { if (pending.delete(id)) resolve("TIMEOUT"); }, deadline);
  });
  const first = await Promise.all([call("A", 2000), call("B", 50), call("C", 50)]);
  console.log("by id,      first round :", first.join(" "));
  const d = await call("D", 10, 3000);
  console.log(`by id,      request D   : asked for D, got "${d}"`);
  await sleep(2500);
  console.log(`by id,      afterwards  : late replies dropped as orphans: ${orphans}, pending map size: ${pending.size}`);
  c.close();
}
await positional(); await tagged(); wss.close();

positional, first round : TIMEOUT TIMEOUT TIMEOUT
positional, request D   : asked for D, got "A"
by id,      first round : TIMEOUT B C
by id,      request D   : asked for D, got "D"
by id,      afterwards  : late replies dropped as orphans: 1, pending map size: 0

// Two long responses (/a, /b), each a 1000-byte chunk every 10 ms for 1.5 s. Plain-text HTTP/2.
import http2 from "node:http2";
const server = http2.createServer();
server.on("stream", (stream) => {
  stream.respond({ ":status": 200 });
  const chunk = Buffer.alloc(1000, 1); let n = 0;
  const t = setInterval(() => { if (stream.destroyed || ++n > 150) { clearInterval(t); stream.end(); } else stream.write(chunk); }, 10);
});
server.listen(9200, "10.1.0.2", () => console.log("server up"));

// mode "single": both requests share one HTTP/2 connection.  mode "two": one connection per request.
// Around t = 500 ms, packets from the server to connection #1 are silently dropped for 30 ms.
import http2 from "node:http2";
import { execFile } from "node:child_process";
import { promisify } from "node:util";
const mode = process.argv[2];
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const run = promisify(execFile);
const nft = (...a) => run("nft", a);          // async: do not block the event loop while measuring
const DROP_MS = Number(process.argv[3] ?? 30);
await nft("add", "table", "inet", "hol"); await nft("add", "chain", "inet", "hol", "in", "{ type filter hook input priority 0; }");

const s1 = http2.connect("http://10.1.0.2:9200");
const s2 = mode === "single" ? s1 : http2.connect("http://10.1.0.2:9200");
await Promise.all([s1, s2].map((s) => new Promise((r) => s.once("connect", r))));
const port1 = s1.socket.localPort;                       // the loss will hit this connection only

const stats = {};
function fetch(session, name) {
  return new Promise((resolve) => {
    const req = session.request({ ":path": `/${name}` });
    const st = (stats[name] = { last: performance.now(), maxGap: 0, at: 0, bytes: 0 });
    const t0 = performance.now();
    req.on("data", (d) => { const now = performance.now(); const gap = now - st.last; if (gap > st.maxGap) { st.maxGap = gap; st.at = st.last - t0; } st.last = now; st.bytes += d.length; });
    req.on("end", resolve); req.resume?.();
  });
}
const done = Promise.all([fetch(s1, "a"), fetch(s2, "b")]);
await sleep(500);
await nft("add", "rule", "inet", "hol", "in", "ip", "saddr", "10.1.0.2", "tcp", "sport", "9200", "tcp", "dport", String(port1), "drop");
await sleep(DROP_MS);
await nft("flush", "chain", "inet", "hol", "in");
await done;
for (const [n, st] of Object.entries(stats)) console.log(`    stream ${n}: longest silence ${st.maxGap.toFixed(0).padStart(4)} ms (starting ${st.at.toFixed(0)} ms in), ${st.bytes} bytes`);
s1.destroy(); s2.destroy();

#!/bin/bash
# root needed. Two namespaces joined by a veth pair; server in "sv", client in "cl".
NODE=$(command -v node); HERE=$(cd "$(dirname "$0")" && pwd)
ip netns del sv 2>/dev/null; ip netns del cl 2>/dev/null
ip netns add sv; ip netns add cl
ip link add name vsv type veth peer name vcl
ip link set vsv netns sv; ip link set vcl netns cl
ip -n sv addr add 10.1.0.2/24 dev vsv; ip -n sv link set vsv up; ip -n sv link set lo up
ip -n cl addr add 10.1.0.1/24 dev vcl; ip -n cl link set vcl up; ip -n cl link set lo up
ip netns exec sv $NODE $HERE/server.mjs & SRV=$!
sleep 1
for round in 1 2 3 4 5; do
  for mode in single two; do
    echo "  [$mode connection(s)] round $round"
    ip netns exec cl $NODE $HERE/client.mjs $mode ${DROP:-30}
    ip netns exec cl nft delete table inet hol 2>/dev/null
  done
done
kill $SRV; ip netns del sv; ip netns del cl

// Is this HTTP/2 connection alive?  PING frame + our own deadline.  Then the path is black-holed.
import http2 from "node:http2";
import { execFile } from "node:child_process";
import { promisify } from "node:util";
const run = promisify(execFile);
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

function alive(session, ms) {                       // resolves true/false, never rejects
  return new Promise((resolve) => {
    const timer = setTimeout(() => resolve(false), ms);
    session.ping((err, rttMs) => { clearTimeout(timer); resolve(err ? false : rttMs); });
  });
}
const session = http2.connect("http://10.1.0.2:9200");
session.on("error", () => {});
await new Promise((r) => session.once("connect", r));
const t0 = performance.now();
console.log("healthy path   : alive() ->", await alive(session, 1000) !== false ? "true" : "false", `(PING round trip ${(await alive(session, 1000)).toFixed(2)} ms)`);
await run("nft", ["add", "table", "inet", "bh"]); await run("nft", ["add", "chain", "inet", "bh", "in", "{ type filter hook input priority 0; }"]);
await run("nft", ["add", "rule", "inet", "bh", "in", "ip", "saddr", "10.1.0.2", "drop"]);
const t1 = performance.now();
const ok = await alive(session, 1000);
console.log(`black-holed    : alive() -> ${ok !== false} after ${(performance.now() - t1).toFixed(0)} ms (deadline 1000 ms)`);
session.destroy();

healthy path   : alive() -> true (PING round trip 0.12 ms)
black-holed    : alive() -> false after 1001 ms (deadline 1000 ms)