refactor: pr feedbacks

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

@@ -583,6 +583,27 @@ describe("RepositoryMutex", () => {
 		expect(repoMutex.isLocked(repoId)).toBe(false);
 	});
 
+	test("should release the lease without invoking the operation when aborted after acquisition", async () => {
+		const repoId = "abort-after-acquire";
+		const abortController = new AbortController();
+		let operationStarted = false;
+		const abortReason = new Error("caller aborted");
+
+		const operation = repoMutex.runShared(
+			repoId,
+			"abort-before-callback",
+			async () => {
+				operationStarted = true;
+			},
+			abortController.signal,
+		);
+		abortController.abort(abortReason);
+
+		await expect(operation).rejects.toThrow("caller aborted");
+		expect(operationStarted).toBe(false);
+		expect(repoMutex.isLocked(repoId)).toBe(false);
+	});
+
 	test("shutdown should release owned lock rows after the timeout when an operation ignores abort", async () => {
 		const repoId = "shutdown-timeout-release";
 		const started = createDeferred<AbortSignal>();

app/server/core/repository-mutex.ts

@@ -147,6 +147,7 @@ export class RepositoryMutex {
 
 	private async runWithLease<T>(lease: RepositoryLease, operation: RepositoryOperation<T>) {
 		try {
+			this.throwIfAborted(lease.signal);
 			return await operation({ signal: lease.signal });
 		} finally {
 			lease.release();

app/server/modules/backups/__tests__/backups.service.execution.test.ts

@@ -26,6 +26,7 @@ import { config } from "~/server/core/config";
 import { Effect } from "effect";
 import { taskStore } from "~/server/modules/tasks/tasks.store";
 import { holdExclusiveLock } from "~/test/helpers/repository-mutex";
+import { repoMutex } from "~/server/core/repository-mutex";
 
 const setup = () => {
 	const resticBackupMock = vi.fn((_: SafeSpawnParams) =>
@@ -871,6 +872,42 @@ describe("stop backup", () => {
 		expect(resticBackupMock).not.toHaveBeenCalled();
 	});
 
+	test("should cancel a queued backup when repository mutex shutdown aborts it", async () => {
+		const { resticBackupMock } = setup();
+		const volume = await createTestVolume();
+		const repository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: repository.id,
+		});
+
+		const releaseLock = await holdExclusiveLock(repository.id, "test");
+		const executePromise = backupsService.executeBackup(schedule.id);
+
+		try {
+			await waitForExpect(async () => {
+				const task = await getBackupTaskForSchedule(schedule.id);
+				expect(task?.status).toBe("queued");
+			});
+
+			expect(resticBackupMock).not.toHaveBeenCalled();
+
+			await repoMutex.shutdown({ timeoutMs: 10 });
+		} finally {
+			await releaseLock();
+		}
+
+		await executePromise;
+
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		const task = await getBackupTaskForSchedule(schedule.id);
+		expect(updatedSchedule.lastBackupStatus).toBe("warning");
+		expect(updatedSchedule.lastBackupError).toBe("Repository mutex is shutting down");
+		expect(task?.status).toBe("cancelled");
+		expect(task?.error).toBe("Repository mutex is shutting down");
+		expect(resticBackupMock).not.toHaveBeenCalled();
+	});
+
 	test("should clear failureRetryCount when a scheduled retry is cancelled", async () => {
 		const { resticBackupMock } = setup();
 		const volume = await createTestVolume();

app/server/modules/backups/backup-executor.ts

@@ -17,6 +17,7 @@ const IGNORE_INODE_FLAG = "--ignore-inode";
 type BackupExecutionRequest = {
 	jobId: string;
 	scheduleId: number;
+	trackedAbortController: AbortController;
 	schedule: BackupSchedule;
 	volume: Volume;
 	repository: Repository;
@@ -109,6 +110,19 @@ const executeBackupWithoutAgent = async (
 	);
 };
 
+const getTrackedExecution = (scheduleId: number, abortController: AbortController) => {
+	const trackedExecution = activeControllersByScheduleId.get(scheduleId);
+	if (!trackedExecution) {
+		throw new Error(`Backup execution for schedule ${scheduleId} was not tracked`);
+	}
+
+	if (trackedExecution.abortController !== abortController) {
+		throw new Error(`Backup execution for schedule ${scheduleId} was superseded`);
+	}
+
+	return trackedExecution;
+};
+
 export const backupExecutor = {
 	track: (scheduleId: number) => {
 		const abortController = new AbortController();
@@ -121,10 +135,7 @@ export const backupExecutor = {
 		}
 	},
 	execute: async (request: BackupExecutionRequest) => {
-		const trackedExecution = activeControllersByScheduleId.get(request.scheduleId);
-		if (!trackedExecution) {
-			throw new Error(`Backup execution for schedule ${request.scheduleId} was not tracked`);
-		}
+		getTrackedExecution(request.scheduleId, request.trackedAbortController);
 
 		if (request.signal.aborted) {
 			throw request.signal.reason || new Error("Operation aborted");
@@ -137,6 +148,7 @@ export const backupExecutor = {
 		}
 
 		const executionAgentId = getBackupExecutionAgentId(request.volume, request.repository);
+		const trackedExecution = getTrackedExecution(request.scheduleId, request.trackedAbortController);
 		trackedExecution.agentId = executionAgentId;
 
 		const executionResult = await agentManager.runBackup(executionAgentId, {

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

@@ -460,6 +460,7 @@ const executeBackup = async (scheduleId: number, manual = false) => {
 				const executionResult = await backupExecutor.execute({
 					jobId: task.id,
 					scheduleId,
+					trackedAbortController: abortController,
 					schedule: ctx.schedule,
 					volume: ctx.volume,
 					repository: ctx.repository,
@@ -514,7 +515,16 @@ const executeBackup = async (scheduleId: number, manual = false) => {
 		);
 	} catch (error) {
 		if (abortController.signal.aborted) {
-			taskStore.cancel(task.id, "Backup was stopped by the user");
+			const message = "Backup was stopped by the user";
+			await handleBackupCancellation(scheduleId, ctx.organizationId, message);
+			taskStore.cancel(task.id, message);
+			return;
+		}
+
+		if (toMessage(error) === "Repository mutex is shutting down") {
+			const message = toMessage(error);
+			await handleBackupCancellation(scheduleId, ctx.organizationId, message);
+			taskStore.cancel(task.id, message);
 			return;
 		}
 

app/server/modules/repositories/__tests__/repositories.service.test.ts

@@ -357,6 +357,37 @@ describe("repositoriesService.listSnapshotFiles", () => {
 	});
 });
 
+describe("repositoriesService.checkHealth", () => {
+	afterEach(() => {
+		vi.restoreAllMocks();
+	});
+
+	test("does not persist an aborted health check as a repository error", async () => {
+		const repository = await createTestRepository(session.organizationId, {
+			status: "healthy",
+			lastError: "previous error",
+		});
+		let shutdownPromise: Promise<void> | null = null;
+		vi.spyOn(restic, "check").mockImplementation(() => {
+			shutdownPromise = repoMutex.shutdown({ timeoutMs: 10 });
+			return Effect.succeed({ success: false, hasErrors: true, output: "", error: "Operation aborted" });
+		});
+
+		await expect(
+			withContext({ organizationId: session.organizationId, userId: session.user.id }, () =>
+				repositoriesService.checkHealth(repository.shortId),
+			),
+		).rejects.toThrow("Repository mutex is shutting down");
+		await shutdownPromise;
+
+		const persistedRepository = await db.query.repositoriesTable.findFirst({
+			where: { id: repository.id },
+		});
+		expect(persistedRepository?.status).toBe("healthy");
+		expect(persistedRepository?.lastError).toBe("previous error");
+	});
+});
+
 describe("repositoriesService.dumpSnapshot", () => {
 	afterEach(() => {
 		vi.restoreAllMocks();

app/server/modules/repositories/repositories.service.ts

@@ -715,6 +715,9 @@ const checkHealth = async (shortId: ShortId) => {
 		const { hasErrors, error } = await runEffectPromise(
 			restic.check(repository.config, { organizationId, signal }),
 		);
+		if (signal.aborted) {
+			throw signal.reason ?? new Error("Operation aborted");
+		}
 
 		await db
 			.update(repositoriesTable)

packages/core/src/restic/commands/__tests__/copy.test.ts

@@ -102,4 +102,24 @@ describe("copy command", () => {
 		expect(separatorIndex).toBeGreaterThan(-1);
 		expect(getArgs().slice(separatorIndex + 1)).toEqual(["latest"]);
 	});
+
+	test("does not treat a cleanup-time abort as a failed successful copy", async () => {
+		const controller = new AbortController();
+		setup();
+		vi.spyOn(cleanupModule, "cleanupTemporaryKeys").mockImplementation(async () => {
+			controller.abort(new Error("aborted during cleanup"));
+		});
+
+		const result = await Effect.runPromise(
+			copy(
+				sourceConfig,
+				destConfig,
+				{ organizationId: "org-1", tag: "daily", signal: controller.signal },
+				mockDeps,
+			),
+		);
+
+		expect(result).toEqual({ success: true, output: "copied" });
+		expect(cleanupModule.cleanupTemporaryKeys).toHaveBeenCalledTimes(2);
+	});
 });

packages/core/src/restic/commands/__tests__/delete-snapshots.test.ts

@@ -68,4 +68,19 @@ describe("deleteSnapshots command", () => {
 		expect(nodeModule.safeExec).not.toHaveBeenCalled();
 		expect(cleanupModule.cleanupTemporaryKeys).not.toHaveBeenCalled();
 	});
+
+	test("does not treat a cleanup-time abort as a failed successful deletion", async () => {
+		const controller = new AbortController();
+		setup();
+		vi.spyOn(cleanupModule, "cleanupTemporaryKeys").mockImplementation(async () => {
+			controller.abort(new Error("aborted during cleanup"));
+		});
+
+		const result = await Effect.runPromise(
+			deleteSnapshots(config, ["snapshot-1"], { organizationId: "org-1", signal: controller.signal }, mockDeps),
+		);
+
+		expect(result).toEqual({ success: true });
+		expect(cleanupModule.cleanupTemporaryKeys).toHaveBeenCalledTimes(1);
+	});
 });

packages/core/src/restic/commands/__tests__/forget.test.ts

@@ -0,0 +1,55 @@
+import { afterEach, describe, expect, test, vi } from "vitest";
+import * as cleanupModule from "../../helpers/cleanup-temporary-keys";
+import * as nodeModule from "../../../node";
+import { forget } from "../forget";
+import type { ResticDeps } from "../../types";
+import { Effect } from "effect";
+
+const mockDeps: ResticDeps = {
+	resolveSecret: async (s) => s,
+	getOrganizationResticPassword: async () => "org-restic-password",
+	resticCacheDir: "/tmp/restic-cache",
+	resticPassFile: "/tmp/restic.pass",
+	defaultExcludes: ["/tmp/restic.pass", "/var/lib/zerobyte/repositories"],
+	rcloneConfigFile: "/root/.config/rclone/rclone.conf",
+};
+
+const config = {
+	backend: "local" as const,
+	path: "/tmp/restic-repo",
+	isExistingRepository: true,
+	customPassword: "custom-password",
+};
+
+const setup = () => {
+	vi.spyOn(cleanupModule, "cleanupTemporaryKeys").mockImplementation(() => Promise.resolve());
+	vi.spyOn(nodeModule, "safeExec").mockImplementation(async () => {
+		return { exitCode: 0, stdout: "", stderr: "", timedOut: false };
+	});
+};
+
+afterEach(() => {
+	vi.restoreAllMocks();
+});
+
+describe("forget command", () => {
+	test("does not treat a cleanup-time abort as a failed successful forget", async () => {
+		const controller = new AbortController();
+		setup();
+		vi.spyOn(cleanupModule, "cleanupTemporaryKeys").mockImplementation(async () => {
+			controller.abort(new Error("aborted during cleanup"));
+		});
+
+		const result = await Effect.runPromise(
+			forget(
+				config,
+				{ keepLast: 1 },
+				{ organizationId: "org-1", tag: "daily", signal: controller.signal },
+				mockDeps,
+			),
+		);
+
+		expect(result).toEqual({ success: true, data: null });
+		expect(cleanupModule.cleanupTemporaryKeys).toHaveBeenCalledTimes(1);
+	});
+});

packages/core/src/restic/commands/copy.ts

@@ -63,19 +63,22 @@ export const copy = (
 			logger.info(`Copying snapshots from ${sourceRepoUrl} to ${destRepoUrl}...`);
 			logger.debug(`Executing: restic ${args.join(" ")}`);
 
-			const res = await safeExec({ command: "restic", args, env, signal: options.signal });
-
-			await cleanupTemporaryKeys(sourceEnv, deps);
-			await cleanupTemporaryKeys(destEnv, deps);
-
-			if (options.signal?.aborted) {
-				logger.warn("Restic copy was aborted by signal.");
-				throw new Error("Operation aborted");
+			let res: Awaited<ReturnType<typeof safeExec>>;
+			try {
+				res = await safeExec({ command: "restic", args, env, signal: options.signal });
+			} finally {
+				await cleanupTemporaryKeys(sourceEnv, deps);
+				await cleanupTemporaryKeys(destEnv, deps);
 			}
 
 			const { stdout, stderr } = res;
 
 			if (res.exitCode !== 0) {
+				if (options.signal?.aborted) {
+					logger.warn("Restic copy was aborted by signal.");
+					throw new Error("Operation aborted");
+				}
+
 				logger.error(`Restic copy failed: ${stderr}`);
 				throw createResticError(res.exitCode, stderr);
 			}

packages/core/src/restic/commands/delete-snapshots.ts

@@ -33,15 +33,19 @@ export const deleteSnapshots = (
 			addCommonArgs(args, env, config);
 			args.push("--", ...snapshotIds);
 
-			const res = await safeExec({ command: "restic", args, env, signal: options.signal });
-			await cleanupTemporaryKeys(env, deps);
-
-			if (options.signal?.aborted) {
-				logger.warn("Restic snapshot deletion was aborted by signal.");
-				throw new Error("Operation aborted");
+			let res: Awaited<ReturnType<typeof safeExec>>;
+			try {
+				res = await safeExec({ command: "restic", args, env, signal: options.signal });
+			} finally {
+				await cleanupTemporaryKeys(env, deps);
 			}
 
 			if (res.exitCode !== 0) {
+				if (options.signal?.aborted) {
+					logger.warn("Restic snapshot deletion was aborted by signal.");
+					throw new Error("Operation aborted");
+				}
+
 				logger.error(`Restic snapshot deletion failed: ${res.stderr}`);
 				throw createResticError(res.exitCode, res.stderr);
 			}

packages/core/src/restic/commands/forget.ts

@@ -60,15 +60,19 @@ export const forget = (
 
 			addCommonArgs(args, env, config);
 
-			const res = await safeExec({ command: "restic", args, env, signal: extra.signal });
-			await cleanupTemporaryKeys(env, deps);
-
-			if (extra.signal?.aborted) {
-				logger.warn("Restic forget was aborted by signal.");
-				throw new Error("Operation aborted");
+			let res: Awaited<ReturnType<typeof safeExec>>;
+			try {
+				res = await safeExec({ command: "restic", args, env, signal: extra.signal });
+			} finally {
+				await cleanupTemporaryKeys(env, deps);
 			}
 
 			if (res.exitCode !== 0) {
+				if (extra.signal?.aborted) {
+					logger.warn("Restic forget was aborted by signal.");
+					throw new Error("Operation aborted");
+				}
+
 				logger.error(`Restic forget failed: ${res.stderr}`);
 				throw createResticError(res.exitCode, res.stderr);
 			}