Miniflare resets the connection when a Worker responds without reading the request body; workerd does this whenever the body crossed a service binding #7634

// A request body that crosses a service binding and is not read gets the connection reset.
// npm i miniflare (which brings workerd), then node repro.mjs. Runs workerd on its own with four
// Workers, Miniflare with one, and a Node HTTP server as a reference. Every request goes over a raw
// socket and declares and sends its body with the headers in one write; every server answers 413.
import { spawn } from "node:child_process";
import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
import http from "node:http";
import net from "node:net";
import os from "node:os";
import path from "node:path";
import { convertV4MiniflareOptions, Miniflare } from "miniflare";
import workerd from "workerd";

const REFUSE = `export default { async fetch() { return new Response("too large", { status: 413 }); } };`;
const READ = `export default { async fetch(request) { await request.arrayBuffer(); return new Response("too large", { status: 413 }); } };`;
const FORWARD = `export default { async fetch(request, env) { return env.TARGET.fetch(request); } };`;
const worker = (script, target) =>
  `(modules = [(name = "worker.js", esModule = ${JSON.stringify(script)})], compatibilityDate = "2026-09-01"` +
  (target ? `, bindings = [(name = "TARGET", service = "${target}")])` : ")");

const PORT = { refuse: 18701, read: 18702, forwardToRefuse: 18703, forwardToRead: 18704, node: 18705 };
const config = `using Workerd = import "/workerd/workerd.capnp";
const config :Workerd.Config = (
  services = [
    (name = "refuse", worker = ${worker(REFUSE)}),
    (name = "read", worker = ${worker(READ)}),
    (name = "forwardToRefuse", worker = ${worker(FORWARD, "refuse")}),
    (name = "forwardToRead", worker = ${worker(FORWARD, "read")}),
  ],
  sockets = [
    (name = "refuse", address = "127.0.0.1:${PORT.refuse}", http = (), service = "refuse"),
    (name = "read", address = "127.0.0.1:${PORT.read}", http = (), service = "read"),
    (name = "forwardToRefuse", address = "127.0.0.1:${PORT.forwardToRefuse}", http = (), service = "forwardToRefuse"),
    (name = "forwardToRead", address = "127.0.0.1:${PORT.forwardToRead}", http = (), service = "forwardToRead"),
  ],
);
`;

// One request with a body of `bytes`. The client starts reading the response `readAfterMs` after
// sending, as a busy client would.
function probe(port, bytes, readAfterMs) {
  return new Promise((resolve) => {
    const socket = net.connect(port, "127.0.0.1");
    const chunks = [];
    let respondedAt;
    let settled = false;
    const finish = (ending) => {
      if (settled) return;
      settled = true;
      const status = /^HTTP\/1\.1 (\d{3})/.exec(Buffer.concat(chunks).toString("latin1"))?.[1];
      const after = respondedAt === undefined ? undefined : performance.now() - respondedAt;
      resolve({ outcome: `${status ?? "no response"}, then ${ending}`, after });
      socket.destroy();
    };
    socket.on("connect", () => {
      socket.write(`POST / HTTP/1.1\r\nHost: localhost\r\nContent-Length: ${bytes}\r\n\r\n${"x".repeat(bytes)}`);
      if (readAfterMs > 0) {
        socket.pause();
        setTimeout(() => socket.resume(), readAfterMs);
      }
    });
    socket.on("data", (chunk) => {
      respondedAt ??= performance.now();
      chunks.push(chunk);
    });
    socket.on("end", () => finish("FIN"));
    socket.on("error", (error) => finish(error.code));
    socket.setTimeout(2000, () => finish("connection still open 2 s later"));
  });
}

async function listening(port) {
  for (let attempt = 0; attempt < 100; attempt++) {
    const open = await new Promise((resolve) => {
      const socket = net.connect(port, "127.0.0.1", () => {
        socket.destroy();
        resolve(true);
      });
      socket.on("error", () => resolve(false));
    });
    if (open) return;
    await new Promise((resolve) => setTimeout(resolve, 100));
  }
  throw new Error(`nothing is listening on ${port}`);
}

const folder = mkdtempSync(path.join(os.tmpdir(), "workerd-repro-"));
writeFileSync(path.join(folder, "config.capnp"), config);
const child = spawn(workerd.default, ["serve", path.join(folder, "config.capnp")], {
  stdio: ["ignore", "inherit", "inherit"],
  windowsHide: true,
});
const reference = http.createServer((request, response) => response.writeHead(413).end("too large"));
await new Promise((resolve) => reference.listen(PORT.node, "127.0.0.1", resolve));
const miniflare = (script) =>
  new Miniflare(convertV4MiniflareOptions({ modules: true, compatibilityDate: "2026-09-01", script }));
const miniflares = { refuse: miniflare(REFUSE), read: miniflare(READ) };

try {
  await Promise.all(Object.values(PORT).map(listening));
  const miniflarePort = async (mf) => Number((await mf.ready).port);
  console.log(`workerd ${workerd.version}, Node ${process.version}, ${os.type()} ${os.release()}; 20 requests per row`);
  const rows = [
    ["workerd: Worker does not read the body", PORT.refuse, 5000],
    ["workerd: Worker forwards through a service binding to that Worker", PORT.forwardToRefuse, 5000],
    ["workerd, control: Worker reads the body", PORT.read, 5000],
    ["workerd, control: Worker forwards through a service binding to the reading Worker", PORT.forwardToRead, 5000],
    ["Miniflare: Worker does not read the body", await miniflarePort(miniflares.refuse), 5000],
    ["Miniflare, control: Worker reads the body", await miniflarePort(miniflares.read), 5000],
    ["Node http server, reference: does not read the body", PORT.node, 5000],
    ["workerd: Worker does not read the body", PORT.refuse, 100000],
  ];
  for (const [name, port, bytes] of rows) {
    for (const readAfterMs of [0, 300]) {
      const outcomes = new Map();
      const after = [];
      for (let i = 0; i < 20; i++) {
        const result = await probe(port, bytes, readAfterMs);
        outcomes.set(result.outcome, (outcomes.get(result.outcome) ?? 0) + 1);
        if (readAfterMs === 0 && result.after !== undefined && !result.outcome.includes("still open")) {
          after.push(result.after);
        }
      }
      const counts = [...outcomes].map(([outcome, n]) => `${n} x ${outcome}`).join("; ");
      const timing = after.length === 0
        ? ""
        : ` (closed ${Math.min(...after).toFixed(1)}-${Math.max(...after).toFixed(1)} ms after the response arrived)`;
      console.log(`${name}, ${bytes}-byte body, client reads after ${readAfterMs} ms: ${counts}${timing}`);
    }
  }
} finally {
  reference.close();
  await Promise.all(Object.values(miniflares).map((mf) => mf.dispose()));
  child.kill();
  await new Promise((resolve) => (child.exitCode !== null ? resolve() : child.once("exit", resolve)));
  rmSync(folder, { recursive: true, force: true });
}