fix: locking around concurrency-1 jobs (#28410)
* fix: locking around concurrency-1 jobs * fix: no lock on queue duplicates (cherry picked from commit 477452b5c51ca71ecdf74205e0004eaec4e18874)
This commit is contained in:
bo0tzz authored and Codex committed
1 parent
9e96fdc627
commit
43848a7d68
3 files changed
+176
-167
No files matched your search
@@ -1018,6 +1018,8 @@ export enum DatabaseLock {
|
||||
IntegrityCheck = 67,
|
||||
VersionCheck = 800,
|
||||
HlsSessionCleanup = 850,
|
||||
FacialRecognition = 900,
|
||||
DuplicateDetection = 1000,
|
||||
}
|
||||
|
||||
export enum MaintenanceAction {
|
||||
|
||||
@@ -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<JobName.AssetDetectDuplicates>): Promise<JobStatus> {
|
||||
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(
|
||||
|
||||
@@ -32,6 +32,7 @@ import {
|
||||
import {
|
||||
AssetVisibility,
|
||||
CacheControl,
|
||||
DatabaseLock,
|
||||
JobName,
|
||||
JobStatus,
|
||||
Permission,
|
||||
@@ -499,150 +500,154 @@ export class PersonService extends BaseService {
|
||||
nightly,
|
||||
clusterGroupId,
|
||||
}: JobOf<JobName.FacialRecognitionQueueAll>): Promise<JobStatus> {
|
||||
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<JobName.FacialRecognition>): Promise<JobStatus> {
|
||||
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 })
|
||||
|
||||
Reference in new issue
Block a user