fix: bound ZIP directory memory for slow downloads

src/core/bounded-zip.ts

@@ -0,0 +1,109 @@
+import archiver = require("archiver");
+import { once } from "events";
+import { createReadStream, closeSync, mkdtempSync, openSync, rmSync, writeSync, utimesSync, ReadStream } from "fs";
+import { setInterval, clearInterval } from "timers";
+import { tmpdir } from "os";
+import { join } from "path";
+
+// archiver 7 / zip-stream 6 / compress-commons 6 retain all directory entries
+// and synchronously emit millions of buffers at finalize(). Keep their ZIP
+// encoding (including ZIP64), but spool directory records as entries finish.
+interface ZipEngine extends NodeJS.WritableStream {
+  offset: number;
+  _entries: { length: number; push(entry: unknown): number };
+  _archive: { centralOffset: number; centralLength: number; processing: boolean; finish: boolean; finished: boolean };
+  _finish(): void;
+  _writeCentralFileHeader(entry: unknown): void;
+  _writeCentralDirectoryZip64(): void;
+  _writeCentralDirectoryEnd(): void;
+  isZip64(): boolean;
+  emit(event: string, error: Error): boolean;
+}
+
+export function boundedZip() {
+  const archive = archiver("zip", {});
+  const engine = (archive as unknown as { _module: { engine: ZipEngine } })._module.engine;
+  if (!Array.isArray(engine._entries) || typeof engine._writeCentralFileHeader !== "function" ||
+      typeof engine._writeCentralDirectoryZip64 !== "function" || typeof engine._writeCentralDirectoryEnd !== "function") {
+    throw new Error("unsupported_zip_engine");
+  }
+  const cancellation = new AbortController();
+  let directory: string | undefined;
+  let fd: number | undefined;
+  let reader: ReadStream | undefined;
+  let heartbeat: NodeJS.Timeout | undefined;
+  let count = 0;
+  let finalizing = false;
+  const cleanup = () => {
+    clearInterval(heartbeat); heartbeat = undefined;
+    reader?.destroy();
+    reader = undefined;
+    if (fd !== undefined) { closeSync(fd); fd = undefined; }
+    if (directory) { rmSync(directory, { recursive: true, force: true }); directory = undefined; }
+  };
+  const stop = () => { cancellation.abort(); cleanup(); };
+  const abort = archive.abort;
+  archive.abort = function () { stop(); return abort.call(this); };
+  archive.once("error", stop);
+  archive.once("close", stop);
+  archive.once("end", cleanup);
+
+  engine._entries = {
+    get length() { return count; },
+    push(entry) {
+      try {
+        if (cancellation.signal.aborted) return count;
+        if (fd === undefined) {
+          directory = mkdtempSync(join(tmpdir(), "anonymous-zip-directory-"));
+          fd = openSync(join(directory, "central"), "wx", 0o600);
+          // A paused download can outlive the stale-spool cleanup threshold.
+          heartbeat = setInterval(() => {
+            if (directory) {
+              const now = new Date();
+              try { utimesSync(join(directory, "central"), now, now); } catch { /* writes/reads report I/O errors */ }
+            }
+          }, 60_000);
+          heartbeat.unref();
+        }
+        // A directory record is bounded by ZIP's 16-bit name/extra/comment
+        // lengths. Serialize only this record, without touching output offsets.
+        const chunks: Buffer[] = [];
+        const write = engine.write;
+        const offset = engine.offset;
+        try {
+          engine.write = (chunk: Uint8Array | string) => { chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); return true; };
+          engine._writeCentralFileHeader(entry);
+        } finally { engine.write = write; engine.offset = offset; }
+        const record = Buffer.concat(chunks);
+        let written = 0;
+        while (written < record.length) written += writeSync(fd, record, written, record.length - written);
+        return ++count;
+      } catch (error) { engine.emit("error", error as Error); return count; }
+    },
+  };
+  engine._finish = () => {
+    if (finalizing || cancellation.signal.aborted) return;
+    finalizing = true;
+    engine._archive.finish = true;
+    engine._archive.centralOffset = engine.offset;
+    void (async () => {
+      if (fd !== undefined) { closeSync(fd); fd = undefined; }
+      if (directory) {
+        reader = createReadStream(join(directory, "central"), { highWaterMark: 64 * 1024 });
+        for await (const chunk of reader) {
+          cancellation.signal.throwIfAborted();
+          if (!engine.write(chunk)) await once(engine, "drain", { signal: cancellation.signal });
+        }
+      }
+      cancellation.signal.throwIfAborted();
+      engine._archive.centralLength = engine.offset - engine._archive.centralOffset;
+      if (engine.isZip64()) engine._writeCentralDirectoryZip64();
+      engine._writeCentralDirectoryEnd();
+      engine._archive.processing = false;
+      engine._archive.finished = true;
+      engine.end();
+      cleanup();
+    })().catch(error => { if (!cancellation.signal.aborted) engine.emit("error", error); cleanup(); });
+  };
+  return archive;
+}

