fix(server): sync files to disk

Ensure that all files are flushed after they've been written.

At current, files are not explicitly flushed to disk, which can cause
data corruption. In extreme circumstances, it's possible that uploaded
files may not ever be persisted at all.

The consequence of this is that it may reduce throughput, but Immich
should prioritise data integrity over performance.

server/src/middleware/file-upload.interceptor.ts

@@ -10,6 +10,7 @@ import { UploadFieldName } from 'src/dtos/asset-media.dto';
 import { RouteKey } from 'src/enum';
 import { AuthRequest } from 'src/middleware/auth.guard';
 import { LoggingRepository } from 'src/repositories/logging.repository';
+import { StorageRepository } from 'src/repositories/storage.repository';
 import { AssetMediaService } from 'src/services/asset-media.service';
 import { ImmichFile, UploadFile, UploadFiles } from 'src/types';
 import { asUploadRequest, mapToUploadFile } from 'src/utils/asset.util';
@@ -54,6 +55,7 @@ export class FileUploadInterceptor implements NestInterceptor {
   constructor(
     private reflect: Reflector,
     private assetService: AssetMediaService,
+    private storageRepository: StorageRepository,
     private logger: LoggingRepository,
   ) {
     this.logger.setContext(FileUploadInterceptor.name);
@@ -125,7 +127,18 @@ export class FileUploadInterceptor implements NestInterceptor {
     });
 
     if (!this.isAssetUploadFile(file)) {
-      this.defaultStorage._handleFile(request, file, callback);
+      this.defaultStorage._handleFile(request, file, (error, info) => {
+        if (error) {
+          return callback(error);
+        }
+        // Multer does not sync files to disk after writing.
+        //
+        // TODO: use `flush: true` in multer when available: https://github.com/expressjs/multer/issues/1381
+        this.storageRepository
+          .datasync(info!.path!)
+          .then(() => callback(null, info!))
+          .catch((error) => callback(error));
+      });
       return;
     }
 
@@ -136,7 +149,13 @@ export class FileUploadInterceptor implements NestInterceptor {
         hash.destroy();
         callback(error);
       } else {
-        callback(null, { ...info, checksum: hash.digest() });
+        this.storageRepository
+          .datasync(info!.path!)
+          .then(() => callback(null, { ...info, checksum: hash.digest() }))
+          .catch((error) => {
+            hash.destroy();
+            callback(error);
+          });
       }
     });
   }

server/src/repositories/media.repository.spec.ts

@@ -6,6 +6,7 @@ import { SourceType } from 'src/enum';
 import { LoggingRepository } from 'src/repositories/logging.repository';
 import { BoundingBox } from 'src/repositories/machine-learning.repository';
 import { MediaRepository } from 'src/repositories/media.repository';
+import { StorageRepository } from 'src/repositories/storage.repository';
 import { checkFaceVisibility, checkOcrVisibility } from 'src/utils/editor';
 import { automock } from 'test/utils';
 
