Refactor unzip functionality to limit concurrency and improve error handling in stream copying

Author: shati-patel

extensions/ql-vscode/src/common/unzip-concurrently.ts

@@ -9,7 +9,7 @@ export async function unzipToDirectoryConcurrently(
   progress?: UnzipProgressCallback,
 ): Promise<void> {
   const queue = new PQueue({
-    concurrency: availableParallelism(),
+    concurrency: Math.min(availableParallelism(), 4),
   });
 
   return unzipToDirectory(

extensions/ql-vscode/src/common/unzip.ts

@@ -2,6 +2,7 @@ import type { Entry as ZipEntry, Options as ZipOptions, ZipFile } from "yauzl";
 import { open } from "yauzl";
 import type { Readable } from "stream";
 import { Transform } from "stream";
+import { pipeline } from "stream/promises";
 import { dirname, join } from "path";
 import type { WriteStream } from "fs";
 import { createWriteStream, ensureDir } from "fs-extra";
@@ -88,31 +89,21 @@ export async function openZipBuffer(
   });
 }
 
-async function copyStream(
+export async function copyStream(
   readable: Readable,
   writeStream: WriteStream,
   bytesExtractedCallback?: (bytesExtracted: number) => void,
 ): Promise<void> {
-  return new Promise((resolve, reject) => {
-    readable.on("error", (err) => {
-      reject(err);
-    });
-    readable.on("end", () => {
-      resolve();
-    });
-
-    readable
-      .pipe(
-        new Transform({
-          transform(chunk, _encoding, callback) {
-            bytesExtractedCallback?.(chunk.length);
-            this.push(chunk);
-            callback();
-          },
-        }),
-      )
-      .pipe(writeStream);
-  });
+  await pipeline(
+    readable,
+    new Transform({
+      transform(chunk, _encoding, callback) {
+        bytesExtractedCallback?.(chunk.length);
+        callback(null, chunk);
+      },
+    }),
+    writeStream,
+  );
 }
 
 type UnzipProgress = {

extensions/ql-vscode/test/unit-tests/common/unzip.test.ts

@@ -1,10 +1,13 @@
 import { createHash } from "crypto";
 import { open } from "fs/promises";
 import { join, relative, resolve, sep } from "path";
+import { Readable } from "stream";
+import { createWriteStream } from "fs";
 import { chmod, pathExists, readdir } from "fs-extra";
 import type { DirectoryResult } from "tmp-promise";
 import { dir } from "tmp-promise";
 import {
+  copyStream,
   excludeDirectories,
   openZip,
   openZipBuffer,
@@ -276,3 +279,71 @@ async function computeHash(contents: Buffer) {
 
   return hash.digest("hex");
 }
+
+describe("copyStream error handling", () => {
+  let tmpDir: DirectoryResult;
+
+  beforeEach(async () => {
+    tmpDir = await dir({
+      unsafeCleanup: true,
+    });
+  });
+
+  afterEach(async () => {
+    await tmpDir?.cleanup();
+  });
+
+  it("rejects when the write stream errors mid-extraction", async () => {
+    // Use a real zip to trigger unzip, but make the destination read-only
+    // so the write stream fails. This verifies the promise rejects rather
+    // than hanging indefinitely.
+    const destPath = join(tmpDir.path, "output");
+
+    // Extract once to create the directory structure
+    await unzipToDirectorySequentially(zipPath, destPath);
+
+    // Make a file read-only so re-extraction will fail on write
+    const targetFile = join(destPath, "directory", "file.txt");
+    await chmod(targetFile, 0o000);
+
+    // On Windows, chmod doesn't prevent writes, so skip assertion there
+    if (process.platform === "win32") {
+      await chmod(targetFile, 0o644);
+      return;
+    }
+
+    // Re-extract — should reject with a write error, not hang
+    await expect(
+      unzipToDirectorySequentially(zipPath, destPath),
+    ).rejects.toThrow();
+
+    // Restore permissions for cleanup
+    await chmod(targetFile, 0o644);
+  });
+
+  it("rejects when the write stream is destroyed mid-copy", async () => {
+    // A destroyed write stream should cause `copyStream` to reject, not hang.
+    const destFile = join(tmpDir.path, "output.bin");
+    const writeStream = createWriteStream(destFile);
+
+    // A readable that emits a chunk and then destroys the write stream, to
+    // simulate a mid-copy write failure.
+    let pushed = false;
+    const readable = new Readable({
+      read() {
+        if (pushed) {
+          return;
+        }
+        pushed = true;
+        this.push(Buffer.alloc(1024, "x"));
+        setImmediate(() => {
+          writeStream.destroy(new Error("simulated write failure"));
+        });
+      },
+    });
+
+    await expect(copyStream(readable, writeStream)).rejects.toThrow(
+      "simulated write failure",
+    );
+  });
+});