fix(gallery): bound and reclaim storage reads in the remaining zip builders (#1410)

* fix(gallery): bound and reclaim storage reads in the remaining zip builders

PR 1402 fixes the cached-zip builder. The same unguarded pattern — one storage
read appended per photo, archiver draining one at a time, nothing destroying
the rest — is still present in three other places, two of which a gallery guest
reaches with no admin credentials:

- routes/gallery/downloads.js, download-all and download-selected
- services/downloadJobService.js, the custom-resolution job builder, which is
  worse in one respect: its per-photo catch skips a bad source without ever
  destroying the stream it had already opened, so every skipped photo leaked a
  socket for the life of the process

An unread S3 response body holds its socket open indefinitely — the SDK arms
its socket timeout on a 3s delay and clears it the moment response headers
land, so a fast response never gets one — and archiver's abort() does not touch
its source streams. That is the mechanism behind the incident described in PR
1402: 43 of 50 pooled sockets held with unread bytes, uploads and gallery reads
starved behind them, a process restart the only way out.

utils/archiveStreamGuard.js caps reads in flight at 2 and destroys whatever is
still open on every exit: an error, a failed append, and — for the two guest
routes — the client closing the tab mid-download, which previously left every
appended-but-undrained read parked forever.

Deliberately does not touch downloadZipService.js, so there is no conflict with
1402. Once that lands, its inline equivalent can move onto this helper.

Local-filesystem installs are unaffected either way: they take archiver's
archive.file(path) branch and open no sockets.

Relates to issue 1399

* fix(gallery): survive a cancelled download and a read that dies while queued

Two failures found reviewing the previous commit, both reachable on an
ordinary download.

A client hanging up mid-download aborted the archive, but the append loop's
`break` still fell through to archive.finalize(), which rejects with ABORTED.
The catch then called errorResponse over a response whose ZIP headers had
already gone out, throwing ERR_HTTP_HEADERS_SENT from an async Express 4
handler with nothing to catch it — an unhandled rejection, on a cancelled
download, which can take the process down. Both bulk routes now return without
finalizing, and the catch stays quiet once headers are sent.

A read that errored while still QUEUED behind another was absorbed by the
guard's own error listener. archiver had not attached its source listener yet,
so it never learned the stream had died, and the dead stream stayed in the
queue: the archive hung when it reached it, and in downloadJobService the build
held its concurrency slot with it. The guard now reports such a failure through
onFatalError so the caller aborts the archive and reclaims the rest. Streams the
guard destroyed itself are deliberately not reported — that is the caller's own
teardown, already running.

Relates to issue 1399

---------

Co-authored-by: Paul Nothaft <[email protected]>

backend/__tests__/utils/archiveStreamGuard.test.js

@@ -0,0 +1,122 @@
+/**
+ * Bounded, reclaimable storage reads for archiver downloads (#1399 follow-up).
+ *
+ * archiver drains the sources it is handed one at a time, so appending a
+ * storage read per photo opens N and drains one. Every other read parks its
+ * socket holding unread bytes, and nothing reclaims them: archiver's abort()
+ * does not touch source streams, and the S3 SDK clears its socket timeout as
+ * soon as response headers land. That is the mechanism behind the incident in
+ * PR #1402 — 43 of 50 pooled sockets held, uploads starved, restart required.
+ *
+ * #1402 fixes the cached-zip builder. These are the guarantees the same guard
+ * has to give the three remaining call sites, two of which need no admin
+ * credentials to reach.
+ */
+const { Readable } = require('stream');
+const { createArchiveStreamGuard } = require('../../src/utils/archiveStreamGuard');
+
+const makeStream = () => new Readable({ read() {} });
+
+describe('archiveStreamGuard (#1399 follow-up)', () => {
+  it('lets the configured number of reads run at once', async () => {
+    const guard = createArchiveStreamGuard({ maxInFlight: 2 });
+    expect(await guard.acquire()).toBe(true);
+    guard.track(makeStream());
+    expect(await guard.acquire()).toBe(true);
+    guard.track(makeStream());
+    expect(guard.openCount).toBe(2);
+  });
+
+  it('parks the next acquire until a read finishes', async () => {
+    const guard = createArchiveStreamGuard({ maxInFlight: 1 });
+    await guard.acquire();
+    const first = guard.track(makeStream());
+
+    let resumed = false;
+    const pending = guard.acquire().then((ok) => { resumed = ok; });
+
+    await new Promise((r) => setImmediate(r));
+    expect(resumed).toBe(false); // still parked — this is the cap doing its job
+
+    first.push(null);
+    first.resume();
+    await pending;
+    expect(resumed).toBe(true);
+  });
+
+  it('releases a slot when a read errors, not just when it ends', async () => {
+    const guard = createArchiveStreamGuard({ maxInFlight: 1 });
+    await guard.acquire();
+    const stream = guard.track(makeStream());
+    stream.on('error', () => {});
+    stream.destroy(new Error('socket died'));
+    // Without the error listener the slot would never come back and the next
+    // photo would park forever.
+    expect(await guard.acquire()).toBe(true);
+  });
+
+  it('reports a failed read so the caller can abort the archive', async () => {
+    // A stream that errors while still QUEUED has no archiver listener on it
+    // yet. Releasing its slot and saying nothing leaves a dead stream in the
+    // queue, and the archive hangs when it reaches it.
+    const seen = [];
+    const guard = createArchiveStreamGuard({ maxInFlight: 2, onFatalError: (e) => seen.push(e) });
+    await guard.acquire();
+    const queued = guard.track(makeStream());
+    queued.on('error', () => {});
+    queued.destroy(new Error('socket died'));
+    await new Promise((r) => setImmediate(r)); // 'error' lands on the next tick
+    expect(seen).toHaveLength(1);
+    expect(seen[0].message).toBe('socket died');
+  });
+
+  it('stays quiet about reads it destroyed itself', async () => {
+    // destroyAll is the caller's own teardown; reporting those back as fatal
+    // would re-enter the abort path it is already running.
+    const seen = [];
+    const guard = createArchiveStreamGuard({ onFatalError: (e) => seen.push(e) });
+    await guard.acquire();
+    const s1 = guard.track(makeStream());
+    s1.on('error', () => {});
+    guard.destroyAll();
+    await new Promise((r) => setImmediate(r));
+    expect(seen).toHaveLength(0);
+  });
+
+  it('destroys every read still holding bytes', async () => {
+    const guard = createArchiveStreamGuard({ maxInFlight: 5 });
+    const streams = [makeStream(), makeStream(), makeStream()];
+    for (const s of streams) { await guard.acquire(); guard.track(s); }
+    expect(guard.openCount).toBe(3);
+
+    guard.destroyAll();
+    expect(streams.every((s) => s.destroyed)).toBe(true);
+    expect(guard.openCount).toBe(0);
+  });
+
+  it('wakes a parked acquire on destroyAll so the loop can exit', async () => {
+    const guard = createArchiveStreamGuard({ maxInFlight: 1 });
+    await guard.acquire();
+    guard.track(makeStream());
+
+    const pending = guard.acquire();
+    guard.destroyAll();
+    // false, so the caller breaks out instead of appending to a dead archive.
+    expect(await pending).toBe(false);
+  });
+
+  it('destroys a stream tracked after shutdown rather than leaking it', () => {
+    const guard = createArchiveStreamGuard();
+    guard.destroyAll();
+    const late = guard.track(makeStream());
+    expect(late.destroyed).toBe(true);
+    expect(guard.openCount).toBe(0);
+  });
+
+  it('tolerates destroyAll twice — exit paths overlap', () => {
+    const guard = createArchiveStreamGuard();
+    guard.track(makeStream());
+    guard.destroyAll();
+    expect(() => guard.destroyAll()).not.toThrow();
+  });
+});

backend/src/routes/gallery/downloads.js

@@ -29,6 +29,7 @@ const {
 } = require('../../services/downloadFilenameService');
 const { buildContentDisposition } = require('../../utils/filenameSanitizer');
 const { getStorage } = require('../../services/storage');
+const { createArchiveStreamGuard } = require('../../utils/archiveStreamGuard');
 const fs = require('fs');
 function parseByteRange(header, size) {
   if (!header || typeof header !== 'string' || !size) return null;
@@ -401,6 +402,13 @@ async function bumpEventDownloadCounts(eventId) {
 }
 
 router.get('/:slug/download-all', verifyGalleryAccess, denySlideshowToken, blockHiddenGallery, async (req, res) => {
+  // Hoisted so the catch can reclaim reads opened before the failure.
+  let guard = null;
+  // The client hung up. An aborted archive rejects finalize() with ABORTED,
+  // and the catch would then try to send JSON over a response whose ZIP
+  // headers already went out — ERR_HTTP_HEADERS_SENT, unhandled, on an
+  // ordinary cancelled download.
+  let cancelled = false;
   try {
     // Check if downloads are allowed for this event
     if (!parseBooleanInput(req.event.allow_downloads, true)) {
@@ -497,6 +505,23 @@ router.get('/:slug/download-all', verifyGalleryAccess, denySlideshowToken, block
       throw err;
     });
 
+    // Reclaim storage reads on every exit (#1399 follow-up). A guest closing
+    // the tab mid-download used to leave every appended-but-undrained read
+    // parked on its socket for the life of the process.
+    guard = createArchiveStreamGuard({
+      // A queued read that dies takes the archive with it: archiver has no
+      // listener on it yet, so it would otherwise sit in the queue and stall
+      // the download forever.
+      onFatalError: () => { cancelled = true; guard.destroyAll(); archive.abort(); },
+    });
+    res.on('close', () => {
+      if (!res.writableFinished) {
+        cancelled = true;
+        guard.destroyAll();
+        archive.abort();
+      }
+    });
+
     archive.pipe(res);
 
     // Get watermark settings - apply if global setting OR event-level setting is enabled
@@ -562,8 +587,9 @@ router.get('/:slug/download-all', verifyGalleryAccess, denySlideshowToken, block
         if (rendered) {
           archive.append(rendered, { name: archiveName });
         } else if (storageKey) {
+          if (!await guard.acquire()) break;
           const stream = await storage.get(storageKey);
-          archive.append(stream, { name: archiveName });
+          archive.append(guard.track(stream), { name: archiveName });
         } else {
           archive.file(resolvePhotoFilePath(req.event, photo), { name: archiveName });
         }
@@ -587,6 +613,7 @@ router.get('/:slug/download-all', verifyGalleryAccess, denySlideshowToken, block
         if (res.statusCode < 400) logActivity('gallery_downloaded', { scope: 'all' }, req.event.id, galleryActor(req));
       });
     }
+    if (cancelled) return;
     await archive.finalize();
 
     if (!req.isAdminPreview) {
@@ -605,12 +632,18 @@ router.get('/:slug/download-all', verifyGalleryAccess, denySlideshowToken, block
       }
     }
   } catch (error) {
+    if (guard) guard.destroyAll();
+    // Nothing to say to a client that already left, and the headers are gone.
+    if (cancelled || res.headersSent) return;
     errorResponse(res, error, 500, 'Failed to create download archive');
   }
 });
 
 // Download selected photos as ZIP
 router.post('/:slug/download-selected', verifyGalleryAccess, denySlideshowToken, blockHiddenGallery, async (req, res) => {
+  // Hoisted so the catch can reclaim reads opened before the failure.
+  let selectedGuard = null;
+  let selectedCancelled = false;
   try {
     // Check if downloads are allowed for this event
     if (!parseBooleanInput(req.event.allow_downloads, true)) {
@@ -679,6 +712,20 @@ router.post('/:slug/download-selected', verifyGalleryAccess, denySlideshowToken,
         // ignore double-send errors
       }
     });
+
+    // Same reclaim contract as download-all above (#1399 follow-up).
+    selectedGuard = createArchiveStreamGuard({
+      onFatalError: () => { selectedCancelled = true; selectedGuard.destroyAll(); archive.abort(); },
+    });
+    archive.on('error', () => selectedGuard.destroyAll());
+    res.on('close', () => {
+      if (!res.writableFinished) {
+        selectedCancelled = true;
+        selectedGuard.destroyAll();
+        archive.abort();
+      }
+    });
+
     archive.pipe(res);
 
     // Check watermark settings - apply if global setting OR event-level setting is enabled
@@ -723,8 +770,9 @@ router.post('/:slug/download-selected', verifyGalleryAccess, denySlideshowToken,
         if (rendered) {
           archive.append(rendered, { name });
         } else if (storageKey) {
+          if (!await selectedGuard.acquire()) break;
           const stream = await selectedStorage.get(storageKey);
-          archive.append(stream, { name });
+          archive.append(selectedGuard.track(stream), { name });
         } else {
           archive.file(resolvePhotoFilePath(req.event, photo), { name });
         }
@@ -746,6 +794,7 @@ router.post('/:slug/download-selected', verifyGalleryAccess, denySlideshowToken,
         if (res.statusCode < 400) logActivity('gallery_downloaded', { scope: 'selected', photo_count: photoIds.length }, req.event.id, galleryActor(req));
       });
     }
+    if (selectedCancelled) return;
     await archive.finalize();
 
     if (!req.isAdminPreview) {
@@ -763,6 +812,8 @@ router.post('/:slug/download-selected', verifyGalleryAccess, denySlideshowToken,
       }
     }
   } catch (error) {
+    if (selectedGuard) selectedGuard.destroyAll();
+    if (selectedCancelled || res.headersSent) return;
     errorResponse(res, error, 500, 'Failed to download selected photos');
   }
 });