@@ -65,8 +66,11 @@ describe(MediaRepository.name, () => {
   let sut: MediaRepository;
 
   beforeEach(() => {
-    // eslint-disable-next-line no-sparse-arrays
-    sut = new MediaRepository(automock(LoggingRepository, { args: [, { getEnv: () => ({}) }], strict: false }));
+    sut = new MediaRepository(
+      // eslint-disable-next-line no-sparse-arrays
+      automock(LoggingRepository, { args: [, { getEnv: () => ({}) }], strict: false }),
+      automock(StorageRepository, { args: [{ setContext: () => {} }], strict: false }),
+    );
   });
 
   describe('applyEdits (single actions)', () => {

server/src/repositories/media.repository.ts

@@ -10,6 +10,7 @@ import { Exif } from 'src/database';
 import { AssetEditActionItem } from 'src/dtos/editing.dto';
 import { Colorspace, LogLevel, RawExtractedFormat } from 'src/enum';
 import { LoggingRepository } from 'src/repositories/logging.repository';
+import { StorageRepository } from 'src/repositories/storage.repository';
 import {
   DecodeToBufferOptions,
   GenerateThumbhashOptions,
@@ -45,7 +46,10 @@ export type ExtractResult = {
 
 @Injectable()
 export class MediaRepository {
-  constructor(private logger: LoggingRepository) {
+  constructor(
+    private logger: LoggingRepository,
+    private storageRepository: StorageRepository,
+  ) {
     this.logger.setContext(MediaRepository.name);
   }
 
@@ -116,6 +120,7 @@ export class MediaRepository {
         ignoreMinorErrors: true,
         writeArgs: ['-overwrite_original'],
       });
+      await this.storageRepository.datasync(output);
       return true;
     } catch (error: any) {
       this.logger.warn(`Could not write exif data to image: ${error.message}`);
@@ -133,6 +138,7 @@ export class MediaRepository {
           writeArgs: ['-TagsFromFile', source, `-${tagGroup}:all>${tagGroup}:all`, '-overwrite_original'],
         },
       );
+      await this.storageRepository.datasync(target);
       return true;
     } catch (error: any) {
       this.logger.warn(`Could not copy tag data to image: ${error.message}`);
@@ -180,6 +186,7 @@ export class MediaRepository {
     });
 
     await decoded.toFile(output);
+    await this.storageRepository.datasync(output);
   }
 
   private async getImageDecodingPipeline(input: string | Buffer, options: DecodeToBufferOptions) {
@@ -274,14 +281,18 @@ export class MediaRepository {
     };
   }
 
-  transcode(input: string, output: string | Writable, options: TranscodeCommand): Promise<void> {
+  async transcode(input: string, output: string | Writable, options: TranscodeCommand): Promise<void> {
     if (!options.twoPass) {
-      return new Promise((resolve, reject) => {
+      await new Promise<void>((resolve, reject) => {
         this.configureFfmpegCall(input, output, options)
           .on('error', reject)
           .on('end', () => resolve())
           .run();
       });
+      if (typeof output === 'string') {
+        await this.storageRepository.datasync(output);
+      }
+      return;
     }
 
     if (typeof output !== 'string') {
@@ -290,7 +301,7 @@ export class MediaRepository {
 
     // two-pass allows for precise control of bitrate at the cost of running twice
     // recommended for vp9 for better quality and compression
-    return new Promise((resolve, reject) => {
+    await new Promise<void>((resolve, reject) => {
       // first pass output is not saved as only the .log file is needed
       this.configureFfmpegCall(input, '/dev/null', options)
         .addOptions('-pass', '1')
@@ -310,6 +321,7 @@ export class MediaRepository {
         })
         .run();
     });
+    await this.storageRepository.datasync(output);
   }
 
   async getImageMetadata(input: string | Buffer): Promise<ImageDimensions & { isTransparent: boolean }> {

server/src/repositories/metadata.repository.ts

@@ -2,6 +2,7 @@ import { Injectable } from '@nestjs/common';
 import { BinaryField, DefaultReadTaskOptions, ExifTool, Tags } from 'exiftool-vendored';
 import geotz from 'geo-tz';
 import { LoggingRepository } from 'src/repositories/logging.repository';
+import { StorageRepository } from 'src/repositories/storage.repository';
 import { mimeTypes } from 'src/utils/mime-types';
 
 interface ExifDuration {
@@ -94,7 +95,10 @@ export class MetadataRepository {
     taskTimeoutMillis: 2 * 60 * 1000,
   });
 
-  constructor(private logger: LoggingRepository) {
+  constructor(
+    private logger: LoggingRepository,
+    private storageRepository: StorageRepository,
+  ) {
     this.logger.setContext(MetadataRepository.name);
   }
 
@@ -121,6 +125,7 @@ export class MetadataRepository {
   async writeTags(path: string, tags: Partial<Tags>): Promise<void> {
     try {
       await this.exiftool.write(path, tags);
+      await this.storageRepository.datasync(path);
     } catch (error) {
       this.logger.warn(`Error writing exif data (${path}): ${error}`);
     }

server/src/repositories/storage.repository.ts

@@ -50,8 +50,18 @@ export class StorageRepository {
     return fs.readdir(folder);
   }
 
-  copyFile(source: string, target: string) {
-    return fs.copyFile(source, target);
+  async copyFile(source: string, target: string) {
+    await fs.copyFile(source, target);
+    await this.datasync(target);
+  }
+
+  async datasync(filepath: string) {
+    const handle = await fs.open(filepath, 'r');
+    try {
+      await handle.datasync();
+    } finally {
+      await handle.close();
+    }
   }
 
   stat(filepath: string) {
@@ -59,19 +69,19 @@ export class StorageRepository {
   }
 
   createFile(filepath: string, buffer: Buffer) {
-    return fs.writeFile(filepath, buffer, { flag: 'wx' });
+    return fs.writeFile(filepath, buffer, { flag: 'wx', flush: true });
   }
 
   createWriteStream(filepath: string): Writable {
-    return createWriteStream(filepath, { flags: 'w' });
+    return createWriteStream(filepath, { flags: 'w', flush: true });
   }
 
   createOrOverwriteFile(filepath: string, buffer: Buffer) {
-    return fs.writeFile(filepath, buffer, { flag: 'w' });
+    return fs.writeFile(filepath, buffer, { flag: 'w', flush: true });
   }
 
   overwriteFile(filepath: string, buffer: Buffer) {
-    return fs.writeFile(filepath, buffer, { flag: 'r+' });
+    return fs.writeFile(filepath, buffer, { flag: 'r+', flush: true });
   }
 
   rename(source: string, target: string) {

server/test/repositories/storage.repository.mock.ts

@@ -72,6 +72,7 @@ export const newStorageRepositoryMock = (): Mocked<RepositoryInterface<StorageRe
     walk: vitest.fn().mockImplementation(async function* () {}),
     rename: vitest.fn(),
     copyFile: vitest.fn(),
+    datasync: vitest.fn(),
     utimes: vitest.fn(),
     watch: vitest.fn().mockImplementation(makeMockWatcher({})),
   };