bufferedAmountを待っても守れるのは送信側だけで、遅い受信側は守れない

socket.send(payload);

// A WebSocket server that accepts connections and then stops reading from the socket.
import { WebSocketServer } from "ws";
const wss = new WebSocketServer({ port: 8081 });
wss.on("connection", (ws) => {
  ws._socket.pause();          // stop reading: the kernel receive buffer fills, then the window closes
  console.log("client connected; not reading");
});
console.log("listening on ws://127.0.0.1:8081");

// Send 64 KiB messages as fast as possible to a server that is not reading.
import WebSocket from "ws";
const ws = new WebSocket("ws://127.0.0.1:8081");
const CHUNK = Buffer.alloc(64 * 1024, 1);
ws.on("open", () => {
  let sent = 0, firstBuffered = null;
  const t0 = process.hrtime.bigint();
  for (let i = 1; i <= 1024; i++) {            // 1024 x 64 KiB = 64 MiB in one synchronous loop
    ws.send(CHUNK);
    sent += CHUNK.length;
    if (firstBuffered === null && ws.bufferedAmount > 0) firstBuffered = sent;
  }
  const ms = Number(process.hrtime.bigint() - t0) / 1e6;
  console.log(`send() x1024 returned after ${ms.toFixed(1)} ms`);
  console.log(`bufferedAmount stayed 0 until ${(firstBuffered / 1048576).toFixed(2)} MiB had been "sent"`);
  console.log(`bufferedAmount now: ${(ws.bufferedAmount / 1048576).toFixed(2)} MiB of ${(sent / 1048576)} MiB`);
  console.log(`rss: ${(process.memoryUsage().rss / 1048576).toFixed(0)} MiB`);
  setTimeout(() => process.exit(0), 200);
});

send() x1024 returned after 193.4 ms   (残り2回は152.3msと152.8ms)
bufferedAmount stayed 0 until 2.56 MiB had been "sent"
bufferedAmount now: 61.51 MiB of 64 MiB
rss: 129 MiB

// Same probe from a real browser WebSocket (headless Chrome). Needs: npm i puppeteer-core, google-chrome.
const puppeteer = require("puppeteer-core");
(async () => {
  const b = await puppeteer.launch({ executablePath: "/usr/bin/google-chrome", headless: "new", args: ["--no-sandbox"] });
  const p = await b.newPage();
  const http = require("http"); const srv = http.createServer((q,r)=>r.end("<!doctype html><title>x</title>")).listen(8082);
  await p.goto("http://127.0.0.1:8082/");
  const r = await p.evaluate(async () => {
    const ws = new WebSocket("ws://127.0.0.1:8081");
    await new Promise((res) => (ws.onopen = res));
    const chunk = new Uint8Array(64 * 1024);
    const t0 = performance.now();
    for (let i = 0; i < 1024; i++) ws.send(chunk);
    const sameTask = ws.bufferedAmount;
    const ms = performance.now() - t0;
    const samples = [];
    for (let k = 0; k < 5; k++) { await new Promise((r) => setTimeout(r, 400)); samples.push(ws.bufferedAmount); }
    return { ms, sameTask, samples, wss: typeof WebSocketStream, readyState: ws.readyState };
  });
  console.log(JSON.stringify(r));
  console.log("sameTask MiB", (r.sameTask / 1048576).toFixed(2), "later MiB", r.samples.map((x) => (x / 1048576).toFixed(2)).join(" "));
  await b.close(); srv.close();
})();

sameTask MiB 64.00 later MiB 61.25 61.25 61.25 61.25 61.25

// Does gating on bufferedAmount protect a slow consumer? Run: node slow_lab.mjs
import { WebSocketServer, WebSocket } from "ws";

const MSGS = 400, SIZE = 64 * 1024, WORK_MS = 5;           // the consumer needs 5 ms per message
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

async function run(mode) {
  const wss = new WebSocketServer({ port: 0 });
  await new Promise((r) => wss.on("listening", r));
  const port = wss.address().port;
  let peak = 0, done;                                         // peak = most messages ever waiting in the consumer's JS memory
  const finished = new Promise((r) => (done = r));
  wss.on("connection", (ws) => {
    const queue = []; let handled = 0, working = false;
    const pump = async () => {
      if (working) return; working = true;
      while (queue.length) {
        await sleep(WORK_MS); queue.shift(); handled++;
        if (mode === "credit") ws.send("ack");                // credit: one ack per message that is really finished
        if (handled === MSGS) done();
      }
      working = false;
    };
    ws.on("message", () => { queue.push(1); peak = Math.max(peak, queue.length); pump(); });
  });

  const c = new WebSocket(`ws://127.0.0.1:${port}`);
  await new Promise((r) => c.on("open", r));
  const payload = Buffer.alloc(SIZE, 7);
  let inFlight = 0, wake = null, peakBuffered = 0;
  c.on("message", () => { inFlight--; wake?.(); });
  const t0 = Date.now();
  for (let i = 0; i < MSGS; i++) {
    if (mode === "bufferedAmount") while (c.bufferedAmount > 256 * 1024) await sleep(5);
    if (mode === "credit") while (inFlight >= 8) await new Promise((r) => (wake = r));   // window of 8 messages
    c.send(payload); inFlight++;
    peakBuffered = Math.max(peakBuffered, c.bufferedAmount);
  }
  const sendMs = Date.now() - t0;
  await finished;
  console.log(`${mode.padEnd(15)} sender done sending after ${String(sendMs).padStart(5)} ms | sender peak bufferedAmount ${(peakBuffered/1048576).toFixed(2)} MiB | consumer peak backlog ${String(peak).padStart(3)} messages (${(peak*SIZE/1048576).toFixed(1)} MiB)`);
  c.close(); wss.close();
}
for (const m of ["naive", "bufferedAmount", "credit"]) await run(m);

// What happens to a message that is larger than the receiver allows? Run: node limit_lab.mjs
import { WebSocketServer, WebSocket } from "ws";
const wss = new WebSocketServer({ port: 0, maxPayload: 1024 * 1024 });   // receiver accepts at most 1 MiB per message
await new Promise((r) => wss.on("listening", r));
wss.on("connection", (ws) => {
  ws.on("message", (m) => console.log(`server got a message of ${m.length} bytes`));
  ws.on("error", (e) => console.log(`server error: ${e.message}`));
});
const c = new WebSocket(`ws://127.0.0.1:${wss.address().port}`);
await new Promise((r) => c.on("open", r));
c.send(Buffer.alloc(512 * 1024));                       // fits
c.send(Buffer.alloc(2 * 1024 * 1024));                  // does not fit
c.send(Buffer.alloc(16));                               // a perfectly small message, sent afterwards
await new Promise((r) => c.on("close", (code, reason) => { console.log(`client: closed with code ${code} ${reason}`); r(); }));
wss.close();

server got a message of 524288 bytes
server error: Max payload size exceeded
client: closed with code 1009