fix(gallery): stop the pre-zip build leaking storage reads (#1402)

Building the download-all archive opened one storage read per photo and
handed each stream to archiver, which uses them one at a time. Every read
past the one being written parked an S3 socket with a full receive buffer,
and both early exits walked away from all of them: archiver's abort() does
not touch its source streams, and the error path only removed the temp
directory. Nothing else reclaims those sockets either, because the SDK arms
its socket timeout on a 3s delay and clears it as soon as the response
headers land. On a live server 43 of the 50 pooled sockets ended up stuck
for days and photo uploads stopped completing, with nothing logged.

The common way in is an ordinary upload. invalidate() runs on every photo
upload, delete and bulk edit, and it aborted an in-flight build.

Track every open read and destroy them on every exit path, cancel the
in-flight build from invalidate() rather than waiting for the loop to reach
its next version check, and cap reads in flight at 2. A build of 120 photos
peaked at 50 concurrent GETs before, the whole agent pool, which starved
uploads and thumbnails on its own.

Only S3 deployments are affected. Local filesystem installs take the
archive.file() branch and open no sockets.

backend/__tests__/services/downloadZipSocketLeak.test.js

@@ -0,0 +1,191 @@
+/**
+ * A failed pre-zip build must not leave storage reads open.
+ *
+ * The builder opened one storage read per photo and handed the raw stream to
+ * archiver. archiver drains its queue one entry at a time, so on an S3 backend
+ * every photo beyond the one being written parked a socket with a full receive
+ * buffer, and the error path (a source stream dying, or a photo upload
+ * invalidating the build) walked away from all of them. archiver's abort()
+ * does not touch the source streams, and the AWS SDK arms its socket timeout
+ * on a 3s delay then clears it once the response headers arrive, so nothing
+ * ever reclaimed those sockets. On a live server 43 of the 50 pooled sockets
+ * ended up stuck for days and photo uploads stopped completing.
+ */
+
+const path = require('path');
+const fs = require('fs');
+const os = require('os');
+
+process.env.NODE_ENV = 'test';
+process.env.TEST_DATABASE_PATH = path.join(
+  fs.mkdtempSync(path.join(os.tmpdir(), 'picpeak-zipleak-')), 'db.sqlite',
+);
+process.env.JWT_SECRET = process.env.JWT_SECRET || 'zipleak-test-secret';
+process.env.STORAGE_PATH = fs.mkdtempSync(path.join(os.tmpdir(), 'picpeak-zipleak-storage-'));
+
+const { Readable } = require('stream');
+
+const PHOTO_COUNT = 6;
+const MAX_INFLIGHT_READS = 2;
+
+// One storage read. It never ends on its own, which is what a large photo
+// looks like to the builder: the bytes only move while archiver pulls them.
+class StoredObject extends Readable {
+  constructor(key, failAfterReads, chunks) {
+    super();
+    this.key = key;
+    this.failAfterReads = failAfterReads;
+    this.chunks = chunks;
+    this.reads = 0;
+  }
+
+  _read() {
+    this.reads += 1;
+    if (this.failAfterReads && this.reads > this.failAfterReads) {
+      // What a dropped connection to S3 looks like in Node.
+      this.destroy(new Error('aborted'));
+      return;
+    }
+    this.push(this.reads > this.chunks ? null : Buffer.alloc(4096, 1));
+  }
+}
+
+const reads = { opened: [], live: 0, peak: 0 };
+const failingKey = { value: null };
+const onOpen = { fn: null };
+// A read only finishes when the build pulls the whole object. Photos big
+// enough to matter never finish inside one archiver turn, and a stream that
+// ends on its own would be auto-destroyed and hide the leak.
+const objectChunks = { value: Number.POSITIVE_INFINITY };
+
+function openStoredObject(key) {
+  const stream = new StoredObject(key, key === failingKey.value ? 1 : 0, objectChunks.value);
+  reads.opened.push(stream);
+  reads.live += 1;
+  if (reads.live > reads.peak) reads.peak = reads.live;
+  let settled = false;
+  const settle = () => { if (!settled) { settled = true; reads.live -= 1; } };
+  stream.once('end', settle);
+  stream.once('close', settle);
+  if (onOpen.fn) onOpen.fn(reads.opened.length);
+  return stream;
+}
+
+const mockStorage = {
+  kind: () => 's3',
+  get: jest.fn(async (key) => openStoredObject(key)),
+  getToFile: jest.fn(async () => undefined),
+  putFromFile: jest.fn(async () => undefined),
+  stat: jest.fn(async () => ({ size: 1234, mtime: new Date() })),
+  delete: jest.fn(async () => undefined),
+  exists: jest.fn(async () => true),
+};
+
+jest.mock('../../src/services/storage', () => ({
+  getStorage: () => mockStorage,
+  initStorage: async () => mockStorage,
+}));
+
+// Nothing to resize or watermark, so the builder takes the stream-from-storage
+// branch, which is the one that holds sockets.
+jest.mock('../../src/services/downloadRendition', () => ({
+  renderPhotoForDownload: jest.fn(async () => null),
+}));
+
+const { bootCrmDb, seedMinimal } = require('../integration/helpers/crmDb');
+const downloadZipService = require('../../src/services/downloadZipService');
+
+describe('pre-zip build releases its storage reads', () => {
+  let db; let cleanup; let eventId;
+
+  beforeAll(async () => {
+    ({ db, cleanup } = await bootCrmDb());
+    await seedMinimal(db);
+
+    const ev = await db('events').insert({
+      slug: 'zipleak',
+      event_type: 'wedding',
+      event_name: 'Zip Leak',
+      event_date: '2026-09-01',
+      host_email: '[email protected]',
+      admin_email: '[email protected]',
+      password_hash: 'x',
+      share_link: '/gallery/zipleak/s',
+      share_token: 'zipleak-share',
+      expires_at: new Date(Date.now() + 7 * 864e5).toISOString(),
+      is_active: 1,
+      is_archived: 0,
+      is_draft: 0,
+      require_password: 0,
+      allow_downloads: 1,
+      created_at: new Date().toISOString(),
+    }).returning('id');
+    eventId = ev[0]?.id ?? ev[0];
+
+    for (let i = 0; i < PHOTO_COUNT; i += 1) {
+      await db('photos').insert({
+        event_id: eventId,
+        filename: `photo-${i}.jpg`,
+        path: `zipleak/photo-${i}.jpg`,
+        type: 'individual',
+        source_origin: 'managed',
+        mime_type: 'image/jpeg',
+        visibility: 'visible',
+        uploaded_at: new Date(Date.now() - i * 1000).toISOString(),
+      });
+    }
+  }, 120000);
+
+  afterAll(async () => { if (cleanup) await cleanup(); });
+
+  beforeEach(() => {
+    reads.opened = [];
+    reads.live = 0;
+    reads.peak = 0;
+    failingKey.value = null;
+    onOpen.fn = null;
+    objectChunks.value = Number.POSITIVE_INFINITY;
+    mockStorage.get.mockClear();
+    downloadZipService.versions.clear();
+    downloadZipService.activeBuilds.clear();
+  });
+
+  it('destroys every open read when a source stream dies mid-build', async () => {
+    // The oldest photo is written first, so failing it strands the rest.
+    failingKey.value = 'events/active/zipleak/photo-0.jpg';
+
+    const result = await downloadZipService.generateZip(eventId);
+
+    expect(result.success).toBe(false);
+    expect(reads.opened.length).toBeGreaterThan(1);
+    const stranded = reads.opened.filter((s) => !s.destroyed);
+    expect(stranded.map((s) => s.key)).toEqual([]);
+  });
+
+  it('destroys every open read when an upload invalidates the build', async () => {
+    // What adminPhotos does on every upload, delete and bulk edit, landing
+    // while the archive is half built.
+    onOpen.fn = (count) => {
+      if (count !== 2) return;
+      downloadZipService.invalidate(eventId);
+      // invalidate() also schedules a rebuild; this test is not about that.
+      clearTimeout(downloadZipService.debounceTimers.get(eventId));
+      downloadZipService.debounceTimers.delete(eventId);
+    };
+
+    const result = await downloadZipService.generateZip(eventId);
+
+    expect(result).toEqual({ success: false, error: 'Build invalidated' });
+    expect(reads.opened.filter((s) => !s.destroyed).map((s) => s.key)).toEqual([]);
+  });
+
+  it('never holds more storage reads open than the build needs', async () => {
+    objectChunks.value = 8;
+
+    const result = await downloadZipService.generateZip(eventId);
+
+    expect(result.success).toBe(true);
+    expect(mockStorage.get).toHaveBeenCalledTimes(PHOTO_COUNT);
+    expect(reads.peak).toBeLessThanOrEqual(MAX_INFLIGHT_READS);
+  });
+});

backend/src/services/downloadZipService.js

@@ -30,11 +30,21 @@ const logger = require('../utils/logger');
 
 const DEBOUNCE_MS = 5000;
 
+// How many storage reads may be open at once while the archive is built.
+// archiver drains its queue one entry at a time, so a stream appended ahead of
+// its turn just parks an S3 socket with a full receive buffer. The SDK agent
+// pool is 50 sockets wide and shared with uploads, thumbnails and gallery
+// reads, so an unbounded loop over a large event starves the whole process.
+// Two keeps the next photo's round trip overlapped with the current write
+// without ever leaving more than one socket idle.
+const MAX_INFLIGHT_READS = 2;
+
 class DownloadZipService {
   constructor() {
     this.activeBuilds = new Map();   // eventId -> { promise, version }
     this.debounceTimers = new Map();  // eventId -> setTimeout handle
     this.versions = new Map();        // eventId -> generation counter
+    this.buildCancellers = new Map(); // eventId -> abort the in-flight build
   }
 
   async stop() {
@@ -42,6 +52,7 @@ class DownloadZipService {
     this.debounceTimers.clear();
     await Promise.allSettled([...this.activeBuilds.values()].map(build => build.promise));
     this.versions.clear();
+    this.buildCancellers.clear();
   }
 
   /**
@@ -117,6 +128,40 @@ class DownloadZipService {
     const storage = getStorage();
     let tmpDir;
 
+    // Storage reads appended to the archive but not yet drained. A failed or
+    // invalidated build must destroy them: archiver's own abort() leaves the
+    // source streams alone, and an unread S3 response body holds its socket
+    // open for the life of the process (the SDK arms its socketTimeout on a
+    // 3s delay and clears it as soon as the response headers land, so
+    // nothing ever reclaims the socket).
+    const openReads = new Set();
+    let cancelled = false;
+    let slotWaiter = null;
+
+    const wakeSlotWaiter = () => {
+      if (!slotWaiter) return;
+      const resume = slotWaiter;
+      slotWaiter = null;
+      resume();
+    };
+    const releaseRead = (stream) => {
+      openReads.delete(stream);
+      wakeSlotWaiter();
+    };
+    // Assigned once the archive exists. Bumping the generation counter only
+    // stops the build the next time the loop looks at it, and the loop can be
+    // parked waiting for a read slot that a stalled archive will never free,
+    // so invalidation cancels the build directly instead of leaving a note.
+    let failBuild = null;
+
+    const trackRead = (stream) => {
+      openReads.add(stream);
+      stream.once('end', () => releaseRead(stream));
+      stream.once('close', () => releaseRead(stream));
+      stream.once('error', () => releaseRead(stream));
+      return stream;
+    };
+
     try {
       const event = await db('events').where({ id: eventId }).first();
       if (!event) return { success: false, error: 'Event not found' };
@@ -162,15 +207,37 @@ class DownloadZipService {
       const useOriginal = await getUseOriginalFilenames();
       const entryNames = getZipEntryNames(photos, useOriginal);
 
+      this.buildCancellers.set(eventId, () => {
+        if (failBuild) failBuild(new Error('Build invalidated'));
+      });
+
       // Build zip — level 0 (store only) since photos are already compressed
       await new Promise((resolve, reject) => {
         const output = fs.createWriteStream(tmpPath);
         const archive = archiver('zip', { zlib: { level: 0 } });
 
+        failBuild = (err) => {
+          if (cancelled) return;
+          cancelled = true;
+          wakeSlotWaiter();
+          // abort() throws if archiver already tore itself down.
+          try { archive.abort(); } catch (_) { /* already aborted */ }
+          reject(err);
+        };
+
         output.on('close', resolve);
-        archive.on('error', reject);
+        archive.on('error', failBuild);
         archive.pipe(output);
 
+        // Block until archiver has drained enough of its queue for another
+        // read. Also returns when the build is cancelled, so a stalled
+        // archive cannot park the loop here forever.
+        const waitForReadSlot = async () => {
+          while (!cancelled && openReads.size >= MAX_INFLIGHT_READS) {
+            await new Promise((resume) => { slotWaiter = resume; });
+          }
+        };
+
         const uniqueTypes = new Set(photos.map(p => p.type)).size;
         const hasMultipleTypes = uniqueTypes > 1;
 
@@ -178,9 +245,9 @@ class DownloadZipService {
           for (let i = 0; i < photos.length; i += 1) {
             const photo = photos[i];
             // Check if build was invalidated
+            if (cancelled) return;
             if (this.versions.get(eventId) !== version) {
-              archive.abort();
+              return failBuild(new Error('Build invalidated'));
-              return reject(new Error('Build invalidated'));
             }
 
             const entryName = entryNames[i];
@@ -212,8 +279,17 @@ class DownloadZipService {
             if (rendered) {
               archive.append(rendered, { name: archiveName });
             } else if (storageKey) {
+              await waitForReadSlot();
+              if (cancelled) return;
               const stream = await storage.get(storageKey);
-              archive.append(stream, { name: archiveName });
+              // The build can be cancelled while the read is in flight, and a
+              // stream nobody appends is a stream nobody closes.
+              if (cancelled || this.versions.get(eventId) !== version) {
+                stream.destroy();
+                if (cancelled) return;
+                return failBuild(new Error('Build invalidated'));
+              }
+              archive.append(trackRead(stream), { name: archiveName });
             } else {
               const filePath = resolvePhotoFilePath(event, photo);
               archive.file(filePath, { name: archiveName });
@@ -223,7 +299,7 @@ class DownloadZipService {
           archive.finalize();
         };
 
-        addPhotos().catch(reject);
+        addPhotos().catch(failBuild);
       });
 
       // Check version again — another invalidation may have arrived
@@ -252,6 +328,11 @@ class DownloadZipService {
       logger.error('downloadZipService._build error', { eventId, error: err.message });
       return { success: false, error: err.message };
     } finally {
+      this.buildCancellers.delete(eventId);
+      for (const stream of openReads) {
+        stream.destroy();
+      }
+      openReads.clear();
       if (tmpDir) {
         await fsp.rm(tmpDir, { recursive: true, force: true }).catch(() => {});
       }
@@ -266,6 +347,11 @@ class DownloadZipService {
     // Bump version to signal any in-flight build is stale
     this.versions.set(eventId, (this.versions.get(eventId) || 0) + 1);
 
+    // Stop the in-flight build now so it releases its storage reads, rather
+    // than when it next reaches the top of its loop.
+    const cancelBuild = this.buildCancellers.get(eventId);
+    if (cancelBuild) cancelBuild();
+
     // Cancel pending debounce
     const timer = this.debounceTimers.get(eventId);
     if (timer) clearTimeout(timer);