Channel.close() while writes await window credit silently drops buffered data and never invokes write callbacks

"use strict";
/**
 * Reproduction: Channel.close() while writes are parked on window credit
 * silently drops the buffered data AND its stream callback, wedging the
 * Writable forever.
 *
 * Scenario (common in port-forwarding proxies): a source stream is piped
 * into a forwarded channel; the source finishes and closes, and the app
 * calls channel.end() + channel.close() (i.e. channel.destroy()) while the
 * channel still holds data the remote hasn't granted window credit for.
 *
 * Expected: the buffered data flushes when WINDOW_ADJUST arrives (or the
 * pending write callback is invoked with an error), 'finish'/'end' fire,
 * and the receiver sees all bytes or an error.
 *
 * Actual: Channel._write() begins with `if (outgoing.state !== 'open')
 * return;` — when the adjust arrives after close() set state='closing',
 * the flush re-invokes _write(), which returns without sending anything
 * and WITHOUT calling the callback. The Writable never finishes, 'finish'
 * never fires, and the receiver hangs mid-stream missing the tail.
 *
 * Run: node repro-close-drops-writes.js   (exits 1 on the bug, 0 if fixed)
 */
const { Server, Client } = require("./lib/index.js");
const { generateKeyPairSync } = require("crypto");

const PAYLOAD = 5 * 1024 * 1024; // > 2 MiB initial window: forces parking
const CHUNK = 64 * 1024;

const hostKey = generateKeyPairSync("rsa", { modulusLength: 2048 })
  .privateKey.export({ type: "pkcs1", format: "pem" });

let received = 0;
let receiverEnded = false;
let senderFinished = false;
let queuedBytes = 0;
let cbsQueued = 0;
let cbsCalled = 0;
let cbErrors = 0;
const senderErrors = [];

const server = new Server({ hostKeys: [hostKey] }, (conn) => {
  conn.on("authentication", (ctx) => ctx.accept());
  conn.on("request", (accept, _reject, name) => {
    if (name !== "tcpip-forward") return;
    accept();
    // Give the client a beat to record the forwarding binding (it accepts
    // forwarded-tcpip opens only for registered bindings), then open the
    // forwarded connection; the client side will be the sender under test.
    setTimeout(() => openForwarded(conn), 100);
  });
  function openForwarded(conn2) {
    conn2.forwardOut("localhost", 9999, "127.0.0.1", 1234, (err, channel) => {
      if (err) throw err;
      // Slow receiver: consume in small paced reads so the sender exhausts
      // its 2 MiB window and parks before the payload is fully sent.
      channel.pause();
      const drain = setInterval(() => {
        let chunk;
        while ((chunk = channel.read(16 * 1024)) !== null) {
          received += chunk.length;
          break; // one small read per tick
        }
      }, 1);
      channel.on("end", () => {
        receiverEnded = true;
        clearInterval(drain);
      });
      channel.on("close", () => clearInterval(drain));
    });
  }
});

server.listen(0, "127.0.0.1", () => {
  const client = new Client();
  client.on("ready", () => {
    client.forwardIn("localhost", 9999, (err) => {
      if (err) throw err;
    });
  });
  client.on("tcp connection", (_info, accept) => {
    const channel = accept(); // sender side under test

    // Write the payload. Everything beyond the 2 MiB window parks inside
    // the channel awaiting WINDOW_ADJUST.
    let sawFalse = false;
    const writeMore = () => {
      while (queuedBytes < PAYLOAD) {
        const size = Math.min(CHUNK, PAYLOAD - queuedBytes);
        queuedBytes += size;
        const ok = channel.write(Buffer.alloc(size, 0x61), (err) => {
          cbsCalled++;
          if (err) cbErrors++;
        });
        cbsQueued++;
        if (!ok && !sawFalse) {
          sawFalse = true;
          // The window is exhausted and writes are parked. This is the
          // moment a proxy's source socket typically closes: the app ends
          // and closes the channel, expecting the buffered tail to flush.
          setImmediate(() => {
            channel.end(); // graceful: should flush buffered data, then EOF
            channel.close(); // what apps (and Channel.destroy()) do next
          });
          return; // stop writing more; the tail is already buffered
        }
        if (!ok) return channel.once("drain", writeMore);
      }
      channel.end();
    };
    channel.on("finish", () => {
      senderFinished = true;
    });
    channel.on("error", (err) => {
      senderErrors.push(err.message);
    });
    writeMore();
  });
  const port = server.address().port;
  client.connect({
    host: "127.0.0.1",
    port,
    username: "test",
    password: "test",
  });
});

setTimeout(() => {
  console.log(`bytes queued into channel:  ${queuedBytes}`);
  console.log(`receiver got:               ${received}`);
  console.log(`receiver saw EOF:           ${receiverEnded}`);
  console.log(`sender 'finish':            ${senderFinished}`);
  console.log(`write cbs queued/called:    ${cbsQueued}/${cbsCalled} (${cbErrors} with error)`);
  console.log(`sender 'error' events:      ${senderErrors.length} ${senderErrors[0] ?? ""}`);
  const dataLost = received < queuedBytes;
  const cbsLost = cbsCalled < cbsQueued;
  if (dataLost && cbsLost && senderErrors.length === 0) {
    console.log(
      "\nBUG: " + (queuedBytes - received) + " buffered bytes silently dropped and " +
        (cbsQueued - cbsCalled) + " write callback(s) never invoked " +
        "(Channel._write returns without calling cb when outgoing.state !== 'open').",
    );
    process.exit(1);
  }
  if (dataLost) {
    console.log("\nPARTIAL: data still dropped, but the failure is observable (cb/error fired).");
    process.exit(2);
  }
  console.log("\nOK: buffered data flushed — bug not present.");
  process.exit(0);
}, 8000);