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

backend/__tests__/utils/archiveStreamGuard.test.js

@@ -0,0 +1,94 @@
+/**
+ * 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('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,8 @@ 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;
   try {
     // Check if downloads are allowed for this event
     if (!parseBooleanInput(req.event.allow_downloads, true)) {
@@ -497,6 +500,17 @@ 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();
+    res.on('close', () => {
+      if (!res.writableFinished) {
+        guard.destroyAll();
+        archive.abort();
+      }
+    });
+
     archive.pipe(res);
 
     // Get watermark settings - apply if global setting OR event-level setting is enabled
@@ -562,8 +576,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 });
         }
@@ -605,12 +620,15 @@ router.get('/:slug/download-all', verifyGalleryAccess, denySlideshowToken, block
       }
     }
   } catch (error) {
+    if (guard) guard.destroyAll();
     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;
   try {
     // Check if downloads are allowed for this event
     if (!parseBooleanInput(req.event.allow_downloads, true)) {
@@ -679,6 +697,17 @@ router.post('/:slug/download-selected', verifyGalleryAccess, denySlideshowToken,
         // ignore double-send errors
       }
     });
+
+    // Same reclaim contract as download-all above (#1399 follow-up).
+    selectedGuard = createArchiveStreamGuard();
+    archive.on('error', () => selectedGuard.destroyAll());
+    res.on('close', () => {
+      if (!res.writableFinished) {
+        selectedGuard.destroyAll();
+        archive.abort();
+      }
+    });
+
     archive.pipe(res);
 
     // Check watermark settings - apply if global setting OR event-level setting is enabled
@@ -723,8 +752,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 });
         }
@@ -763,6 +793,7 @@ router.post('/:slug/download-selected', verifyGalleryAccess, denySlideshowToken,
       }
     }
   } catch (error) {
+    if (selectedGuard) selectedGuard.destroyAll();
     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,13 @@ 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.
+        const guard = createArchiveStreamGuard();
         output.on('close', resolve);
-        archive.on('error', reject);
+        archive.on('error', (err) => { guard.destroyAll(); reject(err); });
         archive.pipe(output);
 
         (async () => {
@@ -312,7 +318,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 +338,7 @@ class DownloadJobService {
             }
           }
           archive.finalize();
-        })().catch(reject);
+        })().catch((err) => { guard.destroyAll(); reject(err); });
       });
 
       if (appended === 0) {

backend/src/utils/archiveStreamGuard.js

@@ -0,0 +1,89 @@
+/**
+ * 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 } = {}) {
+  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', () => release(stream));
+      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 };