diff --git a/app/server/modules/agents/agents-manager.ts b/app/server/modules/agents/agents-manager.ts index 21a5efbb..9875a885 100644 --- a/app/server/modules/agents/agents-manager.ts +++ b/app/server/modules/agents/agents-manager.ts @@ -268,7 +268,7 @@ const createAgentManagerRuntime = () => { (server) => Effect.sync(() => { closeAllSessions(); - server.stop(true); + void server.stop(true); }), ); diff --git a/app/server/modules/backups/__tests__/backups.execution.test.ts b/app/server/modules/backups/__tests__/backups.service.execution.test.ts similarity index 78% rename from app/server/modules/backups/__tests__/backups.execution.test.ts rename to app/server/modules/backups/__tests__/backups.service.execution.test.ts index 8f7ff478..9f883dbe 100644 --- a/app/server/modules/backups/__tests__/backups.execution.test.ts +++ b/app/server/modules/backups/__tests__/backups.service.execution.test.ts @@ -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(); diff --git a/app/server/modules/backups/__tests__/backups.service.test.ts b/app/server/modules/backups/__tests__/backups.service.test.ts index 6918b066..783ab226 100644 --- a/app/server/modules/backups/__tests__/backups.service.test.ts +++ b/app/server/modules/backups/__tests__/backups.service.test.ts @@ -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"); }); }); diff --git a/app/server/modules/backups/backup-executor.ts b/app/server/modules/backups/backup-executor.ts index e7984421..130c2323 100644 --- a/app/server/modules/backups/backup-executor.ts +++ b/app/server/modules/backups/backup-executor.ts @@ -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(); -const activeScheduleIdsByJobId = new Map(); - -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(); +const activeExecutionsByJobId = new Map(); +const activeExecutionJobIdsByScheduleId = new Map(); +const requestedCancellationsByScheduleId = new Set(); 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 => { - const { schedule, volume, repository, organizationId } = params; +const createBackupRunPayload = async ({ + jobId, + schedule, + volume, + repository, + organizationId, +}: BackupExecutionRequest): Promise => { 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 => { - 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) => { + 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((resolve, reject) => { - activeBackup.resolve = resolve; - activeBackup.reject = reject; + const completion = new Promise((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; }, }; diff --git a/app/server/modules/backups/backups.execution.ts b/app/server/modules/backups/backups.execution.ts deleted file mode 100644 index f85388bf..00000000 --- a/app/server/modules/backups/backups.execution.ts +++ /dev/null @@ -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, -}; diff --git a/app/server/modules/backups/backups.service.ts b/app/server/modules/backups/backups.service.ts index 3b9ff793..97e03a58 100644 --- a/app/server/modules/backups/backups.service.ts +++ b/app/server/modules/backups/backups.service.ts @@ -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 { diff --git a/apps/agent/build.ts b/apps/agent/build.ts deleted file mode 100644 index e69de29b..00000000