diff --git a/app/server/core/__tests__/repository-mutex.test.ts b/app/server/core/__tests__/repository-mutex.test.ts index 4ff80e5f..35565a96 100644 --- a/app/server/core/__tests__/repository-mutex.test.ts +++ b/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(); diff --git a/app/server/core/repository-mutex.ts b/app/server/core/repository-mutex.ts index 3ed3c879..d2284cdd 100644 --- a/app/server/core/repository-mutex.ts +++ b/app/server/core/repository-mutex.ts @@ -147,6 +147,7 @@ export class RepositoryMutex { private async runWithLease(lease: RepositoryLease, operation: RepositoryOperation) { try { + this.throwIfAborted(lease.signal); return await operation({ signal: lease.signal }); } finally { lease.release(); diff --git a/app/server/modules/backups/__tests__/backups.service.execution.test.ts b/app/server/modules/backups/__tests__/backups.service.execution.test.ts index ac809295..6ec85c37 100644 --- a/app/server/modules/backups/__tests__/backups.service.execution.test.ts +++ b/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(); diff --git a/app/server/modules/backups/backup-executor.ts b/app/server/modules/backups/backup-executor.ts index 28c16172..d8b0a913 100644 --- a/app/server/modules/backups/backup-executor.ts +++ b/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, { diff --git a/app/server/modules/backups/backups.service.ts b/app/server/modules/backups/backups.service.ts index d01b1ea0..fb05ffb6 100644 --- a/app/server/modules/backups/backups.service.ts +++ b/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; } diff --git a/app/server/modules/repositories/__tests__/repositories.service.test.ts b/app/server/modules/repositories/__tests__/repositories.service.test.ts index eba63d7b..2c868d84 100644 --- a/app/server/modules/repositories/__tests__/repositories.service.test.ts +++ b/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 | 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(); diff --git a/app/server/modules/repositories/repositories.service.ts b/app/server/modules/repositories/repositories.service.ts index 278117f0..d5e5a23e 100644 --- a/app/server/modules/repositories/repositories.service.ts +++ b/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) diff --git a/packages/core/src/restic/commands/__tests__/copy.test.ts b/packages/core/src/restic/commands/__tests__/copy.test.ts index 0c21392f..60d2ce25 100644 --- a/packages/core/src/restic/commands/__tests__/copy.test.ts +++ b/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); + }); }); diff --git a/packages/core/src/restic/commands/__tests__/delete-snapshots.test.ts b/packages/core/src/restic/commands/__tests__/delete-snapshots.test.ts index 1b35680a..525b3292 100644 --- a/packages/core/src/restic/commands/__tests__/delete-snapshots.test.ts +++ b/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); + }); }); diff --git a/packages/core/src/restic/commands/__tests__/forget.test.ts b/packages/core/src/restic/commands/__tests__/forget.test.ts new file mode 100644 index 00000000..817d5a5e --- /dev/null +++ b/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); + }); +}); diff --git a/packages/core/src/restic/commands/copy.ts b/packages/core/src/restic/commands/copy.ts index a7b1d94e..5e0601ef 100644 --- a/packages/core/src/restic/commands/copy.ts +++ b/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>; + 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); } diff --git a/packages/core/src/restic/commands/delete-snapshots.ts b/packages/core/src/restic/commands/delete-snapshots.ts index 39dad0fc..60d69925 100644 --- a/packages/core/src/restic/commands/delete-snapshots.ts +++ b/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>; + 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); } diff --git a/packages/core/src/restic/commands/forget.ts b/packages/core/src/restic/commands/forget.ts index 148fc940..cc4f96aa 100644 --- a/packages/core/src/restic/commands/forget.ts +++ b/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>; + 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); }