fix(file-watcher): bound concurrent photo processing (#846)

* fix(file-watcher): bound concurrent photo processing

chokidar fires 'add' once per file — with no ignoreInitial option the
boot scan fires it for every existing file, and a bulk drop into the
watch folder fires it for every new one at once. Each handler runs DB
lookups plus (for new files) a full sharp pipeline; sharp.concurrency(2)
only caps libvips threads WITHIN one operation, not the number of
parallel pipelines, so unbounded handlers can OOM small hosts.

Gate both 'add' and 'unlink' through a shared p-limit
(FILE_WATCHER_CONCURRENCY, default 2, floor 1) — mass deletes otherwise
burst DB work and ZIP-cache invalidation the same way. p-limit is pinned
to ^3.1.0, the last CommonJS release.

Adapted from the filpgame fork (426ca491) — thanks @filpgame; extended
to cover 'unlink', documented in .env.example, plus a lock-in test for
the existing Sharp cache/concurrency caps this bound relies on.

* chore(compose): pass FILE_WATCHER_CONCURRENCY into the backend container (codex review of #846)

The backend service uses an explicit environment list (no env_file), so
the documented override never reached the container in the default
compose deployments. Added to both compose files + root .env.example.

backend/.env.example

@@ -106,6 +106,12 @@ ARCHIVE_PATH=/app/storage/events/archived
 # EVENTS_PATH=./storage/events
 # ARCHIVE_PATH=./storage/events/archived
 
+# File watcher (auto-import from the events/active folder, local storage only)
+# Max photos processed in parallel by the watcher. The boot scan and bulk
+# folder drops fire one handler per file — this bound keeps thumbnail
+# generation from exhausting memory on small hosts. Default: 2
+# FILE_WATCHER_CONCURRENCY=2
+
 # Analytics Backend Configuration (OPTIONAL)
 # Used for server-side tracking only
 # Primary configuration should be done through Admin UI > Settings > Analytics

backend/__tests__/services/fileWatcher.concurrency.test.js

@@ -0,0 +1,125 @@
+/**
+ * Regression tests for the file-watcher concurrency bound.
+ *
+ * chokidar fires 'add' once per file — with no ignoreInitial option the boot
+ * scan fires it for every existing file, and a bulk drop fires it for every
+ * new one at once. Unbounded handlers each run DB lookups plus a full sharp
+ * pipeline (sharp.concurrency(2) only caps libvips threads WITHIN one
+ * operation), which can OOM small hosts. Both 'add' and 'unlink' must go
+ * through the shared p-limit gate.
+ *
+ * Adapted from the filpgame fork (426ca491), extended to cover 'unlink'.
+ */
+
+const mockLimit = jest.fn((operation) => Promise.resolve().then(operation));
+const mockPLimit = jest.fn(() => mockLimit);
+const mockHandlers = {};
+const mockWatcher = {
+  on: jest.fn((event, handler) => {
+    mockHandlers[event] = handler;
+    return mockWatcher;
+  }),
+};
+
+// Shared instances captured by the mock factories: jest.isolateModules re-runs
+// each factory in a fresh registry, so the factories must return these same
+// objects for the test to observe calls made inside the isolated module.
+const mockLogger = { error: jest.fn(), warn: jest.fn(), info: jest.fn(), debug: jest.fn() };
+// Chainable no-row query — enough for removePhoto's lookup/delete calls.
+const mockDb = jest.fn(() => ({
+  where: jest.fn().mockReturnThis(),
+  first: jest.fn().mockResolvedValue(null),
+  delete: jest.fn().mockResolvedValue(0),
+}));
+
+jest.mock('p-limit', () => mockPLimit);
+jest.mock('chokidar', () => ({
+  watch: jest.fn(() => mockWatcher),
+}));
+jest.mock('../../src/database/db', () => ({ db: mockDb }));
+jest.mock('../../src/utils/logger', () => mockLogger);
+jest.mock('../../src/services/imageProcessor', () => ({
+  generateThumbnail: jest.fn(),
+  generateVideoPlaceholder: jest.fn(),
+}));
+jest.mock('../../src/services/videoProcessor', () => ({
+  isVideoMimeType: jest.fn(() => false),
+}));
+jest.mock('../../src/services/downloadZipService', () => ({ invalidate: jest.fn() }));
+jest.mock('../../src/utils/dbCompat', () => ({
+  formatBoolean: jest.fn((value) => value),
+}));
+
+const loadFileWatcher = () => {
+  let fileWatcher;
+  jest.isolateModules(() => {
+    fileWatcher = require('../../src/services/fileWatcher');
+  });
+  return fileWatcher;
+};
+
+describe('fileWatcher concurrency bound', () => {
+  const originalBackend = process.env.STORAGE_BACKEND;
+  const originalConcurrency = process.env.FILE_WATCHER_CONCURRENCY;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    Object.keys(mockHandlers).forEach((key) => delete mockHandlers[key]);
+    process.env.STORAGE_BACKEND = 'local';
+    delete process.env.FILE_WATCHER_CONCURRENCY;
+  });
+
+  afterAll(() => {
+    if (originalBackend === undefined) delete process.env.STORAGE_BACKEND;
+    else process.env.STORAGE_BACKEND = originalBackend;
+    if (originalConcurrency === undefined) delete process.env.FILE_WATCHER_CONCURRENCY;
+    else process.env.FILE_WATCHER_CONCURRENCY = originalConcurrency;
+  });
+
+  it.each([
+    [undefined, 2],   // default
+    ['3', 3],         // explicit
+    ['0', 1],         // floored to 1
+    ['-4', 1],        // floored to 1
+    ['invalid', 2],   // falls back to default
+  ])('configures the limiter with FILE_WATCHER_CONCURRENCY=%s as %i', (configured, expected) => {
+    if (configured === undefined) delete process.env.FILE_WATCHER_CONCURRENCY;
+    else process.env.FILE_WATCHER_CONCURRENCY = configured;
+
+    loadFileWatcher().startFileWatcher();
+
+    expect(mockPLimit).toHaveBeenCalledWith(expected);
+  });
+
+  it('routes add events through the shared limiter', async () => {
+    loadFileWatcher().startFileWatcher();
+
+    expect(mockHandlers.add).toEqual(expect.any(Function));
+    mockHandlers.add('/outside-watch-root'); // early-returns inside processNewPhoto
+
+    expect(mockLimit).toHaveBeenCalledTimes(1);
+    expect(mockLimit).toHaveBeenCalledWith(expect.any(Function));
+    await mockLimit.mock.results[0].value;
+  });
+
+  it('routes unlink events through the same limiter', async () => {
+    loadFileWatcher().startFileWatcher();
+
+    expect(mockHandlers.unlink).toEqual(expect.any(Function));
+    mockHandlers.unlink('/outside-watch-root'); // early-returns inside removePhoto
+
+    expect(mockLimit).toHaveBeenCalledTimes(1);
+    await mockLimit.mock.results[0].value;
+  });
+
+  it('logs instead of rejecting when a queued handler throws', async () => {
+    loadFileWatcher().startFileWatcher();
+
+    const failure = new Error('boom');
+    mockLimit.mockImplementationOnce(() => Promise.reject(failure));
+    mockHandlers.add('/whatever');
+
+    await new Promise(process.nextTick);
+    expect(mockLogger.error).toHaveBeenCalledWith('Error processing new photo:', failure);
+  });
+});

