From 9a242eb445688ad6d25c619bae655bc6d25db8c5 Mon Sep 17 00:00:00 2001 From: Nicolas Meienberger Date: Sun, 12 Apr 2026 14:49:18 +0200 Subject: [PATCH] fix: rebase conflicts --- .../__tests__/agents-manager.backups.test.ts | 56 +++++ .../agents/__tests__/agents-manager.test.ts | 10 +- .../modules/agents/__tests__/session.test.ts | 24 +- app/server/modules/agents/agents-manager.ts | 214 ++++++++++++++++-- .../modules/agents/controller/server.ts | 8 +- .../modules/agents/controller/session.ts | 54 +++-- .../backups.service.execution.test.ts | 64 +++++- .../backups/__tests__/backups.service.test.ts | 9 +- app/server/modules/backups/backup-executor.ts | 204 +++-------------- app/server/modules/backups/backups.service.ts | 2 +- .../backups/helpers/backup-lifecycle.ts | 2 +- app/test/helpers/agent-mock.ts | 142 ++++++------ apps/agent/tsconfig.json | 4 +- 13 files changed, 461 insertions(+), 332 deletions(-) create mode 100644 app/server/modules/agents/__tests__/agents-manager.backups.test.ts diff --git a/app/server/modules/agents/__tests__/agents-manager.backups.test.ts b/app/server/modules/agents/__tests__/agents-manager.backups.test.ts new file mode 100644 index 00000000..c29ccd2a --- /dev/null +++ b/app/server/modules/agents/__tests__/agents-manager.backups.test.ts @@ -0,0 +1,56 @@ +import { afterEach, expect, test, vi } from "vitest"; +import waitForExpect from "wait-for-expect"; +import { fromPartial } from "@total-typescript/shoehorn"; +import { agentManager } from "../agents-manager"; +import type { AgentManagerRuntime } from "../controller/server"; +import type { BackupRunPayload } from "@zerobyte/contracts/agent-protocol"; + +type ProcessWithAgentRuntime = NodeJS.Process & { + __zerobyteAgentRuntime?: { + agentManager: AgentManagerRuntime | null; + localAgent: null; + isStoppingLocalAgent: boolean; + localAgentRestartTimeout: null; + }; +}; + +const setAgentRuntime = (agentManagerRuntime: Partial | null) => { + (process as ProcessWithAgentRuntime).__zerobyteAgentRuntime = { + agentManager: agentManagerRuntime ? fromPartial(agentManagerRuntime) : null, + localAgent: null, + isStoppingLocalAgent: false, + localAgentRestartTimeout: null, + }; +}; + +afterEach(() => { + delete (process as ProcessWithAgentRuntime).__zerobyteAgentRuntime; + vi.restoreAllMocks(); +}); + +test("cancelBackup resolves a running backup when the cancel command cannot be delivered", async () => { + const sendBackup = vi.fn().mockResolvedValue(true); + const cancelBackup = vi.fn().mockResolvedValue(false); + setAgentRuntime({ sendBackup, cancelBackup }); + + const resultPromise = agentManager.runBackup("local", { + scheduleId: 42, + payload: fromPartial({ + jobId: "job-1", + scheduleId: "schedule-1", + }), + signal: new AbortController().signal, + onProgress: vi.fn(), + }); + + await waitForExpect(() => { + expect(sendBackup).toHaveBeenCalledTimes(1); + }); + + await expect(agentManager.cancelBackup("local", 42)).resolves.toBe(true); + await expect(resultPromise).resolves.toEqual({ status: "cancelled" }); + expect(cancelBackup).toHaveBeenCalledWith("local", { + jobId: "job-1", + scheduleId: "schedule-1", + }); +}); diff --git a/app/server/modules/agents/__tests__/agents-manager.test.ts b/app/server/modules/agents/__tests__/agents-manager.test.ts index dda50938..1d882916 100644 --- a/app/server/modules/agents/__tests__/agents-manager.test.ts +++ b/app/server/modules/agents/__tests__/agents-manager.test.ts @@ -15,10 +15,6 @@ vi.mock("node:child_process", async () => { const { spawnLocalAgent, stopLocalAgent } = await import("../agents-manager"); -const flushMicrotasks = async () => { - await Promise.resolve(); -}; - type FakeChildProcess = EventEmitter & { stdout: PassThrough; stderr: PassThrough; @@ -64,8 +60,7 @@ test("respawns the local agent after an unexpected exit", async () => { firstChild.exitCode = 1; firstChild.emit("exit", 1, null); - vi.advanceTimersByTime(1_000); - await flushMicrotasks(); + await vi.advanceTimersByTimeAsync(1_000); expect(spawnMock).toHaveBeenCalledTimes(2); }); @@ -79,8 +74,7 @@ test("does not respawn the local agent after an intentional stop", async () => { await spawnLocalAgent(); await stopLocalAgent(); - vi.advanceTimersByTime(1_000); - await flushMicrotasks(); + await vi.advanceTimersByTimeAsync(1_000); expect(spawnMock).toHaveBeenCalledTimes(1); expect(child.kill).toHaveBeenCalledTimes(1); diff --git a/app/server/modules/agents/__tests__/session.test.ts b/app/server/modules/agents/__tests__/session.test.ts index 2dbb33d8..e8913fd1 100644 --- a/app/server/modules/agents/__tests__/session.test.ts +++ b/app/server/modules/agents/__tests__/session.test.ts @@ -5,9 +5,7 @@ import { fromPartial } from "@total-typescript/shoehorn"; import { createAgentMessage } from "@zerobyte/contracts/agent-protocol"; import { createControllerAgentSession } from "../controller/session"; -const createSocket = ( - overrides: Partial[0]> = {}, -) => { +const createSocket = (overrides: Partial[0]> = {}) => { return fromPartial[0]>({ data: { id: "connection-1", agentId: "local", organizationId: null, agentName: "Local Agent" }, send: vi.fn(() => 1), @@ -16,10 +14,7 @@ const createSocket = ( }); }; -const createSession = ( - handlers: Parameters[1] = {}, - socket = createSocket(), -) => { +const createSession = (handlers: Parameters[1] = {}, socket = createSocket()) => { const scope = Effect.runSync(Scope.make()); try { @@ -28,12 +23,15 @@ const createSession = ( return { session, run: () => { - Effect.runSync(Scope.extend(Effect.forkScoped(session.run), scope)); + Effect.runFork(Scope.extend(session.run, scope)); }, socket, close: () => { Effect.runSync(Scope.close(scope, Exit.succeed(undefined))); }, + closeAsync: () => { + return Effect.runPromise(Scope.close(scope, Exit.succeed(undefined))); + }, }; } catch (error) { Effect.runSync(Scope.close(scope, Exit.fail(error))); @@ -62,7 +60,8 @@ test("close emits a synthetic backup.cancelled for a started backup", () => { expect(onBackupCancelled).toHaveBeenCalledWith({ jobId: "job-1", scheduleId: "schedule-1", - message: "The connection to the backup agent was lost while this backup was running. Restart the backup to ensure it completes.", + message: + "The connection to the backup agent was lost while this backup was running. Restart the backup to ensure it completes.", }); }); @@ -154,7 +153,8 @@ test("close emits a synthetic backup.cancelled for a queued backup", () => { expect(onBackupCancelled).toHaveBeenCalledWith({ jobId: "job-queued", scheduleId: "schedule-queued", - message: "The connection to the backup agent was lost before this backup started. Restart the backup to ensure it completes.", + message: + "The connection to the backup agent was lost before this backup started. Restart the backup to ensure it completes.", }); }); @@ -162,7 +162,7 @@ test("a dropped backup.cancel closes the session and emits a synthetic backup.ca const send = vi.fn(() => 0); const socket = createSocket({ send, close: vi.fn() }); const onBackupCancelled = vi.fn(); - const { session, run, close: closeSession } = createSession({ onBackupCancelled }, socket); + const { session, run, closeAsync } = createSession({ onBackupCancelled }, socket); try { run(); @@ -193,6 +193,6 @@ test("a dropped backup.cancel closes the session and emits a synthetic backup.ca }); }); } finally { - closeSession(); + await closeAsync(); } }); diff --git a/app/server/modules/agents/agents-manager.ts b/app/server/modules/agents/agents-manager.ts index b97147b5..893ccd1a 100644 --- a/app/server/modules/agents/agents-manager.ts +++ b/app/server/modules/agents/agents-manager.ts @@ -1,9 +1,23 @@ import type { ChildProcess } from "node:child_process"; -import type { BackupCancelPayload, BackupRunPayload } from "@zerobyte/contracts/agent-protocol"; +import { logger } from "@zerobyte/core/node"; +import type { ResticBackupOutputDto } from "@zerobyte/core/restic"; +import type { BackupProgressPayload, 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"; +export type BackupExecutionProgress = BackupProgressPayload["progress"]; +export type BackupExecutionResult = + | { status: "unavailable"; error: Error } + | { status: "completed"; exitCode: number; result: ResticBackupOutputDto | null; warningDetails: string | null } + | { status: "failed"; error: string } + | { status: "cancelled"; message?: string }; + +export type AgentRunBackupRequest = { + scheduleId: number; + payload: BackupRunPayload; + signal: AbortSignal; + onProgress: (progress: BackupExecutionProgress) => void; +}; type AgentRuntimeState = { agentManager: AgentManagerRuntime | null; @@ -37,7 +51,147 @@ const getAgentRuntimeState = () => { const getAgentManagerRuntime = () => getAgentRuntimeState().agentManager; -let backupEventHandlers: AgentBackupEventHandlers = {}; +type ActiveBackupRun = { + scheduleId: number; + jobId: string; + scheduleShortId: string; + onProgress: (progress: BackupExecutionProgress) => void; + resolve: (result: BackupExecutionResult) => void; + cancellationRequested: boolean; +}; + +const activeBackupsByScheduleId = new Map(); +const activeBackupScheduleIdsByJobId = new Map(); + +const getUnavailableError = (agentId: string) => { + if (agentId === "local") { + return new Error("Local backup agent is not connected"); + } + + return new Error(`Backup agent ${agentId} is not connected`); +}; + +const clearActiveBackupRun = (scheduleId: number) => { + const activeBackupRun = activeBackupsByScheduleId.get(scheduleId); + if (!activeBackupRun) { + return null; + } + + activeBackupsByScheduleId.delete(scheduleId); + activeBackupScheduleIdsByJobId.delete(activeBackupRun.jobId); + return activeBackupRun; +}; + +const resolveActiveBackupRun = (scheduleId: number, result: BackupExecutionResult) => { + const activeBackupRun = clearActiveBackupRun(scheduleId); + if (!activeBackupRun) { + return false; + } + + activeBackupRun.resolve(result); + return true; +}; + +const getActiveBackupRun = (jobId: string, scheduleId: string, eventName: string, agentId: string) => { + const trackedScheduleId = activeBackupScheduleIdsByJobId.get(jobId); + if (trackedScheduleId === undefined) { + logger.warn(`Received ${eventName} for unknown job ${jobId} from agent ${agentId}`); + return null; + } + + const activeBackupRun = activeBackupsByScheduleId.get(trackedScheduleId); + if (!activeBackupRun) { + logger.warn(`Received ${eventName} for inactive job ${jobId} from agent ${agentId}`); + return null; + } + + if (activeBackupRun.scheduleShortId !== scheduleId) { + logger.warn(`Ignoring ${eventName} for job ${jobId} due to schedule mismatch ${scheduleId} from agent ${agentId}`); + return null; + } + + return activeBackupRun; +}; + +const requestBackupCancellation = async (agentId: string, scheduleId: number) => { + const activeBackupRun = activeBackupsByScheduleId.get(scheduleId); + if (!activeBackupRun) { + return false; + } + + if (activeBackupRun.cancellationRequested) { + return true; + } + + activeBackupRun.cancellationRequested = true; + + const runtime = getAgentManagerRuntime(); + if (!runtime) { + resolveActiveBackupRun(scheduleId, { status: "cancelled" }); + return true; + } + + if ( + await runtime.cancelBackup(agentId, { + jobId: activeBackupRun.jobId, + scheduleId: activeBackupRun.scheduleShortId, + }) + ) { + return true; + } + + resolveActiveBackupRun(scheduleId, { status: "cancelled" }); + return true; +}; + +const backupEventHandlers: AgentBackupEventHandlers = { + onBackupStarted: ({ agentId, payload }) => { + getActiveBackupRun(payload.jobId, payload.scheduleId, "backup.started", agentId); + }, + onBackupProgress: ({ agentId, payload }) => { + const activeBackupRun = getActiveBackupRun(payload.jobId, payload.scheduleId, "backup.progress", agentId); + if (!activeBackupRun) { + return; + } + + activeBackupRun.onProgress(payload.progress); + }, + onBackupCompleted: ({ agentId, payload }) => { + const activeBackupRun = getActiveBackupRun(payload.jobId, payload.scheduleId, "backup.completed", agentId); + if (!activeBackupRun) { + return; + } + + resolveActiveBackupRun(activeBackupRun.scheduleId, { + status: "completed", + exitCode: payload.exitCode, + result: payload.result, + warningDetails: payload.warningDetails ?? null, + }); + }, + onBackupFailed: ({ agentId, payload }) => { + const activeBackupRun = getActiveBackupRun(payload.jobId, payload.scheduleId, "backup.failed", agentId); + if (!activeBackupRun) { + return; + } + + resolveActiveBackupRun(activeBackupRun.scheduleId, { + status: "failed", + error: payload.errorDetails ?? payload.error, + }); + }, + onBackupCancelled: ({ agentId, payload }) => { + const activeBackupRun = getActiveBackupRun(payload.jobId, payload.scheduleId, "backup.cancelled", agentId); + if (!activeBackupRun) { + return; + } + + resolveActiveBackupRun(activeBackupRun.scheduleId, { + status: "cancelled", + message: activeBackupRun.cancellationRequested ? undefined : payload.message, + }); + }, +}; export const startAgentRuntime = async () => { const runtime = getAgentRuntimeState(); @@ -55,28 +209,52 @@ export const startAgentRuntime = async () => { }; export const agentManager = { - sendBackup: (agentId: string, payload: BackupRunPayload) => { + runBackup: async (agentId: string, request: AgentRunBackupRequest) => { const runtime = getAgentManagerRuntime(); if (!runtime) { - return false; + return { + status: "unavailable", + error: getUnavailableError(agentId), + } satisfies BackupExecutionResult; } - return runtime.sendBackup(agentId, payload); - }, - cancelBackup: (agentId: string, payload: BackupCancelPayload) => { - const runtime = getAgentManagerRuntime(); - if (!runtime) { - return false; + if (request.signal.aborted) { + throw request.signal.reason || new Error("Operation aborted"); } - return runtime.cancelBackup(agentId, payload); + const completion = new Promise((resolve) => { + activeBackupsByScheduleId.set(request.scheduleId, { + scheduleId: request.scheduleId, + jobId: request.payload.jobId, + scheduleShortId: request.payload.scheduleId, + onProgress: request.onProgress, + resolve, + cancellationRequested: false, + }); + activeBackupScheduleIdsByJobId.set(request.payload.jobId, request.scheduleId); + }); + + try { + if (!(await runtime.sendBackup(agentId, request.payload))) { + clearActiveBackupRun(request.scheduleId); + return { + status: "unavailable", + error: getUnavailableError(agentId), + } satisfies BackupExecutionResult; + } + + if (request.signal.aborted) { + await requestBackupCancellation(agentId, request.scheduleId); + } + + return completion; + } catch (error) { + clearActiveBackupRun(request.scheduleId); + throw error; + } }, - setBackupEventHandlers: (handlers: AgentBackupEventHandlers) => { - backupEventHandlers = handlers; - getAgentManagerRuntime()?.setBackupEventHandlers(handlers); - }, - getBackupEventHandlers: () => { - return getAgentManagerRuntime()?.getBackupEventHandlers() ?? backupEventHandlers; + cancelBackup: async (agentId: string, scheduleId: number) => { + return requestBackupCancellation(agentId, scheduleId); }, }; diff --git a/app/server/modules/agents/controller/server.ts b/app/server/modules/agents/controller/server.ts index e5a29b6e..ef7663b1 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: (agentId: string, payload: BackupRunPayload) => { + sendBackup: async (agentId: string, payload: BackupRunPayload) => { const session = getSession(agentId); if (!session) { @@ -244,7 +244,7 @@ export function createAgentManagerRuntime() { return false; } - if (!Effect.runSync(session.sendBackup(payload))) { + if (!(await Effect.runPromise(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: (agentId: string, payload: BackupCancelPayload) => { + cancelBackup: async (agentId: string, payload: BackupCancelPayload) => { const session = getSession(agentId); if (!session) { @@ -260,7 +260,7 @@ export function createAgentManagerRuntime() { return false; } - if (!Effect.runSync(session.sendBackupCancel(payload))) { + if (!(await Effect.runPromise(session.sendBackupCancel(payload)))) { logger.warn(`Cannot cancel backup command. Agent ${agentId} is no longer accepting commands.`); return false; } diff --git a/app/server/modules/agents/controller/session.ts b/app/server/modules/agents/controller/session.ts index e208e1a5..f010f5a5 100644 --- a/app/server/modules/agents/controller/session.ts +++ b/app/server/modules/agents/controller/session.ts @@ -57,6 +57,7 @@ export const createControllerAgentSession = ( handlers: ControllerAgentSessionHandlers = {}, ): Effect.Effect => Effect.gen(function* () { + let isClosed = false; const outboundQueue = yield* Queue.bounded(64); const trackedBackupJobs = yield* Ref.make>(new Map()); const state = yield* Ref.make({ @@ -102,10 +103,7 @@ export const createControllerAgentSession = ( yield* updateState((current) => ({ ...current, isReady: false })); const trackedJobs = yield* takeTrackedBackupJobs; for (const [jobId, trackedJob] of trackedJobs) { - const message = - trackedJob.state === "pending" - ? "The connection to the backup agent was lost before this backup started. Restart the backup to ensure it completes." - : "The connection to the backup agent was lost while this backup was running. Restart the backup to ensure it completes."; + const message = "The connection to the backup agent was lost. Restart the backup to ensure it completes."; yield* Effect.sync(() => { handlers.onBackupCancelled?.({ jobId, scheduleId: trackedJob.scheduleId, message }); @@ -115,36 +113,48 @@ export const createControllerAgentSession = ( yield* Queue.shutdown(outboundQueue); }); - yield* Effect.addFinalizer(() => releaseSession); + const closeSession = () => + Effect.suspend(() => { + if (isClosed) { + return Effect.sync(() => undefined); + } + + isClosed = true; + return releaseSession; + }); + + yield* Effect.addFinalizer(() => closeSession()); const handleSendFailure = (reason: string) => { logger.error( `Closing session for agent ${socket.data.agentId} on ${socket.data.id} after an outbound websocket send failed: ${reason}`, ); - try { - socket.close(); - } catch (error) { - logger.error(`Failed to close socket for agent ${socket.data.agentId} on ${socket.data.id}: ${toMessage(error)}`); - } + socket.close(); + + void Effect.runPromise(closeSession()).catch((error) => { + logger.error( + `Failed to close session for agent ${socket.data.agentId} on ${socket.data.id}: ${toMessage(error)}`, + ); + }); }; const run = Effect.gen(function* () { yield* Effect.forkScoped( Effect.forever( - Effect.gen(function* () { - const message = yield* Queue.take(outboundQueue); - yield* Effect.sync(() => { - try { - const sendResult = socket.send(message); - if (sendResult <= 0) { - handleSendFailure(sendResult === 0 ? "connection issue" : "backpressure"); + Effect.gen(function* () { + const message = yield* Queue.take(outboundQueue); + yield* Effect.sync(() => { + try { + const sendResult = socket.send(message); + if (sendResult <= 0) { + handleSendFailure(sendResult === 0 ? "connection issue" : "backpressure"); + } + } catch (error) { + handleSendFailure(toMessage(error)); } - } catch (error) { - handleSendFailure(toMessage(error)); - } - }); - }), + }); + }), ), ); diff --git a/app/server/modules/backups/__tests__/backups.service.execution.test.ts b/app/server/modules/backups/__tests__/backups.service.execution.test.ts index 3dc64f67..c9b236b9 100644 --- a/app/server/modules/backups/__tests__/backups.service.execution.test.ts +++ b/app/server/modules/backups/__tests__/backups.service.execution.test.ts @@ -28,7 +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 { runBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock); const refreshStatsMock = vi.fn(() => Promise.resolve({ total_size: 0, @@ -44,7 +44,7 @@ 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, "runBackup").mockImplementation(runBackupMock); vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock); vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID); @@ -52,7 +52,7 @@ const setup = () => { resticBackupMock, resticForgetMock, resticCopyMock, - sendBackupMock, + runBackupMock, cancelBackupMock, refreshStatsMock, }; @@ -203,6 +203,27 @@ describe("backup execution - validation failures", () => { errorSpy.mock.calls.some(([message]) => String(message).includes('Failed to parse cron expression ""')), ).toBe(false); }); + + test("should fail backup when the local agent is unavailable", async () => { + const { runBackupMock } = setup(); + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + }); + + runBackupMock.mockResolvedValueOnce({ + status: "unavailable", + error: new Error("Local backup agent is not connected"), + }); + + await backupsService.executeBackup(schedule.id); + + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); + expect(updatedSchedule.lastBackupStatus).toBe("error"); + expect(updatedSchedule.lastBackupError).toBe("Local backup agent is not connected"); + }); }); describe("stop backup", () => { @@ -263,7 +284,7 @@ describe("stop backup", () => { }); test("should block forget on the same repository until the active backup completes", async () => { - const { resticBackupMock, resticForgetMock, sendBackupMock } = setup(); + const { resticBackupMock, resticForgetMock, runBackupMock } = setup(); const volume = await createTestVolume(); const repository = await createTestRepository(); const schedule = await createTestBackupSchedule({ @@ -283,7 +304,7 @@ describe("stop backup", () => { const backupPromise = backupsService.executeBackup(schedule.id); await waitForExpect(() => { - expect(sendBackupMock).toHaveBeenCalledTimes(1); + expect(runBackupMock).toHaveBeenCalledTimes(1); }); let forgetFinished = false; @@ -354,6 +375,33 @@ describe("stop backup", () => { expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user"); }); + test("should stop a running backup when the cancel command cannot be delivered", async () => { + const { resticBackupMock, cancelBackupMock } = setup(); + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + }); + + resticBackupMock.mockImplementation(() => new Promise(() => {})); + cancelBackupMock.mockResolvedValueOnce(false); + + const executePromise = backupsService.executeBackup(schedule.id); + + await waitForExpect(async () => { + const runningSchedule = await getScheduleByIdOrShortId(schedule.id); + expect(runningSchedule.lastBackupStatus).toBe("in_progress"); + }); + + await backupsService.stopBackup(schedule.id); + await executePromise; + + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); + expect(updatedSchedule.lastBackupStatus).toBe("warning"); + expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user"); + }); + test("should stop a queued backup before it acquires the repository lock", async () => { const { resticBackupMock } = setup(); const volume = await createTestVolume(); @@ -405,7 +453,7 @@ describe("stop backup", () => { await backupsService.executeBackup(schedule.id); - const failedSchedule = await backupsService.getScheduleById(schedule.id); + const failedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(failedSchedule.failureRetryCount).toBe(1); resticBackupMock.mockImplementationOnce(({ signal }: SafeSpawnParams) => { @@ -428,14 +476,14 @@ describe("stop backup", () => { const executePromise = backupsService.executeBackup(schedule.id); await waitForExpect(async () => { - const retryingSchedule = await backupsService.getScheduleById(schedule.id); + const retryingSchedule = await getScheduleByIdOrShortId(schedule.id); expect(retryingSchedule.lastBackupStatus).toBe("in_progress"); }); await backupsService.stopBackup(schedule.id); await executePromise; - const cancelledSchedule = await backupsService.getScheduleById(schedule.id); + const cancelledSchedule = await getScheduleByIdOrShortId(schedule.id); expect(cancelledSchedule.lastBackupStatus).toBe("warning"); expect(cancelledSchedule.failureRetryCount).toBe(0); }); diff --git a/app/server/modules/backups/__tests__/backups.service.test.ts b/app/server/modules/backups/__tests__/backups.service.test.ts index 424f2832..6d6c0185 100644 --- a/app/server/modules/backups/__tests__/backups.service.test.ts +++ b/app/server/modules/backups/__tests__/backups.service.test.ts @@ -8,6 +8,7 @@ import { createTestRepository } from "~/test/helpers/repository"; import { generateBackupOutput } from "~/test/helpers/restic"; import { faker } from "@faker-js/faker"; import * as spawnModule from "@zerobyte/core/node"; +import type { SafeSpawnParams } from "@zerobyte/core/node"; import { db } from "~/server/db/db"; import { backupScheduleMirrorsTable, repositoriesTable, volumesTable } from "~/server/db/schema"; import { TEST_ORG_ID } from "~/test/helpers/organization"; @@ -18,8 +19,8 @@ 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: "" })); - const { sendBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock); + const resticBackupMock = vi.fn((_: SafeSpawnParams) => Promise.resolve({ exitCode: 0, summary: "", error: "" })); + const { runBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock); const refreshStatsMock = vi.fn(() => Promise.resolve({ total_size: 0, @@ -32,13 +33,13 @@ const setup = () => { ); vi.spyOn(spawnModule, "safeSpawn").mockImplementation(resticBackupMock); vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock); - vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock); + vi.spyOn(agentManager, "runBackup").mockImplementation(runBackupMock); vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock); vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID); return { resticBackupMock, - sendBackupMock, + runBackupMock, cancelBackupMock, refreshStatsMock, }; diff --git a/app/server/modules/backups/backup-executor.ts b/app/server/modules/backups/backup-executor.ts index 4d9dfea1..6763690d 100644 --- a/app/server/modules/backups/backup-executor.ts +++ b/app/server/modules/backups/backup-executor.ts @@ -1,9 +1,7 @@ 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 type { BackupRunPayload } from "@zerobyte/contracts/agent-protocol"; +import { agentManager, type BackupExecutionProgress } from "../agents/agents-manager"; import { getVolumePath } from "../volumes/helpers"; import { decryptRepositoryConfig } from "../repositories/repository-config-secrets"; import { createBackupOptions } from "./backup.helpers"; @@ -12,7 +10,6 @@ const LOCAL_AGENT_ID = "local"; type BackupExecutionRequest = { scheduleId: number; - jobId: string; schedule: BackupSchedule; volume: Volume; repository: Repository; @@ -21,38 +18,8 @@ type BackupExecutionRequest = { onProgress: (progress: BackupExecutionProgress) => void; }; -type ActiveBackupExecution = { - scheduleId: number; - scheduleShortId: string; - onProgress: (progress: BackupExecutionProgress) => void; - resolve: (result: BackupExecutionResult) => void; -}; +export type { BackupExecutionResult } from "../agents/agents-manager"; -export type BackupExecutionProgress = BackupProgressPayload["progress"]; - -export type BackupExecutionResult = - | { - status: "unavailable"; - error: Error; - } - | { - status: "completed"; - exitCode: number; - result: ResticBackupOutputDto | null; - warningDetails: string | null; - } - | { - status: "failed"; - 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 ({ @@ -61,7 +28,7 @@ const createBackupRunPayload = async ({ volume, repository, organizationId, -}: BackupExecutionRequest): Promise => { +}: BackupExecutionRequest & { jobId: string }): Promise => { const sourcePath = getVolumePath(volume); const { signal: _, ...options } = createBackupOptions(schedule, sourcePath); const repositoryConfig = await decryptRepositoryConfig(repository.config); @@ -89,92 +56,6 @@ const createBackupRunPayload = async ({ }; }; -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(); @@ -187,66 +68,37 @@ export const backupExecutor = { } }, execute: async (request: Omit) => { + const trackedAbortController = activeControllersByScheduleId.get(request.scheduleId); + if (!trackedAbortController || trackedAbortController.signal !== request.signal) { + throw new Error(`Backup execution for schedule ${request.scheduleId} was not tracked`); + } + 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); + if (request.signal.aborted) { + throw request.signal.reason || new Error("Operation aborted"); + } + + const payload = await createBackupRunPayload({ ...request, jobId }); + + if (request.signal.aborted) { + throw request.signal.reason || new Error("Operation aborted"); + } + + return agentManager.runBackup(LOCAL_AGENT_ID, { + scheduleId: request.scheduleId, + payload, + signal: request.signal, + onProgress: request.onProgress, }); - - try { - if (request.signal.aborted) { - throw request.signal.reason || new Error("Operation aborted"); - } - - 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) { - clearExecutionState(jobId, request.scheduleId); - throw error; - } }, - cancel: (scheduleId: number) => { + cancel: async (scheduleId: number) => { const abortController = activeControllersByScheduleId.get(scheduleId); - 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); + if (!abortController) { return false; } - requestedCancellationsByScheduleId.add(scheduleId); - agentManager.cancelBackup(LOCAL_AGENT_ID, { - jobId, - scheduleId: activeExecution.scheduleShortId, - }); - + abortController.abort(); + await agentManager.cancelBackup(LOCAL_AGENT_ID, scheduleId); return true; }, }; diff --git a/app/server/modules/backups/backups.service.ts b/app/server/modules/backups/backups.service.ts index 80e1d0af..d39479f0 100644 --- a/app/server/modules/backups/backups.service.ts +++ b/app/server/modules/backups/backups.service.ts @@ -459,7 +459,7 @@ const stopBackup = async (scheduleId: number) => { } try { - if (!backupExecutor.cancel(scheduleId)) { + if (!(await backupExecutor.cancel(scheduleId))) { throw new ConflictError("No backup is currently running for this schedule"); } diff --git a/app/server/modules/backups/helpers/backup-lifecycle.ts b/app/server/modules/backups/helpers/backup-lifecycle.ts index 4bb33b0f..acc887f1 100644 --- a/app/server/modules/backups/helpers/backup-lifecycle.ts +++ b/app/server/modules/backups/helpers/backup-lifecycle.ts @@ -10,7 +10,7 @@ import { getOrganizationId } from "~/server/core/request-context"; import type { BackupProgressEventDto } from "~/schemas/events-dto"; import { calculateNextRun } from "../backup.helpers"; import { scheduleQueries } from "../backups.queries"; -import type { BackupExecutionProgress } from "../backup-executor"; +import type { BackupExecutionProgress } from "../../agents/agents-manager"; import { repositoriesService } from "../../repositories/repositories.service"; import { copyToMirrors, runForget } from "./backup-maintenance"; diff --git a/app/test/helpers/agent-mock.ts b/app/test/helpers/agent-mock.ts index 10112216..1f00b14f 100644 --- a/app/test/helpers/agent-mock.ts +++ b/app/test/helpers/agent-mock.ts @@ -1,105 +1,95 @@ import { vi } from "vitest"; -import { fromAny } from "@total-typescript/shoehorn"; -import { agentManager } from "~/server/modules/agents/agents-manager"; +import { fromAny, fromPartial } from "@total-typescript/shoehorn"; +import type { SafeSpawnParams } from "@zerobyte/core/node"; +import type { BackupExecutionResult } from "~/server/modules/agents/agents-manager"; export const createAgentBackupMocks = ( - resticBackupMock: (params: never) => Promise<{ + resticBackupMock: (params: SafeSpawnParams) => Promise<{ exitCode: number; summary: string; error: string; stderr?: string; }>, ) => { - const runningJobs = new Map(); + const runningBackups = new Map void; cancelled: boolean }>(); - const sendBackupMock = vi.fn((_agentId: string, payload: { jobId: string; scheduleId: string }) => { - const handlers = agentManager.getBackupEventHandlers(); + const runBackupMock = vi.fn( + async (_agentId: string, request: { scheduleId: number; payload: { jobId: string }; signal: AbortSignal }) => { + return new Promise((resolve) => { + runningBackups.set(request.scheduleId, { resolve, cancelled: false }); - runningJobs.set(payload.jobId, { scheduleId: payload.scheduleId, cancelled: false }); + request.signal.addEventListener( + "abort", + () => { + const running = runningBackups.get(request.scheduleId); + if (!running || running.cancelled) { + return; + } - 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); + running.cancelled = true; + runningBackups.delete(request.scheduleId); + resolve({ status: "cancelled" }); }, - }), - ); - const running = runningJobs.get(payload.jobId); - if (!running || running.cancelled) { - return; - } + { once: true }, + ); - 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; + void (async () => { + const stderrLines: string[] = []; + const result = await resticBackupMock( + fromPartial({ + signal: request.signal, + onStderr: (line: string) => { + stderrLines.push(line); + }, + }), + ); + const running = runningBackups.get(request.scheduleId); + if (!running || running.cancelled) { + return; } - } - 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; + runningBackups.delete(request.scheduleId); - handlers.onBackupFailed?.({ - agentId: "local", - agentName: "local", - payload: { - jobId: payload.jobId, - scheduleId: payload.scheduleId, - error: result.error || `Backup failed with code ${result.exitCode}`, - errorDetails, - }, - }); - } + 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; + } + } - runningJobs.delete(payload.jobId); - })().catch(() => {}); + resolve({ + status: "completed", + exitCode: result.exitCode, + result: fromAny(parsedResult), + warningDetails: stderrLines.join("\n") || null, + }); + return; + } - return true; - }); + const resultWithStderr = result as typeof result & { stderr?: string }; + resolve({ + status: "failed", + error: stderrLines.join("\n") || resultWithStderr.stderr || result.error, + }); + })().catch(() => {}); + }); + }, + ); - const cancelBackupMock = vi.fn((_agentId: string, payload: { jobId: string; scheduleId: string }) => { - const running = runningJobs.get(payload.jobId); + const cancelBackupMock = vi.fn(async (_agentId: string, scheduleId: number) => { + const running = runningBackups.get(scheduleId); 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); + runningBackups.delete(scheduleId); + running.resolve({ status: "cancelled" }); return true; }); - return { sendBackupMock, cancelBackupMock }; + return { runBackupMock, cancelBackupMock }; }; diff --git a/apps/agent/tsconfig.json b/apps/agent/tsconfig.json index 5e4c4d18..4ac5f0a9 100644 --- a/apps/agent/tsconfig.json +++ b/apps/agent/tsconfig.json @@ -2,12 +2,12 @@ "compilerOptions": { "plugins": [{ "name": "@effect/language-service" }], // Environment setup & latest features - "lib": ["DOM", "ESNext"], + "lib": ["ESNext"], "target": "ESNext", "module": "Preserve", "moduleDetection": "force", "jsx": "react-jsx", - "types": ["node", "bun"], + "types": ["bun", "node"], "allowJs": true, // Bundler mode