Merge remote-tracking branch 'origin/main' into codex/test-bun-compatibility

Author: cristianrgreco

packages/testcontainers/src/container-runtime/clients/container/docker-container-client.test.ts

@@ -2,6 +2,29 @@ import { PassThrough, Readable } from "stream";
 import { DockerContainerClient } from "./docker-container-client";
 
 describe("DockerContainerClient", () => {
+  describe("logs", () => {
+    it("should destroy the Docker stream when the consumer closes the log stream", async () => {
+      const actualLogStream = new PassThrough();
+      const demuxStream = vi.fn();
+      const container = {
+        id: "container-id",
+        logs: vi.fn(async () => actualLogStream),
+      };
+      const dockerode = {
+        modem: { demuxStream },
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      } as any;
+      const client = new DockerContainerClient(dockerode);
+
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      const stream = await client.logs(container as any);
+      await vi.waitFor(() => expect(demuxStream).toHaveBeenCalledOnce());
+      stream.destroy();
+
+      await vi.waitFor(() => expect(actualLogStream.destroyed).toBe(true));
+    });
+  });
+
   describe("exec", () => {
     it("should not truncate output when the demuxed streams flush after the raw stream ends", async () => {
       const payload = "the-final-line-that-must-not-be-truncated\n";

packages/testcontainers/src/container-runtime/clients/container/docker-container-client.ts

@@ -194,6 +194,11 @@ export class DockerContainerClient implements ContainerClient {
         actualLogStream.socket?.unref();
 
         const demuxedStream = await this.demuxStream(container.id, actualLogStream);
+        if (proxyStream.destroyed) {
+          demuxedStream.destroy();
+          return;
+        }
+        proxyStream.once("close", () => demuxedStream.destroy());
         demuxedStream.pipe(proxyStream);
         demuxedStream.on("error", (err) => proxyStream.emit("error", err));
         demuxedStream.on("end", () => proxyStream.end());