Skip to content

Commit

Permalink
chore(server): Move library watcher to microservices (#7533)
Browse files Browse the repository at this point in the history 
* move watcher init to micro

* document watcher recovery

* chore: fix lint

* add try lock

* use global library watch lock

* fix: ensure lock stays on

* fix: mocks

* unit test for library watch lock

* move statement to correct test

* fix: correct return type of try lock

* fix: tests

* add library teardown

* add chokidar error handler

* make event strings an enum

* wait for event refactor

* refactor event type mocks

* expect correct error

* don't release lock in teardown

* chore: lint

* use enum

* fix mock

* fix lint

* fix watcher await

* remove await

* simplify typing

* remove async

* Revert "remove async"

This reverts commit 84ab5ab.

* can now change watch settings at runtime

* fix lint

* only watch libraries if enabled

---------

Co-authored-by: mertalev <101130780+mertalev@users.noreply.github.com>
Co-authored-by: Alex Tran <alex.tran1502@gmail.com>
  • Loading branch information
@etnoy @mertalev @alextran1502
3 people committed Mar 7, 2024
1 parent 3278dcb commit 4cb0f37
Show file tree
Hide file tree
Showing 14 changed files with 149 additions and 81 deletions.
10 changes: 10 additions & 0 deletions docs/docs/features/libraries.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,16 @@ This feature - currently hidden in the config file - is considered experimental

If your photos are on a network drive, automatic file watching likely won't work. In that case, you will have to rely on a periodic library refresh to pull in your changes.

#### Troubleshooting

If you encounter an `ENOSPC` error, you need to increase your file watcher limit. In sysctl, this key is called `fs.inotify.max_user_watched` and has a default value of 8192. Increase this number to a suitable value greater than the number of files you will be watching. Note that Immich has to watch all files in your import paths including any ignored files.

```
ERROR [LibraryService] Library watcher for library c69faf55-f96d-4aa0-b83b-2d80cbc27d98 encountered error: Error: ENOSPC: System limit for number of file watchers reached, watch '/media/photo.jpg'
```

In rare cases, the library watcher can hang, preventing Immich from starting up. In this case, disable the library watcher in the configuration file. If the watcher is enabled from within Immich, the app must be started without the microservices. Disable the microservices in the docker compose file, start Immich, disable the library watcher in the admin settings, close Immich, re-enable the microservices, and then Immich can be started normally.

### Nightly job

There is an automatic job that's run once a day and refreshes all modified files in all libraries as well as cleans up any libraries stuck in deletion.
Expand Down
25 changes: 10 additions & 15 deletions server/e2e/jobs/specs/library-watcher.e2e-spec.ts
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { LibraryResponseDto, LibraryService, LoginResponseDto } from '@app/domain';
import { LibraryResponseDto, LibraryService, LoginResponseDto, StorageEventType } from '@app/domain';
import { AssetType, LibraryType } from '@app/infra/entities';
import fs from 'node:fs/promises';
import path from 'node:path';
Expand Down Expand Up @@ -33,7 +33,7 @@ describe(`Library watcher (e2e)`, () => {
});

afterEach(async () => {
await libraryService.unwatchAll();
await libraryService.teardown();
});

afterAll(async () => {
Expand All @@ -57,7 +57,7 @@ describe(`Library watcher (e2e)`, () => {
`${IMMICH_TEST_ASSET_TEMP_PATH}/file.jpg`,
);

await waitForEvent(libraryService, 'add');
await waitForEvent(libraryService, StorageEventType.ADD);

const afterAssets = await api.assetApi.getAllAssets(server, admin.accessToken);
expect(afterAssets.length).toEqual(1);
Expand All @@ -84,10 +84,7 @@ describe(`Library watcher (e2e)`, () => {
`${IMMICH_TEST_ASSET_TEMP_PATH}/file5.jPg`,
);

await waitForEvent(libraryService, 'add');
await waitForEvent(libraryService, 'add');
await waitForEvent(libraryService, 'add');
await waitForEvent(libraryService, 'add');
await waitForEvent(libraryService, StorageEventType.ADD, 4);

const afterAssets = await api.assetApi.getAllAssets(server, admin.accessToken);
expect(afterAssets.length).toEqual(4);
Expand All @@ -99,7 +96,7 @@ describe(`Library watcher (e2e)`, () => {
`${IMMICH_TEST_ASSET_TEMP_PATH}/file.jpg`,
);

await waitForEvent(libraryService, 'add');
await waitForEvent(libraryService, StorageEventType.ADD);

const originalAssets = await api.assetApi.getAllAssets(server, admin.accessToken);
expect(originalAssets.length).toEqual(1);
Expand All @@ -109,7 +106,7 @@ describe(`Library watcher (e2e)`, () => {
`${IMMICH_TEST_ASSET_TEMP_PATH}/file.jpg`,
);

await waitForEvent(libraryService, 'change');
await waitForEvent(libraryService, StorageEventType.CHANGE);

const afterAssets = await api.assetApi.getAllAssets(server, admin.accessToken);
expect(afterAssets).toEqual([
Expand Down Expand Up @@ -161,9 +158,7 @@ describe(`Library watcher (e2e)`, () => {
`${IMMICH_TEST_ASSET_TEMP_PATH}/dir3/file4.jpg`,
);

await waitForEvent(libraryService, 'add');
await waitForEvent(libraryService, 'add');
await waitForEvent(libraryService, 'add');
await waitForEvent(libraryService, StorageEventType.ADD, 3);

const assets = await api.assetApi.getAllAssets(server, admin.accessToken);
expect(assets.length).toEqual(3);
Expand All @@ -175,14 +170,14 @@ describe(`Library watcher (e2e)`, () => {
`${IMMICH_TEST_ASSET_TEMP_PATH}/dir1/file.jpg`,
);

await waitForEvent(libraryService, 'add');
await waitForEvent(libraryService, StorageEventType.ADD);

const addedAssets = await api.assetApi.getAllAssets(server, admin.accessToken);
expect(addedAssets.length).toEqual(1);

await fs.unlink(`${IMMICH_TEST_ASSET_TEMP_PATH}/dir1/file.jpg`);

await waitForEvent(libraryService, 'unlink');
await waitForEvent(libraryService, StorageEventType.UNLINK);

const afterAssets = await api.assetApi.getAllAssets(server, admin.accessToken);
expect(afterAssets[0].isOffline).toEqual(true);
Expand Down Expand Up @@ -220,7 +215,7 @@ describe(`Library watcher (e2e)`, () => {
`${IMMICH_TEST_ASSET_TEMP_PATH}/dir4/file.jpg`,
);

await waitForEvent(libraryService, 'add');
await waitForEvent(libraryService, StorageEventType.ADD);

const afterAssets = await api.assetApi.getAllAssets(server, admin.accessToken);
expect(afterAssets.length).toEqual(1);
Expand Down
2 changes: 1 addition & 1 deletion server/e2e/jobs/specs/library.e2e-spec.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -368,7 +368,7 @@ describe(`${LibraryController.name} (e2e)`, () => {
expect(body).toEqual(errorStub.unauthorized);
});

it('should remvove offline files', async () => {
it('should remove offline files', async () => {
await fs.promises.cp(`${IMMICH_TEST_ASSET_PATH}/albums/nature`, `${IMMICH_TEST_ASSET_TEMP_PATH}/albums/nature`, {
recursive: true,
});
Expand Down
56 changes: 36 additions & 20 deletions server/src/domain/library/library.service.spec.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,11 +9,11 @@ import {
newAccessRepositoryMock,
newAssetRepositoryMock,
newCryptoRepositoryMock,
newDatabaseRepositoryMock,
newJobRepositoryMock,
newLibraryRepositoryMock,
newStorageRepositoryMock,
newSystemConfigRepositoryMock,
newUserRepositoryMock,
systemConfigStub,
userStub,
} from '@test';
Expand All @@ -23,11 +23,12 @@ import { ILibraryFileJob, ILibraryRefreshJob, JobName } from '../job';
import {
IAssetRepository,
ICryptoRepository,
IDatabaseRepository,
IJobRepository,
ILibraryRepository,
IStorageRepository,
ISystemConfigRepository,
IUserRepository,
StorageEventType,
} from '../repositories';
import { SystemConfigCore } from '../system-config/system-config.core';
import { mapLibrary } from './library.dto';
Expand All @@ -40,20 +41,20 @@ describe(LibraryService.name, () => {
let assetMock: jest.Mocked<IAssetRepository>;
let configMock: jest.Mocked<ISystemConfigRepository>;
let cryptoMock: jest.Mocked<ICryptoRepository>;
let userMock: jest.Mocked<IUserRepository>;
let jobMock: jest.Mocked<IJobRepository>;
let libraryMock: jest.Mocked<ILibraryRepository>;
let storageMock: jest.Mocked<IStorageRepository>;
let databaseMock: jest.Mocked<IDatabaseRepository>;

beforeEach(() => {
accessMock = newAccessRepositoryMock();
configMock = newSystemConfigRepositoryMock();
libraryMock = newLibraryRepositoryMock();
userMock = newUserRepositoryMock();
assetMock = newAssetRepositoryMock();
jobMock = newJobRepositoryMock();
cryptoMock = newCryptoRepositoryMock();
storageMock = newStorageRepositoryMock();
databaseMock = newDatabaseRepositoryMock();

// Always validate owner access for library.
accessMock.library.checkOwnerAccess.mockImplementation((_, libraryIds) => Promise.resolve(libraryIds));
Expand All @@ -66,8 +67,10 @@ describe(LibraryService.name, () => {
jobMock,
libraryMock,
storageMock,
userMock,
databaseMock,
);

databaseMock.tryLock.mockResolvedValue(true);
});

it('should work', () => {
Expand Down Expand Up @@ -125,13 +128,22 @@ describe(LibraryService.name, () => {
);
});

it('should not initialize when watching is disabled', async () => {
it('should not initialize watcher when watching is disabled', async () => {
configMock.load.mockResolvedValue(systemConfigStub.libraryWatchDisabled);

await sut.init();

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

it('should not initialize watcher when lock is taken', async () => {
configMock.load.mockResolvedValue(systemConfigStub.libraryWatchEnabled);
databaseMock.tryLock.mockResolvedValue(false);

await sut.init();

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

describe('handleQueueAssetRefresh', () => {
Expand All @@ -146,7 +158,6 @@ describe(LibraryService.name, () => {
storageMock.crawl.mockResolvedValue(['/data/user1/photo.jpg']);
assetMock.getPathsNotInLibrary.mockResolvedValue(['/data/user1/photo.jpg']);
assetMock.getByLibraryId.mockResolvedValue([]);
userMock.get.mockResolvedValue(userStub.admin);

await sut.handleQueueAssetRefresh(mockLibraryJob);

Expand All @@ -173,7 +184,6 @@ describe(LibraryService.name, () => {
libraryMock.get.mockResolvedValue(libraryStub.externalLibrary1);
storageMock.crawl.mockResolvedValue(['/data/user1/photo.jpg']);
assetMock.getByLibraryId.mockResolvedValue([]);
userMock.get.mockResolvedValue(userStub.admin);

await sut.handleQueueAssetRefresh(mockLibraryJob);

Expand Down Expand Up @@ -224,7 +234,6 @@ describe(LibraryService.name, () => {
libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
storageMock.crawl.mockResolvedValue([]);
assetMock.getByLibraryId.mockResolvedValue([]);
userMock.get.mockResolvedValue(userStub.externalPathRoot);

await sut.handleQueueAssetRefresh(mockLibraryJob);

Expand All @@ -240,7 +249,6 @@ describe(LibraryService.name, () => {

beforeEach(() => {
mockUser = userStub.admin;
userMock.get.mockResolvedValue(mockUser);

storageMock.stat.mockResolvedValue({
size: 100,
Expand Down Expand Up @@ -1167,7 +1175,9 @@ describe(LibraryService.name, () => {
it('should handle a new file event', async () => {
libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
libraryMock.getAll.mockResolvedValue([libraryStub.externalLibraryWithImportPaths1]);
storageMock.watch.mockImplementation(makeMockWatcher({ items: [{ event: 'add', value: '/foo/photo.jpg' }] }));
storageMock.watch.mockImplementation(
makeMockWatcher({ items: [{ event: StorageEventType.ADD, value: '/foo/photo.jpg' }] }),
);

await sut.watchAll();

Expand All @@ -1188,7 +1198,7 @@ describe(LibraryService.name, () => {
libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
libraryMock.getAll.mockResolvedValue([libraryStub.externalLibraryWithImportPaths1]);
storageMock.watch.mockImplementation(
makeMockWatcher({ items: [{ event: 'change', value: '/foo/photo.jpg' }] }),
makeMockWatcher({ items: [{ event: StorageEventType.CHANGE, value: '/foo/photo.jpg' }] }),
);

await sut.watchAll();
Expand All @@ -1211,7 +1221,7 @@ describe(LibraryService.name, () => {
libraryMock.getAll.mockResolvedValue([libraryStub.externalLibraryWithImportPaths1]);
assetMock.getByLibraryIdAndOriginalPath.mockResolvedValue(assetStub.external);
storageMock.watch.mockImplementation(
makeMockWatcher({ items: [{ event: 'unlink', value: '/foo/photo.jpg' }] }),
makeMockWatcher({ items: [{ event: StorageEventType.UNLINK, value: '/foo/photo.jpg' }] }),
);

await sut.watchAll();
Expand All @@ -1225,17 +1235,19 @@ describe(LibraryService.name, () => {
libraryMock.getAll.mockResolvedValue([libraryStub.externalLibraryWithImportPaths1]);
storageMock.watch.mockImplementation(
makeMockWatcher({
items: [{ event: 'error', value: 'Error!' }],
items: [{ event: StorageEventType.ERROR, value: 'Error!' }],
}),
);

await sut.watchAll();
await expect(sut.watchAll()).rejects.toThrow('Error!');
});

it('should ignore unknown extensions', async () => {
libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
libraryMock.getAll.mockResolvedValue([libraryStub.externalLibraryWithImportPaths1]);
storageMock.watch.mockImplementation(makeMockWatcher({ items: [{ event: 'add', value: '/foo/photo.jpg' }] }));
storageMock.watch.mockImplementation(
makeMockWatcher({ items: [{ event: StorageEventType.ADD, value: '/foo/photo.jpg' }] }),
);

await sut.watchAll();

Expand All @@ -1245,7 +1257,9 @@ describe(LibraryService.name, () => {
it('should ignore excluded paths', async () => {
libraryMock.get.mockResolvedValue(libraryStub.patternPath);
libraryMock.getAll.mockResolvedValue([libraryStub.patternPath]);
storageMock.watch.mockImplementation(makeMockWatcher({ items: [{ event: 'add', value: '/dir1/photo.txt' }] }));
storageMock.watch.mockImplementation(
makeMockWatcher({ items: [{ event: StorageEventType.ADD, value: '/dir1/photo.txt' }] }),
);

await sut.watchAll();

Expand All @@ -1255,7 +1269,9 @@ describe(LibraryService.name, () => {
it('should ignore excluded paths without case sensitivity', async () => {
libraryMock.get.mockResolvedValue(libraryStub.patternPath);
libraryMock.getAll.mockResolvedValue([libraryStub.patternPath]);
storageMock.watch.mockImplementation(makeMockWatcher({ items: [{ event: 'add', value: '/DIR1/photo.txt' }] }));
storageMock.watch.mockImplementation(
makeMockWatcher({ items: [{ event: StorageEventType.ADD, value: '/DIR1/photo.txt' }] }),
);

await sut.watchAll();

Expand All @@ -1264,7 +1280,7 @@ describe(LibraryService.name, () => {
});
});

describe('tearDown', () => {
describe('teardown', () => {
it('should tear down all watchers', async () => {
libraryMock.getAll.mockResolvedValue([
libraryStub.externalLibraryWithImportPaths1,
Expand All @@ -1286,7 +1302,7 @@ describe(LibraryService.name, () => {
storageMock.watch.mockImplementation(makeMockWatcher({ close: mockClose }));

await sut.init();
await sut.unwatchAll();
await sut.teardown();

expect(mockClose).toHaveBeenCalledTimes(2);
});
Expand Down

0 comments on commit 4cb0f37

Please sign in to comment.