Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: duplicated library scan jobs and api server library watch #13734

Merged
merged 2 commits into from
Oct 25, 2024
Merged

fix: duplicated library scan jobs and api server library watch #13734

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fa7dfc0
fix: library scanning cron job created on every instance instead of j…
zackpollard Oct 25, 2024
80f6aaf
fix: only run library scan and file watches on microservices
zackpollard Oct 25, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion server/src/interfaces/database.interface.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ export enum DatabaseLock {
StorageTemplateMigration = 420,
VersionHistory = 500,
CLIPDimSize = 512,
LibraryWatch = 1337,
Library = 1337,
GetSystemConfig = 69,
}

Expand Down
41 changes: 28 additions & 13 deletions server/src/services/library.service.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ import { Stats } from 'node:fs';
import { defaults, SystemConfig } from 'src/config';
import { mapLibrary } from 'src/dtos/library.dto';
import { UserEntity } from 'src/entities/user.entity';
import { AssetType } from 'src/enum';
import { AssetType, ImmichWorker } from 'src/enum';
import { IAssetRepository } from 'src/interfaces/asset.interface';
import { IDatabaseRepository } from 'src/interfaces/database.interface';
import {
Expand Down Expand Up @@ -55,7 +55,7 @@ describe(LibraryService.name, () => {
it('should init cron job and handle config changes', async () => {
systemMock.get.mockResolvedValue(systemConfigStub.libraryScan);

await sut.onBootstrap();
await sut.onBootstrap(ImmichWorker.MICROSERVICES);

expect(jobMock.addCronJob).toHaveBeenCalled();
expect(systemMock.get).toHaveBeenCalled();
Expand Down Expand Up @@ -91,7 +91,7 @@ describe(LibraryService.name, () => {
),
);

await sut.onBootstrap();
await sut.onBootstrap(ImmichWorker.MICROSERVICES);

expect(storageMock.watch.mock.calls).toEqual(
expect.arrayContaining([
Expand All @@ -104,7 +104,7 @@ describe(LibraryService.name, () => {
it('should not initialize watcher when watching is disabled', async () => {
systemMock.get.mockResolvedValue(systemConfigStub.libraryWatchDisabled);

await sut.onBootstrap();
await sut.onBootstrap(ImmichWorker.MICROSERVICES);

expect(storageMock.watch).not.toHaveBeenCalled();
});
Expand All @@ -113,17 +113,32 @@ describe(LibraryService.name, () => {
systemMock.get.mockResolvedValue(systemConfigStub.libraryWatchEnabled);
databaseMock.tryLock.mockResolvedValue(false);

await sut.onBootstrap();
await sut.onBootstrap(ImmichWorker.MICROSERVICES);

expect(storageMock.watch).not.toHaveBeenCalled();
});

it('should not initialize library scan cron job when lock is taken', async () => {
systemMock.get.mockResolvedValue(systemConfigStub.libraryWatchEnabled);
databaseMock.tryLock.mockResolvedValue(false);

await sut.onBootstrap(ImmichWorker.MICROSERVICES);

expect(jobMock.addCronJob).not.toHaveBeenCalled();
});

it('should not initialize watcher or library scan job when running on api', async () => {
await sut.onBootstrap(ImmichWorker.API);

expect(jobMock.addCronJob).not.toHaveBeenCalled();
});
});

describe('onConfigUpdateEvent', () => {
beforeEach(async () => {
systemMock.get.mockResolvedValue(defaults);
databaseMock.tryLock.mockResolvedValue(true);
await sut.onBootstrap();
await sut.onBootstrap(ImmichWorker.MICROSERVICES);
});

it('should do nothing if oldConfig is not provided', async () => {
Expand All @@ -133,7 +148,7 @@ describe(LibraryService.name, () => {

it('should do nothing if instance does not have the watch lock', async () => {
databaseMock.tryLock.mockResolvedValue(false);
await sut.onBootstrap();
await sut.onBootstrap(ImmichWorker.MICROSERVICES);
await sut.onConfigUpdate({ newConfig: systemConfigStub.libraryScan as SystemConfig, oldConfig: defaults });
expect(jobMock.updateCronJob).not.toHaveBeenCalled();
});
Expand Down Expand Up @@ -693,7 +708,7 @@ describe(LibraryService.name, () => {
const mockClose = vitest.fn();
storageMock.watch.mockImplementation(makeMockWatcher({ close: mockClose }));

await sut.onBootstrap();
await sut.onBootstrap(ImmichWorker.MICROSERVICES);
await sut.delete(libraryStub.externalLibraryWithImportPaths1.id);

expect(mockClose).toHaveBeenCalled();
Expand Down Expand Up @@ -827,7 +842,7 @@ describe(LibraryService.name, () => {
libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
libraryMock.getAll.mockResolvedValue([]);

await sut.onBootstrap();
await sut.onBootstrap(ImmichWorker.MICROSERVICES);
await sut.create({
ownerId: authStub.admin.user.id,
importPaths: libraryStub.externalLibraryWithImportPaths1.importPaths,
Expand Down Expand Up @@ -890,7 +905,7 @@ describe(LibraryService.name, () => {
systemMock.get.mockResolvedValue(systemConfigStub.libraryWatchEnabled);
libraryMock.getAll.mockResolvedValue([]);

await sut.onBootstrap();
await sut.onBootstrap(ImmichWorker.MICROSERVICES);
});

it('should throw an error if an import path is invalid', async () => {
Expand Down Expand Up @@ -931,7 +946,7 @@ describe(LibraryService.name, () => {
beforeEach(async () => {
systemMock.get.mockResolvedValue(systemConfigStub.libraryWatchDisabled);

await sut.onBootstrap();
await sut.onBootstrap(ImmichWorker.MICROSERVICES);
});

it('should not watch library', async () => {
Expand All @@ -947,7 +962,7 @@ describe(LibraryService.name, () => {
beforeEach(async () => {
systemMock.get.mockResolvedValue(systemConfigStub.libraryWatchEnabled);
libraryMock.getAll.mockResolvedValue([]);
await sut.onBootstrap();
await sut.onBootstrap(ImmichWorker.MICROSERVICES);
});

it('should watch library', async () => {
Expand Down Expand Up @@ -1113,7 +1128,7 @@ describe(LibraryService.name, () => {
const mockClose = vitest.fn();
storageMock.watch.mockImplementation(makeMockWatcher({ close: mockClose }));

await sut.onBootstrap();
await sut.onBootstrap(ImmichWorker.MICROSERVICES);
await sut.onShutdown();

expect(mockClose).toHaveBeenCalledTimes(2);
Expand Down
35 changes: 20 additions & 15 deletions server/src/services/library.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ import {
} from 'src/dtos/library.dto';
import { AssetEntity } from 'src/entities/asset.entity';
import { LibraryEntity } from 'src/entities/library.entity';
import { AssetType } from 'src/enum';
import { AssetType, ImmichWorker } from 'src/enum';
import { DatabaseLock } from 'src/interfaces/database.interface';
import { ArgOf } from 'src/interfaces/event.interface';
import {
Expand All @@ -36,27 +36,32 @@ import { validateCronExpression } from 'src/validation';
@Injectable()
export class LibraryService extends BaseService {
private watchLibraries = false;
private watchLock = false;
private lock = false;
private watchers: Record<string, () => Promise<void>> = {};

@OnEvent({ name: 'app.bootstrap' })
async onBootstrap() {
async onBootstrap(workerType: ImmichWorker) {
if (workerType !== ImmichWorker.MICROSERVICES) {
return;
}

const config = await this.getConfig({ withCache: false });

const { watch, scan } = config.library;

// This ensures that library watching only occurs in one microservice
// TODO: we could make the lock be per-library instead of global
this.watchLock = await this.databaseRepository.tryLock(DatabaseLock.LibraryWatch);
this.lock = await this.databaseRepository.tryLock(DatabaseLock.Library);

this.watchLibraries = this.watchLock && watch.enabled;
this.watchLibraries = this.lock && watch.enabled;

this.jobRepository.addCronJob(
'libraryScan',
scan.cronExpression,
() => handlePromiseError(this.jobRepository.queue({ name: JobName.LIBRARY_QUEUE_SYNC_ALL }), this.logger),
scan.enabled,
);
if (this.lock) {
this.jobRepository.addCronJob(
'libraryScan',
scan.cronExpression,
() => handlePromiseError(this.jobRepository.queue({ name: JobName.LIBRARY_QUEUE_SYNC_ALL }), this.logger),
scan.enabled,
);
}

if (this.watchLibraries) {
await this.watchAll();
Expand All @@ -65,7 +70,7 @@ export class LibraryService extends BaseService {

@OnEvent({ name: 'config.update', server: true })
async onConfigUpdate({ newConfig: { library }, oldConfig }: ArgOf<'config.update'>) {
if (!oldConfig || !this.watchLock) {
if (!oldConfig || !this.lock) {
return;
}

Expand Down Expand Up @@ -180,7 +185,7 @@ export class LibraryService extends BaseService {
}

private async unwatchAll() {
if (!this.watchLock) {
if (!this.lock) {
return false;
}

Expand All @@ -190,7 +195,7 @@ export class LibraryService extends BaseService {
}

async watchAll() {
if (!this.watchLock) {
if (!this.lock) {
return false;
}

Expand Down
Loading