diff --git a/backend/__tests__/services/downloadZipSocketLeak.test.js b/backend/__tests__/services/downloadZipSocketLeak.test.js new file mode 100644 index 00000000..61696d65 --- /dev/null +++ b/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: 'h@example.com', + admin_email: 'a@example.com', + 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); + }); +}); diff --git a/backend/src/services/downloadZipService.js b/backend/src/services/downloadZipService.js index ffc5f42d..cd5d7e31 100644 --- a/backend/src/services/downloadZipService.js +++ b/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 reject(new Error('Build invalidated')); + return failBuild(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);