chore(server): Move library watcher to microservices

docs/docs/features/libraries.md

@@ -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.

server/e2e/jobs/specs/library-watcher.e2e-spec.ts

@@ -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';
@@ -33,7 +33,7 @@ describe(`Library watcher (e2e)`, () => {
   });
 
   afterEach(async () => {
-    await libraryService.unwatchAll();
+    await libraryService.teardown();
   });
 
   afterAll(async () => {
@@ -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);
@@ -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);
@@ -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);
@@ -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([
@@ -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);
@@ -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);
@@ -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);

server/src/domain/library/library.service.spec.ts

@@ -9,11 +9,11 @@ import {
   newAccessRepositoryMock,
   newAssetRepositoryMock,
   newCryptoRepositoryMock,
+  newDatabaseRepositoryMock,
   newJobRepositoryMock,
   newLibraryRepositoryMock,
   newStorageRepositoryMock,
   newSystemConfigRepositoryMock,
-  newUserRepositoryMock,
   systemConfigStub,
   userStub,
 } from '@test';
@@ -22,11 +22,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';
@@ -39,20 +40,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(async (_, libraryIds) => libraryIds);
@@ -65,8 +66,10 @@ describe(LibraryService.name, () => {
       jobMock,
       libraryMock,
       storageMock,
-      userMock,
+      databaseMock,
     );
+
+    databaseMock.tryLock.mockResolvedValue(true);
   });
 
   it('should work', () => {
@@ -130,13 +133,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', () => {
@@ -150,7 +162,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);
 
@@ -177,7 +188,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);
 
@@ -228,7 +238,6 @@ describe(LibraryService.name, () => {
       libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
       storageMock.crawl.mockResolvedValue([]);
       assetMock.getByLibraryId.mockResolvedValue([]);
-      userMock.get.mockResolvedValue(userStub.externalPathRoot);
 
       await sut.handleQueueAssetRefresh(mockLibraryJob);
 
@@ -244,7 +253,6 @@ describe(LibraryService.name, () => {
 
     beforeEach(() => {
       mockUser = userStub.admin;
-      userMock.get.mockResolvedValue(mockUser);
 
       storageMock.stat.mockResolvedValue({
         size: 100,
@@ -1171,7 +1179,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();
 
@@ -1192,7 +1202,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();
@@ -1215,7 +1225,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();
@@ -1229,17 +1239,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();
 
@@ -1249,7 +1261,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();
 
@@ -1259,7 +1273,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();
 
@@ -1268,7 +1284,7 @@ describe(LibraryService.name, () => {
     });
   });
 
-  describe('tearDown', () => {
+  describe('teardown', () => {
     it('should tear down all watchers', async () => {
       libraryMock.getAll.mockResolvedValue([
         libraryStub.externalLibraryWithImportPaths1,
@@ -1296,7 +1312,7 @@ describe(LibraryService.name, () => {
       storageMock.watch.mockImplementation(makeMockWatcher({ close: mockClose }));
 
       await sut.init();
-      await sut.unwatchAll();
+      await sut.teardown();
 
       expect(mockClose).toHaveBeenCalledTimes(2);
     });