refactor(doctor): support abort signal in all operations

app/server/core/__tests__/repository-mutex.test.ts

@@ -35,4 +35,63 @@ describe("RepositoryMutex", () => {
 
 		releaseShared2();
 	});
+
+	test("should remove aborted acquisitions from the wait queue", async () => {
+		const repoId = "abort-test";
+		const results: string[] = [];
+
+		const releaseShared1 = await repoMutex.acquireShared(repoId, "backup-1");
+		results.push("acquired-shared-1");
+
+		const controller = new AbortController();
+		const exclusivePromise = repoMutex.acquireExclusive(repoId, "unlock", controller.signal).catch((err) => {
+			results.push("aborted-exclusive");
+			throw err;
+		});
+
+		const shared2Promise = repoMutex.acquireShared(repoId, "backup-2").then((release) => {
+			results.push("acquired-shared-2");
+			return release;
+		});
+
+		await new Promise((resolve) => setTimeout(resolve, 50));
+		expect(results).toEqual(["acquired-shared-1"]);
+
+		controller.abort();
+
+		expect(exclusivePromise).rejects.toThrow();
+		expect(results).toEqual(["acquired-shared-1", "aborted-exclusive"]);
+
+		releaseShared1();
+
+		// After exclusive is aborted, shared-2 should be next
+		const releaseShared2 = await shared2Promise;
+		expect(results).toEqual(["acquired-shared-1", "aborted-exclusive", "acquired-shared-2"]);
+
+		releaseShared2();
+	});
+
+	test("should handle multiple aborts correctly", async () => {
+		const repoId = "multi-abort";
+		const releaseShared1 = await repoMutex.acquireShared(repoId, "backup-1");
+
+		const controller1 = new AbortController();
+		const controller2 = new AbortController();
+
+		const p1 = repoMutex.acquireExclusive(repoId, "ex-1", controller1.signal);
+		const p2 = repoMutex.acquireExclusive(repoId, "ex-2", controller2.signal);
+		const p3 = repoMutex.acquireExclusive(repoId, "ex-3");
+
+		controller2.abort();
+		expect(p2).rejects.toThrow();
+
+		controller1.abort();
+		expect(p1).rejects.toThrow();
+
+		releaseShared1();
+
+		const releaseEx3 = await p3;
+		expect(releaseEx3).toBeDefined();
+		releaseEx3();
+	});
 });

app/server/core/repository-mutex.ts

@@ -46,7 +46,11 @@ class RepositoryMutex {
 		}
 	}
 
