fix: handle shutdown aborts without persisting failures
This commit is contained in:
parent
e674a8f6bf
commit
6f0bac785b
8 changed files with 179 additions and 17 deletions
|
|
@ -1256,6 +1256,32 @@ describe("mirror operations", () => {
|
||||||
expect(updatedMirror?.lastCopyAt).not.toBeNull();
|
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 () => {
|
test("should run forget on mirror after successful copy when retention policy exists", async () => {
|
||||||
// arrange
|
// arrange
|
||||||
const { resticCopyMock, resticForgetMock } = setup();
|
const { resticCopyMock, resticForgetMock } = setup();
|
||||||
|
|
|
||||||
|
|
@ -7,7 +7,7 @@ type MirrorStatusType = "in_progress" | "success" | "error";
|
||||||
|
|
||||||
type MirrorStatusUpdate = {
|
type MirrorStatusUpdate = {
|
||||||
lastCopyAt?: number | null;
|
lastCopyAt?: number | null;
|
||||||
lastCopyStatus?: MirrorStatusType;
|
lastCopyStatus?: MirrorStatusType | null;
|
||||||
lastCopyError?: string | null;
|
lastCopyError?: string | null;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
@ -48,7 +48,9 @@ export const scheduleQueries = {
|
||||||
return db
|
return db
|
||||||
.update(backupSchedulesTable)
|
.update(backupSchedulesTable)
|
||||||
.set({ ...status, updatedAt: Date.now() })
|
.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)),
|
||||||
|
);
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -9,6 +9,11 @@ import { runEffectPromise, toMessage } from "../../../utils/errors";
|
||||||
import { getOrganizationId } from "~/server/core/request-context";
|
import { getOrganizationId } from "~/server/core/request-context";
|
||||||
import { mirrorQueries, repositoryQueries, scheduleQueries } from "../backups.queries";
|
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) {
|
export async function runForget(scheduleId: number, repositoryId?: string, organizationIdOverride?: string) {
|
||||||
const organizationId = organizationIdOverride ?? getOrganizationId();
|
const organizationId = organizationIdOverride ?? getOrganizationId();
|
||||||
const schedule = await scheduleQueries.findById(scheduleId, organizationId);
|
const schedule = await scheduleQueries.findById(scheduleId, organizationId);
|
||||||
|
|
@ -154,6 +159,17 @@ export async function syncSnapshotsToMirror(
|
||||||
});
|
});
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
const errorMessage = toMessage(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(
|
logger.error(
|
||||||
`[Background] Failed to sync all snapshots to mirror repository ${mirrorRepository.name}: ${errorMessage}`,
|
`[Background] Failed to sync all snapshots to mirror repository ${mirrorRepository.name}: ${errorMessage}`,
|
||||||
);
|
);
|
||||||
|
|
@ -243,6 +259,17 @@ async function copyToSingleMirror(
|
||||||
});
|
});
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
const errorMessage = toMessage(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}`);
|
logger.error(`[Background] Failed to copy to mirror repository ${mirror.repository.name}: ${errorMessage}`);
|
||||||
|
|
||||||
await mirrorQueries.updateStatus(scheduleId, mirror.repositoryId, {
|
await mirrorQueries.updateStatus(scheduleId, mirror.repositoryId, {
|
||||||
|
|
|
||||||
|
|
@ -22,7 +22,7 @@ import { ResticError } from "@zerobyte/core/restic/server";
|
||||||
import { repoMutex } from "~/server/core/repository-mutex";
|
import { repoMutex } from "~/server/core/repository-mutex";
|
||||||
import { taskStore } from "~/server/modules/tasks/tasks.store";
|
import { taskStore } from "~/server/modules/tasks/tasks.store";
|
||||||
import { repositoriesService } from "../repositories.service";
|
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<RepositoryInsert> = {}) => {
|
const createTestRepository = async (organizationId: string, overrides: Partial<RepositoryInsert> = {}) => {
|
||||||
const id = randomUUID();
|
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", () => {
|
describe("repositoriesService.dumpSnapshot", () => {
|
||||||
afterEach(() => {
|
afterEach(() => {
|
||||||
vi.restoreAllMocks();
|
vi.restoreAllMocks();
|
||||||
|
|
@ -699,7 +740,8 @@ describe("repositoriesService.restoreSnapshot", () => {
|
||||||
};
|
};
|
||||||
|
|
||||||
test("rejects protected targets even when the local agent is enabled", async () => {
|
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");
|
const targetPath = nodePath.join(os.tmpdir(), "zerobyte-restore-target");
|
||||||
|
|
||||||
await expect(
|
await expect(
|
||||||
|
|
@ -714,7 +756,8 @@ describe("repositoriesService.restoreSnapshot", () => {
|
||||||
});
|
});
|
||||||
|
|
||||||
test("restores to a custom target outside protected roots", async () => {
|
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-"));
|
const targetPath = await fs.mkdtemp(nodePath.join(process.cwd(), "restore-target-"));
|
||||||
|
|
||||||
try {
|
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 () => {
|
test("routes restore to the requested target agent", async () => {
|
||||||
const organizationId = session.organizationId;
|
const organizationId = session.organizationId;
|
||||||
const agentId = `agent-${randomUUID()}`;
|
const agentId = `agent-${randomUUID()}`;
|
||||||
|
|
|
||||||
|
|
@ -167,7 +167,13 @@ export const executeDoctor = async (
|
||||||
completedAt: Date.now(),
|
completedAt: Date.now(),
|
||||||
});
|
});
|
||||||
} catch (error) {
|
} 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 = {
|
const doctorResult: DoctorResult = {
|
||||||
success: false,
|
success: false,
|
||||||
steps,
|
steps,
|
||||||
|
|
@ -179,7 +185,7 @@ export const executeDoctor = async (
|
||||||
.set({
|
.set({
|
||||||
status: "cancelled",
|
status: "cancelled",
|
||||||
lastChecked: Date.now(),
|
lastChecked: Date.now(),
|
||||||
lastError: toMessage(error),
|
lastError: errorMessage,
|
||||||
doctorResult,
|
doctorResult,
|
||||||
})
|
})
|
||||||
.where(eq(repositoriesTable.id, repositoryId));
|
.where(eq(repositoriesTable.id, repositoryId));
|
||||||
|
|
@ -188,7 +194,7 @@ export const executeDoctor = async (
|
||||||
organizationId,
|
organizationId,
|
||||||
repositoryId: repositoryShortId,
|
repositoryId: repositoryShortId,
|
||||||
repositoryName,
|
repositoryName,
|
||||||
error: toMessage(error),
|
error: errorMessage,
|
||||||
});
|
});
|
||||||
} else {
|
} else {
|
||||||
await db
|
await db
|
||||||
|
|
|
||||||
|
|
@ -29,6 +29,11 @@ type RestoreExecutionHandle = {
|
||||||
|
|
||||||
const shouldRunInController = (agentId: string) => agentId === LOCAL_AGENT_ID && !appConfig.flags.enableLocalAgent;
|
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<RestoreRunPayload> => {
|
const createRestoreRunPayload = async (request: RestoreExecutionRequest): Promise<RestoreRunPayload> => {
|
||||||
const encryptedResticPassword = await resticDeps.getOrganizationResticPassword(request.organizationId);
|
const encryptedResticPassword = await resticDeps.getOrganizationResticPassword(request.organizationId);
|
||||||
const resticPassword = await resticDeps.resolveSecret(encryptedResticPassword);
|
const resticPassword = await resticDeps.resolveSecret(encryptedResticPassword);
|
||||||
|
|
@ -70,7 +75,7 @@ const executeControllerRestore = async (
|
||||||
|
|
||||||
return { status: "completed", result };
|
return { status: "completed", result };
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
if (signal.aborted) {
|
if (signal.aborted || isAbortLikeError(error)) {
|
||||||
return { status: "cancelled", message: "Restore was cancelled" };
|
return { status: "cancelled", message: "Restore was cancelled" };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -101,7 +106,7 @@ const executeAgentRestore = async (
|
||||||
|
|
||||||
return await started.result;
|
return await started.result;
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
if (signal.aborted) {
|
if (signal.aborted || isAbortLikeError(error)) {
|
||||||
return { status: "cancelled", message: "Restore was cancelled" };
|
return { status: "cancelled", message: "Restore was cancelled" };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -127,7 +132,7 @@ const executeRestoreWithRepositoryLock = async (
|
||||||
signal,
|
signal,
|
||||||
);
|
);
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
if (signal.aborted) {
|
if (signal.aborted || isAbortLikeError(error)) {
|
||||||
return { status: "cancelled", message: "Restore was cancelled" };
|
return { status: "cancelled", message: "Restore was cancelled" };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -52,4 +52,25 @@ describe("tagSnapshots command", () => {
|
||||||
expect(separatorIndex).toBeGreaterThan(-1);
|
expect(separatorIndex).toBeGreaterThan(-1);
|
||||||
expect(getArgs().slice(separatorIndex + 1)).toEqual(snapshotIds);
|
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);
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|
|
||||||
|
|
@ -53,15 +53,19 @@ export const tagSnapshots = (
|
||||||
addCommonArgs(args, env, config);
|
addCommonArgs(args, env, config);
|
||||||
args.push("--", ...snapshotIds);
|
args.push("--", ...snapshotIds);
|
||||||
|
|
||||||
const res = await safeExec({ command: "restic", args, env, signal: options.signal });
|
let res: Awaited<ReturnType<typeof safeExec>>;
|
||||||
await cleanupTemporaryKeys(env, deps);
|
try {
|
||||||
|
res = await safeExec({ command: "restic", args, env, signal: options.signal });
|
||||||
if (options.signal?.aborted) {
|
} finally {
|
||||||
logger.warn("Restic snapshot tagging was aborted by signal.");
|
await cleanupTemporaryKeys(env, deps);
|
||||||
throw new Error("Operation aborted");
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if (res.exitCode !== 0) {
|
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}`);
|
logger.error(`Restic snapshot tagging failed: ${res.stderr}`);
|
||||||
throw createResticError(res.exitCode, res.stderr);
|
throw createResticError(res.exitCode, res.stderr);
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue