Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
308 changes: 149 additions & 159 deletions test/regression/issue/13696.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,185 +5,175 @@
// is why testcontainers' default HostPortWaitStrategy hung until timeout.

import { expect, test } from "bun:test";
import { bunEnv, bunExe, tempDir } from "harness";
import { tempDir } from "harness";
import { once } from "node:events";
import http from "node:http";
import net from "node:net";
import { join } from "node:path";

// Runs a fixture that simulates docker-modem's chunked POST with an open
// request body, against a raw TCP server that responds before the request is
// finished. Prints "recv:<text>" for each response chunk as it arrives, and
// "request-seen" once the server has received the headers.
const fixture = (socketPath: string | undefined) => `
const net = require("net");
const http = require("http");

const socketPath = ${JSON.stringify(socketPath)};

const wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

const server = net.createServer((sock) => {
let buf = "";
let responded = false;
sock.on("data", async (d) => {
buf += d.toString("latin1");
if (responded || !buf.includes("\\r\\n\\r\\n")) return;
responded = true;
console.log("request-seen");
// Docker's exec response has no Content-Length and no chunked encoding;
// it just writes raw frames until the connection closes.
sock.write(
"HTTP/1.1 200 OK\\r\\n" +
"Content-Type: application/vnd.docker.raw-stream\\r\\n" +
"\\r\\n",
);
for (let i = 0; i < 3; i++) {
sock.write("chunk-" + i + "\\n");
await wait(50);
}
sock.end();
// Resolves with everything the socket has received once `marker` appears in it.
function readUntil(sock: net.Socket, marker: string): Promise<string> {
return new Promise(resolve => {
let buf = "";
const onData = (d: Buffer) => {
buf += d.toString("latin1");
if (!buf.includes(marker)) return;
sock.off("data", onData);
resolve(buf);
};
sock.on("data", onData);
});
});

const listenArgs = socketPath ? [socketPath] : [0, "127.0.0.1"];

server.listen(...listenArgs, () => {
const requestOpts = socketPath
? { socketPath, path: "/exec/abc/start" }
: { host: "127.0.0.1", port: server.address().port, path: "/exec/abc/start" };
}

// docker-modem passes an empty callback here and attaches 'response'
// separately via req.on('response', ...).
const req = http.request(
{
...requestOpts,
method: "POST",
headers: {
"Content-Type": "application/json",
"Transfer-Encoding": "chunked",
},
// Simulates docker-modem's chunked POST with an open request body, against a
// raw TCP server that responds before the request is finished. The server
// writes the next body chunk only after the client has observed the previous
// one, so every chunk is proven to stream while the request body is open.
for (const socketMode of ["tcp", "unix"] as const) {
test.concurrent(
`http.request delivers response while request body stream is still open (${socketMode})`,
async () => {
using dir = tempDir("issue-13696", {});
const socketPath = socketMode === "unix" ? join(String(dir), "docker.sock") : undefined;
const payload = JSON.stringify({ Detach: false, Tty: true });
const chunkedFrame = `${payload.length.toString(16)}\r\n${payload}\r\n`;

const events: string[] = [];
let request = "";
let chunkReceived = Promise.withResolvers<void>();

await using server = net.createServer(async sock => {
// req.destroy() at the end of the test may reset the connection.
sock.on("error", () => {});
// The single write() must dispatch the headers and the body chunk
// without req.end().
request = await readUntil(sock, chunkedFrame);
events.push("request-seen");

// Docker's exec response has no Content-Length and no chunked encoding;
// it just writes raw frames until the connection closes.
sock.write("HTTP/1.1 200 OK\r\nContent-Type: application/vnd.docker.raw-stream\r\n\r\n");
for (let i = 0; i < 3; i++) {
chunkReceived = Promise.withResolvers<void>();
sock.write(`chunk-${i}\n`);
await chunkReceived.promise;
}
sock.end();
});
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
if (socketPath) {
server.listen(socketPath);
} else {
server.listen(0, "127.0.0.1");
}
await once(server, "listening");

const requestOpts = socketPath
? { socketPath, path: "/exec/abc/start" }
: { host: "127.0.0.1", port: (server.address() as net.AddressInfo).port, path: "/exec/abc/start" };

// docker-modem passes an empty callback here and attaches 'response'
// separately via req.on('response', ...).
const req = http.request(
{
...requestOpts,
method: "POST",
headers: {
"Content-Type": "application/json",
"Transfer-Encoding": "chunked",
},
},
function () {},
);
try {
const responseEnd = new Promise<void>((resolve, reject) => {
req.on("error", reject);
req.on("response", res => {
events.push(`response-status:${res.statusCode}`);
res.setEncoding("utf8");
// Only a complete line counts as a received chunk: a data event
// can carry part of a line.
let pending = "";
res.on("data", (chunk: string) => {
pending += chunk;
const lines = pending.split("\n");
pending = lines.pop()!;
for (const line of lines) {
events.push(`recv:${line}`);
chunkReceived.resolve();
}
});
Comment thread
robobun marked this conversation as resolved.
Resolved
res.on("end", () => {
events.push("response-end");
resolve();
});
});
});

// Single write, no req.end(). docker-modem does exactly this for
// openStdin: true.
req.write(payload);
await responseEnd;

// The whole response arrived while the request body stream is still open.
expect({ events, writableEnded: req.writableEnded }).toEqual({
events: [
"request-seen",
"response-status:200",
"recv:chunk-0",
"recv:chunk-1",
"recv:chunk-2",
"response-end",
],
writableEnded: false,
});
expect(request).toStartWith("POST /exec/abc/start HTTP/1.1\r\n");
expect(request.toLowerCase()).toContain("transfer-encoding: chunked\r\n");
expect(request).toEndWith(chunkedFrame);
} finally {
req.destroy();
}
},
function () {},
);

req.on("response", (res) => {
console.log("response-status:" + res.statusCode);
res.setEncoding("utf8");
res.on("data", (chunk) => {
for (const line of chunk.split("\\n")) {
if (line) console.log("recv:" + line);
}
});
res.on("end", () => {
console.log("response-end");
server.close();
// The request body stream is still open; end it now so the process
// can exit cleanly.
req.end();
});
});

req.on("error", (err) => {
console.error("request-error:" + err.message);
process.exit(1);
});

// Single write, no req.end(). docker-modem does exactly this for
// openStdin: true.
req.write(JSON.stringify({ Detach: false, Tty: true }));
});

// Exit with whatever we've collected so far if the response never arrives,
// so the parent test gets a clean assertion failure instead of a timeout.
setTimeout(() => process.exit(0), 3000).unref();
`;

