fix(faces): defer on unreachable storage, and commit the import path first (#1097)
* fix(faces): defer on unreachable storage, and commit the import path first The two follow-ups left open by #1091, both consequences of external photos becoming scannable at all. **A dropped mount no longer burns the gallery.** ensurePreviewImage returns null for "this JPEG is corrupt" and "the NFS share is gone" alike, and faceProcessor marked both 'failed'. Nothing re-queues a failure automatically and the queue only ever claims 'pending', so a mount that blinked mid-scan cost the whole event a manual Re-scan — on external libraries, where network storage drops far more often than local disk, that is the common case rather than the corner one. faceProcessor now probes the containing DIRECTORY before failing, and throws TransientSourceError when it cannot be reached; faceQueue treats that exactly like SidecarUnavailableError — release to pending, back off, retry — with the warning rate-limited to one per five minutes, since an outage hits every photo in the event. The directory rather than the file is the whole point: a missing file inside a healthy directory is a broken photo and should still fail, and it still does. Anything that goes wrong deciding which case it is falls through to 'failed', because guessing 'transient' on an unknown condition would retry forever. **The import commits its path before inserting rows.** enqueueEvent accepts processing_status NULL (faceProcessor.js:243-246), which these inserts leave unset, so an admin hitting the toggle or Re-scan during a long import could queue partial rows while the event still resolved against the old directory — and burn them to 'failed'. Moving events.external_path ahead of the loop closes that, and fixes a pre-existing bug on the same line: an import that died at photo 500 of 1000 used to leave those 500 rows pointing into the new tree while the event still resolved against the old one, making every one of them unreadable. Safe to do first because the path is already validated above, and existing photos are unaffected — photo.source_origin takes precedence over event.source_mode in both resolvers and is NOT NULL defaulting to 'managed'. The #1090 test that asserted a missing source fails needed its setup corrected rather than its intent: it created no directory at all, which is now (correctly) a dropped mount. It now creates one, so it tests the case it always meant to — healthy storage, dead photo. Verified both fixes discriminate: removing the probe fails the defer test, and moving the event update back after the loop fails the ordering test. Face suite 59 passed across 8 suites; the full backend suite fails the same 10 pre-existing suites as unmodified main, no more. * fix(faces): stop a dead mount stalling the whole queue External review, and the first finding is one my own change created. Deferring by returning the row to 'pending' was a trap: claimNextPhoto orders by id ascending and the queue defaults to a single worker, so the same unreachable row becomes the oldest pending one after every backoff and the worker never reaches a higher id. One dead mount would have stalled face scanning for the entire install — unrelated events, fresh uploads, everything. Strictly worse than the permanent 'failed' this set out to replace. The row is now left parked in 'processing' with face_started_at intact. It is not claimable, so the worker moves straight on; the janitor that already exists returns it to 'pending' past STUCK_TIMEOUT_MS, which is the retry. No new column and no new timer. The sidecar branch still releases, because a down sidecar blocks every photo anyway — there is no other work to get on with. Second: probing existence was not enough. Unmounting an NFS or SMB share usually leaves the mountpoint behind as an ordinary empty directory, so fs.access succeeded on storage that was entirely gone and the photo was failed anyway — the exact case this was written for. An empty directory where the photo should live now counts as unreachable. The trade is deliberate and documented: a directory an admin genuinely emptied is retried rather than failed, which now costs one attempt per janitor sweep and nothing else. Third: a comment in adminExternalMedia claimed source_origin isolates existing photos from the early external_path update. That is true of managed rows and false of external ones — resolveExternalPath prefixes every external row with event.external_path, so importing folder B into an event referencing folder A rebases the A rows. Pre-existing rather than introduced here (the update always did this, just later), but the comment asserted otherwise, so it now says what actually happens and names the underlying single-base-path limitation. The deferral test initially passed against the blocking version too — database state alone cannot tell the fix from the bug. It now inspects the branch directly, the way the #596 contract tests do, and fails when releaseToPending is put back or the two branches are merged. * fix(faces): back off per event, and stop clobbering concurrent scans Round two of external review. Parking a row in 'processing' fixed the head-of-line block but not the cost: every janitor sweep handed the whole dead gallery back, and the worker walked all of it again — one stat per photo against storage that may be hard-mounted and slow to time out — before reaching any healthy event. Every one of those attempts also went through generatePreviewImage first, which logs an error per photo, so a down mount produced a recurring flood that the rate-limited warning did nothing about. So the backoff is now per EVENT and separate from the janitor: TransientSourceError carries the event id, the queue records a cooldown, and claimNextPhoto excludes those events while it lasts. The janitor keeps doing its own job, which is rescuing rows a crashed worker abandoned. Cooldown is in memory on purpose — a restart is usually what follows fixing a mount, so it should retry at once. Second: committing the event path before the loop means a toggle or Re-scan firing mid-import can now genuinely queue and finish some of those rows. The final bulk update was unconditional, so it dragged 'done' rows back to 'pending' for a duplicate sidecar scan and knocked 'processing' rows out from under the worker. It is now whereNull — only rows nothing has touched are ours to queue. Also corrected a comment of mine that had gone stale in the same file: it still described the enqueue as happening after the event path was written "below", which stopped being true when that update moved above the loop. * fix(faces): judge the mount, the path and the file separately Round three of external review. The probe was too coarse in both directions. It read any ENOENT on the photo's own directory as a mount-wide outage, so a deleted or renamed subfolder — individual/ gone while collages/ is healthy — deferred the entire event and starved every sibling folder, renewing the cooldown on each retry. It now judges the EVENT ROOT for that verdict: root missing, or present-but-empty, is an outage; anything below a populated root is a broken path and fails. And it read a listable directory as proof the photo was at fault, so EACCES on a reconnected share, EIO, or the classic NFS ESTALE handle were burnt as permanent failures. Only ENOENT now means genuinely gone; any other error opening the file defers. The event-wide backoff was also too broad. A reference event can hold managed uploads alongside imported external ones, and those live in local storage that is fine — excluding the whole event id left them unscanned for as long as external rows kept renewing the cooldown, which during a real outage is indefinitely. The exclusion is now scoped to external and reference rows. One of my own tests had modelled the unmount wrongly: it emptied the photo's subdirectory rather than the event root, which under the corrected logic is a populated mount with a missing folder — a failure, not an outage. It now empties the root, which is what an unmount actually leaves behind. Dropped the path require the first version of this probe needed; the event-root form does not. All three fixes mutation-checked: reverting each one fails the test written for it. Face suite 69 passed across 9 suites; full backend suite fails the same 10 pre-existing suites as main, no more. --------- Co-authored-by: Paul Nothaft <paul@MacStudio-von-Paul.local>
This commit is contained in:
Paul Nothaft
@@ -33,6 +33,8 @@ describe('external import queues faces (#1090)', () => {
|
||||
// When set to an event id, the mocked thumbnail call turns detection on
|
||||
// mid-loop, standing in for an admin flipping the toggle during an import.
|
||||
let flipFacesOnDuringLoop = null;
|
||||
// Stands in for a concurrent Re-scan completing a row mid-import.
|
||||
let markDoneDuringLoop = false;
|
||||
|
||||
beforeAll(async () => {
|
||||
tmpDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), 'picpeak-extenq-'));
|
||||
@@ -70,6 +72,12 @@ describe('external import queues faces (#1090)', () => {
|
||||
pendingSeenDuringLoop += rows.length;
|
||||
const ev = await liveDb('events').first();
|
||||
externalPathDuringLoop = ev ? ev.external_path : undefined;
|
||||
if (markDoneDuringLoop) {
|
||||
const rows = await liveDb('photos').orderBy('id', 'asc').limit(1);
|
||||
if (rows.length) {
|
||||
await liveDb('photos').where({ id: rows[0].id }).update({ face_status: 'done' });
|
||||
}
|
||||
}
|
||||
if (flipFacesOnDuringLoop) {
|
||||
await liveDb('events').where({ id: flipFacesOnDuringLoop })
|
||||
.update({ face_recognition_enabled: true });
|
||||
@@ -123,6 +131,7 @@ describe('external import queues faces (#1090)', () => {
|
||||
|
||||
pendingSeenDuringLoop = 0;
|
||||
externalPathDuringLoop = undefined;
|
||||
markDoneDuringLoop = false;
|
||||
return typeof e === 'object' ? e.id : e;
|
||||
}
|
||||
|
||||
@@ -149,11 +158,11 @@ describe('external import queues faces (#1090)', () => {
|
||||
|
||||
await runImport(eventId);
|
||||
|
||||
// Observed from inside the loop: nothing may be claimable yet, because the
|
||||
// event still resolves to the old (here: empty) directory. Marking rows on
|
||||
// insert would make this non-zero and leave photos permanently 'failed'.
|
||||
// Observed from inside the loop: nothing is claimable yet. The event path
|
||||
// is already committed (see the test above), so this is no longer load
|
||||
// bearing for correctness — but keeping the enqueue at the end is what lets
|
||||
// the feature setting be read after the loop, so the invariant stays.
|
||||
expect(pendingSeenDuringLoop).toBe(0);
|
||||
expect(externalPathDuringLoop).toBeFalsy();
|
||||
|
||||
// ...and afterwards both are in place.
|
||||
const ev = await db('events').where({ id: eventId }).first();
|
||||
@@ -178,6 +187,38 @@ describe('external import queues faces (#1090)', () => {
|
||||
expect(photos.every((p) => p.face_status === 'pending')).toBe(true);
|
||||
});
|
||||
|
||||
it('commits events.external_path before the first row is inserted', async () => {
|
||||
// enqueueEvent accepts processing_status NULL (faceProcessor.js:243-246),
|
||||
// which these inserts leave unset — so a toggle or Re-scan firing mid-import
|
||||
// can queue partial rows. If the event still pointed at the old directory
|
||||
// they would resolve against it and burn to 'failed'. Setting the path
|
||||
// first also means a half-finished import leaves rows that still resolve,
|
||||
// instead of rows stranded against the previous path.
|
||||
const eventId = await seedEvent({ facesEnabled: true, flagOn: true });
|
||||
|
||||
await runImport(eventId);
|
||||
|
||||
// Sampled from inside the per-photo thumbnail call, i.e. while rows are
|
||||
// still being inserted.
|
||||
expect(externalPathDuringLoop).toBe('nas');
|
||||
});
|
||||
|
||||
it('does not re-queue rows a concurrent scan already handled', async () => {
|
||||
// Committing the event path before the loop means a toggle or Re-scan
|
||||
// firing mid-import can now genuinely queue and even finish some of these
|
||||
// rows. A blanket update at the end would drag 'done' rows back to
|
||||
// 'pending' for a duplicate sidecar scan and knock 'processing' rows out
|
||||
// from under the worker.
|
||||
const eventId = await seedEvent({ facesEnabled: true, flagOn: true });
|
||||
markDoneDuringLoop = true;
|
||||
|
||||
await runImport(eventId);
|
||||
markDoneDuringLoop = false;
|
||||
|
||||
const done = await db('photos').where({ event_id: eventId, face_status: 'done' });
|
||||
expect(done.length).toBeGreaterThan(0); // the concurrent scan's work survived
|
||||
});
|
||||
|
||||
it('leaves face_status untouched when the per-event toggle is off', async () => {
|
||||
const eventId = await seedEvent({ facesEnabled: false, flagOn: true });
|
||||
await runImport(eventId);
|
||||
|
||||
@@ -25,6 +25,15 @@ process.env.TEST_DATABASE_PATH = path.join(
|
||||
fs.mkdtempSync(path.join(os.tmpdir(), 'picpeak-faceext-')), 'db.sqlite',
|
||||
);
|
||||
process.env.JWT_SECRET = process.env.JWT_SECRET || 'faceext-test-secret';
|
||||
// A real, existing media root. getExternalMediaRoot only honours the env var
|
||||
// if the directory exists and caches it on first call, so this has to be set
|
||||
// up before anything requires externalMediaService.
|
||||
process.env.EXTERNAL_MEDIA_ROOT = path.join(path.dirname(process.env.TEST_DATABASE_PATH), 'media');
|
||||
fs.mkdirSync(path.join(process.env.EXTERNAL_MEDIA_ROOT, 'share', 'individual'), { recursive: true });
|
||||
// Non-empty on purpose: an empty directory is read as an unmounted share
|
||||
// (faceTransientSource.test.js), so a "healthy storage, dead photo" fixture
|
||||
// needs a sibling present or it defers instead of failing.
|
||||
fs.writeFileSync(path.join(process.env.EXTERNAL_MEDIA_ROOT, 'share', 'individual', 'sibling.jpg'), 'x');
|
||||
|
||||
const sharp = require('sharp');
|
||||
|
||||
@@ -65,6 +74,7 @@ async function seedPhoto({ sourceOrigin = 'managed', sourceMode = 'managed' } =
|
||||
expires_at: new Date().toISOString(),
|
||||
face_recognition_enabled: true,
|
||||
source_mode: sourceMode,
|
||||
external_path: 'share',
|
||||
}).returning('id');
|
||||
const eventId = typeof e === 'object' ? e.id : e;
|
||||
|
||||
@@ -78,7 +88,7 @@ async function seedPhoto({ sourceOrigin = 'managed', sourceMode = 'managed' } =
|
||||
processing_status: 'complete',
|
||||
face_status: 'processing',
|
||||
source_origin: sourceOrigin,
|
||||
external_relpath: sourceOrigin === 'managed' ? null : 'nas/ext.jpg',
|
||||
external_relpath: sourceOrigin === 'managed' ? null : 'individual/ext.jpg',
|
||||
}).returning('id');
|
||||
return { eventId, photoId: typeof p === 'object' ? p.id : p };
|
||||
}
|
||||
@@ -134,6 +144,11 @@ describe('face scanning of external/reference photos (#1090)', () => {
|
||||
// A missing file is a property of that photo, so it should be visible as a
|
||||
// failure the admin can act on — not silently absorbed the way the old
|
||||
// blanket skip did.
|
||||
//
|
||||
// The containing directory exists here on purpose. An absent directory is
|
||||
// a dropped mount, which defers rather than fails
|
||||
// (faceTransientSource.test.js); this is the other case — healthy storage,
|
||||
// dead photo.
|
||||
previewKeyResult = null;
|
||||
const { photoId } = await seedPhoto({ sourceOrigin: 'external', sourceMode: 'reference' });
|
||||
|
||||
|
||||
@@ -0,0 +1,226 @@
|
||||
/**
|
||||
* A deferred photo must not stall the queue.
|
||||
*
|
||||
* claimNextPhoto orders by id ascending, and the queue defaults to a single
|
||||
* worker. So returning an unreachable photo to 'pending' — the obvious way to
|
||||
* say "try again later" — makes that same row the oldest pending one forever:
|
||||
* the worker reclaims it after every backoff and never reaches a higher id.
|
||||
* One dead mount would stall face scanning for the entire install, including
|
||||
* unrelated events and fresh uploads.
|
||||
*
|
||||
* The row is instead left parked in 'processing' with its face_started_at
|
||||
* intact. It is not claimable, so the worker advances; the existing janitor
|
||||
* returns it to 'pending' after STUCK_TIMEOUT_MS, which is the retry.
|
||||
*/
|
||||
|
||||
const path = require('path');
|
||||
const fs = require('fs');
|
||||
const os = require('os');
|
||||
|
||||
process.env.NODE_ENV = 'test';
|
||||
const tmpRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'picpeak-defer-'));
|
||||
process.env.TEST_DATABASE_PATH = path.join(tmpRoot, 'db.sqlite');
|
||||
process.env.JWT_SECRET = process.env.JWT_SECRET || 'defer-test-secret';
|
||||
|
||||
const { bootCrmDb } = require('./helpers/crmDb');
|
||||
|
||||
let db; let cleanup; let faceQueue; let faceProcessor;
|
||||
|
||||
describe('deferred photos do not block the queue', () => {
|
||||
beforeAll(async () => {
|
||||
({ db, cleanup } = await bootCrmDb());
|
||||
faceQueue = require('../../src/services/faceQueue');
|
||||
faceProcessor = require('../../src/services/faceProcessor');
|
||||
}, 120000);
|
||||
|
||||
afterAll(async () => {
|
||||
if (cleanup) await cleanup();
|
||||
await fs.promises.rm(tmpRoot, { recursive: true, force: true }).catch(() => {});
|
||||
});
|
||||
|
||||
it('exports TransientSourceError for the queue to branch on', () => {
|
||||
// The queue imports this from faceProcessor; if the export is dropped the
|
||||
// instanceof check silently becomes false and every deferral turns back
|
||||
// into a permanent failure.
|
||||
expect(typeof faceProcessor.TransientSourceError).toBe('function');
|
||||
expect(new faceProcessor.TransientSourceError(1, 'x'))
|
||||
.toBeInstanceOf(Error);
|
||||
});
|
||||
|
||||
it('does NOT return a deferred row to pending', () => {
|
||||
// Source inspection, deliberately. workerLoop is an unexported infinite
|
||||
// loop, so the branch cannot be driven directly, and asserting on database
|
||||
// state alone does not distinguish the fix from the bug — a version that
|
||||
// re-queues the row passes every state assertion in this file. What
|
||||
// actually matters is that this one branch does not call releaseToPending,
|
||||
// so that is what is pinned. Same approach as the contract tests added for
|
||||
// #596.
|
||||
const src = fs.readFileSync(
|
||||
path.join(__dirname, '..', '..', 'src', 'services', 'faceQueue.js'), 'utf8'
|
||||
);
|
||||
|
||||
const marker = 'if (err instanceof TransientSourceError) {';
|
||||
const start = src.indexOf(marker);
|
||||
expect(start).toBeGreaterThan(-1);
|
||||
|
||||
// The branch body, up to its closing brace.
|
||||
const body = src.slice(start, src.indexOf('\n }', start));
|
||||
expect(body).not.toMatch(/releaseToPending/);
|
||||
expect(body).toMatch(/continue/);
|
||||
|
||||
// And the sidecar branch, which SHOULD still release, so this test fails
|
||||
// if the two branches are ever collapsed back together.
|
||||
const sideStart = src.indexOf('if (err instanceof SidecarUnavailableError) {');
|
||||
expect(sideStart).toBeGreaterThan(-1);
|
||||
const sideBody = src.slice(sideStart, src.indexOf('\n }', sideStart));
|
||||
expect(sideBody).toMatch(/releaseToPending/);
|
||||
});
|
||||
|
||||
it('leaves a deferred row claimable-later, not claimable-now', async () => {
|
||||
// A row parked in 'processing' is invisible to claimNextPhoto, which only
|
||||
// ever selects face_status='pending' — that is what lets the worker move
|
||||
// past it instead of spinning on it.
|
||||
const [e] = await db('events').insert({
|
||||
slug: `defer-${Math.random().toString(36).slice(2, 8)}`,
|
||||
event_type: 'wedding',
|
||||
event_name: 'defer',
|
||||
event_date: '2026-01-01',
|
||||
host_email: 'h@example.com',
|
||||
admin_email: 'a@example.com',
|
||||
password_hash: 'x',
|
||||
share_link: `defer-${Math.random()}`,
|
||||
expires_at: new Date().toISOString(),
|
||||
face_recognition_enabled: true,
|
||||
}).returning('id');
|
||||
const eventId = typeof e === 'object' ? e.id : e;
|
||||
|
||||
const [stuck] = await db('photos').insert({
|
||||
event_id: eventId,
|
||||
filename: 'stuck.jpg',
|
||||
path: 'd/stuck.jpg',
|
||||
type: 'individual',
|
||||
processing_status: 'complete',
|
||||
face_status: 'processing',
|
||||
face_started_at: new Date().toISOString(),
|
||||
source_origin: 'external',
|
||||
}).returning('id');
|
||||
const stuckId = typeof stuck === 'object' ? stuck.id : stuck;
|
||||
|
||||
const parked = await db('photos')
|
||||
.where({ id: stuckId, face_status: 'pending' })
|
||||
.first();
|
||||
expect(parked).toBeUndefined(); // not claimable while parked
|
||||
|
||||
// The janitor's contract is what turns the park into a retry: it resets
|
||||
// 'processing' rows whose face_started_at is older than the stuck timeout.
|
||||
// Backdate past it and the row becomes claimable again.
|
||||
const longAgo = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString();
|
||||
await db('photos').where({ id: stuckId }).update({ face_started_at: longAgo });
|
||||
|
||||
const cutoff = new Date(Date.now() - 600000).toISOString();
|
||||
const reset = await db('photos')
|
||||
.where('face_status', 'processing')
|
||||
.where('face_started_at', '<', cutoff)
|
||||
.update({ face_status: 'pending', face_started_at: null });
|
||||
|
||||
expect(reset).toBeGreaterThan(0);
|
||||
const after = await db('photos').where({ id: stuckId }).first();
|
||||
expect(after.face_status).toBe('pending');
|
||||
});
|
||||
|
||||
it('claimNextPhoto skips events inside their backoff window', async () => {
|
||||
// The per-event cooldown is what stops the janitor handing a whole dead
|
||||
// gallery back every sweep. Without the exclusion the worker walks all of
|
||||
// it again — one slow stat per photo against a possibly hard-mounted
|
||||
// share — before reaching any healthy event.
|
||||
const mk = async (name) => {
|
||||
const [e] = await db('events').insert({
|
||||
slug: `cd-${name}-${Math.random().toString(36).slice(2, 8)}`,
|
||||
event_type: 'wedding',
|
||||
event_name: name,
|
||||
event_date: '2026-01-01',
|
||||
host_email: 'h@example.com',
|
||||
admin_email: 'a@example.com',
|
||||
password_hash: 'x',
|
||||
share_link: `cd-${name}-${Math.random()}`,
|
||||
expires_at: new Date().toISOString(),
|
||||
face_recognition_enabled: true,
|
||||
}).returning('id');
|
||||
const eventId = typeof e === 'object' ? e.id : e;
|
||||
const [p2] = await db('photos').insert({
|
||||
event_id: eventId,
|
||||
filename: `${name}.jpg`,
|
||||
path: `cd/${name}.jpg`,
|
||||
type: 'individual',
|
||||
processing_status: 'complete',
|
||||
face_status: 'pending',
|
||||
source_origin: 'external',
|
||||
}).returning('id');
|
||||
return { eventId, photoId: typeof p2 === 'object' ? p2.id : p2 };
|
||||
};
|
||||
|
||||
await db('photos').del();
|
||||
const dead = await mk('dead'); // lower id -> would win the FIFO
|
||||
const healthy = await mk('healthy');
|
||||
|
||||
// Without exclusion the dead event's row is claimed first...
|
||||
const first = await faceQueue.claimNextPhoto([]);
|
||||
expect(first.id).toBe(dead.photoId);
|
||||
await db('photos').where({ id: dead.photoId }).update({ face_status: 'pending' });
|
||||
|
||||
// ...and with it, the worker reaches the healthy event instead.
|
||||
const second = await faceQueue.claimNextPhoto([dead.eventId]);
|
||||
expect(second.id).toBe(healthy.photoId);
|
||||
});
|
||||
|
||||
it('backoff spares managed rows in a mixed-source event', async () => {
|
||||
// A reference event can hold managed uploads alongside imported external
|
||||
// ones. Excluding the whole event id would leave those unscanned for as
|
||||
// long as external rows keep renewing the cooldown — indefinitely, during
|
||||
// a real outage — even though their local source is fine.
|
||||
await db('photos').del();
|
||||
const [e] = await db('events').insert({
|
||||
slug: `mix-${Math.random().toString(36).slice(2, 8)}`,
|
||||
event_type: 'wedding',
|
||||
event_name: 'mix',
|
||||
event_date: '2026-01-01',
|
||||
host_email: 'h@example.com',
|
||||
admin_email: 'a@example.com',
|
||||
password_hash: 'x',
|
||||
share_link: `mix-${Math.random()}`,
|
||||
expires_at: new Date().toISOString(),
|
||||
face_recognition_enabled: true,
|
||||
source_mode: 'reference',
|
||||
}).returning('id');
|
||||
const eventId = typeof e === 'object' ? e.id : e;
|
||||
|
||||
const add = async (origin, name) => {
|
||||
const [p2] = await db('photos').insert({
|
||||
event_id: eventId,
|
||||
filename: name,
|
||||
path: `mix/${name}`,
|
||||
type: 'individual',
|
||||
processing_status: 'complete',
|
||||
face_status: 'pending',
|
||||
source_origin: origin,
|
||||
}).returning('id');
|
||||
return typeof p2 === 'object' ? p2.id : p2;
|
||||
};
|
||||
await add('external', 'ext.jpg'); // lower id, would win the FIFO
|
||||
const managedId = await add('managed', 'man.jpg');
|
||||
|
||||
// Event is in backoff: the external row is skipped, the managed one is not.
|
||||
const claimed = await faceQueue.claimNextPhoto([eventId]);
|
||||
expect(claimed).toBeTruthy();
|
||||
expect(claimed.id).toBe(managedId);
|
||||
});
|
||||
|
||||
it('startQueue is exported and does not throw on import', () => {
|
||||
// faceQueue requires faceProcessor for TransientSourceError while
|
||||
// faceProcessor is itself required by the routes — a circular require here
|
||||
// would surface as an undefined export rather than a crash, so assert the
|
||||
// module actually loaded something usable.
|
||||
expect(faceQueue).toBeTruthy();
|
||||
expect(Object.keys(faceQueue).length).toBeGreaterThan(0);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,209 @@
|
||||
/**
|
||||
* A dropped mount defers a scan; a dead photo fails it.
|
||||
*
|
||||
* ensurePreviewImage returns null for both "this JPEG is corrupt" and "the
|
||||
* NFS share is gone", and #1090 made that distinction matter: external
|
||||
* libraries now reach this path, and network mounts drop far more often than
|
||||
* local disks. Failing on an outage strands the photo — faceQueue only ever
|
||||
* claims 'pending', and nothing re-queues a failure automatically, so a mount
|
||||
* that blinked mid-scan would cost an entire gallery a manual Re-scan.
|
||||
*
|
||||
* The probe checks the containing DIRECTORY rather than the file, because that
|
||||
* is what separates the two cases: a missing file inside a healthy directory
|
||||
* is a broken photo, an unreachable directory is broken storage.
|
||||
*/
|
||||
|
||||
const path = require('path');
|
||||
const fs = require('fs');
|
||||
const os = require('os');
|
||||
|
||||
process.env.NODE_ENV = 'test';
|
||||
const tmpRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'picpeak-transient-'));
|
||||
process.env.TEST_DATABASE_PATH = path.join(tmpRoot, 'db.sqlite');
|
||||
process.env.JWT_SECRET = process.env.JWT_SECRET || 'transient-test-secret';
|
||||
// Created BEFORE anything requires externalMediaService: getExternalMediaRoot
|
||||
// only honours the env var if the directory already exists, and caches the
|
||||
// result on first call — set it later and every path silently resolves
|
||||
// against a fallback root instead.
|
||||
process.env.EXTERNAL_MEDIA_ROOT = path.join(tmpRoot, 'media');
|
||||
fs.mkdirSync(process.env.EXTERNAL_MEDIA_ROOT, { recursive: true });
|
||||
|
||||
let previewKeyResult = null;
|
||||
const mockEnsurePreviewImage = jest.fn(async () => previewKeyResult);
|
||||
|
||||
jest.mock('../../src/services/imageProcessor', () => ({
|
||||
...jest.requireActual('../../src/services/imageProcessor'),
|
||||
ensurePreviewImage: (...args) => mockEnsurePreviewImage(...args),
|
||||
}));
|
||||
|
||||
jest.mock('../../src/services/faceClient', () => ({
|
||||
detectFaces: jest.fn(async () => ({ model_version: 'test-v1', faces: [] })),
|
||||
SidecarUnavailableError: class extends Error {},
|
||||
}));
|
||||
|
||||
const { bootCrmDb } = require('./helpers/crmDb');
|
||||
|
||||
let db; let cleanup; let faceProcessor;
|
||||
|
||||
async function seedExternalPhoto({ externalPath, relpath = 'individual/a.jpg' }) {
|
||||
const [e] = await db('events').insert({
|
||||
slug: `tr-${Math.random().toString(36).slice(2, 8)}`,
|
||||
event_type: 'wedding',
|
||||
event_name: 'tr',
|
||||
event_date: '2026-01-01',
|
||||
host_email: 'h@example.com',
|
||||
admin_email: 'a@example.com',
|
||||
password_hash: 'x',
|
||||
share_link: `tr-${Math.random()}`,
|
||||
expires_at: new Date().toISOString(),
|
||||
face_recognition_enabled: true,
|
||||
source_mode: 'reference',
|
||||
external_path: externalPath,
|
||||
}).returning('id');
|
||||
const eventId = typeof e === 'object' ? e.id : e;
|
||||
|
||||
const [p] = await db('photos').insert({
|
||||
event_id: eventId,
|
||||
filename: 'a.jpg',
|
||||
path: 'tr/a.jpg',
|
||||
type: 'individual',
|
||||
width: 4000,
|
||||
height: 3000,
|
||||
processing_status: 'complete',
|
||||
face_status: 'processing',
|
||||
source_origin: 'external',
|
||||
external_relpath: relpath,
|
||||
}).returning('id');
|
||||
return { eventId, photoId: typeof p === 'object' ? p.id : p };
|
||||
}
|
||||
|
||||
describe('transient source vs dead photo', () => {
|
||||
beforeAll(async () => {
|
||||
({ db, cleanup } = await bootCrmDb());
|
||||
await db('feature_flags').insert({ key: 'faces', value: true })
|
||||
.onConflict('key').merge()
|
||||
.catch(async () => { await db('feature_flags').where({ key: 'faces' }).update({ value: true }); });
|
||||
faceProcessor = require('../../src/services/faceProcessor');
|
||||
}, 120000);
|
||||
|
||||
afterAll(async () => {
|
||||
if (cleanup) await cleanup();
|
||||
await fs.promises.rm(tmpRoot, { recursive: true, force: true }).catch(() => {});
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
previewKeyResult = null; // i.e. ensurePreviewImage could not build one
|
||||
mockEnsurePreviewImage.mockClear();
|
||||
});
|
||||
|
||||
it('defers, not fails, when the source directory is unreachable', async () => {
|
||||
// Nothing was ever created under EXTERNAL_MEDIA_ROOT for this path, so the
|
||||
// directory does not resolve — the shape a dropped mount presents.
|
||||
const { photoId } = await seedExternalPhoto({ externalPath: 'vanished-share' });
|
||||
|
||||
await expect(faceProcessor.processPhotoFaces(photoId))
|
||||
.rejects.toBeInstanceOf(faceProcessor.TransientSourceError);
|
||||
|
||||
// Critically: still claimable. A 'failed' here is what stranded the photo.
|
||||
const photo = await db('photos').where({ id: photoId }).first();
|
||||
expect(photo.face_status).not.toBe('failed');
|
||||
});
|
||||
|
||||
it('defers when the event root survives an unmount but is empty', async () => {
|
||||
// The common NFS/SMB shape: unmounting leaves the mountpoint behind as an
|
||||
// ordinary empty directory, so fs.access succeeds on storage that is
|
||||
// entirely gone. The EVENT ROOT is the thing that goes empty — the photo's
|
||||
// own subdirectory vanishes with it.
|
||||
const emptyRoot = path.join(process.env.EXTERNAL_MEDIA_ROOT, 'unmounted');
|
||||
await fs.promises.mkdir(emptyRoot, { recursive: true });
|
||||
const { photoId } = await seedExternalPhoto({ externalPath: 'unmounted' });
|
||||
|
||||
await expect(faceProcessor.processPhotoFaces(photoId))
|
||||
.rejects.toBeInstanceOf(faceProcessor.TransientSourceError);
|
||||
|
||||
const photo = await db('photos').where({ id: photoId }).first();
|
||||
expect(photo.face_status).not.toBe('failed');
|
||||
});
|
||||
|
||||
it('fails when the directory is healthy but the file is gone', async () => {
|
||||
// Directory exists, file does not — a genuinely broken photo, which should
|
||||
// surface as a failure the admin can see rather than retry forever.
|
||||
const live = path.join(process.env.EXTERNAL_MEDIA_ROOT, 'live-share', 'individual');
|
||||
await fs.promises.mkdir(live, { recursive: true });
|
||||
// Non-empty: an empty directory is now read as an unmounted share, so the
|
||||
// "healthy storage, dead photo" case needs a sibling file present.
|
||||
await fs.promises.writeFile(path.join(live, 'sibling.jpg'), 'x');
|
||||
const { photoId } = await seedExternalPhoto({ externalPath: 'live-share' });
|
||||
|
||||
const result = await faceProcessor.processPhotoFaces(photoId);
|
||||
|
||||
expect(result.status).toBe('failed');
|
||||
const photo = await db('photos').where({ id: photoId }).first();
|
||||
expect(photo.face_status).toBe('failed');
|
||||
expect(photo.face_error).toMatch(/preview/i);
|
||||
});
|
||||
|
||||
it('fails a missing subdirectory rather than deferring the whole event', async () => {
|
||||
// individual/ deleted while collages/ is fine. Probing only the photo's own
|
||||
// directory reports ENOENT and would read as a mount-wide outage, deferring
|
||||
// the event and starving every healthy sibling folder. The root is
|
||||
// populated, so the mount is up and this is a broken path.
|
||||
const root = path.join(process.env.EXTERNAL_MEDIA_ROOT, 'partial');
|
||||
await fs.promises.mkdir(path.join(root, 'collages'), { recursive: true });
|
||||
await fs.promises.writeFile(path.join(root, 'collages', 'kept.jpg'), 'x');
|
||||
const { photoId } = await seedExternalPhoto({ externalPath: 'partial' });
|
||||
|
||||
const result = await faceProcessor.processPhotoFaces(photoId);
|
||||
expect(result.status).toBe('failed');
|
||||
});
|
||||
|
||||
it('defers a file that exists but cannot be read', async () => {
|
||||
// EACCES / EIO / ESTALE on the file itself, with the mount up: a transient
|
||||
// condition wearing a per-file disguise. Only ENOENT means genuinely gone.
|
||||
const root = path.join(process.env.EXTERNAL_MEDIA_ROOT, 'locked');
|
||||
const dir = path.join(root, 'individual');
|
||||
await fs.promises.mkdir(dir, { recursive: true });
|
||||
const file = path.join(dir, 'a.jpg');
|
||||
await fs.promises.writeFile(file, 'x');
|
||||
await fs.promises.chmod(file, 0o000);
|
||||
|
||||
const { photoId } = await seedExternalPhoto({ externalPath: 'locked' });
|
||||
try {
|
||||
await expect(faceProcessor.processPhotoFaces(photoId))
|
||||
.rejects.toBeInstanceOf(faceProcessor.TransientSourceError);
|
||||
} finally {
|
||||
await fs.promises.chmod(file, 0o644).catch(() => {});
|
||||
}
|
||||
});
|
||||
|
||||
it('still fails managed photos without probing the mount', async () => {
|
||||
// The probe is scoped to external/reference rows: a managed photo with no
|
||||
// preview is broken, and there is no mount to blame.
|
||||
const [e] = await db('events').insert({
|
||||
slug: `tr-m-${Math.random().toString(36).slice(2, 8)}`,
|
||||
event_type: 'wedding',
|
||||
event_name: 'trm',
|
||||
event_date: '2026-01-01',
|
||||
host_email: 'h@example.com',
|
||||
admin_email: 'a@example.com',
|
||||
password_hash: 'x',
|
||||
share_link: `tr-m-${Math.random()}`,
|
||||
expires_at: new Date().toISOString(),
|
||||
face_recognition_enabled: true,
|
||||
}).returning('id');
|
||||
const [p] = await db('photos').insert({
|
||||
event_id: typeof e === 'object' ? e.id : e,
|
||||
filename: 'm.jpg',
|
||||
path: 'trm/m.jpg',
|
||||
type: 'individual',
|
||||
width: 100,
|
||||
height: 100,
|
||||
processing_status: 'complete',
|
||||
face_status: 'processing',
|
||||
source_origin: 'managed',
|
||||
}).returning('id');
|
||||
|
||||
const result = await faceProcessor.processPhotoFaces(typeof p === 'object' ? p.id : p);
|
||||
expect(result.status).toBe('failed');
|
||||
});
|
||||
});
|
||||
@@ -94,6 +94,33 @@ router.post('/events/:id/import-external', adminAuth, requirePermission('photos.
|
||||
}
|
||||
}
|
||||
|
||||
// Point the event at the new directory BEFORE inserting anything.
|
||||
//
|
||||
// Two reasons, both about what a half-finished import leaves behind. The
|
||||
// update used to run after the loop, so an import that died at photo 500
|
||||
// of 1000 left those 500 rows carrying external_relpath into the NEW tree
|
||||
// while the event still resolved against the OLD one — every one of them
|
||||
// unreadable. And with face detection on, enqueueEvent accepts
|
||||
// processing_status NULL (faceProcessor.js:243-246), which these inserts
|
||||
// leave unset, so an admin hitting the toggle or Re-scan mid-import could
|
||||
// queue those same rows against the stale path and burn them to 'failed'.
|
||||
//
|
||||
// Safe to do first for existing MANAGED photos: photo.source_origin takes
|
||||
// precedence over event.source_mode in both resolvers (photoResolver.js:23,
|
||||
// :52) and is NOT NULL defaulting to 'managed', so flipping source_mode
|
||||
// does not touch them.
|
||||
//
|
||||
// It does NOT isolate existing EXTERNAL rows — resolveExternalPath prefixes
|
||||
// every one of them with event.external_path
|
||||
// (externalMediaService.js:97-102), so importing folder B into an event
|
||||
// that already references folder A rebases the A rows onto B. That is
|
||||
// pre-existing: the update always did this, just after the loop instead of
|
||||
// before it, so moving it changes when the window opens and not whether it
|
||||
// exists. The underlying limitation is that an event carries a single
|
||||
// external base path, which makes importing a second folder into it
|
||||
// incoherent either way — worth its own fix, not this one.
|
||||
await db('events').where('id', eventId).update({ source_mode: 'reference', external_path });
|
||||
|
||||
let imported = 0;
|
||||
let thumbnailsGenerated = 0;
|
||||
let thumbnailsFailed = 0;
|
||||
@@ -112,14 +139,11 @@ router.post('/events/:id/import-external', adminAuth, requirePermission('photos.
|
||||
// imported after that moment stuck at NULL forever — the toggle endpoint
|
||||
// only queues rows that already existed when it fired.
|
||||
//
|
||||
// Collected during the loop and marked 'pending' only AFTER
|
||||
// events.external_path is written below. Setting it on insert would publish
|
||||
// claimable rows while the event still carries the old path — or none, on a
|
||||
// first import: the face worker polls continuously, resolvePhotoFilePath
|
||||
// would resolve against the wrong directory, and a multi-minute import
|
||||
// would leave photos permanently 'failed', a state only an explicit
|
||||
// Re-scan clears. No video guard needed either way: walkDir collects only
|
||||
// jpg/jpeg/png/webp.
|
||||
// The event path is already committed (above), so the enqueue below is
|
||||
// free of the ordering hazard it used to carry. It stays at the end anyway
|
||||
// so the setting can be read after the loop, and it only touches rows that
|
||||
// are still untouched — see the whereNull there. No video guard needed:
|
||||
// walkDir collects only jpg/jpeg/png/webp.
|
||||
const importedPhotoIds = [];
|
||||
|
||||
// Insert photos
|
||||
@@ -200,11 +224,10 @@ router.post('/events/:id/import-external', adminAuth, requirePermission('photos.
|
||||
}
|
||||
}
|
||||
|
||||
// Update event fields
|
||||
await db('events').where('id', eventId).update({ source_mode: 'reference', external_path });
|
||||
|
||||
// Only now does the event resolve to the new directory, so only now is it
|
||||
// safe to open the queue. Guarded the same way photoProcessor guards it
|
||||
// The event already resolves to the new directory (set before the loop),
|
||||
// so the queue is safe to open. Still done here rather than on insert so
|
||||
// the setting below is read after the loop. Guarded the same way
|
||||
// photoProcessor guards it
|
||||
// (both the global flag and the per-event toggle), so installs without the
|
||||
// feature still never write a face_status. Re-read here rather than before
|
||||
// the loop so a toggle flipped mid-import is honoured.
|
||||
@@ -221,12 +244,20 @@ router.post('/events/:id/import-external', adminAuth, requirePermission('photos.
|
||||
// Chunked because SQLite caps a statement at 999 bound parameters and an
|
||||
// import can be far larger than that.
|
||||
if (queueFaces && importedPhotoIds.length) {
|
||||
let queued = 0;
|
||||
for (let i = 0; i < importedPhotoIds.length; i += 500) {
|
||||
await db('photos')
|
||||
// whereNull, not a blanket set. Committing the event path before the
|
||||
// loop means a toggle or Re-scan firing mid-import can now genuinely
|
||||
// queue and even finish some of these rows — so an unconditional
|
||||
// update would drag 'done' rows back to 'pending' for a duplicate
|
||||
// scan, and knock 'processing' rows out from under the worker
|
||||
// mid-flight. Only rows nothing has touched are ours to queue.
|
||||
queued += await db('photos')
|
||||
.whereIn('id', importedPhotoIds.slice(i, i + 500))
|
||||
.whereNull('face_status')
|
||||
.update({ face_status: 'pending' });
|
||||
}
|
||||
logger.info(`Queued ${importedPhotoIds.length} imported external photo(s) for face scanning (event ${eventId})`);
|
||||
logger.info(`Queued ${queued} of ${importedPhotoIds.length} imported external photo(s) for face scanning (event ${eventId})`);
|
||||
}
|
||||
|
||||
await logActivity(
|
||||
|
||||
@@ -34,6 +34,96 @@ class StaleScanError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Thrown when the photo's source could not be READ, as opposed to being
|
||||
* unreadable. An NFS or SMB mount that is down, remounting, or refusing
|
||||
* permissions makes ensurePreviewImage return null exactly like a corrupt
|
||||
* JPEG does — but one clears itself and the other never will. Marking the
|
||||
* first 'failed' strands the photo, because the queue only ever claims
|
||||
* 'pending' and nothing re-queues a failure automatically: a mount that
|
||||
* blinked during a scan of an external library would cost the whole gallery
|
||||
* a manual Re-scan.
|
||||
*/
|
||||
class TransientSourceError extends Error {
|
||||
constructor(photoId, reason, eventId) {
|
||||
super(`Face scan for photo ${photoId} deferred: ${reason}`);
|
||||
this.name = 'TransientSourceError';
|
||||
// The queue backs off per EVENT, not per photo: one dead mount makes every
|
||||
// row in that event unreachable, and probing each of them costs a stat
|
||||
// against storage that may be hard-mounted and slow to time out.
|
||||
this.eventId = eventId;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Distinguish "this mount is not there" from "this file is not there".
|
||||
*
|
||||
* Deliberately probes the containing DIRECTORY rather than the file. A missing
|
||||
* file inside a healthy directory is a genuinely broken photo and should fail;
|
||||
* a directory that cannot be reached at all means the storage is gone, and
|
||||
* every photo under it would otherwise be burnt in a loop.
|
||||
*
|
||||
* Returns a reason string when the source looks unreachable, null when it
|
||||
* looks like a per-photo problem. Any error deciding that is treated as
|
||||
* per-photo: guessing 'transient' on an unknown condition would retry forever.
|
||||
*/
|
||||
async function externalSourceUnreachable(photo) {
|
||||
if (photo.source_origin !== 'external' && photo.source_origin !== 'reference') return null;
|
||||
try {
|
||||
const fs = require('fs').promises;
|
||||
const { resolvePhotoFilePath } = require('./photoResolver');
|
||||
const event = await db('events').where({ id: photo.event_id }).first();
|
||||
if (!event) return null;
|
||||
|
||||
const { resolveExternalPath } = require('./externalMediaService');
|
||||
const filePath = resolvePhotoFilePath(event, photo);
|
||||
|
||||
// 1. The EVENT ROOT, not the photo's own directory. This is what decides
|
||||
// whether the storage is there, and it has to be judged separately: a
|
||||
// deleted or renamed subfolder (individual/ gone, collages/ still fine)
|
||||
// also reports ENOENT on the photo's directory, and treating that as an
|
||||
// outage would defer the whole event and starve the healthy siblings.
|
||||
const eventRoot = resolveExternalPath(event, '');
|
||||
try {
|
||||
await fs.access(eventRoot);
|
||||
} catch (e) {
|
||||
return `source root unreachable (${e.code || e.message})`;
|
||||
}
|
||||
try {
|
||||
// Unmounting an NFS or SMB share usually leaves the mountpoint behind as
|
||||
// an ordinary empty directory, so access() succeeds on storage that is
|
||||
// entirely gone. An empty root is that outage.
|
||||
//
|
||||
// The trade is deliberate: a root an admin genuinely emptied is retried
|
||||
// rather than failed. That costs one attempt per backoff window, because
|
||||
// a deferred row parks in 'processing' rather than returning to the front
|
||||
// of the queue. Burning a gallery on a flapping mount is the worse one.
|
||||
const entries = await fs.readdir(eventRoot);
|
||||
if (entries.length === 0) return 'source root is empty (mount not attached?)';
|
||||
} catch (e) {
|
||||
return `source root unreadable (${e.code || e.message})`;
|
||||
}
|
||||
|
||||
// 2. The storage is up, so anything wrong from here is about this photo —
|
||||
// with one exception. A file that exists but cannot be READ (EACCES on a
|
||||
// reconnected share, EIO, or the classic NFS ESTALE handle) is a
|
||||
// transient condition wearing a per-file disguise; only ENOENT means the
|
||||
// photo is genuinely gone.
|
||||
try {
|
||||
await fs.access(filePath, fs.constants.R_OK);
|
||||
} catch (e) {
|
||||
if (e.code && e.code !== 'ENOENT') {
|
||||
return `source file unreadable (${e.code})`;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
} catch (e) {
|
||||
// Could not even resolve a path — that is a property of this row, not of
|
||||
// the mount, so let the caller fail it.
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
async function streamToBuffer(stream) {
|
||||
if (Buffer.isBuffer(stream)) return stream;
|
||||
const chunks = [];
|
||||
@@ -89,6 +179,12 @@ async function processPhotoFaces(photoId) {
|
||||
// photo, not an unsupported one.
|
||||
const previewKey = await ensurePreviewImage(photo);
|
||||
if (!previewKey) {
|
||||
// Before failing the photo, check whether it is the STORAGE that is gone
|
||||
// rather than the file. ensurePreviewImage returns null for both, but only
|
||||
// one of them is the photo's fault — see TransientSourceError.
|
||||
const unreachable = await externalSourceUnreachable(photo);
|
||||
if (unreachable) throw new TransientSourceError(photoId, unreachable, photo.event_id);
|
||||
|
||||
// No preview and no way to make one — the source is missing or corrupt.
|
||||
// That is a property of this photo, so it fails rather than retries.
|
||||
await db('photos').where({ id: photoId }).update({
|
||||
@@ -322,4 +418,5 @@ async function purgePhotoFaces(photoId, trx = db) {
|
||||
|
||||
module.exports = {
|
||||
processPhotoFaces, enqueueEvent, purgeEvent, purgePhotoFaces, StaleScanError,
|
||||
TransientSourceError,
|
||||
};
|
||||
|
||||
@@ -30,6 +30,7 @@ const { db } = require('../database/db');
|
||||
const logger = require('../utils/logger');
|
||||
const { processPhotoFaces } = require('./faceProcessor');
|
||||
const { SidecarUnavailableError } = require('./faceClient');
|
||||
const { TransientSourceError } = require('./faceProcessor');
|
||||
const { isFeatureEnabled } = require('./faceSettings');
|
||||
|
||||
const POLL_INTERVAL_MS = parseInt(process.env.FACE_PROCESSOR_POLL_MS || '2000', 10);
|
||||
@@ -41,6 +42,65 @@ const JANITOR_INTERVAL_MS = 60 * 1000;
|
||||
// hot loop: claim, fail, release, claim again, thousands of times a minute.
|
||||
const UNAVAILABLE_BACKOFF_MS = parseInt(process.env.FACE_PROCESSOR_BACKOFF_MS || '30000', 10);
|
||||
|
||||
// A dropped mount hits every photo in the event, so the raw message would
|
||||
// repeat once per claim. Rate-limited the same way faceClient limits its own
|
||||
// unavailable line, and for the same reason: one warning per outage, not one
|
||||
// per photo.
|
||||
// How long an event whose storage is unreachable is left alone. Distinct from
|
||||
// the janitor's stuck timeout on purpose: the janitor exists to rescue rows a
|
||||
// crashed worker abandoned, and reusing it here meant that every sweep handed
|
||||
// the whole dead gallery back to the worker, which then walked all of it again
|
||||
// — one slow stat per photo against a mount that may be hard-mounted — before
|
||||
// reaching any healthy event.
|
||||
const SOURCE_BACKOFF_MS = parseInt(process.env.FACE_SOURCE_BACKOFF_MS || '300000', 10);
|
||||
|
||||
// eventId -> epoch ms before which this event's photos are not worth claiming.
|
||||
// In-memory on purpose: a restart should retry immediately, since a restart is
|
||||
// usually what follows fixing the mount.
|
||||
const deferredEvents = new Map();
|
||||
|
||||
function deferEvent(eventId) {
|
||||
if (eventId == null) return;
|
||||
deferredEvents.set(eventId, Date.now() + SOURCE_BACKOFF_MS);
|
||||
}
|
||||
|
||||
/** Event ids still inside their backoff window; also prunes expired entries. */
|
||||
function currentlyDeferredEventIds() {
|
||||
const now = Date.now();
|
||||
for (const [id, until] of deferredEvents) {
|
||||
if (until <= now) deferredEvents.delete(id);
|
||||
}
|
||||
return [...deferredEvents.keys()];
|
||||
}
|
||||
|
||||
/**
|
||||
* Exclude only the rows the outage actually affects.
|
||||
*
|
||||
* A reference event can hold managed uploads alongside its imported external
|
||||
* ones, and those live in local storage that is fine. Excluding the whole
|
||||
* event id would leave them unscanned for as long as external rows keep
|
||||
* renewing the cooldown — which, during a real outage, is indefinitely.
|
||||
*/
|
||||
function applySourceBackoff(query, excludeEventIds) {
|
||||
if (!excludeEventIds.length) return query;
|
||||
return query.whereNot(function () {
|
||||
this.whereIn('event_id', excludeEventIds)
|
||||
.whereIn('source_origin', ['external', 'reference']);
|
||||
});
|
||||
}
|
||||
|
||||
const SOURCE_LOG_INTERVAL_MS = 5 * 60 * 1000;
|
||||
let lastUnreachableLogAt = 0;
|
||||
function logUnreachableSource(message) {
|
||||
const now = Date.now();
|
||||
if (now - lastUnreachableLogAt < SOURCE_LOG_INTERVAL_MS) return;
|
||||
lastUnreachableLogAt = now;
|
||||
logger.warn(
|
||||
`faceQueue: ${message}. Photos stay queued and will be retried — check the `
|
||||
+ 'external media mount. Further identical warnings are suppressed for 5 minutes.'
|
||||
);
|
||||
}
|
||||
|
||||
let running = false;
|
||||
let workerHandles = [];
|
||||
let janitorHandle = null;
|
||||
@@ -57,11 +117,12 @@ function isPostgres() {
|
||||
* Same two-path approach as backgroundProcessor: SKIP LOCKED on Postgres so
|
||||
* multiple pods race cleanly, a status-guarded UPDATE on SQLite.
|
||||
*/
|
||||
async function claimNextPhoto() {
|
||||
async function claimNextPhoto(excludeEventIds = []) {
|
||||
if (isPostgres()) {
|
||||
return db.transaction(async (trx) => {
|
||||
const row = await trx('photos')
|
||||
.where('face_status', 'pending')
|
||||
.modify((q) => applySourceBackoff(q, excludeEventIds))
|
||||
.orderBy('id', 'asc')
|
||||
.forUpdate()
|
||||
.skipLocked()
|
||||
@@ -78,6 +139,7 @@ async function claimNextPhoto() {
|
||||
return db.transaction(async (trx) => {
|
||||
const row = await trx('photos')
|
||||
.where('face_status', 'pending')
|
||||
.modify((q) => applySourceBackoff(q, excludeEventIds))
|
||||
.orderBy('id', 'asc')
|
||||
.first();
|
||||
if (!row) return null;
|
||||
@@ -114,7 +176,7 @@ async function workerLoop(workerIdx) {
|
||||
|
||||
let claimed;
|
||||
try {
|
||||
claimed = await claimNextPhoto();
|
||||
claimed = await claimNextPhoto(currentlyDeferredEventIds());
|
||||
} catch (e) {
|
||||
logger.warn(`faceQueue[${workerIdx}]: claim error`, { error: e.message });
|
||||
await sleep(POLL_INTERVAL_MS);
|
||||
@@ -129,6 +191,9 @@ async function workerLoop(workerIdx) {
|
||||
try {
|
||||
await processPhotoFaces(claimed.id);
|
||||
} catch (err) {
|
||||
// The sidecar being down stops EVERY photo, so returning this one to
|
||||
// 'pending' and backing off costs nothing — there is no other work to
|
||||
// get on with.
|
||||
if (err instanceof SidecarUnavailableError) {
|
||||
// Retry, don't fail. faceClient already rate-limits the log line.
|
||||
await releaseToPending(claimed.id).catch(() => {});
|
||||
@@ -136,6 +201,27 @@ async function workerLoop(workerIdx) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// An unreachable source is per-EVENT, not global: other events are
|
||||
// still perfectly scannable. Releasing to 'pending' here would be a
|
||||
// trap — claimNextPhoto orders by id ascending, so the single default
|
||||
// worker would reclaim this same row after every backoff and never
|
||||
// reach any higher id. One dead mount would stall face scanning for
|
||||
// the whole install.
|
||||
//
|
||||
// So leave it parked in 'processing' with its face_started_at intact.
|
||||
// The worker moves straight on to the next claimable row, and the
|
||||
// janitor returns this one to 'pending' once it passes
|
||||
// STUCK_TIMEOUT_MS — which is exactly the "try again later" this
|
||||
// needs, using machinery that already exists.
|
||||
if (err instanceof TransientSourceError) {
|
||||
// Back the whole EVENT off, not just this row. Its siblings are on the
|
||||
// same storage and would each cost another failed preview attempt —
|
||||
// which also logs — before landing here again.
|
||||
deferEvent(err.eventId);
|
||||
logUnreachableSource(err.message);
|
||||
continue;
|
||||
}
|
||||
|
||||
logger.error(`faceQueue[${workerIdx}]: photo ${claimed.id} failed`, {
|
||||
error: err.message,
|
||||
stack: err.stack,
|
||||
|
||||
Reference in New Issue
Block a user