From 43848a7d680f8184142dbd1694ddfce64002f7a2 Mon Sep 17 00:00:00 2001 From: bo0tzz Date: Tue, 20 Oct 2026 10:06:08 +0200 Subject: [PATCH] fix: locking around concurrency-1 jobs (#28410) * fix: locking around concurrency-1 jobs * fix: no lock on queue duplicates (cherry picked from commit 477452b5c51ca71ecdf74205e0004eaec4e18874) --- server/src/enum.ts | 2 + server/src/services/duplicate.service.ts | 98 ++++----- server/src/services/person.service.ts | 243 ++++++++++++----------- 3 files changed, 176 insertions(+), 167 deletions(-) diff --git a/server/src/enum.ts b/server/src/enum.ts index 4151c88d7..843339d22 100644 --- a/server/src/enum.ts +++ b/server/src/enum.ts @@ -1018,6 +1018,8 @@ export enum DatabaseLock { IntegrityCheck = 67, VersionCheck = 800, HlsSessionCleanup = 850, + FacialRecognition = 900, + DuplicateDetection = 1000, } export enum MaintenanceAction { diff --git a/server/src/services/duplicate.service.ts b/server/src/services/duplicate.service.ts index 7bcb371dd..1d01e195a 100644 --- a/server/src/services/duplicate.service.ts +++ b/server/src/services/duplicate.service.ts @@ -5,7 +5,7 @@ import { BulkIdErrorReason, BulkIdResponseDto, BulkIdsDto } from 'src/dtos/asset import { MapAsset, mapAsset } from 'src/dtos/asset-response.dto.js'; import { AuthDto } from 'src/dtos/auth.dto.js'; import { DuplicateResolveDto, DuplicateResolveGroupDto, DuplicateResponseDto } from 'src/dtos/duplicate.dto.js'; -import { AssetStatus, AssetVisibility, JobName, JobStatus, Permission, QueueName } from 'src/enum.js'; +import { AssetStatus, AssetVisibility, DatabaseLock, JobName, JobStatus, Permission, QueueName } from 'src/enum.js'; import { AssetDuplicateResult } from 'src/repositories/search.repository.js'; import { BaseService } from 'src/services/base.service.js'; import { suggestDuplicateKeepAssetIds } from 'src/utils/duplicate.js'; @@ -317,60 +317,62 @@ export class DuplicateService extends BaseService { @OnJob({ name: JobName.AssetDetectDuplicates, queue: QueueName.DuplicateDetection }) async handleSearchDuplicates({ id }: JobOf): Promise { - const { machineLearning } = await this.getConfig({ withCache: true }); - if (!isDuplicateDetectionEnabled(machineLearning)) { - return JobStatus.Skipped; - } + return this.databaseRepository.withLock(DatabaseLock.DuplicateDetection, async () => { + const { machineLearning } = await this.getConfig({ withCache: true }); + if (!isDuplicateDetectionEnabled(machineLearning)) { + return JobStatus.Skipped; + } - const asset = await this.assetJobRepository.getForSearchDuplicatesJob(id); - if (!asset) { - this.logger.error(`Asset ${id} not found`); - return JobStatus.Failed; - } + const asset = await this.assetJobRepository.getForSearchDuplicatesJob(id); + if (!asset) { + this.logger.error(`Asset ${id} not found`); + return JobStatus.Failed; + } - if (asset.stackId) { - this.logger.debug(`Asset ${id} is part of a stack, skipping`); - return JobStatus.Skipped; - } + if (asset.stackId) { + this.logger.debug(`Asset ${id} is part of a stack, skipping`); + return JobStatus.Skipped; + } - if (asset.visibility === AssetVisibility.Hidden) { - this.logger.debug(`Asset ${id} is not visible, skipping`); - return JobStatus.Skipped; - } + if (asset.visibility === AssetVisibility.Hidden) { + this.logger.debug(`Asset ${id} is not visible, skipping`); + return JobStatus.Skipped; + } - if (asset.visibility === AssetVisibility.Locked) { - this.logger.debug(`Asset ${id} is locked, skipping`); - return JobStatus.Skipped; - } + if (asset.visibility === AssetVisibility.Locked) { + this.logger.debug(`Asset ${id} is locked, skipping`); + return JobStatus.Skipped; + } - if (!asset.embedding) { - this.logger.debug(`Asset ${id} is missing embedding`); - return JobStatus.Failed; - } + if (!asset.embedding) { + this.logger.debug(`Asset ${id} is missing embedding`); + return JobStatus.Failed; + } - const duplicateAssets = await this.duplicateRepository.search({ - assetId: asset.id, - embedding: asset.embedding, - maxDistance: machineLearning.duplicateDetection.maxDistance, - type: asset.type, - userIds: [asset.ownerId], + const duplicateAssets = await this.duplicateRepository.search({ + assetId: asset.id, + embedding: asset.embedding, + maxDistance: machineLearning.duplicateDetection.maxDistance, + type: asset.type, + userIds: [asset.ownerId], + }); + + let assetIds = [asset.id]; + if (duplicateAssets.length > 0) { + this.logger.debug( + `Found ${duplicateAssets.length} duplicate${duplicateAssets.length === 1 ? '' : 's'} for asset ${asset.id}`, + ); + assetIds = await this.updateDuplicates(asset, duplicateAssets); + } else if (asset.duplicateId) { + this.logger.debug(`No duplicates found for asset ${asset.id}, removing duplicateId`); + await this.assetRepository.update({ id: asset.id, duplicateId: null }); + } + + const duplicatesDetectedAt = new Date(); + await this.assetRepository.upsertJobStatus(...assetIds.map((assetId) => ({ assetId, duplicatesDetectedAt }))); + + return JobStatus.Success; }); - - let assetIds = [asset.id]; - if (duplicateAssets.length > 0) { - this.logger.debug( - `Found ${duplicateAssets.length} duplicate${duplicateAssets.length === 1 ? '' : 's'} for asset ${asset.id}`, - ); - assetIds = await this.updateDuplicates(asset, duplicateAssets); - } else if (asset.duplicateId) { - this.logger.debug(`No duplicates found for asset ${asset.id}, removing duplicateId`); - await this.assetRepository.update({ id: asset.id, duplicateId: null }); - } - - const duplicatesDetectedAt = new Date(); - await this.assetRepository.upsertJobStatus(...assetIds.map((assetId) => ({ assetId, duplicatesDetectedAt }))); - - return JobStatus.Success; } private async updateDuplicates( diff --git a/server/src/services/person.service.ts b/server/src/services/person.service.ts index 4109b019b..2e20cd01c 100644 --- a/server/src/services/person.service.ts +++ b/server/src/services/person.service.ts @@ -32,6 +32,7 @@ import { import { AssetVisibility, CacheControl, + DatabaseLock, JobName, JobStatus, Permission, @@ -499,150 +500,154 @@ export class PersonService extends BaseService { nightly, clusterGroupId, }: JobOf): Promise { - const { machineLearning } = await this.getConfig({ withCache: false }); - if (!isFacialRecognitionEnabled(machineLearning)) { - return JobStatus.Skipped; - } - - await this.jobRepository.waitForQueueCompletion(QueueName.ThumbnailGeneration, QueueName.FaceDetection); - - if (nightly) { - const [state, latestFaceDate] = await Promise.all([ - this.systemMetadataRepository.get(SystemMetadataKey.FacialRecognitionState), - this.personRepository.getLatestFaceDate(), - ]); - - if (state?.lastRun && latestFaceDate && state.lastRun > latestFaceDate) { - this.logger.debug('Skipping facial recognition nightly since no face has been added since the last run'); + return this.databaseRepository.withLock(DatabaseLock.FacialRecognition, async () => { + const { machineLearning } = await this.getConfig({ withCache: false }); + if (!isFacialRecognitionEnabled(machineLearning)) { return JobStatus.Skipped; } - } - const { waiting } = await this.jobRepository.getJobCounts(QueueName.FacialRecognition); + await this.jobRepository.waitForQueueCompletion(QueueName.ThumbnailGeneration, QueueName.FaceDetection); - if (force) { - await this.personRepository.unassignFaces({ clusterGroupId, sourceType: SourceType.MachineLearning }); - await this.handlePersonCleanup(); - await this.vacuum('asset_face', 'person'); - } else if (waiting) { - this.logger.debug( - `Skipping facial recognition queueing because ${waiting} job${waiting > 1 ? 's are' : ' is'} already queued`, + if (nightly) { + const [state, latestFaceDate] = await Promise.all([ + this.systemMetadataRepository.get(SystemMetadataKey.FacialRecognitionState), + this.personRepository.getLatestFaceDate(), + ]); + + if (state?.lastRun && latestFaceDate && state.lastRun > latestFaceDate) { + this.logger.debug('Skipping facial recognition nightly since no face has been added since the last run'); + return JobStatus.Skipped; + } + } + + const { waiting } = await this.jobRepository.getJobCounts(QueueName.FacialRecognition); + + if (force) { + await this.personRepository.unassignFaces({ clusterGroupId, sourceType: SourceType.MachineLearning }); + await this.handlePersonCleanup(); + await this.vacuum('asset_face', 'person'); + } else if (waiting) { + this.logger.debug( + `Skipping facial recognition queueing because ${waiting} job${waiting > 1 ? 's are' : ' is'} already queued`, + ); + return JobStatus.Skipped; + } + + await this.databaseRepository.prewarm(VectorIndex.Face); + + const lastRun = new Date().toISOString(); + + const faces = this.personRepository.getAllFaces( + force + ? { clusterGroupId, sourceType: clusterGroupId ? SourceType.MachineLearning : undefined } + : { personGroupId: null, clusterGroupId, sourceType: SourceType.MachineLearning }, ); - return JobStatus.Skipped; - } + for await (const batch of batched(faces)) { + await this.jobRepository.queueAll( + batch.map((face) => ({ name: JobName.FacialRecognition, data: { id: face.id, deferred: false } })), + ); + } - await this.databaseRepository.prewarm(VectorIndex.Face); + await this.systemMetadataRepository.set(SystemMetadataKey.FacialRecognitionState, { lastRun }); - const lastRun = new Date().toISOString(); - - const faces = this.personRepository.getAllFaces( - force - ? { clusterGroupId, sourceType: clusterGroupId ? SourceType.MachineLearning : undefined } - : { personGroupId: null, clusterGroupId, sourceType: SourceType.MachineLearning }, - ); - for await (const batch of batched(faces)) { - await this.jobRepository.queueAll( - batch.map((face) => ({ name: JobName.FacialRecognition, data: { id: face.id, deferred: false } })), - ); - } - - await this.systemMetadataRepository.set(SystemMetadataKey.FacialRecognitionState, { lastRun }); - - return JobStatus.Success; + return JobStatus.Success; + }); } @OnJob({ name: JobName.FacialRecognition, queue: QueueName.FacialRecognition }) async handleRecognizeFaces({ id, deferred }: JobOf): Promise { - const { machineLearning } = await this.getConfig({ withCache: true }); - if (!isFacialRecognitionEnabled(machineLearning)) { - return JobStatus.Skipped; - } + return this.databaseRepository.withLock(DatabaseLock.FacialRecognition, async () => { + const { machineLearning } = await this.getConfig({ withCache: true }); + if (!isFacialRecognitionEnabled(machineLearning)) { + return JobStatus.Skipped; + } - const face = await this.personRepository.getFaceForFacialRecognitionJob(id); - if (!face || !face.asset) { - this.logger.warn(`Face ${id} not found`); - return JobStatus.Failed; - } + const face = await this.personRepository.getFaceForFacialRecognitionJob(id); + if (!face || !face.asset) { + this.logger.warn(`Face ${id} not found`); + return JobStatus.Failed; + } - if (face.sourceType !== SourceType.MachineLearning) { - this.logger.warn(`Skipping face ${id} due to source ${face.sourceType}`); - return JobStatus.Skipped; - } + if (face.sourceType !== SourceType.MachineLearning) { + this.logger.warn(`Skipping face ${id} due to source ${face.sourceType}`); + return JobStatus.Skipped; + } - if (!face.faceSearch?.embedding) { - this.logger.warn(`Face ${id} does not have an embedding`); - return JobStatus.Failed; - } + if (!face.faceSearch?.embedding) { + this.logger.warn(`Face ${id} does not have an embedding`); + return JobStatus.Failed; + } - if (face.personGroupId) { - this.logger.debug(`Face ${id} already has a person assigned`); - return JobStatus.Skipped; - } + if (face.personGroupId) { + this.logger.debug(`Face ${id} already has a person assigned`); + return JobStatus.Skipped; + } - const { ownerId, clusterGroupId } = face.asset; - const matches = await this.searchRepository.searchFaces({ - clusterGroupId, - embedding: face.faceSearch.embedding, - maxDistance: machineLearning.facialRecognition.maxDistance, - numResults: machineLearning.facialRecognition.minFaces, - minBirthDate: new Date(face.asset.fileCreatedAt), - }); - - // `matches` also includes the face itself - if (machineLearning.facialRecognition.minFaces > 1 && matches.length <= 1) { - this.logger.debug(`Face ${id} only matched the face itself, skipping`); - return JobStatus.Skipped; - } - - this.logger.debug(`Face ${id} has ${matches.length} matches`); - - const isCore = - matches.length >= machineLearning.facialRecognition.minFaces && - face.asset.visibility === AssetVisibility.Timeline; - if (!isCore && !deferred) { - this.logger.debug(`Deferring non-core face ${id} for later processing`); - await this.jobRepository.queue({ name: JobName.FacialRecognition, data: { id, deferred: true } }); - return JobStatus.Skipped; - } - - let personGroupId = matches.find((match) => match.personGroupId)?.personGroupId; - if (!personGroupId) { - const [matchWithPerson] = await this.searchRepository.searchFaces({ + const { ownerId, clusterGroupId } = face.asset; + const matches = await this.searchRepository.searchFaces({ clusterGroupId, embedding: face.faceSearch.embedding, maxDistance: machineLearning.facialRecognition.maxDistance, - numResults: 1, - hasPerson: true, + numResults: machineLearning.facialRecognition.minFaces, minBirthDate: new Date(face.asset.fileCreatedAt), }); - personGroupId = matchWithPerson?.personGroupId ?? undefined; - } - - if (!personGroupId && isCore) { - const group = await this.personRepository.createGroup(ownerId); - personGroupId = group.id; - this.logger.log(`Created person group ${personGroupId} for face ${id}`); - } - - if (personGroupId) { - const person = await this.personRepository.getByGroupId({ ownerId, personGroupId }); - if (person) { - this.logger.debug(`Face ${id} matched person ${person.personGroupId}`); - } else { - await this.personRepository.create({ ownerId, faceAssetId: face.id, personGroupId }); - this.logger.log(`Created person for face ${id} in group ${personGroupId}`); - await this.jobRepository.queue({ - name: JobName.PersonGenerateThumbnail, - data: { ownerId, personGroupId }, - }); + // `matches` also includes the face itself + if (machineLearning.facialRecognition.minFaces > 1 && matches.length <= 1) { + this.logger.debug(`Face ${id} only matched the face itself, skipping`); + return JobStatus.Skipped; } - this.logger.debug(`Assigning face ${id} to person group ${personGroupId}`); - await this.personRepository.reassignFaces({ faceIds: [id], newPersonGroupId: personGroupId }); - } + this.logger.debug(`Face ${id} has ${matches.length} matches`); - return JobStatus.Success; + const isCore = + matches.length >= machineLearning.facialRecognition.minFaces && + face.asset.visibility === AssetVisibility.Timeline; + if (!isCore && !deferred) { + this.logger.debug(`Deferring non-core face ${id} for later processing`); + await this.jobRepository.queue({ name: JobName.FacialRecognition, data: { id, deferred: true } }); + return JobStatus.Skipped; + } + + let personGroupId = matches.find((match) => match.personGroupId)?.personGroupId; + if (!personGroupId) { + const [matchWithPerson] = await this.searchRepository.searchFaces({ + clusterGroupId, + embedding: face.faceSearch.embedding, + maxDistance: machineLearning.facialRecognition.maxDistance, + numResults: 1, + hasPerson: true, + minBirthDate: new Date(face.asset.fileCreatedAt), + }); + + personGroupId = matchWithPerson?.personGroupId ?? undefined; + } + + if (!personGroupId && isCore) { + const group = await this.personRepository.createGroup(ownerId); + personGroupId = group.id; + this.logger.log(`Created person group ${personGroupId} for face ${id}`); + } + + if (personGroupId) { + const person = await this.personRepository.getByGroupId({ ownerId, personGroupId }); + if (person) { + this.logger.debug(`Face ${id} matched person ${person.personGroupId}`); + } else { + await this.personRepository.create({ ownerId, faceAssetId: face.id, personGroupId }); + this.logger.log(`Created person for face ${id} in group ${personGroupId}`); + await this.jobRepository.queue({ + name: JobName.PersonGenerateThumbnail, + data: { ownerId, personGroupId }, + }); + } + + this.logger.debug(`Assigning face ${id} to person group ${personGroupId}`); + await this.personRepository.reassignFaces({ faceIds: [id], newPersonGroupId: personGroupId }); + } + + return JobStatus.Success; + }); } @OnJob({ name: JobName.PersonFileMigration, queue: QueueName.Migration })