for (const socketMode of ["tcp", "unix"] as const) {
test(`http.request delivers response while request body stream is still open (${socketMode})`, async () => {
using dir = tempDir("issue-13696", {});
const socketPath = socketMode === "unix" ? join(String(dir), "docker.sock") : undefined;

await using proc = Bun.spawn({
cmd: [bunExe(), "-e", fixture(socketPath)],
env: bunEnv,
stdout: "pipe",
stderr: "pipe",
});

const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);

const lines = stdout.trim().split("\n");
// The server must have received the request (a single write() dispatches
// it), the response must be emitted with status 200, every body chunk must
// be delivered, and the response must end cleanly.
expect({ lines, stderr }).toEqual({
lines: ["request-seen", "response-status:200", "recv:chunk-0", "recv:chunk-1", "recv:chunk-2", "response-end"],
stderr: expect.not.stringContaining("request-error"),
});
expect(exitCode).toBe(0);
});
}

// Also cover the case where flushHeaders() is called explicitly (which already
// started the fetch in duplex mode) but the response was still being held back
// until req.end().
test("http.request emits 'response' in duplex mode after flushHeaders() without end()", async () => {
const src = `
const net = require("net");
const http = require("http");

const server = net.createServer((sock) => {
let buf = "";
let responded = false;
sock.on("data", (d) => {
buf += d.toString("latin1");
if (responded || !buf.includes("\\r\\n\\r\\n")) return;
responded = true;
sock.write("HTTP/1.1 200 OK\\r\\nContent-Type: text/plain\\r\\n\\r\\nhello");
test.concurrent("http.request emits 'response' in duplex mode after flushHeaders() without end()", async () => {
let request = "";
await using server = net.createServer(async sock => {
sock.on("error", () => {});
request = await readUntil(sock, "\r\n\r\n");
sock.write("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nhello");
sock.end();
});
});
server.listen(0, "127.0.0.1");
await once(server, "listening");
const { port } = server.address() as net.AddressInfo;

server.listen(0, "127.0.0.1", () => {
const { port } = server.address();
const req = http.request({
host: "127.0.0.1",
port,
path: "/",
method: "POST",
headers: { "Transfer-Encoding": "chunked" },
});
req.on("response", (res) => {
let body = "";
res.setEncoding("utf8");
res.on("data", (c) => (body += c));
res.on("end", () => {
console.log("body:" + body);
server.close();
req.end();
try {
const body = new Promise<string>((resolve, reject) => {
req.on("error", reject);
req.on("response", res => {
let body = "";
res.setEncoding("utf8");
res.on("data", (c: string) => (body += c));
res.on("end", () => resolve(body));
});
});
});
req.on("error", (err) => {
console.error("request-error:" + err.message);
process.exit(1);
});
req.flushHeaders();
req.write("payload");
});

setTimeout(() => process.exit(0), 3000).unref();
`;

await using proc = Bun.spawn({
cmd: [bunExe(), "-e", src],
env: bunEnv,
stdout: "pipe",
stderr: "pipe",
});

const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);

expect({ stdout: stdout.trim(), stderr }).toEqual({
stdout: "body:hello",
stderr: expect.not.stringContaining("request-error"),
});
expect(exitCode).toBe(0);
req.flushHeaders();
req.write("payload");

expect({ body: await body, writableEnded: req.writableEnded }).toEqual({ body: "hello", writableEnded: false });
expect(request).toStartWith("POST / HTTP/1.1\r\n");
} finally {
req.destroy();
}
});