fix(server): kill pg_dump when a backup fails so it stops holding table locks (#30851)

* fix(server): kill the child process when a spawned duplex stream is destroyed

A failed database backup left pg_dump running. The stream was destroyed but
the child was not, so the dump sat idle in transaction holding AccessShareLock
on every table until someone terminated it by hand. One report had it survive
five days and block DDL from another application on the same Postgres.

Destroying the duplex now kills the child, which also covers pipeline()
tearing every stream down when one of them fails.

The backup service additionally holds references to the streams it spawns so
the error path can destroy them. createWriteStream can throw before pipeline()
takes ownership, which is how the reported ENOENT on an unmounted backup volume
left nothing referencing the running pg_dump.

* chore(server): drop the explanatory comments from the backup teardown fix

Requested in review. The reasoning lives in the commit message and the PR
body, which is where it belongs.

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

@@ -81,5 +81,34 @@ describe(ProcessRepository.name, () => {
 
       await pipeline(Readable.from(data()), process);
     });
+
+    it('should kill the child process when the stream is destroyed', async () => {
+      const process = sut.spawnDuplexStream('bash', ['-c', 'sleep 60']);
+      const realProcess = (process as never as { _process: ChildProcessWithoutNullStreams })._process;
+
+      expect(realProcess.exitCode).toBeNull();
+
+      const exited = new Promise<void>((resolve) => realProcess.once('exit', () => resolve()));
+      process.destroy();
+      await exited;
+
+      expect(realProcess.killed).toBe(true);
+    });
+
+    it('should kill the child when pipeline tears the stream down after a failure', async () => {
+      const process = sut.spawnDuplexStream('yes');
+      const realProcess = (process as never as { _process: ChildProcessWithoutNullStreams })._process;
+      const failingSink = new Writable({
+        write(_chunk, _encoding, callback) {
+          callback(new Error('sink exploded'));
+        },
+      });
+
+      const exited = new Promise<void>((resolve) => realProcess.once('exit', () => resolve()));
+      await expect(pipeline(process, failingSink)).rejects.toThrow('sink exploded');
+      await exited;
+
+      expect(realProcess.killed).toBe(true);
+    });
   });
 });

server/src/repositories/process.repository.ts

@@ -42,6 +42,13 @@ export class ProcessRepository {
           process.stdin.end(callback);
         }
       },
+
+      destroy(error, callback) {
+        if (process.exitCode === null && process.signalCode === null) {
+          process.kill();
+        }
+        callback(error);
+      },
     });
 
     // stdout -> duplex

server/src/services/database-backup.service.spec.ts

@@ -1,6 +1,6 @@
 import { BadRequestException } from '@nestjs/common';
 import { DateTime } from 'luxon';
-import { PassThrough, Readable } from 'node:stream';
+import { Duplex, PassThrough, Readable } from 'node:stream';
 import { StorageCore } from 'src/cores/storage.core';
 import { defaults, SystemConfig } from 'src/dtos/config.dto';
 import { ImmichWorker, JobStatus, StorageFolder } from 'src/enum';
@@ -252,6 +252,25 @@ describe(DatabaseBackupService.name, () => {
       await expect(sut.handleBackupDatabase()).rejects.toThrow('error');
     });
 
+    it('should destroy the spawned processes if the write stream fails', async () => {
+      const spawned: Duplex[] = [];
+      mocks.process.spawnDuplexStream.mockImplementation(() => {
+        const duplex = mockDuplex()('command', 0, 'data', '');
+        spawned.push(duplex);
+        return duplex;
+      });
+      mocks.storage.createWriteStream.mockImplementation(() => {
+        throw new Error('ENOENT: no such file or directory');
+      });
+
+      await expect(sut.handleBackupDatabase()).rejects.toThrow('ENOENT');
+
+      expect(spawned).toHaveLength(2);
+      for (const stream of spawned) {
+        expect(stream.destroyed).toBe(true);
+      }
+    });
+
     it('should fail if rename fails', async () => {
       mocks.storage.rename.mockRejectedValue(new Error('error'));
       await expect(sut.handleBackupDatabase()).rejects.toThrow('error');

server/src/services/database-backup.service.ts

@@ -2,7 +2,7 @@ import { BadRequestException, Injectable, Optional } from '@nestjs/common';
 import { debounce } from 'lodash';
 import { DateTime } from 'luxon';
 import path, { basename } from 'node:path';
-import { PassThrough, Readable, Writable } from 'node:stream';
+import { Duplex, PassThrough, Readable, Writable } from 'node:stream';
 import { pipeline } from 'node:stream/promises';
 import semver from 'semver';
 import { serverVersion } from 'src/constants';
@@ -236,21 +236,26 @@ export class DatabaseBackupService {
     const backupFilePath = path.join(StorageCore.getBaseFolder(StorageFolder.Backups), filename);
     const temporaryFilePath = `${backupFilePath}.tmp`;
 
+    let pgdump: Duplex | undefined;
+    let gzip: Duplex | undefined;
+
     try {
-      const pgdump = this.processRepository.spawnDuplexStream(bin, args, {
+      pgdump = this.processRepository.spawnDuplexStream(bin, args, {
         env: {
           PATH: process.env.PATH,
           PGPASSWORD: databasePassword,
         },
       });
 
-      const gzip = this.processRepository.spawnDuplexStream('gzip', ['--rsyncable']);
+      gzip = this.processRepository.spawnDuplexStream('gzip', ['--rsyncable']);
       const fileStream = this.storageRepository.createWriteStream(temporaryFilePath);
 
       await pipeline(pgdump, gzip, fileStream);
       await this.storageRepository.rename(temporaryFilePath, backupFilePath);
     } catch (error) {
       this.logger.error(`Database Backup Failure: ${error}`);
+      pgdump?.destroy();
+      gzip?.destroy();
       await this.storageRepository
         .unlink(temporaryFilePath)