backend/__tests__/services/imageProcessor.sharpConfig.test.js

@@ -0,0 +1,30 @@
+/**
+ * Locks the process-wide Sharp memory guards. The file-watcher concurrency
+ * bound (FILE_WATCHER_CONCURRENCY) assumes these caps stay in place — they
+ * limit libvips threads/cache WITHIN one operation while p-limit bounds the
+ * number of parallel pipelines. From the filpgame fork (426ca491).
+ */
+
+const mockSharp = jest.fn();
+mockSharp.cache = jest.fn();
+mockSharp.concurrency = jest.fn();
+
+jest.mock('sharp', () => mockSharp);
+
+jest.mock('../../src/utils/logger', () => ({
+  error: jest.fn(),
+  warn: jest.fn(),
+  info: jest.fn(),
+  debug: jest.fn(),
+}));
+
+describe('imageProcessor Sharp configuration', () => {
+  it('disables the Sharp cache and caps libvips concurrency', () => {
+    jest.isolateModules(() => {
+      require('../../src/services/imageProcessor');
+    });
+
+    expect(mockSharp.cache).toHaveBeenCalledWith(false);
+    expect(mockSharp.concurrency).toHaveBeenCalledWith(2);
+  });
+});

backend/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "picpeak-backend",
-  "version": "3.88.0-beta.0",
+  "version": "3.92.1-beta.0",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "picpeak-backend",
-      "version": "3.88.0-beta.0",
+      "version": "3.92.1-beta.0",
       "dependencies": {
         "@aws-sdk/client-s3": "^3.850.0",
         "@aws-sdk/lib-storage": "^3.850.0",
@@ -42,6 +42,7 @@
         "nodemailer": "^9.0.1",
         "openid-client": "^5.7.1",
         "otplib": "^12.0.1",
+        "p-limit": "^3.1.0",
         "pdf-lib": "^1.17.1",
         "pdfkit": "^0.17.2",
         "pg": "^8.16.3",
@@ -9498,7 +9499,6 @@
       "version": "3.1.0",
       "resolved": "https://registry.npmjs.org/p-limit/-/p-limit-3.1.0.tgz",
       "integrity": "sha512-TYOanM3wGwNGsZN2cVTYPArw454xnXj5qmWF1bEoAc4+cU/ol7GVh7odevjp1FNHduHc3KZMcFduxU5Xc6uJRQ==",
-      "dev": true,
       "license": "MIT",
       "dependencies": {
         "yocto-queue": "^0.1.0"
@@ -12511,7 +12511,6 @@
       "version": "0.1.0",
       "resolved": "https://registry.npmjs.org/yocto-queue/-/yocto-queue-0.1.0.tgz",
       "integrity": "sha512-rVksvsnNCdJ/ohGc6xgPwyN8eheCxsiLM8mxuE/t/mOVqJewPuO1miLpTHQiRgTKCLexL4MeAFVagts7HmNZ2Q==",
-      "dev": true,
       "license": "MIT",
       "engines": {
         "node": ">=10"

backend/package.json

@@ -49,6 +49,7 @@
     "nodemailer": "^9.0.1",
     "openid-client": "^5.7.1",
     "otplib": "^12.0.1",
+    "p-limit": "^3.1.0",
     "pdf-lib": "^1.17.1",
     "pdfkit": "^0.17.2",
     "pg": "^8.16.3",

backend/src/services/fileWatcher.js

@@ -2,6 +2,7 @@ const chokidar = require('chokidar');
 const path = require('path');
 const fs = require('fs').promises;
 const sharp = require('sharp');
+const pLimit = require('p-limit');
 const { db } = require('../database/db');
 const { formatBoolean } = require('../utils/dbCompat');
 const { generateThumbnail, generateVideoPlaceholder } = require('./imageProcessor');
@@ -13,6 +14,20 @@ const downloadZipService = require('./downloadZipService');
 const getStoragePath = () => process.env.STORAGE_PATH || path.join(__dirname, '../../../storage');
 const WATCH_PATH = () => path.join(getStoragePath(), 'events/active');
 
+// Bound concurrent watcher work. chokidar fires 'add' once per file — with no
+// ignoreInitial option the boot scan fires it for EVERY existing file, and a
+// bulk drop into the watch folder fires it for every new one at once. Each
+// handler runs DB lookups and (for new files) a full sharp pipeline;
+// sharp.concurrency(2) only caps libvips threads WITHIN one operation, not the
+// number of parallel pipelines, so unbounded handlers can OOM small hosts.
+// 'unlink' shares the limiter: mass deletes otherwise burst DB work and
+// ZIP-cache invalidation the same way.
+const configuredConcurrency = Number.parseInt(process.env.FILE_WATCHER_CONCURRENCY || '2', 10);
+const watcherConcurrency = Number.isFinite(configuredConcurrency)
+  ? Math.max(1, configuredConcurrency)
+  : 2;
+const processLimit = pLimit(watcherConcurrency);
+
 function startFileWatcher() {
   // Auto-import via filesystem watching only works with the local storage
   // backend. In S3 mode there is no local directory to watch — every photo
@@ -34,19 +49,15 @@ function startFileWatcher() {
   });
 
   watcher
-    .on('add', async (filePath) => {
-      try {
-        await processNewPhoto(filePath);
-      } catch (error) {
+    .on('add', (filePath) => {
+      processLimit(() => processNewPhoto(filePath)).catch((error) => {
         logger.error('Error processing new photo:', error);
-      }
+      });
     })
-    .on('unlink', async (filePath) => {
-      try {
-        await removePhoto(filePath);
-      } catch (error) {
+    .on('unlink', (filePath) => {
+      processLimit(() => removePhoto(filePath)).catch((error) => {
         logger.error('Error removing photo:', error);
-      }
+      });
     });
 
   logger.info('File watcher started');