diff --git a/app/server/modules/agents/agents-manager.ts b/app/server/modules/agents/agents-manager.ts index c1cd9635..1d912878 100644 --- a/app/server/modules/agents/agents-manager.ts +++ b/app/server/modules/agents/agents-manager.ts @@ -1,5 +1,6 @@ import type { ChildProcess } from "node:child_process"; -import type { AgentManagerRuntime } from "./controller/server"; +import type { BackupCancelPayload, BackupRunPayload } from "@zerobyte/contracts/agent-protocol"; +import type { AgentBackupEventHandlers, AgentManagerRuntime } from "./controller/server"; import { spawnLocalAgentProcess, stopLocalAgentProcess } from "./local/process"; export type { AgentBackupEventHandlers } from "./controller/server"; @@ -32,6 +33,8 @@ const getAgentRuntimeState = () => { const getAgentManagerRuntime = () => getAgentRuntimeState().agentManager; +let backupEventHandlers: AgentBackupEventHandlers = {}; + export const startAgentRuntime = async () => { const runtime = getAgentRuntimeState(); @@ -41,11 +44,38 @@ export const startAgentRuntime = async () => { const { createAgentManagerRuntime } = await import("./controller/server"); const agentManager = createAgentManagerRuntime(); + agentManager.setBackupEventHandlers(backupEventHandlers); await agentManager.start(); runtime.agentManager = agentManager; }; +export const agentManager = { + sendBackup: (agentId: string, payload: BackupRunPayload) => { + const runtime = getAgentManagerRuntime(); + if (!runtime) { + return false; + } + + return runtime.sendBackup(agentId, payload); + }, + cancelBackup: (agentId: string, payload: BackupCancelPayload) => { + const runtime = getAgentManagerRuntime(); + if (!runtime) { + return false; + } + + return runtime.cancelBackup(agentId, payload); + }, + setBackupEventHandlers: (handlers: AgentBackupEventHandlers) => { + backupEventHandlers = handlers; + getAgentManagerRuntime()?.setBackupEventHandlers(handlers); + }, + getBackupEventHandlers: () => { + return getAgentManagerRuntime()?.getBackupEventHandlers() ?? backupEventHandlers; + }, +}; + export const spawnLocalAgent = async () => { await spawnLocalAgentProcess(getAgentRuntimeState()); }; diff --git a/app/server/modules/agents/controller/server.ts b/app/server/modules/agents/controller/server.ts index ef7663b1..e5a29b6e 100644 --- a/app/server/modules/agents/controller/server.ts +++ b/app/server/modules/agents/controller/server.ts @@ -231,7 +231,7 @@ export function createAgentManagerRuntime() { return { start, - sendBackup: async (agentId: string, payload: BackupRunPayload) => { + sendBackup: (agentId: string, payload: BackupRunPayload) => { const session = getSession(agentId); if (!session) { @@ -244,7 +244,7 @@ export function createAgentManagerRuntime() { return false; } - if (!(await Effect.runPromise(session.sendBackup(payload)))) { + if (!Effect.runSync(session.sendBackup(payload))) { logger.warn(`Cannot send backup command. Agent ${agentId} is no longer accepting commands.`); return false; } @@ -252,7 +252,7 @@ export function createAgentManagerRuntime() { logger.info(`Sent backup command ${payload.jobId} to agent ${agentId} for schedule ${payload.scheduleId}`); return true; }, - cancelBackup: async (agentId: string, payload: BackupCancelPayload) => { + cancelBackup: (agentId: string, payload: BackupCancelPayload) => { const session = getSession(agentId); if (!session) { @@ -260,7 +260,7 @@ export function createAgentManagerRuntime() { return false; } - if (!(await Effect.runPromise(session.sendBackupCancel(payload)))) { + if (!Effect.runSync(session.sendBackupCancel(payload))) { logger.warn(`Cannot cancel backup command. Agent ${agentId} is no longer accepting commands.`); return false; } 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 72% rename from app/server/modules/backups/__tests__/backups.execution.test.ts rename to app/server/modules/backups/__tests__/backups.service.execution.test.ts index fbb20003..3dc64f67 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"; @@ -14,9 +13,14 @@ import type { SafeSpawnParams } from "@zerobyte/core/node"; import { logger } from "@zerobyte/core/node"; import { restic } from "~/server/core/restic"; import { NotFoundError, BadRequestError } from "http-errors-enhanced"; +import { fromAny } from "@total-typescript/shoehorn"; +import { scheduleQueries } from "../backups.queries"; import { repositoriesService } from "~/server/modules/repositories/repositories.service"; import { repoMutex } from "~/server/core/repository-mutex"; import { notificationsService } from "~/server/modules/notifications/notifications.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((_: SafeSpawnParams) => @@ -24,6 +28,7 @@ const setup = () => { ); const resticForgetMock = vi.fn(() => Promise.resolve({ success: true, data: null })); const resticCopyMock = vi.fn(() => Promise.resolve({ success: true, output: "" })); + const { sendBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock); const refreshStatsMock = vi.fn(() => Promise.resolve({ total_size: 0, @@ -39,12 +44,16 @@ const setup = () => { vi.spyOn(restic, "forget").mockImplementation(resticForgetMock); vi.spyOn(restic, "copy").mockImplementation(resticCopyMock); vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock); + vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock); + vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock); vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID); return { resticBackupMock, resticForgetMock, resticCopyMock, + sendBackupMock, + cancelBackupMock, refreshStatsMock, }; }; @@ -65,7 +74,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"); @@ -76,10 +85,70 @@ 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"); @@ -106,7 +175,7 @@ describe("backup execution - validation failures", () => { Promise.resolve({ exitCode: 1, summary: generateBackupOutput(), error: "failed" }), ); - await backupsExecutionService.executeBackup(schedule.id); + await backupsService.executeBackup(schedule.id); expect(notificationSpy).toHaveBeenCalled(); expect(notificationSpy.mock.calls.at(-1)?.[2]?.error).toBe("failed"); @@ -128,7 +197,7 @@ describe("backup execution - validation failures", () => { Promise.resolve({ exitCode: 1, summary: generateBackupOutput(), error: "manual failure" }), ); - await backupsExecutionService.executeBackup(schedule.id, true); + await backupsService.executeBackup(schedule.id, true); expect( errorSpy.mock.calls.some(([message]) => String(message).includes('Failed to parse cron expression ""')), @@ -156,9 +225,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"); }); @@ -184,15 +253,63 @@ 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.", ); }); + test("should block forget on the same repository until the active backup completes", async () => { + const { resticBackupMock, resticForgetMock, sendBackupMock } = setup(); + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + retentionPolicy: { keepHourly: 24 }, + }); + + let completeBackup: (() => void) | undefined; + resticBackupMock.mockImplementationOnce( + () => + new Promise((resolve) => { + completeBackup = () => resolve({ exitCode: 0, summary: generateBackupOutput(), error: "" }); + }), + ); + + const backupPromise = backupsService.executeBackup(schedule.id); + + await waitForExpect(() => { + expect(sendBackupMock).toHaveBeenCalledTimes(1); + }); + + let forgetFinished = false; + const forgetPromise = backupsService.runForget(schedule.id).finally(() => { + forgetFinished = true; + }); + + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(resticForgetMock).not.toHaveBeenCalled(); + expect(forgetFinished).toBe(false); + + expect(completeBackup).toBeDefined(); + completeBackup?.(); + + await backupPromise; + await forgetPromise; + + expect(resticForgetMock).toHaveBeenCalled(); + expect(resticForgetMock).toHaveBeenCalledWith( + repository.config, + expect.objectContaining({ keepHourly: 24 }), + expect.objectContaining({ tag: schedule.shortId, organizationId: TEST_ORG_ID }), + ); + }); + test("should stop a running backup", async () => { // arrange const { resticBackupMock } = setup(); @@ -220,19 +337,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"); }); @@ -246,36 +363,25 @@ describe("stop backup", () => { repositoryId: repository.id, }); - vi.spyOn(repoMutex, "acquireShared").mockImplementation((_repositoryId, _operation, signal) => { - return new Promise((_, reject) => { - if (signal?.aborted) { - reject(signal.reason instanceof Error ? signal.reason : new Error("Operation aborted")); - return; - } + const releaseLock = await repoMutex.acquireExclusive(repository.id, "test"); + const executePromise = backupsService.executeBackup(schedule.id); - signal?.addEventListener( - "abort", - () => { - reject(signal.reason instanceof Error ? signal.reason : new Error("Operation aborted")); - }, - { once: true }, - ); + try { + await waitForExpect(async () => { + const queuedSchedule = await getScheduleByIdOrShortId(schedule.id); + expect(queuedSchedule.lastBackupStatus).toBe("in_progress"); }); - }); - const executePromise = backupsExecutionService.executeBackup(schedule.id); + expect(resticBackupMock).not.toHaveBeenCalled(); - await waitForExpect(async () => { - const queuedSchedule = await backupsService.getScheduleById(schedule.id); - expect(queuedSchedule.lastBackupStatus).toBe("in_progress"); - }); + await backupsService.stopBackup(schedule.id); + } finally { + releaseLock(); + } - expect(resticBackupMock).not.toHaveBeenCalled(); - - await backupsExecutionService.stopBackup(schedule.id); await executePromise; - 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"); expect(resticBackupMock).not.toHaveBeenCalled(); @@ -297,7 +403,7 @@ describe("stop backup", () => { Promise.resolve({ exitCode: 1, summary: generateBackupOutput(), error: "retry me" }), ); - await backupsExecutionService.executeBackup(schedule.id); + await backupsService.executeBackup(schedule.id); const failedSchedule = await backupsService.getScheduleById(schedule.id); expect(failedSchedule.failureRetryCount).toBe(1); @@ -319,14 +425,14 @@ describe("stop backup", () => { }); }); - const executePromise = backupsExecutionService.executeBackup(schedule.id); + const executePromise = backupsService.executeBackup(schedule.id); await waitForExpect(async () => { const retryingSchedule = await backupsService.getScheduleById(schedule.id); expect(retryingSchedule.lastBackupStatus).toBe("in_progress"); }); - await backupsExecutionService.stopBackup(schedule.id); + await backupsService.stopBackup(schedule.id); await executePromise; const cancelledSchedule = await backupsService.getScheduleById(schedule.id); @@ -348,20 +454,40 @@ 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", ); - const updatedSchedule = await backupsService.getScheduleById(schedule.id); + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(updatedSchedule.lastBackupAt).toBe(previousLastBackupAt); expect(updatedSchedule.lastBackupStatus).toBe("warning"); 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"); }); }); @@ -382,7 +508,7 @@ describe("retention policy - runForget", () => { }); // act - await backupsExecutionService.runForget(schedule.id); + await backupsService.runForget(schedule.id); // assert expect(resticForgetMock).toHaveBeenCalledWith( @@ -411,7 +537,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", ); }); @@ -419,7 +545,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 () => { @@ -432,9 +558,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"); }); }); @@ -453,7 +577,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( @@ -480,7 +604,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(); @@ -500,7 +624,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); @@ -531,7 +655,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); @@ -558,7 +682,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); @@ -586,7 +710,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(); @@ -617,7 +741,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(); @@ -660,10 +784,10 @@ describe("mirror operations", () => { ); resticCopyMock.mockImplementation(() => Promise.resolve({ success: true, output: "" })); - const firstCopyPromise = backupsExecutionService.copyToMirrors(firstSchedule.id, sourceRepository, null); + const firstCopyPromise = backupsService.copyToMirrors(firstSchedule.id, sourceRepository, null); await firstCopyStarted; - const secondCopyPromise = backupsExecutionService.copyToMirrors(secondSchedule.id, sourceRepository, null); + const secondCopyPromise = backupsService.copyToMirrors(secondSchedule.id, sourceRepository, null); try { const secondCopyState = await Promise.race<"resolved" | "timeout">([ diff --git a/app/server/modules/backups/__tests__/backups.service.test.ts b/app/server/modules/backups/__tests__/backups.service.test.ts index e0d82282..424f2832 100644 --- a/app/server/modules/backups/__tests__/backups.service.test.ts +++ b/app/server/modules/backups/__tests__/backups.service.test.ts @@ -12,11 +12,14 @@ 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(() => Promise.resolve({ exitCode: 0, summary: "", error: "" })); + const resticBackupMock = vi.fn((_: unknown) => Promise.resolve({ exitCode: 0, summary: "", error: "" })); + const { sendBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock); const refreshStatsMock = vi.fn(() => Promise.resolve({ total_size: 0, @@ -29,10 +32,14 @@ const setup = () => { ); vi.spyOn(spawnModule, "safeSpawn").mockImplementation(resticBackupMock); vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock); + vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock); + vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock); vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID); return { resticBackupMock, + sendBackupMock, + cancelBackupMock, refreshStatsMock, }; }; @@ -59,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); @@ -84,7 +91,7 @@ describe("execute backup", () => { }); // act - await backupsExecutionService.executeBackup(schedule.id); + await backupsService.executeBackup(schedule.id); // assert expect(resticBackupMock).not.toHaveBeenCalled(); @@ -106,11 +113,11 @@ describe("execute backup", () => { ); // act - await backupsExecutionService.executeBackup(schedule.id, true); + await backupsService.executeBackup(schedule.id, true); // assert expect(resticBackupMock).toHaveBeenCalled(); - const updatedSchedule = await backupsService.getScheduleById(schedule.id); + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(updatedSchedule.lastBackupStatus).toBe("success"); expect(updatedSchedule.lastBackupAt).not.toBeNull(); }); @@ -132,10 +139,10 @@ describe("execute backup", () => { ); // act - await backupsExecutionService.executeBackup(schedule.id, true); + await backupsService.executeBackup(schedule.id, true); // assert - const updatedSchedule = await backupsService.getScheduleById(schedule.id); + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(updatedSchedule.nextBackupAt).toBeNull(); }); @@ -155,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); @@ -182,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"); }); @@ -204,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"); }); }); @@ -229,7 +236,7 @@ describe("getSchedulesToExecute", () => { }); // act - const schedulesToExecute = await backupsExecutionService.getSchedulesToExecute(); + const schedulesToExecute = await backupsService.getSchedulesToExecute(); // assert expect(schedulesToExecute).toContain(schedule.id); @@ -246,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); @@ -261,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); @@ -274,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 60c9918c..4d9dfea1 100644 --- a/app/server/modules/backups/backup-executor.ts +++ b/app/server/modules/backups/backup-executor.ts @@ -1,12 +1,18 @@ -import { Effect } from "effect"; -import { restic } 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 type { BackupSchedule, Volume, Repository } from "../../db/schema"; +import { logger } from "@zerobyte/core/node"; +import { resticDeps } from "../../core/restic"; +import type { ResticBackupOutputDto } from "@zerobyte/core/restic"; +import type { BackupProgressPayload, BackupRunPayload } from "@zerobyte/contracts/agent-protocol"; +import { agentManager } from "../agents/agents-manager"; import { getVolumePath } from "../volumes/helpers"; +import { decryptRepositoryConfig } from "../repositories/repository-config-secrets"; +import { createBackupOptions } from "./backup.helpers"; + +const LOCAL_AGENT_ID = "local"; type BackupExecutionRequest = { scheduleId: number; + jobId: string; schedule: BackupSchedule; volume: Volume; repository: Repository; @@ -15,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 = | { @@ -30,15 +43,138 @@ export type BackupExecutionResult = } | { status: "failed"; - error: unknown; + error: string; } | { status: "cancelled"; message?: string; }; +const activeExecutionsByJobId = new Map(); +const activeExecutionJobIdsByScheduleId = new Map(); +const requestedCancellationsByScheduleId = new Set(); const activeControllersByScheduleId = new Map(); +const createBackupRunPayload = async ({ + jobId, + schedule, + volume, + repository, + organizationId, +}: BackupExecutionRequest): Promise => { + const sourcePath = getVolumePath(volume); + const { signal: _, ...options } = createBackupOptions(schedule, sourcePath); + const repositoryConfig = await decryptRepositoryConfig(repository.config); + const encryptedResticPassword = await resticDeps.getOrganizationResticPassword(organizationId); + const resticPassword = await resticDeps.resolveSecret(encryptedResticPassword); + + return { + jobId, + scheduleId: schedule.shortId, + organizationId, + sourcePath, + repositoryConfig, + options: { + ...options, + compressionMode: repository.compressionMode ?? "auto", + }, + runtime: { + password: resticPassword, + cacheDir: resticDeps.resticCacheDir, + passFile: resticDeps.resticPassFile, + defaultExcludes: resticDeps.defaultExcludes, + rcloneConfigFile: resticDeps.rcloneConfigFile, + hostname: resticDeps.hostname, + }, + }; +}; + +const clearActiveExecution = (jobId: string) => { + const activeExecution = activeExecutionsByJobId.get(jobId); + if (!activeExecution) { + return null; + } + + activeExecutionsByJobId.delete(jobId); + activeExecutionJobIdsByScheduleId.delete(activeExecution.scheduleId); + return activeExecution; +}; + +const clearExecutionState = (jobId: string, scheduleId: number) => { + requestedCancellationsByScheduleId.delete(scheduleId); + clearActiveExecution(jobId); +}; + +const resolveExecution = (jobId: string, activeExecution: ActiveBackupExecution, result: BackupExecutionResult) => { + clearExecutionState(jobId, activeExecution.scheduleId); + activeExecution.resolve(result); +}; + +const getActiveExecution = (jobId: string, scheduleId: string, eventName: string, agentId: string) => { + const activeExecution = activeExecutionsByJobId.get(jobId); + if (!activeExecution) { + logger.warn(`Received ${eventName} for unknown job ${jobId} from agent ${agentId}`); + return null; + } + + if (activeExecution.scheduleShortId !== scheduleId) { + logger.warn(`Ignoring ${eventName} for job ${jobId} due to schedule mismatch ${scheduleId} from agent ${agentId}`); + return null; + } + + return activeExecution; +}; + +agentManager.setBackupEventHandlers({ + onBackupStarted: ({ agentId, payload }) => { + getActiveExecution(payload.jobId, payload.scheduleId, "backup.started", agentId); + }, + onBackupProgress: ({ agentId, payload }) => { + const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.progress", agentId); + if (!activeExecution) { + return; + } + + activeExecution.onProgress(payload.progress); + }, + onBackupCompleted: ({ agentId, payload }) => { + const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.completed", agentId); + if (!activeExecution) { + return; + } + + resolveExecution(payload.jobId, activeExecution, { + status: "completed", + exitCode: payload.exitCode, + result: payload.result, + warningDetails: payload.warningDetails ?? null, + }); + }, + onBackupFailed: ({ agentId, payload }) => { + const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.failed", agentId); + if (!activeExecution) { + return; + } + + resolveExecution(payload.jobId, activeExecution, { + status: "failed", + error: payload.errorDetails ?? payload.error, + }); + }, + onBackupCancelled: ({ agentId, payload }) => { + const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.cancelled", agentId); + if (!activeExecution) { + return; + } + + const wasRequested = requestedCancellationsByScheduleId.has(activeExecution.scheduleId); + resolveExecution(payload.jobId, activeExecution, { + status: "cancelled", + message: wasRequested ? undefined : payload.message, + }); + }, +}); + export const backupExecutor = { track: (scheduleId: number) => { const abortController = new AbortController(); @@ -50,43 +186,67 @@ export const backupExecutor = { activeControllersByScheduleId.delete(scheduleId); } }, - execute: async (params: BackupExecutionRequest): Promise => { - const { schedule, volume, repository, organizationId, signal, onProgress } = params; + execute: async (request: Omit) => { + const jobId = Bun.randomUUIDv7(); + const completion = new Promise((resolve) => { + activeExecutionsByJobId.set(jobId, { + scheduleId: request.scheduleId, + scheduleShortId: request.schedule.shortId, + onProgress: request.onProgress, + resolve, + }); + activeExecutionJobIdsByScheduleId.set(request.scheduleId, jobId); + }); + try { - const volumePath = getVolumePath(volume); - const backupOptions = createBackupOptions(schedule, volumePath, signal); - - const execution = await Effect.runPromise( - restic - .backup(repository.config, volumePath, { - ...backupOptions, - compressionMode: repository.compressionMode ?? "auto", - organizationId, - onProgress, - }) - .pipe( - Effect.map((result) => ({ success: true as const, result })), - Effect.catchAll((error) => Effect.succeed({ success: false as const, error })), - ), - ); - - if (!execution.success) { - throw execution.error; + if (request.signal.aborted) { + throw request.signal.reason || new Error("Operation aborted"); } - const { exitCode, result, warningDetails } = execution.result; - return { status: "completed", exitCode, result, warningDetails }; + const payload = await createBackupRunPayload({ ...request, jobId }); + + if (request.signal.aborted) { + throw request.signal.reason || new Error("Operation aborted"); + } + + if (!agentManager.sendBackup(LOCAL_AGENT_ID, payload)) { + clearExecutionState(jobId, request.scheduleId); + return { + status: "unavailable", + error: new Error("Local backup agent is not connected"), + } satisfies BackupExecutionResult; + } + + return completion; } catch (error) { - return { status: "failed", error }; + clearExecutionState(jobId, request.scheduleId); + throw error; } }, cancel: (scheduleId: number) => { const abortController = activeControllersByScheduleId.get(scheduleId); - if (!abortController) { + if (abortController) { + abortController.abort(); + } + + const jobId = activeExecutionJobIdsByScheduleId.get(scheduleId); + if (!jobId) { + return abortController !== undefined; + } + + const activeExecution = activeExecutionsByJobId.get(jobId); + if (!activeExecution) { + activeExecutionJobIdsByScheduleId.delete(scheduleId); + requestedCancellationsByScheduleId.delete(scheduleId); return false; } - abortController.abort(); + requestedCancellationsByScheduleId.add(scheduleId); + agentManager.cancelBackup(LOCAL_AGENT_ID, { + jobId, + scheduleId: activeExecution.scheduleShortId, + }); + 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 a01047b3..80e1d0af 100644 --- a/app/server/modules/backups/backups.service.ts +++ b/app/server/modules/backups/backups.service.ts @@ -37,14 +37,6 @@ const listSchedules = async () => { return schedules.filter((schedule) => schedule.volume && schedule.repository); }; -const getScheduleById = async (scheduleId: number) => { - return getScheduleByIdOrShortId(scheduleId); -}; - -const getScheduleByShortId = async (shortId: ShortId) => { - return getScheduleByIdOrShortId(shortId); -}; - const createSchedule = async (data: CreateBackupScheduleBody) => { const organizationId = getOrganizationId(); if (data.cronExpression && !isValidCron(data.cronExpression)) { @@ -479,9 +471,6 @@ const stopBackup = async (scheduleId: number) => { export const backupsService = { listSchedules, - getScheduleById, - getScheduleByShortId, - getScheduleByIdOrShortId, createSchedule, updateSchedule, deleteSchedule, diff --git a/app/test/helpers/agent-mock.ts b/app/test/helpers/agent-mock.ts new file mode 100644 index 00000000..10112216 --- /dev/null +++ b/app/test/helpers/agent-mock.ts @@ -0,0 +1,105 @@ +import { vi } from "vitest"; +import { fromAny } from "@total-typescript/shoehorn"; +import { agentManager } from "~/server/modules/agents/agents-manager"; + +export const createAgentBackupMocks = ( + resticBackupMock: (params: never) => Promise<{ + exitCode: number; + summary: string; + error: string; + stderr?: string; + }>, +) => { + const runningJobs = new Map(); + + const sendBackupMock = vi.fn((_agentId: string, payload: { jobId: string; scheduleId: string }) => { + const handlers = agentManager.getBackupEventHandlers(); + + runningJobs.set(payload.jobId, { scheduleId: payload.scheduleId, cancelled: false }); + + handlers.onBackupStarted?.({ + agentId: "local", + agentName: "local", + payload: { jobId: payload.jobId, scheduleId: payload.scheduleId }, + }); + + void (async () => { + const stderrLines: string[] = []; + const result = await resticBackupMock( + fromAny({ + onStderr: (line: string) => { + stderrLines.push(line); + }, + }), + ); + const running = runningJobs.get(payload.jobId); + if (!running || running.cancelled) { + return; + } + + if (result.exitCode === 0 || result.exitCode === 3) { + let parsedResult: Record | null = null; + if (result.summary) { + try { + parsedResult = JSON.parse(result.summary) as Record; + } catch { + parsedResult = null; + } + } + + handlers.onBackupCompleted?.({ + agentId: "local", + agentName: "local", + payload: { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + exitCode: result.exitCode, + result: fromAny(parsedResult), + warningDetails: stderrLines.join("\n") || undefined, + }, + }); + } else { + const resultWithStderr = result as typeof result & { stderr?: string }; + const errorDetails = stderrLines.join("\n") || resultWithStderr.stderr || result.error; + + handlers.onBackupFailed?.({ + agentId: "local", + agentName: "local", + payload: { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + error: result.error || `Backup failed with code ${result.exitCode}`, + errorDetails, + }, + }); + } + + runningJobs.delete(payload.jobId); + })().catch(() => {}); + + return true; + }); + + const cancelBackupMock = vi.fn((_agentId: string, payload: { jobId: string; scheduleId: string }) => { + const running = runningJobs.get(payload.jobId); + if (!running) { + return false; + } + + running.cancelled = true; + const handlers = agentManager.getBackupEventHandlers(); + handlers.onBackupCancelled?.({ + agentId: "local", + agentName: "local", + payload: { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + message: "Backup was stopped by user", + }, + }); + runningJobs.delete(payload.jobId); + return true; + }); + + return { sendBackupMock, cancelBackupMock }; +};