-	async acquireShared(repositoryId: string, operation: string): Promise<() => void> {
+	async acquireShared(repositoryId: string, operation: string, signal?: AbortSignal): Promise<() => void> {
+		if (signal?.aborted) {
+			throw signal.reason || new Error("Operation aborted");
+		}
+
 		const state = this.getOrCreateState(repositoryId);
 
 		const hasExclusiveInQueue = state.waitQueue.some((item) => item.type === "exclusive");
@@ -64,14 +68,42 @@ class RepositoryMutex {
 		logger.debug(
 			`[Mutex] Waiting for shared lock on repo ${repositoryId}: ${operation} (exclusive held by: ${state.exclusiveHolder?.operation ?? "none"}, queue: ${state.waitQueue.length})`,
 		);
-		const lockId = await new Promise<string>((resolve) => {
+
-			state.waitQueue.push({ type: "shared", operation, resolve });
+		let onAbort: () => void = () => {};
+		const lockId = await new Promise<string>((resolve, reject) => {
+			const waiter = { type: "shared" as const, operation, resolve };
+			state.waitQueue.push(waiter);
+
+			if (signal) {
+				onAbort = () => {
+					const index = state.waitQueue.indexOf(waiter);
+					if (index !== -1) {
+						state.waitQueue.splice(index, 1);
+						this.cleanupStateIfEmpty(repositoryId);
+						reject(signal.reason || new Error("Operation aborted"));
+					}
+				};
+				signal.addEventListener("abort", onAbort);
+			}
 		});
 
+		if (onAbort) {
+			signal?.removeEventListener("abort", onAbort);
+		}
+
+		if (signal?.aborted) {
+			this.releaseShared(repositoryId, lockId);
+			throw signal.reason || new Error("Operation aborted");
+		}
+
 		return () => this.releaseShared(repositoryId, lockId);
 	}
 
-	async acquireExclusive(repositoryId: string, operation: string): Promise<() => void> {
+	async acquireExclusive(repositoryId: string, operation: string, signal?: AbortSignal): Promise<() => void> {
+		if (signal?.aborted) {
+			throw signal.reason || new Error("Operation aborted");
+		}
+
 		const state = this.getOrCreateState(repositoryId);
 
 		if (!state.exclusiveHolder && state.sharedHolders.size === 0 && state.waitQueue.length === 0) {
@@ -87,10 +119,34 @@ class RepositoryMutex {
 		logger.debug(
 			`[Mutex] Waiting for exclusive lock on repo ${repositoryId}: ${operation} (shared: ${state.sharedHolders.size}, exclusive: ${state.exclusiveHolder ? "yes" : "no"}, queue: ${state.waitQueue.length})`,
 		);
-		const lockId = await new Promise<string>((resolve) => {
+
-			state.waitQueue.push({ type: "exclusive", operation, resolve });
+		let onAbort: () => void = () => {};
+		const lockId = await new Promise<string>((resolve, reject) => {
+			const waiter = { type: "exclusive" as const, operation, resolve };
+			state.waitQueue.push(waiter);
+
+			if (signal) {
+				onAbort = () => {
+					const index = state.waitQueue.indexOf(waiter);
+					if (index !== -1) {
+						state.waitQueue.splice(index, 1);
+						this.cleanupStateIfEmpty(repositoryId);
+						reject(signal.reason || new Error("Operation aborted"));
+					}
+				};
+				signal.addEventListener("abort", onAbort);
+			}
 		});
 
+		if (onAbort) {
+			signal?.removeEventListener("abort", onAbort);
+		}
+
+		if (signal?.aborted) {
+			this.releaseExclusive(repositoryId, lockId);
+			throw signal.reason || new Error("Operation aborted");
+		}
+
 		logger.debug(`[Mutex] Acquired exclusive lock for repo ${repositoryId}: ${operation} (${lockId})`);
 		return () => this.releaseExclusive(repositoryId, lockId);
 	}

app/server/modules/backups/backups.service.ts

@@ -309,7 +309,7 @@ const executeBackup = async (scheduleId: number, manual = false) => {
 			backupOptions.include = schedule.includePatterns.map((p) => processPattern(p, volumePath));
 		}
 
-		const releaseBackupLock = await repoMutex.acquireShared(repository.id, `backup:${volume.name}`);
+		const releaseBackupLock = await repoMutex.acquireShared(repository.id, `backup:${volume.name}`, abortController.signal);
 		let exitCode: number;
 		try {
 			const result = await restic.backup(repository.config, volumePath, {

app/server/modules/repositories/doctor.ts

@@ -9,8 +9,8 @@ import { type } from "arktype";
 import { serverEvents } from "../../core/events";
 import { logger } from "../../utils/logger";
 
-const runUnlockStep = async (config: RepositoryConfig): Promise<DoctorStep> => {
+const runUnlockStep = async (config: RepositoryConfig, signal?: AbortSignal) => {
-	const result = await restic.unlock(config).then(
+	const result = await restic.unlock(config, { signal }).then(
 		(result) => ({ success: true, message: result.message, error: null }),
 		(error) => ({ success: false, message: null, error: toMessage(error) }),
 	);
@@ -23,8 +23,8 @@ const runUnlockStep = async (config: RepositoryConfig): Promise<DoctorStep> => {
 	};
 };
 
-const runCheckStep = async (config: RepositoryConfig): Promise<DoctorStep> => {
+const runCheckStep = async (config: RepositoryConfig, signal: AbortSignal) => {
-	const result = await restic.check(config, { readData: true }).then(
+	const result = await restic.check(config, { readData: true, signal }).then(
 		(result) => result,
 		(error) => ({ success: false, output: null, error: toMessage(error), hasErrors: true }),
 	);
@@ -37,8 +37,8 @@ const runCheckStep = async (config: RepositoryConfig): Promise<DoctorStep> => {
 	};
 };
 
-const runRepairIndexStep = async (config: RepositoryConfig) => {
+const runRepairIndexStep = async (config: RepositoryConfig, signal: AbortSignal) => {
-	const result = await restic.repairIndex(config).then(
+	const result = await restic.repairIndex(config, { signal }).then(
 		(result) => ({ success: true, output: result.output, error: null }),
 		(error) => ({ success: false, output: null, error: toMessage(error) }),
 	);
@@ -103,28 +103,28 @@ export const executeDoctor = async (
 	repositoryId: string,
 	repositoryConfig: RepositoryConfig,
 	repositoryName: string,
-	signal: AbortSignal | undefined,
+	signal: AbortSignal,
 ) => {
 	const steps: DoctorStep[] = [];
 
 	try {
-		const releaseLock = await repoMutex.acquireExclusive(repositoryId, "doctor");
+		const releaseLock = await repoMutex.acquireExclusive(repositoryId, "doctor", signal);
 
+		try {
 			// Step 1: Unlock repository
-		const unlockStep = await runUnlockStep(repositoryConfig);
+			const unlockStep = await runUnlockStep(repositoryConfig, signal);
 			steps.push(unlockStep);
 			checkAbortSignal(signal);
 
-		// Step 2: Check repository with exclusive lock
+			// Step 2: Check repository
-		try {
+			const checkStep = await runCheckStep(repositoryConfig, signal);
-			const checkStep = await runCheckStep(repositoryConfig);
 			steps.push(checkStep);
 			checkAbortSignal(signal);
 
 			// Step 3: Repair index if suggested
 			const checkOutput = parseCheckOutput(checkStep.output);
 			if (checkOutput?.suggest_repair_index) {
-				const repairStep = await runRepairIndexStep(repositoryConfig);
+				const repairStep = await runRepairIndexStep(repositoryConfig, signal);
 				steps.push(repairStep);
 				checkAbortSignal(signal);
 			}

app/server/utils/restic.ts

@@ -736,14 +736,14 @@ const ls = async (config: RepositoryConfig, snapshotId: string, organizationId:
 	return { snapshot, nodes };
 };
 
-const unlock = async (config: RepositoryConfig, organizationId: string) => {
+const unlock = async (config: RepositoryConfig, options: { signal?: AbortSignal, organizationId: string }) => {
 	const repoUrl = buildRepoUrl(config);
-	const env = await buildEnv(config, organizationId);
+	const env = await buildEnv(config, options.organizationId);
 
 	const args = ["unlock", "--repo", repoUrl, "--remove-all"];
 	addCommonArgs(args, env, config);
 
-	const res = await exec({ command: "restic", args, env });
+	const res = await exec({ command: "restic", args, env, signal: options?.signal });
 	await cleanupTemporaryKeys(env);
 
 	if (res.exitCode !== 0) {
@@ -755,7 +755,7 @@ const unlock = async (config: RepositoryConfig, organizationId: string) => {
 	return { success: true, message: "Repository unlocked successfully" };
 };
 
-const check = async (config: RepositoryConfig, options: { readData?: boolean; organizationId: string }) => {
+const check = async (config: RepositoryConfig, options?: { readData?: boolean; signal?: AbortSignal }) => {
 	const repoUrl = buildRepoUrl(config);
 	const env = await buildEnv(config, options.organizationId);
 
@@ -767,7 +767,8 @@ const check = async (config: RepositoryConfig, options: { readData?: boolean; or
 
 	addCommonArgs(args, env, config);
 
-	const res = await exec({ command: "restic", args, env });
+	const res = await exec({ command: "restic", args, env, signal: options?.signal });
+
 	await cleanupTemporaryKeys(env);
 
 	const { stdout, stderr } = res;
@@ -793,14 +794,14 @@ const check = async (config: RepositoryConfig, options: { readData?: boolean; or
 	};
 };
 
-const repairIndex = async (config: RepositoryConfig, organizationId: string) => {
+const repairIndex = async (config: RepositoryConfig, options?: { signal?: AbortSignal }) => {
 	const repoUrl = buildRepoUrl(config);
 	const env = await buildEnv(config, organizationId);
 
 	const args = ["repair", "index", "--repo", repoUrl];
 	addCommonArgs(args, env, config);
 
-	const res = await exec({ command: "restic", args, env });
+	const res = await exec({ command: "restic", args, env, signal: options?.signal });
 	await cleanupTemporaryKeys(env);
 
 	const { stdout, stderr } = res;