fix: keep ZIP downloads flowing through compression (#872)

docs/performance-plan.md

@@ -55,9 +55,11 @@ The ninth review follow-up registers renewable staged-tree leases before inserti
 
 The tenth review follow-up makes conference removal attempt every repository sequentially before reporting all failures. Legacy repository, gist and pull-request lists retain one bulk expiration update and then reload affected documents, preserving owner extensions, current statuses and options, and omitting concurrently deleted artifacts. Four new regressions cover multiple removal failures and concurrent list changes across all three artifact types.
 
+A production rollout of PR #866 on 8 October 2026 exposed a ZIP transport regression: Express compression does not forward response write callbacks, so downloads stalled after the initial ZIP signature. Application containers were rolled back; the additive indexes remain installed. The follow-up uses the write return value and drain events for bounded backpressure, and removes listeners from the actual emitter used by compression. Two real HTTP regressions cover plain and gzip responses with forced backpressure and chunked upstream data. After rollback, the public example ZIP downloaded in 3.0 seconds and all 699 entries passed integrity checks.
+
 ## Reproduce and deploy
 
-Run `npm run build`, `npm run lint`, and `npm test`. On 8 October 2026, a clean committed checkout passed build, lint and the full suite: 948 passing, 51 pending, and no failures, with disposable MongoDB and Redis enabled. The streamer startup, lifecycle and idle-cleanup tests used compiled application code in separate processes. This run used temporary memory-backed database and spool storage after local disk I/O stalls caused integration connection and large-file deadline failures. Lint reports seven existing warnings. The full suite includes the dashboard UI regressions; both focused rename tests and five status-polling UI tests also passed. An earlier working-tree run included unrelated untracked runtime tests with two pre-existing missing-export failures; those files are excluded from this PR. The clean run includes the final generation guards for file counts and readiness.
+Run `npm run build`, `npm run lint`, and `npm test`. On 8 October 2026, a clean committed checkout passed build, lint and the full suite: 950 passing, 51 pending, and no failures, with disposable MongoDB and Redis enabled. The streamer startup, lifecycle and idle-cleanup tests used compiled application code in separate processes. This run used temporary memory-backed database and spool storage after local disk I/O stalls caused integration connection and large-file deadline failures. Lint reports seven existing warnings. The full suite includes the dashboard UI regressions; both focused rename tests and five status-polling UI tests also passed. An earlier working-tree run included unrelated untracked runtime tests with two pre-existing missing-export failures; those files are excluded from this PR. The clean run includes the final generation guards for file counts and readiness.
 
 The full run includes all 122 focused performance tests with disposable MongoDB and Redis. Run them separately with `npm run test:performance`. To include database and cross-process lease checks, set `PERFORMANCE_MONGO_URI` to a disposable MongoDB database named `perf_test_*`, and `PERFORMANCE_REDIS_PORT` to a disposable Redis port. The MongoDB suite drops its test database afterward. The S3 tests use a local HTTP protocol fixture by default; set `PERFORMANCE_S3_ENDPOINT` to test a disposable compatible service instead.
 

docs/performance-results.json

@@ -108,10 +108,10 @@
   ],
   "tests": {
     "fullSuite": {
-      "passing": 948,
+      "passing": 950,
       "pending": 51,
       "failing": 0,
-      "note": "Clean committed checkout after ten PR review follow-ups, with disposable MongoDB/Redis and memory-backed temporary storage. Streamer startup/lifecycle/idle-cleanup integration uses compiled application code in separate processes; unrelated workspace files excluded. Includes bulk expiration races across repositories, gists and pull requests, and conference removal after multiple storage failures.",
+      "note": "Clean committed checkout after the ZIP HTTP rollout follow-up, with disposable MongoDB/Redis and memory-backed temporary storage. Compiled streamer startup/lifecycle/idle-cleanup integration remains enabled. Two real HTTP tests cover Express compression, plain and gzip output, forced backpressure, listener cleanup and chunked upstream archives.",
       "validationDate": "2026-10-08"
     },
     "performanceSuite": {
@@ -147,6 +147,12 @@
       "idleReplicaCacheAndSpoolCleanup": "passed",
       "startupFailureExitCode": 1,
       "scope": "Disposable MongoDB and local filesystem; controlled download/publication fixtures. Pool settings are asserted, not production socket measurements."
+    },
+    "zipHttp": {
+      "plain": "passed",
+      "gzip": "passed",
+      "forcedBackpressure": "passed",
+      "drainListenerCleanup": "passed"
     }
   },
   "sockets": {

src/core/zipStream.ts

@@ -119,19 +119,31 @@ export async function streamAnonymizedZip(
   const compiledTerms = compileTerms(opt.anonymizerOptions.terms || []);
   let stopped = false;
   let entryWaiting = false;
+  let cleanupDrain: (() => void) | undefined;
   const resumeParser = () => {
     if (!stopped && entryWaiting && archive.readableLength === 0 && output.writableLength === 0) {
       entryWaiting = false;
       parser.resume();
     }
   };
-  // Wait for response writes to complete before admitting the next entry.
+  // Middleware can omit write callbacks. Honor the writable backpressure
+  // contract instead so queued output stays bounded.
   const output = new Writable({
     write(chunk, encoding, callback) {
-      res.write(chunk, encoding, (error?: Error | null) => {
-        callback(error);
+      const written = () => {
+        cleanupDrain?.();
+        cleanupDrain = undefined;
+        callback();
         setImmediate(resumeParser);
-      });
+      };
+      if (res.write(chunk, encoding)) written();
+      else {
+        // Compression forwards drain subscriptions to its underlying stream.
+        // Remove the listener from the emitter returned by on(), rather than
+        // accumulating wrappers on that stream through response.once().
+        const emitter = res.on("drain", written) as NodeJS.EventEmitter;
+        cleanupDrain = () => emitter.removeListener("drain", written);
+      }
     },
     final(callback) { res.end(); callback(); },
   });
@@ -140,6 +152,8 @@ export async function streamAnonymizedZip(
   const stop = () => {
     if (stopped) return;
     stopped = true;
+    cleanupDrain?.();
+    cleanupDrain = undefined;
     downloadStream.destroy();
     parser.destroy();
     for (const stream of activeStreams) stream.destroy();

test/large-notebook-download.test.js

@@ -1,6 +1,11 @@
 const { expect } = require("chai");
 const { Readable, PassThrough } = require("stream");
 const { once } = require("events");
+const http = require("node:http");
+const express = require("express");
+const compression = require("compression");
+const { randomBytes } = require("node:crypto");
+const { gunzipSync } = require("node:zlib");
 const archiver = require("archiver");
 const { Parse } = require("unzip-stream");
 require("ts-node/register/transpile-only");
@@ -81,17 +86,69 @@ describe("large notebook downloads", function () {
       GitHubDownload.prototype.getZipUrl = async () => ({ url: "https://example.invalid/source.zip" });
       got.stream = () => upstream;
     });
-    afterEach(function () {
+    afterEach(async function () {
       got.stream = originalStream;
       GitHubDownload.prototype.getZipUrl = originalUrl;
       config.MAX_FILE_SIZE = originalLimit;
       upstream?.destroy();
+      await require("../src/core/cache-coordination").closeCacheRedis();
     });
     const options = {
       repoId: "fixture", organization: "owner", repoName: "repo", commit: "HEAD",
       getToken: () => "test", anonymizerOptions: { filePath: "", terms: ["Alice"] },
     };
 
+    for (const compressed of [false, true]) {
+      it(`finishes ZIP responses through Express compression with gzip ${compressed}`, async function () {
+        const text = "Alice Σ 研究\n".repeat(25000), binary = randomBytes(128 * 1024);
+        const input = await zip({ "root/empty.txt": "", "root/large.txt": text, "root/image.png": binary, "root/after.txt": "Alice" });
+        upstream = Readable.from((async function* () {
+          for (let offset = 0; offset < input.length; offset += 1021) {
+            yield input.subarray(offset, offset + 1021);
+          }
+        })());
+        const app = express();
+        app.use(compression({ threshold: 0, highWaterMark: 1024, filter: () => compressed }));
+        const drainEmitters = new Set();
+        let drainSubscriptions = 0;
+        app.get("/archive", async (_req, res) => {
+          const on = res.on;
+          res.on = function (event, listener) {
+            const emitter = on.call(this, event, listener);
+            if (event === "drain") { drainSubscriptions++; drainEmitters.add(emitter); }
+            return emitter;
+          };
+          res.type("application/zip");
+          await streamAnonymizedZip(options, res);
+        });
+        const server = await new Promise(resolve => {
+          const listening = http.createServer({ highWaterMark: 1024 }, app).listen(0, "127.0.0.1", () => resolve(listening));
+        });
+        let client;
+        try {
+          const response = await new Promise((resolve, reject) => {
+            client = http.get(`http://127.0.0.1:${server.address().port}/archive`, { headers: { "Accept-Encoding": "gzip" } }, resolve);
+            client.setTimeout(8000, () => client.destroy(new Error("ZIP response stalled")));
+            client.on("error", reject);
+          });
+          const bytes = await collect(response);
+          expect(response.statusCode).to.equal(200);
+          expect(response.headers["content-encoding"]).to.equal(compressed ? "gzip" : undefined);
+          const entries = await unzip(compressed ? gunzipSync(bytes) : bytes);
+          expect(Object.keys(entries)).to.have.members(["empty.txt", "large.txt", "image.png", "after.txt"]);
+          expect(entries["empty.txt"]).to.have.length(0);
+          expect(entries["large.txt"].toString()).to.equal(new ContentAnonimizer(options.anonymizerOptions).anonymize(text));
+          expect(entries["image.png"]).to.deep.equal(binary);
+          expect(entries["after.txt"].toString()).to.equal("XXXX-1");
+          expect(drainSubscriptions).to.be.greaterThan(0);
+          for (const emitter of drainEmitters) expect(emitter.listenerCount("drain")).to.equal(0);
+        } finally {
+          client?.destroy(); server.closeAllConnections();
+          await new Promise(resolve => server.close(resolve));
+        }
+      });
+    }
+
     it("completes an archive containing a large notebook and subsequent entries", async function () {
       upstream = Readable.from([await zip({ "root/example.ipynb": notebook(), "root/after.txt": "Alice" })]);
       const response = new PassThrough();