diff --git a/backend/__tests__/utils/archiveStreamGuard.test.js b/backend/__tests__/utils/archiveStreamGuard.test.js new file mode 100644 index 00000000..1752286b --- /dev/null +++ b/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(); + }); +}); diff --git a/backend/src/routes/gallery/downloads.js b/backend/src/routes/gallery/downloads.js index 1dee1a08..59a284d3 100644 --- a/backend/src/routes/gallery/downloads.js +++ b/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'); } }); diff --git a/backend/src/services/downloadJobService.js b/backend/src/services/downloadJobService.js index 87a94e30..c3abd096 100644 --- a/backend/src/services/downloadJobService.js +++ b/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) { diff --git a/backend/src/utils/archiveStreamGuard.js b/backend/src/utils/archiveStreamGuard.js new file mode 100644 index 00000000..c9154792 --- /dev/null +++ b/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 };