src/core/storage/FileSystem.ts

@@ -1,3 +1,4 @@
+import { boundedZip } from "../bounded-zip";
 import { appendArchiveEntry, zipEntryOptions } from "../archive-entry";
 import config from "../../config";
 import * as fs from "fs";
@@ -261,7 +262,7 @@ export default class FileSystem extends StorageBase {
     }
   ) {
     this.assertSafePath(dir);
-    const archive = archiver(opt?.format || "zip", {});
+    const archive = opt?.format === "tar" ? archiver("tar", {}) : boundedZip();
     let active: Readable | undefined;
     let stopped = false;
     archive.once("close", () => { stopped = true; active?.destroy(); });

src/core/storage/S3.ts

@@ -1,3 +1,4 @@
+import { boundedZip } from "../bounded-zip";
 import { appendArchiveEntry, zipEntryOptions } from "../archive-entry";
 import {
   GetObjectCommand,
@@ -371,7 +372,7 @@ export default class S3Storage extends StorageBase {
   ) {
     if (!config.S3_BUCKET) throw new Error("S3_BUCKET not set");
     this.assertSafePath(dir);
-    const archive = archiver(opt?.format || "zip", {});
+    const archive = opt?.format === "tar" ? archiver("tar", {}) : boundedZip();
     if (dir && dir[dir.length - 1] != "/") dir = dir + "/";
 
     let active: Readable | undefined;

src/core/temporary-storage.ts

@@ -13,10 +13,11 @@ let cleanup: Promise<void> | undefined;
 export async function removeStaleTextSpools(directory = tmpdir(), now = Date.now()) {
   const entries = await fs.opendir(directory);
   for await (const entry of entries) {
-    if (!entry.isDirectory() || !entry.name.startsWith("anonymous-text-")) continue;
+    if (!entry.isDirectory() || (!entry.name.startsWith("anonymous-text-") && !entry.name.startsWith("anonymous-zip-directory-"))) continue;
     const path = join(directory, entry.name);
     try {
-      if (now - (await fs.stat(path)).mtimeMs > 3600_000) await fs.rm(path, { recursive: true, force: true });
+      const agePath = entry.name.startsWith("anonymous-zip-directory-") ? join(path, "central") : path;
+      if (now - (await fs.stat(agePath)).mtimeMs > 3600_000) await fs.rm(path, { recursive: true, force: true });
     } catch { /* another process can finish cleanup first */ }
   }
 }

src/core/zipStream.ts

@@ -3,7 +3,7 @@ import config from "../config";
 import { MAX_SUBMODULE_ENTRIES, SubmoduleMount, safeRepositoryPath, validateMounts } from "./source/github-submodules";
 import { Readable, Transform, Writable } from "stream";
 import { Parse } from "unzip-stream";
-import archiver = require("archiver");
+import { boundedZip } from "./bounded-zip";
 
 import GitHubDownload from "./source/GitHubDownload";
 import { classifyGitHubMissError } from "./source/GitHubBase";
@@ -118,7 +118,7 @@ export async function streamAnonymizedZip(
     });
   }
   let downloadStream: Readable;
-  const archive = archiver("zip", {});
+  const archive = boundedZip();
   let parser: Transform;
   let childEntries = 0;
   const activeStreams = new Set<Readable>();