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 6ec85c37..437e33c6 100644 --- a/app/server/modules/backups/__tests__/backups.service.execution.test.ts +++ b/app/server/modules/backups/__tests__/backups.service.execution.test.ts @@ -1256,6 +1256,32 @@ describe("mirror operations", () => { expect(updatedMirror?.lastCopyAt).not.toBeNull(); }); + test("should clear mirror in-progress status when shutdown aborts copy", async () => { + const { resticCopyMock } = setup(); + const volume = await createTestVolume(); + const sourceRepository = await createTestRepository(); + const mirrorRepository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: sourceRepository.id, + }); + + const mirror = await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id); + resticCopyMock.mockImplementationOnce(() => + Effect.sync(() => { + throw new Error("Repository mutex is shutting down"); + }), + ); + + await backupsService.copyToMirrors(schedule.id, sourceRepository, null); + + const mirrors = await backupsService.getMirrors(schedule.id); + const updatedMirror = mirrors.find((m) => m.id === mirror.id); + expect(updatedMirror?.lastCopyStatus).toBeNull(); + expect(updatedMirror?.lastCopyError).toBeNull(); + expect(resticCopyMock).toHaveBeenCalledTimes(1); + }); + test("should run forget on mirror after successful copy when retention policy exists", async () => { // arrange const { resticCopyMock, resticForgetMock } = setup(); diff --git a/app/server/modules/backups/backups.queries.ts b/app/server/modules/backups/backups.queries.ts index 00b2641a..ffa1b0c8 100644 --- a/app/server/modules/backups/backups.queries.ts +++ b/app/server/modules/backups/backups.queries.ts @@ -7,7 +7,7 @@ type MirrorStatusType = "in_progress" | "success" | "error"; type MirrorStatusUpdate = { lastCopyAt?: number | null; - lastCopyStatus?: MirrorStatusType; + lastCopyStatus?: MirrorStatusType | null; lastCopyError?: string | null; }; @@ -48,7 +48,9 @@ export const scheduleQueries = { return db .update(backupSchedulesTable) .set({ ...status, updatedAt: Date.now() }) - .where(and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId))); + .where( + and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId)), + ); }, }; diff --git a/app/server/modules/backups/helpers/backup-maintenance.ts b/app/server/modules/backups/helpers/backup-maintenance.ts index ae668d4c..9e5dd631 100644 --- a/app/server/modules/backups/helpers/backup-maintenance.ts +++ b/app/server/modules/backups/helpers/backup-maintenance.ts @@ -9,6 +9,11 @@ import { runEffectPromise, toMessage } from "../../../utils/errors"; import { getOrganizationId } from "~/server/core/request-context"; import { mirrorQueries, repositoryQueries, scheduleQueries } from "../backups.queries"; +const isMirrorCopyCancellation = (error: unknown) => { + const message = toMessage(error); + return message === "Repository mutex is shutting down" || message === "Operation aborted"; +}; + export async function runForget(scheduleId: number, repositoryId?: string, organizationIdOverride?: string) { const organizationId = organizationIdOverride ?? getOrganizationId(); const schedule = await scheduleQueries.findById(scheduleId, organizationId); @@ -154,6 +159,17 @@ export async function syncSnapshotsToMirror( }); } catch (error) { const errorMessage = toMessage(error); + if (isMirrorCopyCancellation(error)) { + logger.info( + `[Background] Mirror sync to repository ${mirrorRepository.name} was cancelled: ${errorMessage}`, + ); + await mirrorQueries.updateStatus(scheduleId, mirrorRepositoryId, { + lastCopyStatus: null, + lastCopyError: null, + }); + return; + } + logger.error( `[Background] Failed to sync all snapshots to mirror repository ${mirrorRepository.name}: ${errorMessage}`, ); @@ -243,6 +259,17 @@ async function copyToSingleMirror( }); } catch (error) { const errorMessage = toMessage(error); + if (isMirrorCopyCancellation(error)) { + logger.info( + `[Background] Mirror copy to repository ${mirror.repository.name} was cancelled: ${errorMessage}`, + ); + await mirrorQueries.updateStatus(scheduleId, mirror.repositoryId, { + lastCopyStatus: null, + lastCopyError: null, + }); + return; + } + logger.error(`[Background] Failed to copy to mirror repository ${mirror.repository.name}: ${errorMessage}`); await mirrorQueries.updateStatus(scheduleId, mirror.repositoryId, { diff --git a/app/server/modules/repositories/__tests__/repositories.service.test.ts b/app/server/modules/repositories/__tests__/repositories.service.test.ts index 2c868d84..11cf054b 100644 --- a/app/server/modules/repositories/__tests__/repositories.service.test.ts +++ b/app/server/modules/repositories/__tests__/repositories.service.test.ts @@ -22,7 +22,7 @@ import { ResticError } from "@zerobyte/core/restic/server"; import { repoMutex } from "~/server/core/repository-mutex"; import { taskStore } from "~/server/modules/tasks/tasks.store"; import { repositoriesService } from "../repositories.service"; -import { holdExclusiveLock } from "~/test/helpers/repository-mutex"; +import { holdExclusiveLock, holdSharedLock } from "~/test/helpers/repository-mutex"; const createTestRepository = async (organizationId: string, overrides: Partial = {}) => { const id = randomUUID(); @@ -388,6 +388,47 @@ describe("repositoriesService.checkHealth", () => { }); }); +describe("repositoriesService.startDoctor", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + test("marks a queued doctor as cancelled when repository mutex shutdown aborts it", async () => { + const repository = await createTestRepository(session.organizationId, { + status: "healthy", + lastError: "previous error", + }); + const releaseLock = await holdSharedLock(repository.id, "backup"); + + try { + await withContext({ organizationId: session.organizationId, userId: session.user.id }, () => + repositoriesService.startDoctor(repository.shortId), + ); + + await waitForExpect(async () => { + const runningRepository = await db.query.repositoriesTable.findFirst({ where: { id: repository.id } }); + expect(runningRepository?.status).toBe("doctor"); + }); + await waitForExpect(async () => { + const waiters = await db.query.repositoryLockWaitersTable.findMany({ + where: { repositoryId: repository.id }, + }); + expect(waiters).toHaveLength(1); + }); + + await repoMutex.shutdown({ timeoutMs: 10 }); + } finally { + await releaseLock(); + } + + await waitForExpect(async () => { + const cancelledRepository = await db.query.repositoriesTable.findFirst({ where: { id: repository.id } }); + expect(cancelledRepository?.status).toBe("cancelled"); + expect(cancelledRepository?.lastError).toBe("Repository mutex is shutting down"); + }); + }); +}); + describe("repositoriesService.dumpSnapshot", () => { afterEach(() => { vi.restoreAllMocks(); @@ -699,7 +740,8 @@ describe("repositoriesService.restoreSnapshot", () => { }; test("rejects protected targets even when the local agent is enabled", async () => { - const { organizationId, userId, repositoryShortId, restoreMock } = await setupRestoreSnapshotScenario(); + const { organizationId, userId, repositoryId, repositoryShortId, restoreMock } = + await setupRestoreSnapshotScenario(); const targetPath = nodePath.join(os.tmpdir(), "zerobyte-restore-target"); await expect( @@ -714,7 +756,8 @@ describe("repositoriesService.restoreSnapshot", () => { }); test("restores to a custom target outside protected roots", async () => { - const { organizationId, userId, repositoryShortId, restoreMock } = await setupRestoreSnapshotScenario(); + const { organizationId, userId, repositoryId, repositoryShortId, restoreMock } = + await setupRestoreSnapshotScenario(); const targetPath = await fs.mkdtemp(nodePath.join(process.cwd(), "restore-target-")); try { @@ -835,6 +878,34 @@ describe("repositoriesService.restoreSnapshot", () => { }); }); + test("cancels a mutex-aborted restore before it starts", async () => { + const { organizationId, userId, repositoryShortId, restoreMock } = await setupRestoreSnapshotScenario(); + const targetPath = await fs.mkdtemp(nodePath.join(process.cwd(), "restore-target-")); + await withContext({ organizationId, userId }, () => + repositoriesService.getSnapshotDetails(repositoryShortId, "snapshot-restore"), + ); + vi.spyOn(repoMutex, "runShared").mockRejectedValueOnce(new Error("Repository mutex is shutting down")); + + try { + const result = await withContext({ organizationId, userId }, () => + repositoriesService.restoreSnapshot(repositoryShortId, "snapshot-restore", { + targetPath, + }), + ); + + expect(restoreMock).not.toHaveBeenCalled(); + await waitForExpect(async () => { + const task = await db.query.tasksTable.findFirst({ where: { id: result.restoreId } }); + expect(task?.status).toBe("cancelled"); + expect(task?.error).toBe("Restore was cancelled"); + }); + } finally { + await fs.rm(targetPath, { recursive: true, force: true }); + } + + expect(restoreMock).not.toHaveBeenCalled(); + }); + test("routes restore to the requested target agent", async () => { const organizationId = session.organizationId; const agentId = `agent-${randomUUID()}`; diff --git a/app/server/modules/repositories/helpers/doctor.ts b/app/server/modules/repositories/helpers/doctor.ts index 36ca7aeb..c35063dd 100644 --- a/app/server/modules/repositories/helpers/doctor.ts +++ b/app/server/modules/repositories/helpers/doctor.ts @@ -167,7 +167,13 @@ export const executeDoctor = async ( completedAt: Date.now(), }); } catch (error) { - if (error instanceof AbortError) { + const errorMessage = toMessage(error); + const isCancellation = + error instanceof AbortError || + errorMessage === "Repository mutex is shutting down" || + errorMessage === "Operation aborted"; + + if (isCancellation) { const doctorResult: DoctorResult = { success: false, steps, @@ -179,7 +185,7 @@ export const executeDoctor = async ( .set({ status: "cancelled", lastChecked: Date.now(), - lastError: toMessage(error), + lastError: errorMessage, doctorResult, }) .where(eq(repositoriesTable.id, repositoryId)); @@ -188,7 +194,7 @@ export const executeDoctor = async ( organizationId, repositoryId: repositoryShortId, repositoryName, - error: toMessage(error), + error: errorMessage, }); } else { await db diff --git a/app/server/modules/repositories/restore-executor.ts b/app/server/modules/repositories/restore-executor.ts index f89b4a66..93145c33 100644 --- a/app/server/modules/repositories/restore-executor.ts +++ b/app/server/modules/repositories/restore-executor.ts @@ -29,6 +29,11 @@ type RestoreExecutionHandle = { const shouldRunInController = (agentId: string) => agentId === LOCAL_AGENT_ID && !appConfig.flags.enableLocalAgent; +const isAbortLikeError = (error: unknown) => { + const message = toMessage(error); + return message === "Repository mutex is shutting down" || message === "Operation aborted"; +}; + const createRestoreRunPayload = async (request: RestoreExecutionRequest): Promise => { const encryptedResticPassword = await resticDeps.getOrganizationResticPassword(request.organizationId); const resticPassword = await resticDeps.resolveSecret(encryptedResticPassword); @@ -70,7 +75,7 @@ const executeControllerRestore = async ( return { status: "completed", result }; } catch (error) { - if (signal.aborted) { + if (signal.aborted || isAbortLikeError(error)) { return { status: "cancelled", message: "Restore was cancelled" }; } @@ -101,7 +106,7 @@ const executeAgentRestore = async ( return await started.result; } catch (error) { - if (signal.aborted) { + if (signal.aborted || isAbortLikeError(error)) { return { status: "cancelled", message: "Restore was cancelled" }; } @@ -127,7 +132,7 @@ const executeRestoreWithRepositoryLock = async ( signal, ); } catch (error) { - if (signal.aborted) { + if (signal.aborted || isAbortLikeError(error)) { return { status: "cancelled", message: "Restore was cancelled" }; } diff --git a/packages/core/src/restic/commands/__tests__/tag-snapshots.test.ts b/packages/core/src/restic/commands/__tests__/tag-snapshots.test.ts index 64a682be..f8139004 100644 --- a/packages/core/src/restic/commands/__tests__/tag-snapshots.test.ts +++ b/packages/core/src/restic/commands/__tests__/tag-snapshots.test.ts @@ -52,4 +52,25 @@ describe("tagSnapshots command", () => { expect(separatorIndex).toBeGreaterThan(-1); expect(getArgs().slice(separatorIndex + 1)).toEqual(snapshotIds); }); + + test("does not treat a cleanup-time abort as a failed successful tag", async () => { + const controller = new AbortController(); + setup(); + vi.spyOn(cleanupModule, "cleanupTemporaryKeys").mockImplementation(async () => { + controller.abort(new Error("aborted during cleanup")); + }); + + const result = await Effect.runPromise( + tagSnapshots( + config, + ["snapshot-1"], + { add: ["keep"] }, + { 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/tag-snapshots.ts b/packages/core/src/restic/commands/tag-snapshots.ts index 8a24503c..8866040e 100644 --- a/packages/core/src/restic/commands/tag-snapshots.ts +++ b/packages/core/src/restic/commands/tag-snapshots.ts @@ -53,15 +53,19 @@ export const tagSnapshots = ( 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 tagging 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 tagging was aborted by signal."); + throw new Error("Operation aborted"); + } + logger.error(`Restic snapshot tagging failed: ${res.stderr}`); throw createResticError(res.exitCode, res.stderr); }