danvics/zerobyte
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor(backups): split into helpers

Browse source
This commit is contained in:
Nicolas Meienberger 2026-04-01 23:24:51 +02:00
parent af4ac1c39c
commit d5021566ac
7 changed files with 275 additions and 211 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
app/server/modules/agents/agents-manager.ts
View file

@ -268,7 +268,7 @@ const createAgentManagerRuntime = () => {
(server) =>
Effect.sync(() => {
closeAllSessions();
server.stop(true);
void server.stop(true);
}),
);

141
app/server/modules/backups/__tests__/backups.execution.test.ts → app/server/modules/backups/__tests__/backups.service.execution.test.ts
View file

@ -1,7 +1,6 @@
import waitForExpect from "wait-for-expect";
import { afterEach, describe, expect, test, vi } from "vitest";
import { backupsService } from "../backups.service";
import { backupsExecutionService } from "../backups.execution";
import { createTestVolume } from "~/test/helpers/volume";
import { createTestBackupSchedule } from "~/test/helpers/backup";
import { createTestRepository } from "~/test/helpers/repository";
@ -17,6 +16,9 @@ import { repositoriesService } from "~/server/modules/repositories/repositories.
import { repoMutex } from "~/server/core/repository-mutex";
import { agentManager } from "~/server/modules/agents/agents-manager";
import { createAgentBackupMocks } from "~/test/helpers/agent-mock";
import { fromAny } from "@total-typescript/shoehorn";
import { scheduleQueries } from "../backups.queries";
import { getScheduleByIdOrShortId } from "../helpers/backup-schedule-lookups";
const setup = () => {
const resticBackupMock = vi.fn((_: SafeSpawnParams) =>
@ -70,7 +72,7 @@ describe("backup execution - validation failures", () => {
});
// act
const result = await backupsExecutionService.validateBackupExecution(schedule.id);
const result = await backupsService.validateBackupExecution(schedule.id);
// assert
expect(result.type).toBe("failure");
@ -81,10 +83,71 @@ describe("backup execution - validation failures", () => {
expect(resticBackupMock).not.toHaveBeenCalled();
});
test("should fail backup when volume does not exist", async () => {
// arrange
setup();
const volume = await createTestVolume();
const repository = await createTestRepository();
const schedule = await createTestBackupSchedule({
volumeId: volume.id,
repositoryId: repository.id,
});
const hydratedSchedule = await scheduleQueries.findById(schedule.id, TEST_ORG_ID);
expect(hydratedSchedule).toBeDefined();
const scheduleWithoutVolume = {
...hydratedSchedule,
volume: null,
};
vi.spyOn(scheduleQueries, "findById").mockResolvedValueOnce(fromAny(scheduleWithoutVolume));
// act
const result = await backupsService.validateBackupExecution(schedule.id);
// assert
expect(result.type).toBe("failure");
if (result.type === "failure") {
expect(result.error).toBeInstanceOf(NotFoundError);
expect(result.error.message).toBe("Volume not found");
expect(result.partialContext?.schedule).toBeDefined();
}
});
test("should fail backup when repository does not exist", async () => {
// arrange
setup();
const volume = await createTestVolume();
const repository = await createTestRepository();
const schedule = await createTestBackupSchedule({
volumeId: volume.id,
repositoryId: repository.id,
});
const hydratedSchedule = await scheduleQueries.findById(schedule.id, TEST_ORG_ID);
expect(hydratedSchedule).toBeDefined();
const scheduleWithoutRepository = {
...hydratedSchedule,
repository: null,
};
vi.spyOn(scheduleQueries, "findById").mockResolvedValueOnce(fromAny(scheduleWithoutRepository));
// act
const result = await backupsService.validateBackupExecution(schedule.id);
// assert
expect(result.type).toBe("failure");
if (result.type === "failure") {
expect(result.error).toBeInstanceOf(NotFoundError);
expect(result.error.message).toBe("Repository not found");
expect(result.partialContext?.schedule).toBeDefined();
expect(result.partialContext?.volume).toBeDefined();
}
});
test("should fail backup when schedule does not exist", async () => {
setup();
// act
const result = await backupsExecutionService.validateBackupExecution(99999);
const result = await backupsService.validateBackupExecution(99999);
// assert
expect(result.type).toBe("failure");
@ -115,9 +178,9 @@ describe("stop backup", () => {
});
});
await backupsExecutionService.executeBackup(schedule.id);
await backupsService.executeBackup(schedule.id);
const updatedSchedule = await backupsService.getScheduleById(schedule.id);
const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
expect(updatedSchedule.lastBackupStatus).toBe("warning");
expect(updatedSchedule.lastBackupError).toBe("error: open /mnt/data/private.db: permission denied");
});
@ -143,9 +206,9 @@ describe("stop backup", () => {
});
});
await backupsExecutionService.executeBackup(schedule.id);
await backupsService.executeBackup(schedule.id);
const updatedSchedule = await backupsService.getScheduleById(schedule.id);
const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
expect(updatedSchedule.lastBackupStatus).toBe("error");
expect(updatedSchedule.lastBackupError).toBe(
"Permissions 0755 for '/tmp/zerobyte-ssh-key' are too open.\nThis private key will be ignored.",
@ -170,14 +233,14 @@ describe("stop backup", () => {
}),
);
const backupPromise = backupsExecutionService.executeBackup(schedule.id);
const backupPromise = backupsService.executeBackup(schedule.id);
await waitForExpect(() => {
expect(sendBackupMock).toHaveBeenCalledTimes(1);
});
let forgetFinished = false;
const forgetPromise = backupsExecutionService.runForget(schedule.id).finally(() => {
const forgetPromise = backupsService.runForget(schedule.id).finally(() => {
forgetFinished = true;
});
@ -227,19 +290,19 @@ describe("stop backup", () => {
});
});
const executePromise = backupsExecutionService.executeBackup(schedule.id);
const executePromise = backupsService.executeBackup(schedule.id);
await waitForExpect(async () => {
const runningSchedule = await backupsService.getScheduleById(schedule.id);
const runningSchedule = await getScheduleByIdOrShortId(schedule.id);
expect(runningSchedule.lastBackupStatus).toBe("in_progress");
});
// act
await backupsExecutionService.stopBackup(schedule.id);
await backupsService.stopBackup(schedule.id);
await executePromise;
// assert
const updatedSchedule = await backupsService.getScheduleById(schedule.id);
const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
expect(updatedSchedule.lastBackupStatus).toBe("warning");
expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
});
@ -270,7 +333,7 @@ describe("stop backup", () => {
});
});
const executePromise = backupsExecutionService.executeBackup(schedule.id);
const executePromise = backupsService.executeBackup(schedule.id);
await waitForExpect(async () => {
const queuedSchedule = await backupsService.getScheduleById(schedule.id);
@ -279,7 +342,7 @@ describe("stop backup", () => {
expect(resticBackupMock).not.toHaveBeenCalled();
await backupsExecutionService.stopBackup(schedule.id);
await backupsService.stopBackup(schedule.id);
await executePromise;
const updatedSchedule = await backupsService.getScheduleById(schedule.id);
@ -302,7 +365,7 @@ describe("stop backup", () => {
});
// act & assert
await expect(backupsExecutionService.stopBackup(schedule.id)).rejects.toThrow(
await expect(backupsService.stopBackup(schedule.id)).rejects.toThrow(
"No backup is currently running for this schedule",
);
@ -312,10 +375,30 @@ describe("stop backup", () => {
expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
});
test("should reset a stuck in_progress status even when no backup is running", async () => {
// arrange
setup();
const volume = await createTestVolume();
const repository = await createTestRepository();
const schedule = await createTestBackupSchedule({
volumeId: volume.id,
repositoryId: repository.id,
lastBackupStatus: "in_progress",
});
// act
await backupsService.stopBackup(schedule.id).catch(() => {});
// assert
const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
expect(updatedSchedule.lastBackupStatus).toBe("warning");
expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
});
test("should throw NotFoundError when schedule does not exist", async () => {
setup();
// act & assert
await expect(backupsExecutionService.stopBackup(99999)).rejects.toThrow("Backup schedule not found");
await expect(backupsService.stopBackup(99999)).rejects.toThrow("Backup schedule not found");
});
});
@ -336,7 +419,7 @@ describe("retention policy - runForget", () => {
});
// act
await backupsExecutionService.runForget(schedule.id);
await backupsService.runForget(schedule.id);
// assert
expect(resticForgetMock).toHaveBeenCalledWith(
@ -365,7 +448,7 @@ describe("retention policy - runForget", () => {
});
// act & assert
await expect(backupsExecutionService.runForget(schedule.id)).rejects.toThrow(
await expect(backupsService.runForget(schedule.id)).rejects.toThrow(
"No retention policy configured for this schedule",
);
});
@ -373,7 +456,7 @@ describe("retention policy - runForget", () => {
test("should throw NotFoundError when schedule does not exist", async () => {
setup();
// act & assert
await expect(backupsExecutionService.runForget(99999)).rejects.toThrow("Backup schedule not found");
await expect(backupsService.runForget(99999)).rejects.toThrow("Backup schedule not found");
});
test("should throw NotFoundError when repository does not exist", async () => {
@ -386,9 +469,7 @@ describe("retention policy - runForget", () => {
});
// act & assert
await expect(backupsExecutionService.runForget(schedule.id, "non-existent-repo")).rejects.toThrow(
"Repository not found",
);
await expect(backupsService.runForget(schedule.id, "non-existent-repo")).rejects.toThrow("Repository not found");
});
});
@ -407,7 +488,7 @@ describe("mirror operations", () => {
await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id);
// act
await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
// assert
expect(resticCopyMock).toHaveBeenCalledWith(
@ -434,7 +515,7 @@ describe("mirror operations", () => {
await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id, { enabled: false });
// act
await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
// assert
expect(resticCopyMock).not.toHaveBeenCalled();
@ -454,7 +535,7 @@ describe("mirror operations", () => {
const mirror = await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id);
// act
await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
// assert
const mirrors = await backupsService.getMirrors(schedule.id);
@ -485,7 +566,7 @@ describe("mirror operations", () => {
});
// act
await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
// assert
const mirrors = await backupsService.getMirrors(schedule.id);
@ -512,7 +593,7 @@ describe("mirror operations", () => {
resticCopyMock.mockImplementationOnce(() => Promise.reject(new Error("Copy failed")));
// act
await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
// assert
const mirrors = await backupsService.getMirrors(schedule.id);
@ -540,7 +621,7 @@ describe("mirror operations", () => {
resticCopyMock.mockImplementation(() => Promise.resolve({ success: true, output: "" }));
// act
await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
await backupsService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
await waitForExpect(() => {
expect(resticCopyMock).toHaveBeenCalled();
@ -571,7 +652,7 @@ describe("mirror operations", () => {
resticForgetMock.mockClear();
// act
await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
await backupsService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
await waitForExpect(() => {
expect(resticCopyMock).toHaveBeenCalled();

36
app/server/modules/backups/__tests__/backups.service.test.ts
View file

@ -12,10 +12,10 @@ import { db } from "~/server/db/db";
import { backupScheduleMirrorsTable, repositoriesTable, volumesTable } from "~/server/db/schema";
import { TEST_ORG_ID } from "~/test/helpers/organization";
import * as context from "~/server/core/request-context";
import { backupsExecutionService } from "../backups.execution";
import { repositoriesService } from "~/server/modules/repositories/repositories.service";
import { agentManager } from "~/server/modules/agents/agents-manager";
import { createAgentBackupMocks } from "~/test/helpers/agent-mock";
import { getScheduleByIdOrShortId } from "../helpers/backup-schedule-lookups";
const setup = () => {
const resticBackupMock = vi.fn((_: unknown) => Promise.resolve({ exitCode: 0, summary: "", error: "" }));
@ -66,10 +66,10 @@ describe("execute backup", () => {
);
// act
await backupsExecutionService.executeBackup(schedule.id);
await backupsService.executeBackup(schedule.id);
// assert
const updatedSchedule = await backupsService.getScheduleById(schedule.id);
const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
expect(updatedSchedule.nextBackupAt).not.toBeNull();
const nextBackupAt = new Date(updatedSchedule.nextBackupAt ?? 0);
@ -91,7 +91,7 @@ describe("execute backup", () => {
});
// act
await backupsExecutionService.executeBackup(schedule.id);
await backupsService.executeBackup(schedule.id);
// assert
expect(resticBackupMock).not.toHaveBeenCalled();
@ -113,7 +113,7 @@ describe("execute backup", () => {
);
// act
await backupsExecutionService.executeBackup(schedule.id, true);
await backupsService.executeBackup(schedule.id, true);
// assert
expect(resticBackupMock).toHaveBeenCalled();
@ -139,7 +139,7 @@ describe("execute backup", () => {
);
// act
await backupsExecutionService.executeBackup(schedule.id, true);
await backupsService.executeBackup(schedule.id, true);
// assert
const updatedSchedule = await backupsService.getScheduleById(schedule.id);
@ -162,13 +162,13 @@ describe("execute backup", () => {
});
// act
void backupsExecutionService.executeBackup(schedule.id);
void backupsService.executeBackup(schedule.id);
await waitForExpect(() => {
expect(resticBackupMock).toHaveBeenCalledTimes(1);
});
await backupsExecutionService.executeBackup(schedule.id);
await backupsService.executeBackup(schedule.id);
// assert
expect(resticBackupMock).toHaveBeenCalledTimes(1);
@ -189,10 +189,10 @@ describe("execute backup", () => {
);
// act
await backupsExecutionService.executeBackup(schedule.id);
await backupsService.executeBackup(schedule.id);
// assert
const updatedSchedule = await backupsService.getScheduleById(schedule.id);
const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
expect(updatedSchedule.lastBackupStatus).toBe("warning");
});
@ -211,10 +211,10 @@ describe("execute backup", () => {
);
// act
await backupsExecutionService.executeBackup(schedule.id);
await backupsService.executeBackup(schedule.id);
// assert
const updatedSchedule = await backupsService.getScheduleById(schedule.id);
const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
expect(updatedSchedule.lastBackupStatus).toBe("error");
});
});
@ -236,7 +236,7 @@ describe("getSchedulesToExecute", () => {
});
// act
const schedulesToExecute = await backupsExecutionService.getSchedulesToExecute();
const schedulesToExecute = await backupsService.getSchedulesToExecute();
// assert
expect(schedulesToExecute).toContain(schedule.id);
@ -253,7 +253,7 @@ describe("getScheduleByIdOrShortId", () => {
repositoryId: repository.id,
});
const found = await backupsService.getScheduleByIdOrShortId(String(schedule.id));
const found = await getScheduleByIdOrShortId(String(schedule.id));
expect(found.id).toBe(schedule.id);
expect(found.shortId).toBe(schedule.shortId);
@ -268,7 +268,7 @@ describe("getScheduleByIdOrShortId", () => {
repositoryId: repository.id,
});
const found = await backupsService.getScheduleByIdOrShortId(schedule.shortId);
const found = await getScheduleByIdOrShortId(schedule.shortId);
expect(found.id).toBe(schedule.id);
expect(found.shortId).toBe(schedule.shortId);
@ -281,10 +281,8 @@ describe("getScheduleByIdOrShortId", () => {
organizationId: otherOrgId,
});
await expect(backupsService.getScheduleByIdOrShortId(schedule.shortId)).rejects.toThrow(
"Backup schedule not found",
);
await expect(backupsService.getScheduleByIdOrShortId(schedule.id)).rejects.toThrow("Backup schedule not found");
await expect(getScheduleByIdOrShortId(schedule.shortId)).rejects.toThrow("Backup schedule not found");
await expect(getScheduleByIdOrShortId(schedule.id)).rejects.toThrow("Backup schedule not found");
});
});

291
app/server/modules/backups/backup-executor.ts
View file

@ -1,14 +1,18 @@
import type { BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
import { agentManager } from "../agents/agents-manager";
import { logger } from "@zerobyte/core/node";
import type { BackupProgressPayload, BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
import type { ResticBackupOutputDto } from "@zerobyte/core/restic";
import { resticDeps } from "../../core/restic";
import type { BackupSchedule, Repository, Volume } from "../../db/schema";
import type { ResticBackupOutputDto, ResticBackupProgressDto } from "@zerobyte/core/restic";
import { createBackupOptions } from "./backup.helpers";
import { getVolumePath } from "../volumes/helpers";
import { agentManager } from "../agents/agents-manager";
import { decryptRepositoryConfig } from "../repositories/repository-config-secrets";
import { getVolumePath } from "../volumes/helpers";
import { createBackupOptions } from "./backup.helpers";
const LOCAL_AGENT_ID = "local";
type BackupExecutionRequest = {
scheduleId: number;
jobId: string;
schedule: BackupSchedule;
volume: Volume;
repository: Repository;
@ -17,7 +21,14 @@ type BackupExecutionRequest = {
onProgress: (progress: BackupExecutionProgress) => void;
};
export type BackupExecutionProgress = ResticBackupProgressDto;
type ActiveBackupExecution = {
scheduleId: number;
scheduleShortId: string;
onProgress: (progress: BackupExecutionProgress) => void;
resolve: (result: BackupExecutionResult) => void;
};
export type BackupExecutionProgress = BackupProgressPayload["progress"];
export type BackupExecutionResult =
| {
@ -32,75 +43,28 @@ export type BackupExecutionResult =
}
| {
status: "failed";
error: unknown;
error: string;
}
| {
status: "cancelled";
message?: string;
};
type ActiveBackupExecution = {
abortController: AbortController;
jobId?: string;
scheduleShortId?: string;
onProgress?: (progress: BackupExecutionProgress) => void;
resolve?: (result: BackupExecutionResult) => void;
reject?: (error: unknown) => void;
settled: boolean;
};
const LOCAL_AGENT_ID = "local";
const activeBackupsByScheduleId = new Map<number, ActiveBackupExecution>();
const activeScheduleIdsByJobId = new Map<string, number>();
const resetPendingExecution = (activeBackup: ActiveBackupExecution) => {
activeBackup.jobId = undefined;
activeBackup.scheduleShortId = undefined;
activeBackup.onProgress = undefined;
activeBackup.resolve = undefined;
activeBackup.reject = undefined;
activeBackup.settled = false;
};
const getActiveBackupByJobId = (jobId: string) => {
const scheduleId = activeScheduleIdsByJobId.get(jobId);
if (scheduleId === undefined) {
return null;
}
const activeBackup = activeBackupsByScheduleId.get(scheduleId);
if (!activeBackup || activeBackup.jobId !== jobId) {
activeScheduleIdsByJobId.delete(jobId);
return null;
}
return { scheduleId, activeBackup };
};
const resolveActiveBackup = (activeBackup: ActiveBackupExecution, result: BackupExecutionResult) => {
if (activeBackup.settled) {
return;
}
activeBackup.settled = true;
activeBackup.resolve?.(result);
};
const rejectActiveBackup = (activeBackup: ActiveBackupExecution, error: unknown) => {
if (activeBackup.settled) {
return;
}
activeBackup.settled = true;
activeBackup.reject?.(error);
};
const trackedAbortControllersByScheduleId = new Map<number, AbortController>();
const activeExecutionsByJobId = new Map<string, ActiveBackupExecution>();
const activeExecutionJobIdsByScheduleId = new Map<number, string>();
const requestedCancellationsByScheduleId = new Set<number>();
const getCancellationError = (signal: AbortSignal, message?: string) =>
signal.reason instanceof Error ? signal.reason : new Error(message ?? "Backup was stopped by the user");
const buildAgentBackupPayload = async (params: BackupExecutionRequest, jobId: string): Promise<BackupRunPayload> => {
const { schedule, volume, repository, organizationId } = params;
const createBackupRunPayload = async ({
jobId,
schedule,
volume,
repository,
organizationId,
}: BackupExecutionRequest): Promise<BackupRunPayload> => {
const sourcePath = getVolumePath(volume);
const { signal: _ignoredSignal, ...options } = createBackupOptions(schedule, sourcePath);
const repositoryConfig = await decryptRepositoryConfig(repository.config);
@ -114,13 +78,7 @@ const buildAgentBackupPayload = async (params: BackupExecutionRequest, jobId: st
sourcePath,
repositoryConfig,
options: {
tags: options.tags,
oneFileSystem: options.oneFileSystem,
exclude: options.exclude,
excludeIfPresent: options.excludeIfPresent,
includePaths: options.includePaths,
includePatterns: options.includePatterns,
customResticParams: options.customResticParams,
...options,
compressionMode: repository.compressionMode ?? "auto",
},
runtime: {
@ -133,60 +91,86 @@ const buildAgentBackupPayload = async (params: BackupExecutionRequest, jobId: st
};
};
const clearActiveExecution = (jobId: string) => {
const activeExecution = activeExecutionsByJobId.get(jobId);
if (!activeExecution) {
return null;
}
activeExecutionsByJobId.delete(jobId);
activeExecutionJobIdsByScheduleId.delete(activeExecution.scheduleId);
return activeExecution;
};
const getActiveExecution = (jobId: string, scheduleId: string, eventName: string, executorId: string) => {
const activeExecution = activeExecutionsByJobId.get(jobId);
if (!activeExecution) {
logger.warn(`Received ${eventName} for unknown job ${jobId} from executor ${executorId}`);
return null;
}
if (activeExecution.scheduleShortId !== scheduleId) {
logger.warn(
`Ignoring ${eventName} for job ${jobId} due to schedule mismatch ${scheduleId} from executor ${executorId}`,
);
return null;
}
return activeExecution;
};
agentManager.setBackupEventHandlers({
onBackupProgress: ({ payload }) => {
const running = getActiveBackupByJobId(payload.jobId);
if (!running || running.activeBackup.scheduleShortId !== payload.scheduleId) {
return;
}
running.activeBackup.onProgress?.(payload.progress);
onBackupStarted: ({ agentId, payload }) => {
getActiveExecution(payload.jobId, payload.scheduleId, "backup.started", agentId);
},
onBackupCompleted: ({ payload }) => {
const running = getActiveBackupByJobId(payload.jobId);
if (!running || running.activeBackup.scheduleShortId !== payload.scheduleId) {
onBackupProgress: ({ agentId, payload }) => {
const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.progress", agentId);
if (!activeExecution) {
return;
}
activeScheduleIdsByJobId.delete(payload.jobId);
resolveActiveBackup(running.activeBackup, {
activeExecution.onProgress(payload.progress);
},
onBackupCompleted: ({ agentId, payload }) => {
const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.completed", agentId);
if (!activeExecution) {
return;
}
requestedCancellationsByScheduleId.delete(activeExecution.scheduleId);
clearActiveExecution(payload.jobId);
activeExecution.resolve({
status: "completed",
exitCode: payload.exitCode,
result: payload.result,
warningDetails: payload.warningDetails ?? null,
});
},
onBackupFailed: ({ payload }) => {
const running = getActiveBackupByJobId(payload.jobId);
if (!running || running.activeBackup.scheduleShortId !== payload.scheduleId) {
onBackupFailed: ({ agentId, payload }) => {
const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.failed", agentId);
if (!activeExecution) {
return;
}
activeScheduleIdsByJobId.delete(payload.jobId);
resolveActiveBackup(running.activeBackup, {
requestedCancellationsByScheduleId.delete(activeExecution.scheduleId);
clearActiveExecution(payload.jobId);
activeExecution.resolve({
status: "failed",
error: payload.errorDetails ?? payload.error,
});
},
onBackupCancelled: ({ payload }) => {
const running = getActiveBackupByJobId(payload.jobId);
if (!running || running.activeBackup.scheduleShortId !== payload.scheduleId) {
onBackupCancelled: ({ agentId, payload }) => {
const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.cancelled", agentId);
if (!activeExecution) {
return;
}
activeScheduleIdsByJobId.delete(payload.jobId);
if (running.activeBackup.abortController.signal.aborted) {
rejectActiveBackup(
running.activeBackup,
getCancellationError(running.activeBackup.abortController.signal, payload.message),
);
return;
}
resolveActiveBackup(running.activeBackup, {
const wasRequested = requestedCancellationsByScheduleId.has(activeExecution.scheduleId);
requestedCancellationsByScheduleId.delete(activeExecution.scheduleId);
clearActiveExecution(payload.jobId);
activeExecution.resolve({
status: "cancelled",
message: payload.message,
message: wasRequested ? undefined : payload.message,
});
},
});
@ -194,65 +178,74 @@ agentManager.setBackupEventHandlers({
export const backupExecutor = {
track: (scheduleId: number) => {
const abortController = new AbortController();
activeBackupsByScheduleId.set(scheduleId, {
abortController,
settled: false,
});
trackedAbortControllersByScheduleId.set(scheduleId, abortController);
return abortController;
},
untrack: (scheduleId: number, abortController: AbortController) => {
const activeBackup = activeBackupsByScheduleId.get(scheduleId);
if (activeBackup?.abortController === abortController) {
if (activeBackup.jobId) {
activeScheduleIdsByJobId.delete(activeBackup.jobId);
}
activeBackupsByScheduleId.delete(scheduleId);
if (trackedAbortControllersByScheduleId.get(scheduleId) === abortController) {
trackedAbortControllersByScheduleId.delete(scheduleId);
}
requestedCancellationsByScheduleId.delete(scheduleId);
},
execute: async (params: BackupExecutionRequest): Promise<BackupExecutionResult> => {
const activeBackup = activeBackupsByScheduleId.get(params.scheduleId);
if (!activeBackup) {
throw new Error(`Backup ${params.scheduleId} is not tracked`);
}
if (params.signal.aborted) {
throw getCancellationError(params.signal);
execute: async (request: Omit<BackupExecutionRequest, "jobId">) => {
if (request.signal.aborted) {
throw getCancellationError(request.signal);
}
const jobId = Bun.randomUUIDv7();
const payload = await buildAgentBackupPayload(params, jobId);
const payload = await createBackupRunPayload({ ...request, jobId });
if (params.signal.aborted) {
throw getCancellationError(params.signal);
if (request.signal.aborted) {
throw getCancellationError(request.signal);
}
activeBackup.jobId = jobId;
activeBackup.scheduleShortId = params.schedule.shortId;
activeBackup.onProgress = params.onProgress;
activeBackup.settled = false;
const completion = new Promise<BackupExecutionResult>((resolve, reject) => {
activeBackup.resolve = resolve;
activeBackup.reject = reject;
const completion = new Promise<BackupExecutionResult>((resolve) => {
activeExecutionsByJobId.set(jobId, {
scheduleId: request.scheduleId,
scheduleShortId: request.schedule.shortId,
onProgress: request.onProgress,
resolve,
});
activeExecutionJobIdsByScheduleId.set(request.scheduleId, jobId);
});
let dispatched = false;
const handleAbort = () => {
activeScheduleIdsByJobId.delete(jobId);
agentManager.cancelBackup(LOCAL_AGENT_ID, {
jobId,
scheduleId: params.schedule.shortId,
});
rejectActiveBackup(activeBackup, getCancellationError(params.signal));
const activeExecution = activeExecutionsByJobId.get(jobId);
if (!activeExecution) {
return;
}
if (!dispatched) {
clearActiveExecution(jobId);
activeExecution.resolve({ status: "cancelled" });
return;
}
requestedCancellationsByScheduleId.add(request.scheduleId);
if (
!agentManager.cancelBackup(LOCAL_AGENT_ID, {
jobId,
scheduleId: activeExecution.scheduleShortId,
})
) {
requestedCancellationsByScheduleId.delete(request.scheduleId);
clearActiveExecution(jobId);
activeExecution.resolve({ status: "cancelled" });
}
};
params.signal.addEventListener("abort", handleAbort, { once: true });
activeScheduleIdsByJobId.set(jobId, params.scheduleId);
request.signal.addEventListener("abort", handleAbort, { once: true });
const dispatched = agentManager.sendBackup(LOCAL_AGENT_ID, payload);
if (request.signal.aborted) {
return completion;
}
dispatched = agentManager.sendBackup(LOCAL_AGENT_ID, payload);
if (!dispatched) {
params.signal.removeEventListener("abort", handleAbort);
activeScheduleIdsByJobId.delete(jobId);
resetPendingExecution(activeBackup);
request.signal.removeEventListener("abort", handleAbort);
requestedCancellationsByScheduleId.delete(request.scheduleId);
clearActiveExecution(jobId);
return {
status: "unavailable",
error: new Error("Local backup agent is not connected"),
@ -262,16 +255,16 @@ export const backupExecutor = {
try {
return await completion;
} finally {
params.signal.removeEventListener("abort", handleAbort);
request.signal.removeEventListener("abort", handleAbort);
}
},
cancel: (scheduleId: number) => {
const activeBackup = activeBackupsByScheduleId.get(scheduleId);
if (!activeBackup) {
const abortController = trackedAbortControllersByScheduleId.get(scheduleId);
if (!abortController) {
return false;
}
activeBackup.abortController.abort(new Error("Backup was stopped by the user"));
abortController.abort(new Error("Backup was stopped by the user"));
return true;
},
};

11
app/server/modules/backups/backups.execution.ts
View file

@ -1,11 +0,0 @@
import { backupsService } from "./backups.service";
export const backupsExecutionService = {
executeBackup: backupsService.executeBackup,
validateBackupExecution: backupsService.validateBackupExecution,
getSchedulesToExecute: backupsService.getSchedulesToExecute,
stopBackup: backupsService.stopBackup,
runForget: backupsService.runForget,
copyToMirrors: backupsService.copyToMirrors,
getBackupProgress: backupsService.getBackupProgress,
};

5
app/server/modules/backups/backups.service.ts
View file

@ -44,7 +44,6 @@ const getScheduleById = async (scheduleId: number) => {
const getScheduleByShortId = async (shortId: ShortId) => {
return getScheduleByIdOrShortId(shortId);
};
const createSchedule = async (data: CreateBackupScheduleBody) => {
const organizationId = getOrganizationId();
if (data.cronExpression && !isValidCron(data.cronExpression)) {
@ -434,6 +433,10 @@ const executeBackup = async (scheduleId: number, manual = false) => {
case "failed":
return handleBackupFailure(scheduleId, ctx.organizationId, executionResult.error, ctx);
case "cancelled":
if (abortController.signal.aborted) {
return;
}
return handleBackupCancellation(scheduleId, ctx.organizationId, executionResult.message);
}
} finally {

0
apps/agent/build.ts
View file