backend/src/services/downloadJobService.js

@@ -28,6 +28,7 @@ const crypto = require('crypto');
 const archiver = require('archiver');
 const { db } = require('../database/db');
 const { getStorage } = require('./storage');
+const { createArchiveStreamGuard } = require('../utils/archiveStreamGuard');
 const { getUseOriginalFilenames, getZipEntryNames } = require('./downloadFilenameService');
 const { renderPhotoForDownload, resolveWatermarkSettings } = require('./downloadRendition');
 const { resolvePhotoStorageKey, resolvePhotoFilePath } = require('./photoResolver');
@@ -297,8 +298,17 @@ class DownloadJobService {
         const output = fs.createWriteStream(tmpPath);
         // level 0 — photos are already compressed, so deflate only burns CPU.
         const archive = archiver('zip', { zlib: { level: 0 } });
+        // Bound and reclaim the storage reads (#1399 follow-up). The per-photo
+        // catch below deliberately skips a bad source, but it never destroyed
+        // the stream it had already opened, so every skipped photo leaked a
+        // socket for the life of the process.
+        // A queued read that dies would otherwise stall the archive and hold
+        // its slot for the life of the build, so it fails the job instead.
+        const guard = createArchiveStreamGuard({
+          onFatalError: (err) => { guard.destroyAll(); archive.abort(); reject(err); },
+        });
         output.on('close', resolve);
-        archive.on('error', reject);
+        archive.on('error', (err) => { guard.destroyAll(); reject(err); });
         archive.pipe(output);
 
         (async () => {
@@ -312,7 +322,8 @@ class DownloadJobService {
               } else {
                 const key = resolvePhotoStorageKey(event, photo);
                 if (key) {
-                  archive.append(await storage.get(key), { name });
+                  if (!await guard.acquire()) break;
+                  archive.append(guard.track(await storage.get(key)), { name });
                 } else {
                   archive.file(resolvePhotoFilePath(event, photo), { name });
                 }
@@ -331,7 +342,7 @@ class DownloadJobService {
             }
           }
           archive.finalize();
-        })().catch(reject);
+        })().catch((err) => { guard.destroyAll(); reject(err); });
       });
 
       if (appended === 0) {

backend/src/utils/archiveStreamGuard.js

@@ -0,0 +1,100 @@
+/**
+ * Bounded, reclaimable storage reads for archiver-based downloads.
+ *
+ * archiver consumes the sources it is handed one at a time. Appending a
+ * storage read per photo in a tight loop therefore opens N reads and drains
+ * one, and every other one parks an S3 socket holding megabytes of unread
+ * body. Nothing reclaims them on its own: archiver's abort() does not touch
+ * its source streams, and the SDK arms its socket timeout on a 3s delay and
+ * clears it the moment response headers land, so a fast response never gets
+ * one at all.
+ *
+ * That is the shape of the incident in PR #1402 — 43 of 50 pooled sockets
+ * ESTABLISHED with unread bytes, uploads and gallery reads starved behind
+ * them, a process restart the only way out. #1402 fixes the cached-zip
+ * builder. This is the same guard for the other three call sites, two of
+ * which a gallery guest can reach with no admin credentials at all.
+ *
+ * Local-filesystem installs are unaffected — they take archiver's
+ * `archive.file(path)` branch and open no sockets — which is most likely why
+ * this went unnoticed for so long.
+ */
+
+// Two in flight: one being drained, one ready to go. Enough to keep archiver
+// fed, few enough that a build cannot monopolise the agent pool.
+const DEFAULT_MAX_IN_FLIGHT = 2;
+
+function createArchiveStreamGuard({ maxInFlight = DEFAULT_MAX_IN_FLIGHT, onFatalError } = {}) {
+  const openReads = new Set();
+  let waiter = null;
+  let closed = false;
+
+  const wake = () => {
+    if (!waiter) return;
+    const resume = waiter;
+    waiter = null;
+    resume();
+  };
+
+  const release = (stream) => {
+    openReads.delete(stream);
+    wake();
+  };
+
+  return {
+    /** Park until a read slot frees up. Returns false once destroyAll ran. */
+    async acquire() {
+      while (!closed && openReads.size >= maxInFlight) {
+        await new Promise((resolve) => { waiter = resolve; });
+      }
+      return !closed;
+    },
+
+    /** Register a stream and hand it straight back, for inline use. */
+    track(stream) {
+      if (closed) {
+        stream.destroy();
+        return stream;
+      }
+      openReads.add(stream);
+      stream.once('end', () => release(stream));
+      stream.once('close', () => release(stream));
+      stream.once('error', (err) => {
+        release(stream);
+        // A stream that errors while still QUEUED behind another has no
+        // archiver listener on it yet, so archiver never learns it failed.
+        // Absorbing the error here and leaving the dead stream in the queue
+        // makes the archive hang forever when it reaches it — and in
+        // downloadJobService the build keeps its slot with it. Hand the
+        // failure to the caller, which aborts the archive.
+        if (!closed && typeof onFatalError === 'function') {
+          onFatalError(err);
+        }
+      });
+      return stream;
+    },
+
+    /**
+     * Destroy every read still holding bytes. Safe to call more than once —
+     * the exit paths overlap (client disconnect and an error can both fire).
+     */
+    destroyAll() {
+      closed = true;
+      for (const stream of openReads) {
+        try {
+          stream.destroy();
+        } catch {
+          // Already gone; nothing to reclaim.
+        }
+      }
+      openReads.clear();
+      wake();
+    },
+
+    get openCount() {
+      return openReads.size;
+    },
+  };
+}
+
+module.exports = { createArchiveStreamGuard, DEFAULT_MAX_IN_FLIGHT };