feat(faces): consolidate look-alike clusters after a scan, and suggest the rest (#1107)
consolidate() has existed since #1074 and described this exact symptom in its own comment, but its only caller was recluster() — i.e. when an admin pressed Re-group people. After a normal background scan the centroids converged and nobody looked, so a gallery settled with 14 people that should have been 8. It now runs when a scan drains. There is no scan-finished event to hook, so an idle worker asks whether the events it touched have actually drained — 'a worker went idle' is deliberately not treated as sufficient, because with concurrency above one the others may still be working. The uncertain band asks instead of acting: pairs between the assignment threshold and the stricter auto-merge one surface as accept/dismiss suggestions, with sticky dismissals. Nothing merges silently — a pass that merged anything reports it and points at Split. Review rounds hardened it against overruling explicit decisions: it no longer absorbs ignored clusters (mergePeople ORs is_ignored onto the survivor, which would have hidden a real person), no longer merges dismissed pairs, no longer undoes a manual Split (which now records a separation), and no longer runs after detection is switched off. The dismissal read fails closed, a failed pass is retried with backoff rather than lost or hot-looped, and the new table follows event_people out of exports and backups. Name autocomplete needs no endpoint — the people list already open is the source, and it is event-scoped on purpose. Known limitation, tracked in #1132: separations are keyed on person ids, so a full re-scan loses them. Reported by @BraynArts.
This commit is contained in:
Paul Nothaft
@@ -99,6 +99,14 @@ module.exports = (router) => {
|
||||
enabled: event.face_recognition_enabled === true || event.face_recognition_enabled === 1,
|
||||
visible_to_guests: faceSettings.areFacesVisibleToGuests(event),
|
||||
last_scan_at: event.faces_last_scan_at || null,
|
||||
// What the automatic consolidation pass did when this event last
|
||||
// drained (#1107). Surfaced because merging biometric clusters
|
||||
// without saying so is the wrong default, however confident the
|
||||
// similarity was.
|
||||
consolidation: {
|
||||
merged: Number(event.faces_last_consolidated_count || 0),
|
||||
at: event.faces_last_consolidated_at || null,
|
||||
},
|
||||
status,
|
||||
});
|
||||
} catch (error) {
|
||||
@@ -189,6 +197,71 @@ module.exports = (router) => {
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* Look-alike pairs in the band BELOW the auto-merge threshold (#1107).
|
||||
*
|
||||
* Registered before the '/:id/people/:personId' family so the literal
|
||||
* 'suggestions' segment can never be read as a person id — see ./index.js on
|
||||
* why registration order is load-bearing here. It would not collide today
|
||||
* (that route has no GET), but relying on that is one refactor away from a
|
||||
* person id of "suggestions" reaching the database.
|
||||
*
|
||||
* Returns ids and a score only. Every caller already holds the people list
|
||||
* with its covers, so re-sending face crops here would duplicate a payload
|
||||
* the modal has open in front of it.
|
||||
*/
|
||||
router.get('/:id/people/suggestions',
|
||||
adminAuth, requirePermission('events.view'), requireFaces, requireEventOwnership,
|
||||
async (req, res) => {
|
||||
try {
|
||||
const event = await loadOwnedEvent(req);
|
||||
if (!event) return res.status(404).json({ error: 'Event not found' });
|
||||
|
||||
const suggestions = await faceClustering.suggestMerges(event.id);
|
||||
res.json({ suggestions });
|
||||
} catch (error) {
|
||||
errorResponse(res, error, 500, 'Failed to fetch merge suggestions');
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* "These two are not the same person." Sticky, so the pair stops coming back
|
||||
* after every scan.
|
||||
*/
|
||||
router.post('/:id/people/suggestions/dismiss',
|
||||
adminAuth,
|
||||
requirePermission('events.edit'),
|
||||
requireFaces,
|
||||
requireEventOwnership,
|
||||
[body('person_a_id').isInt(), body('person_b_id').isInt()],
|
||||
async (req, res) => {
|
||||
const errors = validationResult(req);
|
||||
if (!errors.isEmpty()) return res.status(400).json({ errors: errors.array() });
|
||||
|
||||
try {
|
||||
const event = await loadOwnedEvent(req);
|
||||
if (!event) return res.status(404).json({ error: 'Event not found' });
|
||||
|
||||
// Both ids must belong to this event. Without this an admin could
|
||||
// write dismissal rows naming people in someone else's gallery —
|
||||
// harmless on its own, but it is other tenants' data in our table.
|
||||
const ids = [Number(req.body.person_a_id), Number(req.body.person_b_id)];
|
||||
if (ids[0] === ids[1]) {
|
||||
return res.status(400).json({ error: 'A person cannot be dismissed against itself' });
|
||||
}
|
||||
const owned = await db('event_people')
|
||||
.where({ event_id: event.id }).whereIn('id', ids).pluck('id');
|
||||
if (owned.length !== 2) {
|
||||
return res.status(400).json({ error: 'One or more people do not belong to this event' });
|
||||
}
|
||||
|
||||
const result = await faceClustering.dismissMergeSuggestion(event.id, ids[0], ids[1]);
|
||||
res.json({ success: true, ...result });
|
||||
} catch (error) {
|
||||
errorResponse(res, error, 500, 'Failed to dismiss suggestion');
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* Rename / hide / ignore / set cover.
|
||||
*/
|
||||
|
||||
@@ -28,7 +28,7 @@ const PROGRESS_INTERVAL = 100; // Report progress every 100 rows
|
||||
// SQLite cannot filter at all — `sqlite3 .backup` is a whole-file binary copy
|
||||
// — so the rows are deleted from the temp copy before it is finalised. See
|
||||
// createSQLiteBackup below.
|
||||
const FACE_TABLES = ['photo_faces', 'event_people'];
|
||||
const FACE_TABLES = ['photo_faces', 'event_people', 'event_people_merge_dismissals'];
|
||||
|
||||
/**
|
||||
* Database Backup Service
|
||||
|
||||
@@ -247,6 +247,60 @@ async function assignFacesLocked(eventId, faceRows, thresholds, trx) {
|
||||
return assignments;
|
||||
}
|
||||
|
||||
/**
|
||||
* Does this error mean the table isn't there yet, as opposed to the query
|
||||
* failing? Postgres reports SQLSTATE 42P01; SQLite says so in the message.
|
||||
* The distinction decides whether a dismissal read may fail open.
|
||||
*/
|
||||
function isMissingTable(err) {
|
||||
if (!err) return false;
|
||||
// Postgres: undefined_table. Deliberately NOT matched on the message — its
|
||||
// "does not exist" wording also covers a missing COLUMN, which is a broken
|
||||
// query rather than a pre-migration install and must not fail open.
|
||||
if (err.code === '42P01') return true;
|
||||
return /no such table/i.test(err.message || '');
|
||||
}
|
||||
|
||||
/**
|
||||
* One identity for a pair regardless of which order it was produced in.
|
||||
* Rows are stored the same way, so the two always agree.
|
||||
*/
|
||||
function pairKey(a, b) {
|
||||
return a < b ? `${a}:${b}` : `${b}:${a}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Pairs the photographer has explicitly kept apart (#1107).
|
||||
*
|
||||
* Read by BOTH the automatic pass and the suggestion list: "not the same
|
||||
* person" has to bind the thing that acts on its own even more than it binds
|
||||
* the thing that asks. Missing table (pre-migration) reads as "nothing
|
||||
* dismissed" rather than failing the merge that called this.
|
||||
*/
|
||||
async function loadDismissedPairs(eventId) {
|
||||
try {
|
||||
const rows = await db('event_people_merge_dismissals')
|
||||
.where({ event_id: eventId })
|
||||
.select('person_a_id', 'person_b_id');
|
||||
return new Set(rows.map((d) => pairKey(d.person_a_id, d.person_b_id)));
|
||||
} catch (err) {
|
||||
// ONLY a missing table reads as "nothing dismissed" — that is a
|
||||
// pre-migration install, where by definition nothing has been dismissed.
|
||||
//
|
||||
// Everything else FAILS CLOSED. Treating a timeout or a permission error
|
||||
// as an empty set would let the automatic pass merge pairs the
|
||||
// photographer explicitly separated, which is precisely the decision this
|
||||
// set exists to protect. The caller defers instead: drainConsolidation
|
||||
// backs the event off and retries.
|
||||
if (!isMissingTable(err)) throw err;
|
||||
logger.warn(
|
||||
`faceClustering: merge-dismissal table absent for event ${eventId} — treating as none`,
|
||||
{ error: err.message }
|
||||
);
|
||||
return new Set();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Merge people whose centroids have drifted together.
|
||||
*
|
||||
@@ -264,40 +318,214 @@ async function consolidate(eventId, options = {}) {
|
||||
|
||||
const people = await db('event_people')
|
||||
.where({ event_id: eventId })
|
||||
.select('id', 'centroid', 'face_count_total', 'model_version', 'label');
|
||||
.select('id', 'centroid', 'face_count_total', 'model_version', 'label', 'is_ignored');
|
||||
|
||||
const state = people
|
||||
// "Not a real person" must never be merged INTO one. mergePeople ORs
|
||||
// is_ignored onto the survivor, so absorbing a false-positive cluster
|
||||
// would mark a real person ignored and drop them out of the guest-facing
|
||||
// strip entirely. Cheap to skip, expensive to discover.
|
||||
.filter((p) => !(p.is_ignored === true || p.is_ignored === 1))
|
||||
.map((p) => ({ ...p, vec: unpackEmbedding(p.centroid) }))
|
||||
.filter((p) => p.vec);
|
||||
|
||||
// A pair the photographer answered "not the same" about stays not the same,
|
||||
// however far the centroids drift afterwards. Without this the automatic
|
||||
// pass silently overturns an explicit human decision the moment new faces
|
||||
// push the pair over the threshold — or the moment someone tunes it.
|
||||
const dismissed = await loadDismissedPairs(eventId);
|
||||
|
||||
const merged = [];
|
||||
const absorbed = new Set();
|
||||
|
||||
// Each mergePeople is its own transaction, so a pass that dies halfway has
|
||||
// still committed what it did. Reporting has to survive that: recorded in a
|
||||
// finally, or a failure on the second pair would leave the first one merged
|
||||
// and unreported — silent, which is the one thing this must never be. The
|
||||
// retry then re-reports against whatever is left.
|
||||
try {
|
||||
for (let i = 0; i < state.length; i++) {
|
||||
if (absorbed.has(state[i].id)) continue;
|
||||
for (let j = i + 1; j < state.length; j++) {
|
||||
if (absorbed.has(state[j].id)) continue;
|
||||
const a = state[i];
|
||||
const b = state[j];
|
||||
if (a.model_version !== b.model_version) continue;
|
||||
|
||||
// Never silently merge two people the photographer has NAMED
|
||||
// differently — that is a human assertion this heuristic does not get
|
||||
// to overrule.
|
||||
if (a.label && b.label && a.label !== b.label) continue;
|
||||
|
||||
if (dismissed.has(pairKey(a.id, b.id))) continue;
|
||||
|
||||
if (dot(a.vec, b.vec) >= mergeThreshold) {
|
||||
// Re-read this one pair immediately before acting. The set above was
|
||||
// loaded once for the whole pass, and a photographer pressing "Not the
|
||||
// same" during it would otherwise be overruled by a decision that was
|
||||
// already stale when it was made. Narrows the window to a single
|
||||
// statement rather than the length of the pass; one extra query per
|
||||
// pair that is actually about to merge, which is rare.
|
||||
const justDismissed = await db('event_people_merge_dismissals')
|
||||
.where({
|
||||
event_id: eventId,
|
||||
person_a_id: Math.min(a.id, b.id),
|
||||
person_b_id: Math.max(a.id, b.id),
|
||||
})
|
||||
.first()
|
||||
.catch((err) => {
|
||||
if (isMissingTable(err)) return null;
|
||||
throw err;
|
||||
});
|
||||
if (justDismissed) continue;
|
||||
|
||||
await mergePeople(eventId, [b.id], a.id);
|
||||
absorbed.add(b.id);
|
||||
merged.push({ from: b.id, into: a.id });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
} finally {
|
||||
if (merged.length) {
|
||||
logger.info(`faceClustering: consolidated ${merged.length} person pair(s) in event ${eventId}`);
|
||||
}
|
||||
|
||||
// Record the outcome even when it is zero. Merging biometric clusters
|
||||
// silently is the wrong default however confident the maths is (#1107), so
|
||||
// the admin card reports what this pass did — and a run that merged nothing
|
||||
// has to clear a previous run's count rather than leave it standing.
|
||||
await db('events').where({ id: eventId }).update({
|
||||
faces_last_consolidated_count: merged.length,
|
||||
faces_last_consolidated_at: new Date().toISOString(),
|
||||
}).catch((err) => {
|
||||
// Pre-migration installs simply do not report. Never let bookkeeping fail
|
||||
// a merge that already happened.
|
||||
logger.warn(`faceClustering: could not record consolidation for event ${eventId}`, {
|
||||
error: err.message,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
return merged;
|
||||
}
|
||||
|
||||
/**
|
||||
* Pairs that look like the same person but not confidently enough to merge
|
||||
* automatically (#1107).
|
||||
*
|
||||
* The band is [match_threshold, merge_threshold): above the top of it
|
||||
* `consolidate()` has already merged the pair, and below the bottom the two
|
||||
* centroids are further apart than the distance at which a single face would
|
||||
* have joined the cluster at all — which is not a claim worth putting in front
|
||||
* of anyone.
|
||||
*
|
||||
* This is the "with a warning" half of the request. An over-eager merge is much
|
||||
* harder to unpick than a missed one, so the uncertain band never merges by
|
||||
* itself; it asks.
|
||||
*/
|
||||
async function suggestMerges(eventId, options = {}) {
|
||||
const thresholds = options.thresholds || (await getThresholds());
|
||||
const mergeThreshold = Math.min(0.95, thresholds.face_match_threshold + 0.08);
|
||||
const floor = thresholds.face_match_threshold;
|
||||
const limit = options.limit || 20;
|
||||
|
||||
const people = await db('event_people')
|
||||
.where({ event_id: eventId })
|
||||
.select('id', 'centroid', 'face_count_total', 'model_version', 'label', 'is_ignored');
|
||||
|
||||
const state = people
|
||||
// "Not a real person" is an answer already given — never ask about it again.
|
||||
.filter((p) => !(p.is_ignored === true || p.is_ignored === 1))
|
||||
.map((p) => ({ ...p, vec: unpackEmbedding(p.centroid) }))
|
||||
.filter((p) => p.vec);
|
||||
|
||||
if (state.length < 2) return [];
|
||||
|
||||
const dismissed = await loadDismissedPairs(eventId);
|
||||
|
||||
const pairs = [];
|
||||
for (let i = 0; i < state.length; i++) {
|
||||
if (absorbed.has(state[i].id)) continue;
|
||||
for (let j = i + 1; j < state.length; j++) {
|
||||
if (absorbed.has(state[j].id)) continue;
|
||||
const a = state[i];
|
||||
const b = state[j];
|
||||
if (a.model_version !== b.model_version) continue;
|
||||
|
||||
// Never silently merge two people the photographer has NAMED
|
||||
// differently — that is a human assertion this heuristic does not get
|
||||
// to overrule.
|
||||
// Same rule consolidate() applies: two different names is a human
|
||||
// assertion, not a question.
|
||||
if (a.label && b.label && a.label !== b.label) continue;
|
||||
|
||||
if (dot(a.vec, b.vec) >= mergeThreshold) {
|
||||
await mergePeople(eventId, [b.id], a.id);
|
||||
absorbed.add(b.id);
|
||||
merged.push({ from: b.id, into: a.id });
|
||||
}
|
||||
if (dismissed.has(pairKey(a.id, b.id))) continue;
|
||||
|
||||
const score = dot(a.vec, b.vec);
|
||||
if (score < floor || score >= mergeThreshold) continue;
|
||||
|
||||
pairs.push({
|
||||
person_a_id: Math.min(a.id, b.id),
|
||||
person_b_id: Math.max(a.id, b.id),
|
||||
score,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if (merged.length) {
|
||||
logger.info(`faceClustering: consolidated ${merged.length} person pair(s) in event ${eventId}`);
|
||||
// Most-similar first: the strongest suggestion is the one most likely to be
|
||||
// accepted, and a photographer working down the list should meet it first.
|
||||
pairs.sort((x, y) => y.score - x.score);
|
||||
|
||||
// One suggestion per person per round. Without this a cluster that genuinely
|
||||
// has three fragments produces A-B, A-C and B-C, and accepting A-B leaves two
|
||||
// suggestions pointing at a person that no longer exists.
|
||||
const used = new Set();
|
||||
const result = [];
|
||||
for (const pair of pairs) {
|
||||
if (used.has(pair.person_a_id) || used.has(pair.person_b_id)) continue;
|
||||
used.add(pair.person_a_id);
|
||||
used.add(pair.person_b_id);
|
||||
result.push(pair);
|
||||
if (result.length >= limit) break;
|
||||
}
|
||||
return merged;
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Is this the UNIQUE constraint firing, as opposed to a real write failure?
|
||||
*
|
||||
* Both engines have to be recognised: Postgres reports SQLSTATE 23505, and
|
||||
* sqlite3 reports SQLITE_CONSTRAINT with the specific constraint named in the
|
||||
* message (better-sqlite3 narrows the code itself). Matching too broadly here
|
||||
* would put us back to swallowing genuine failures.
|
||||
*/
|
||||
function isUniqueViolation(err) {
|
||||
if (!err) return false;
|
||||
if (err.code === '23505') return true;
|
||||
if (typeof err.code === 'string' && err.code.startsWith('SQLITE_CONSTRAINT')) {
|
||||
return /unique/i.test(err.message || '');
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Remember that these two are NOT the same person, so the pair stops being
|
||||
* suggested. Normalized to (lower id, higher id) so the pair has one identity.
|
||||
*/
|
||||
async function dismissMergeSuggestion(eventId, personAId, personBId) {
|
||||
const lo = Math.min(personAId, personBId);
|
||||
const hi = Math.max(personAId, personBId);
|
||||
try {
|
||||
await db('event_people_merge_dismissals').insert({
|
||||
event_id: eventId,
|
||||
person_a_id: lo,
|
||||
person_b_id: hi,
|
||||
created_at: new Date().toISOString(),
|
||||
});
|
||||
} catch (err) {
|
||||
// ONLY the UNIQUE constraint doing its job — dismissing twice is a
|
||||
// double-click, not an error. Anything else (missing table on a
|
||||
// pre-migration install, read-only database) must reach the caller:
|
||||
// reporting "kept separate" for a decision that was never stored is worse
|
||||
// than an error, because the pair silently comes back next scan.
|
||||
if (!isUniqueViolation(err)) throw err;
|
||||
}
|
||||
return { dismissed: true };
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -380,6 +608,22 @@ async function splitPerson(eventId, personId, faceIds) {
|
||||
.whereIn('id', faces.map((f) => f.id))
|
||||
.update({ person_id: newPersonId });
|
||||
|
||||
// A split IS a "these are not the same person" decision, and it has to be
|
||||
// recorded as one (#1107). Consolidation now runs automatically after
|
||||
// every scan, and two clusters a photographer pulled apart are look-alikes
|
||||
// by construction — their centroids usually sit above the merge threshold,
|
||||
// so the very next scan would put them straight back together and the
|
||||
// manual correction would look like it never happened.
|
||||
await trx('event_people_merge_dismissals').insert({
|
||||
event_id: eventId,
|
||||
person_a_id: Math.min(personId, newPersonId),
|
||||
person_b_id: Math.max(personId, newPersonId),
|
||||
created_at: new Date().toISOString(),
|
||||
}).catch((err) => {
|
||||
// Pre-migration install, or the pair was already separated once before.
|
||||
if (!isMissingTable(err) && !isUniqueViolation(err)) throw err;
|
||||
});
|
||||
|
||||
await recomputeCentroid(newPersonId, trx);
|
||||
await recomputeCentroid(personId, trx);
|
||||
return newPersonId;
|
||||
@@ -557,6 +801,12 @@ module.exports = {
|
||||
meetsQualityFloor,
|
||||
assignFaces,
|
||||
consolidate,
|
||||
suggestMerges,
|
||||
dismissMergeSuggestion,
|
||||
// Exported for the tests that pin which failures may be swallowed and which
|
||||
// must stop the pass.
|
||||
isUniqueViolation,
|
||||
isMissingTable,
|
||||
mergePeople,
|
||||
splitPerson,
|
||||
recomputeCentroid,
|
||||
|
||||
@@ -371,6 +371,18 @@ async function purgeEvent(eventId) {
|
||||
await trx('photos').where({ event_id: eventId }).update({
|
||||
face_status: null, face_count: null, face_started_at: null, face_error: null,
|
||||
});
|
||||
|
||||
// Everything the erasure was about is gone, so the records ABOUT that
|
||||
// grouping go too (#1107): dismissal rows name people that no longer
|
||||
// exist, and a consolidation count left standing would have the card
|
||||
// reporting merges beside an empty people list. Inside purgeEvent rather
|
||||
// than the route so archival gets the same treatment.
|
||||
await trx('event_people_merge_dismissals').where({ event_id: eventId }).del()
|
||||
.catch(() => { /* pre-migration install — nothing to clear */ });
|
||||
await trx('events').where({ id: eventId }).update({
|
||||
faces_last_consolidated_count: 0,
|
||||
faces_last_consolidated_at: null,
|
||||
}).catch(() => { /* pre-migration install */ });
|
||||
logger.info(`faceProcessor: purged ${faces} face(s) and ${people} person(s) from event ${eventId}`);
|
||||
return { faces, people };
|
||||
});
|
||||
|
||||
@@ -31,7 +31,7 @@ const logger = require('../utils/logger');
|
||||
const { processPhotoFaces } = require('./faceProcessor');
|
||||
const { SidecarUnavailableError } = require('./faceClient');
|
||||
const { TransientSourceError } = require('./faceProcessor');
|
||||
const { isFeatureEnabled } = require('./faceSettings');
|
||||
const { isFeatureEnabled, isEnabledForEvent } = require('./faceSettings');
|
||||
|
||||
const POLL_INTERVAL_MS = parseInt(process.env.FACE_PROCESSOR_POLL_MS || '2000', 10);
|
||||
const CONCURRENCY = Math.max(1, parseInt(process.env.FACE_PROCESSOR_CONCURRENCY || '1', 10));
|
||||
@@ -101,6 +101,121 @@ function logUnreachableSource(message) {
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Events that have had a photo scanned since their last consolidation pass
|
||||
* (#1107).
|
||||
*
|
||||
* There is no "scan finished" event to hook: the queue is per-photo, and a
|
||||
* backfill is just a lot of independent claims. So a worker that finds nothing
|
||||
* left to claim asks whether the events it touched have actually drained, and
|
||||
* consolidates the ones that have — greedy assignment leaves "Anna in daylight"
|
||||
* and "Anna at the party" as two clusters whose centroids have since converged,
|
||||
* and until now nothing looked unless an admin pressed "Re-group people".
|
||||
*
|
||||
* In-memory, like `deferredEvents`: a restart loses at most a consolidation
|
||||
* pass, and the next scan of that event schedules another one.
|
||||
*/
|
||||
const touchedEvents = new Set();
|
||||
|
||||
// A consolidation that fails keeps its place so a transient database error
|
||||
// does not cost the gallery its pass — but the worker goes idle every
|
||||
// POLL_INTERVAL_MS, so retrying immediately would hot-loop a permanently
|
||||
// broken event and emit a warning every couple of seconds. Same shape as
|
||||
// `deferredEvents` above: back it off, then try again.
|
||||
const CONSOLIDATE_RETRY_MS = parseInt(process.env.FACE_CONSOLIDATE_RETRY_MS || '60000', 10);
|
||||
const consolidationRetryAt = new Map();
|
||||
|
||||
// eventId -> how many workers are currently inside processPhotoFaces for it.
|
||||
//
|
||||
// `face_status` alone cannot answer "is anyone still working on this event":
|
||||
// the photo is committed 'done' inside the transaction, and auto-categorisation
|
||||
// then runs before the call returns. During that window the row looks drained
|
||||
// to every other worker. Counting the callers closes it.
|
||||
const inFlightByEvent = new Map();
|
||||
|
||||
function markInFlight(eventId, delta) {
|
||||
if (eventId == null) return;
|
||||
const next = (inFlightByEvent.get(eventId) || 0) + delta;
|
||||
if (next > 0) inFlightByEvent.set(eventId, next);
|
||||
else inFlightByEvent.delete(eventId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Consolidate every touched event that has genuinely drained.
|
||||
*
|
||||
* "A worker went idle" is not the same as "the scan is done" — with
|
||||
* FACE_PROCESSOR_CONCURRENCY > 1 the others may still be working, and photos
|
||||
* released back to `pending` by a down sidecar are still owed. So the drain is
|
||||
* tested directly against the queue rather than inferred, and an event that is
|
||||
* still busy simply stays in the set for the next idle tick.
|
||||
*
|
||||
* Across multiple pods two workers can consolidate the same event at once.
|
||||
* That is safe rather than coordinated: `mergePeople` is transactional, and a
|
||||
* pair whose source was already absorbed merges nothing.
|
||||
*/
|
||||
async function drainConsolidation() {
|
||||
if (!touchedEvents.size) return;
|
||||
|
||||
for (const eventId of [...touchedEvents]) {
|
||||
const retryAt = consolidationRetryAt.get(eventId);
|
||||
if (retryAt && Date.now() < retryAt) continue;
|
||||
|
||||
try {
|
||||
const outstanding = await db('photos')
|
||||
.where({ event_id: eventId })
|
||||
.whereIn('face_status', ['pending', 'processing'])
|
||||
.count({ c: '*' })
|
||||
.first();
|
||||
|
||||
if (Number(outstanding?.c ?? 0) > 0) continue;
|
||||
|
||||
// A worker may still be inside processPhotoFaces for this event: the
|
||||
// photo is committed 'done' before auto-categorisation runs, so the row
|
||||
// stops counting as outstanding while the call is still going. Without
|
||||
// this, an idle worker consolidates and records its count, the busy one
|
||||
// then re-marks the event, and the next pass overwrites the real number
|
||||
// with zero — losing exactly the report this feature exists to give.
|
||||
if ((inFlightByEvent.get(eventId) || 0) > 0) continue;
|
||||
|
||||
// Detection may have been switched off mid-drain, after an earlier
|
||||
// photo already marked this event. Merging someone's clusters just
|
||||
// after they turned the feature off is not a thing to do quietly.
|
||||
const event = await db('events').where({ id: eventId }).first();
|
||||
if (!(await isEnabledForEvent(event))) {
|
||||
touchedEvents.delete(eventId);
|
||||
consolidationRetryAt.delete(eventId);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Required here rather than at module load, matching faceProcessor's
|
||||
// call into faceAutoCategories: the binding stays late, which keeps the
|
||||
// seam this is tested through honest.
|
||||
const { consolidate } = require('./faceClustering');
|
||||
const merged = await consolidate(eventId);
|
||||
// Dropped only once it has actually run. Removing it first meant a
|
||||
// transient database error lost the pass entirely — no retry until the
|
||||
// gallery happened to be scanned again.
|
||||
touchedEvents.delete(eventId);
|
||||
consolidationRetryAt.delete(eventId);
|
||||
if (merged.length) {
|
||||
logger.info(
|
||||
`faceQueue: scan of event ${eventId} drained — consolidated ${merged.length} look-alike pair(s)`
|
||||
);
|
||||
}
|
||||
} catch (e) {
|
||||
// Never let a consolidation failure stop the queue. The event keeps its
|
||||
// place so a transient error is retried, but not before the backoff —
|
||||
// otherwise a permanently failing event warns on every idle tick.
|
||||
consolidationRetryAt.set(eventId, Date.now() + CONSOLIDATE_RETRY_MS);
|
||||
logger.warn(
|
||||
`faceQueue: consolidation failed for event ${eventId}, retrying in `
|
||||
+ `${Math.round(CONSOLIDATE_RETRY_MS / 1000)}s`,
|
||||
{ error: e.message }
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let running = false;
|
||||
let workerHandles = [];
|
||||
let janitorHandle = null;
|
||||
@@ -184,12 +299,27 @@ async function workerLoop(workerIdx) {
|
||||
}
|
||||
|
||||
if (!claimed) {
|
||||
// Nothing left to claim is the only signal this queue has that a scan
|
||||
// may have finished. Cheap when idle: no-ops unless work happened.
|
||||
await drainConsolidation();
|
||||
await sleep(POLL_INTERVAL_MS);
|
||||
continue;
|
||||
}
|
||||
|
||||
markInFlight(claimed.event_id, +1);
|
||||
try {
|
||||
await processPhotoFaces(claimed.id);
|
||||
const outcome = await processPhotoFaces(claimed.id);
|
||||
// Only a photo that actually went through detection can have moved this
|
||||
// event's clusters. 'skipped' covers videos, a purge that raced the
|
||||
// scan, and — the one that matters — detection being switched OFF
|
||||
// mid-drain: consolidating there would merge clusters moments after the
|
||||
// admin turned the feature off. 'failed' produced no faces either.
|
||||
//
|
||||
// A photo containing no faces still reports 'done', so an event whose
|
||||
// photos are all empty still gets its (harmless) pass.
|
||||
if (outcome?.status === 'done' && claimed.event_id != null) {
|
||||
touchedEvents.add(claimed.event_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
|
||||
@@ -237,6 +367,11 @@ async function workerLoop(workerIdx) {
|
||||
error: updateErr.message,
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
// In a finally because the catch above returns to the loop via
|
||||
// `continue` on two paths — a leaked count would block this event's
|
||||
// consolidation for the lifetime of the process.
|
||||
markInFlight(claimed.event_id, -1);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -255,6 +390,18 @@ async function janitorLoop() {
|
||||
} catch (e) {
|
||||
logger.warn('faceQueue: janitor error', { error: e.message });
|
||||
}
|
||||
|
||||
// The worker only reaches its drain when it can claim NOTHING, anywhere.
|
||||
// With the default single worker that means one gallery finishing during a
|
||||
// large backfill waits for every other gallery — and under continuous
|
||||
// ingestion it could wait indefinitely. Running the drain here too makes
|
||||
// consolidation depend on the event being finished rather than the whole
|
||||
// install being idle. It is per-event guarded, so this is a no-op for
|
||||
// anything still in flight.
|
||||
await drainConsolidation().catch((e) =>
|
||||
logger.warn('faceQueue: janitor drain error', { error: e.message })
|
||||
);
|
||||
|
||||
await sleep(JANITOR_INTERVAL_MS);
|
||||
}
|
||||
}
|
||||
@@ -293,4 +440,11 @@ async function stop() {
|
||||
janitorHandle = null;
|
||||
}
|
||||
|
||||
module.exports = { start, stop, claimNextPhoto };
|
||||
// drainConsolidation, touchedEvents and consolidationRetryAt are exported for
|
||||
// the same reason claimNextPhoto is: the drain condition and its failure
|
||||
// backoff are the subtle parts of #1107 and are worth pinning directly, rather
|
||||
// than through a running worker loop.
|
||||
module.exports = {
|
||||
start, stop, claimNextPhoto, drainConsolidation,
|
||||
touchedEvents, consolidationRetryAt, inFlightByEvent,
|
||||
};
|
||||
|
||||
@@ -42,6 +42,11 @@ const EXCLUDED_TABLES = new Set([
|
||||
'knex_migrations_lock',
|
||||
'photo_faces',
|
||||
'event_people',
|
||||
// Follows event_people out of the export (#1107): these rows are nothing but
|
||||
// references to person ids the target will never receive. Carried across,
|
||||
// they would attach to whatever ids the target's own re-scan happens to
|
||||
// mint, silently suppressing merge suggestions in unrelated galleries.
|
||||
'event_people_merge_dismissals',
|
||||
]);
|
||||
|
||||
// Storage subdirs holding non-recalculable blobs — always included.
|
||||
|
||||
@@ -358,7 +358,7 @@ async function replaceAllTables(tables, dataDir, currentAdmin, roleSnapshot, { c
|
||||
// orphans can end up attached to reused photo/event ids from the incoming
|
||||
// archive: one instance's biometric data silently adopted by another's
|
||||
// galleries. Purge them explicitly.
|
||||
for (const faceTable of ['photo_faces', 'event_people']) {
|
||||
for (const faceTable of ['photo_faces', 'event_people', 'event_people_merge_dismissals']) {
|
||||
try {
|
||||
await trx(faceTable).del();
|
||||
} catch (err) {
|
||||
|
||||
Reference in New Issue
Block a user