From 5eb1d2b727f022f59c995117505dc365f54b1ca6 Mon Sep 17 00:00:00 2001 From: Nicolas Meienberger Date: Tue, 17 Mar 2026 21:13:22 +0100 Subject: [PATCH] feat: backup through agent wip --- Dockerfile | 1 + .../agents/__tests__/agents-manager.test.ts | 33 +++ app/server/modules/agents/agents-manager.ts | 109 +++++++- .../agents/controller-agent-session.ts | 58 ++++- .../__tests__/backups.execution.test.ts | 94 +++++++ .../backups/__tests__/backups.service.test.ts | 96 ++++++- app/server/modules/backups/backup-executor.ts | 242 ++++++++++++++++-- app/server/modules/lifecycle/bootstrap.ts | 2 +- apps/agent/src/controller-session.ts | 177 ++++++++++++- packages/contracts/src/agent-protocol.ts | 131 +++++++++- 10 files changed, 880 insertions(+), 63 deletions(-) create mode 100644 app/server/modules/agents/__tests__/agents-manager.test.ts diff --git a/Dockerfile b/Dockerfile index 2feb66be..e2be2862 100644 --- a/Dockerfile +++ b/Dockerfile @@ -91,6 +91,7 @@ RUN bun install --frozen-lockfile COPY . . RUN bun run build +RUN bun build apps/agent/src/index.ts --outfile .output/agent/index.mjs --target bun FROM base AS production diff --git a/app/server/modules/agents/__tests__/agents-manager.test.ts b/app/server/modules/agents/__tests__/agents-manager.test.ts new file mode 100644 index 00000000..c8dc4af6 --- /dev/null +++ b/app/server/modules/agents/__tests__/agents-manager.test.ts @@ -0,0 +1,33 @@ +import { afterEach, describe, expect, mock, test } from "bun:test"; + +const loadAgentsManagerModule = async () => { + const moduleUrl = new URL("../agents-manager.ts", import.meta.url); + moduleUrl.searchParams.set("test", crypto.randomUUID()); + return import(moduleUrl.href); +}; + +afterEach(() => { + delete (globalThis as Record).__agentManager; + mock.restore(); +}); + +describe("agents-manager module", () => { + test("reuses the existing global agent manager without stopping it", async () => { + const existingManager = { + start: mock(() => {}), + sendBackup: mock(() => false), + cancelBackup: mock(() => false), + setBackupEventHandlers: mock(() => {}), + getBackupEventHandlers: mock(() => ({})), + stop: mock(() => {}), + }; + + (globalThis as Record).__agentManager = existingManager; + + const { agentManager } = await loadAgentsManagerModule(); + + expect(agentManager).toBe(existingManager); + expect(existingManager.stop).not.toHaveBeenCalled(); + expect((globalThis as Record).__agentManager).toBe(existingManager); + }); +}); diff --git a/app/server/modules/agents/agents-manager.ts b/app/server/modules/agents/agents-manager.ts index c5f87fef..67ee95ce 100644 --- a/app/server/modules/agents/agents-manager.ts +++ b/app/server/modules/agents/agents-manager.ts @@ -1,14 +1,20 @@ import { type ChildProcess, spawn } from "node:child_process"; +import { existsSync } from "node:fs"; import path from "node:path"; import { Effect, Exit, Ref, Scope } from "effect"; import { logger } from "@zerobyte/core/node"; +import type { + BackupCancelPayload, + BackupCancelledPayload, + BackupCompletedPayload, + BackupFailedPayload, + BackupProgressPayload, + BackupRunPayload, + BackupStartedPayload, +} from "@zerobyte/contracts/agent-protocol"; import { config } from "../../core/config"; import { validateAgentToken, deriveLocalAgentToken } from "./agent-tokens"; -import { - createControllerAgentSession, - type BackupDispatchPayload, - type ControllerAgentSession, -} from "./controller-agent-session"; +import { createControllerAgentSession, type ControllerAgentSession } from "./controller-agent-session"; type AgentConnectionData = { id: string; @@ -17,19 +23,45 @@ type AgentConnectionData = { agentName: string; }; +type AgentBackupEventContext = { + agentId: string; + agentName: string; + payload: + | BackupStartedPayload + | BackupProgressPayload + | BackupCompletedPayload + | BackupFailedPayload + | BackupCancelledPayload; +}; + +export type AgentBackupEventHandlers = { + onBackupStarted?: (context: AgentBackupEventContext & { payload: BackupStartedPayload }) => void; + onBackupProgress?: (context: AgentBackupEventContext & { payload: BackupProgressPayload }) => void; + onBackupCompleted?: (context: AgentBackupEventContext & { payload: BackupCompletedPayload }) => void; + onBackupFailed?: (context: AgentBackupEventContext & { payload: BackupFailedPayload }) => void; + onBackupCancelled?: (context: AgentBackupEventContext & { payload: BackupCancelledPayload }) => void; +}; + export const spawnLocalAgent = async () => { const previousAgent = (globalThis as Record).__localAgent as ChildProcess | undefined; if (previousAgent) { previousAgent.kill(); } - const agentEntryPoint = path.join(process.cwd(), "apps", "agent", "src", "index.ts"); + const sourceEntryPoint = path.join(process.cwd(), "apps", "agent", "src", "index.ts"); + const productionEntryPoint = path.join(process.cwd(), ".output", "agent", "index.mjs"); + + if (config.__prod__ && !existsSync(productionEntryPoint)) { + throw new Error(`Local agent entrypoint not found at ${productionEntryPoint}`); + } + + const agentEntryPoint = config.__prod__ ? productionEntryPoint : sourceEntryPoint; const agentToken = await deriveLocalAgentToken(); const args = config.__prod__ ? ["run", agentEntryPoint] : ["run", "--watch", agentEntryPoint]; const localAgent = spawn("bun", args, { env: { - PATH: process.env.PATH, + ...process.env, ZEROBYTE_CONTROLLER_URL: "ws://localhost:3001", ZEROBYTE_AGENT_TOKEN: agentToken, }, @@ -55,12 +87,17 @@ export const spawnLocalAgent = async () => { const createAgentManagerRuntime = () => { const sessionsRef = Effect.runSync(Ref.make>(new Map())); + const backupHandlersRef = Effect.runSync(Ref.make({})); let runtimeScope: Scope.CloseableScope | null = null; const getSessions = () => Effect.runSync(Ref.get(sessionsRef)); + const getBackupHandlers = () => Effect.runSync(Ref.get(backupHandlersRef)); const setSessions = (sessions: Map) => { Effect.runSync(Ref.set(sessionsRef, sessions)); }; + const setBackupHandlers = (handlers: AgentBackupEventHandlers) => { + Effect.runSync(Ref.set(backupHandlersRef, handlers)); + }; const closeAllSessions = () => { const sessions = getSessions(); @@ -95,6 +132,26 @@ const createAgentManagerRuntime = () => { setSessions(nextSessions); }; + const handleBackupStarted = (ws: Bun.ServerWebSocket, payload: BackupStartedPayload) => { + getBackupHandlers().onBackupStarted?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload }); + }; + + const handleBackupProgress = (ws: Bun.ServerWebSocket, payload: BackupProgressPayload) => { + getBackupHandlers().onBackupProgress?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload }); + }; + + const handleBackupCompleted = (ws: Bun.ServerWebSocket, payload: BackupCompletedPayload) => { + getBackupHandlers().onBackupCompleted?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload }); + }; + + const handleBackupFailed = (ws: Bun.ServerWebSocket, payload: BackupFailedPayload) => { + getBackupHandlers().onBackupFailed?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload }); + }; + + const handleBackupCancelled = (ws: Bun.ServerWebSocket, payload: BackupCancelledPayload) => { + getBackupHandlers().onBackupCancelled?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload }); + }; + const acquireServer = Effect.acquireRelease( Effect.sync(() => Bun.serve({ @@ -125,7 +182,16 @@ const createAgentManagerRuntime = () => { }, websocket: { open: (ws) => { - setSession(ws.data.agentId, createControllerAgentSession(ws)); + setSession( + ws.data.agentId, + createControllerAgentSession(ws, { + onBackupStarted: (payload) => handleBackupStarted(ws, payload), + onBackupProgress: (payload) => handleBackupProgress(ws, payload), + onBackupCompleted: (payload) => handleBackupCompleted(ws, payload), + onBackupFailed: (payload) => handleBackupFailed(ws, payload), + onBackupCancelled: (payload) => handleBackupCancelled(ws, payload), + }), + ); logger.info(`Agent "${ws.data.agentName}" (${ws.data.agentId}) connected on ${ws.data.id}`); }, message: (ws, data) => { @@ -187,7 +253,7 @@ const createAgentManagerRuntime = () => { return { start, - sendBackup: (agentId: string, payload: BackupDispatchPayload) => { + sendBackup: (agentId: string, payload: BackupRunPayload) => { const session = getSession(agentId); if (!session) { @@ -195,10 +261,31 @@ const createAgentManagerRuntime = () => { return false; } + if (!session.isReady()) { + logger.warn(`Cannot send backup command. Agent ${agentId} is not ready.`); + return false; + } + const jobId = session.sendBackup(payload); logger.info(`Sent backup command ${jobId} to agent ${agentId} for schedule ${payload.scheduleId}`); return true; }, + cancelBackup: (agentId: string, payload: BackupCancelPayload) => { + const session = getSession(agentId); + + if (!session) { + logger.warn(`Cannot cancel backup command. Agent ${agentId} is not connected.`); + return false; + } + + session.sendBackupCancel(payload); + logger.info(`Sent backup cancel for command ${payload.jobId} to agent ${agentId}`); + return true; + }, + setBackupEventHandlers: (handlers: AgentBackupEventHandlers) => { + setBackupHandlers(handlers); + }, + getBackupEventHandlers: () => getBackupHandlers(), stop, }; }; @@ -207,7 +294,5 @@ const previous = (globalThis as Record).__agentManager as | ReturnType | undefined; -previous?.stop(); - -export const agentManager = createAgentManagerRuntime(); +export const agentManager = previous ?? createAgentManagerRuntime(); (globalThis as Record).__agentManager = agentManager; diff --git a/app/server/modules/agents/controller-agent-session.ts b/app/server/modules/agents/controller-agent-session.ts index 5e7cc723..6ec0e0e4 100644 --- a/app/server/modules/agents/controller-agent-session.ts +++ b/app/server/modules/agents/controller-agent-session.ts @@ -3,6 +3,13 @@ import { createControllerMessage, parseAgentMessage, type AgentMessage, + type BackupCancelledPayload, + type BackupCompletedPayload, + type BackupFailedPayload, + type BackupProgressPayload, + type BackupRunPayload, + type BackupCancelPayload, + type BackupStartedPayload, type ControllerWireMessage, } from "@zerobyte/contracts/agent-protocol"; import { logger } from "@zerobyte/core/node"; @@ -30,20 +37,27 @@ const toMessage = (error: unknown) => { return String(error); }; -export type BackupDispatchPayload = { - scheduleId: string; - jobId?: string; +type ControllerAgentSessionHandlers = { + onBackupStarted?: (payload: BackupStartedPayload) => void; + onBackupProgress?: (payload: BackupProgressPayload) => void; + onBackupCompleted?: (payload: BackupCompletedPayload) => void; + onBackupFailed?: (payload: BackupFailedPayload) => void; + onBackupCancelled?: (payload: BackupCancelledPayload) => void; }; export type ControllerAgentSession = { readonly connectionId: string; handleMessage: (data: string) => void; - sendBackup: (payload: BackupDispatchPayload) => string; + sendBackup: (payload: BackupRunPayload) => string; + sendBackupCancel: (payload: BackupCancelPayload) => void; isReady: () => boolean; close: () => void; }; -export const createControllerAgentSession = (socket: AgentSocket): ControllerAgentSession => { +export const createControllerAgentSession = ( + socket: AgentSocket, + handlers: ControllerAgentSessionHandlers = {}, +): ControllerAgentSession => { const outboundQueue = Effect.runSync(Queue.bounded(64)); const state = Effect.runSync( Ref.make({ @@ -106,6 +120,27 @@ export const createControllerAgentSession = (socket: AgentSocket): ControllerAge logger.info( `Backup ${message.payload.jobId} started on agent ${socket.data.agentId} for schedule ${message.payload.scheduleId}`, ); + handlers.onBackupStarted?.(message.payload); + break; + } + case "backup.progress": { + updateState((current) => ({ ...current, lastSeenAt: Date.now() })); + handlers.onBackupProgress?.(message.payload); + break; + } + case "backup.completed": { + updateState((current) => ({ ...current, lastSeenAt: Date.now() })); + handlers.onBackupCompleted?.(message.payload); + break; + } + case "backup.failed": { + updateState((current) => ({ ...current, lastSeenAt: Date.now() })); + handlers.onBackupFailed?.(message.payload); + break; + } + case "backup.cancelled": { + updateState((current) => ({ ...current, lastSeenAt: Date.now() })); + handlers.onBackupCancelled?.(message.payload); break; } case "heartbeat.pong": { @@ -137,14 +172,19 @@ export const createControllerAgentSession = (socket: AgentSocket): ControllerAge handleAgentMessage(parsed.data); }, sendBackup: (payload) => { - const jobId = payload.jobId ?? Bun.randomUUIDv7(); offerOutbound( createControllerMessage("backup.run", { - jobId, - scheduleId: payload.scheduleId, + ...payload, + }), + ); + return payload.jobId; + }, + sendBackupCancel: (payload) => { + offerOutbound( + createControllerMessage("backup.cancel", { + ...payload, }), ); - return jobId; }, isReady: () => Effect.runSync(Ref.get(state)).isReady, close: () => { diff --git a/app/server/modules/backups/__tests__/backups.execution.test.ts b/app/server/modules/backups/__tests__/backups.execution.test.ts index feb444ee..efd8a660 100644 --- a/app/server/modules/backups/__tests__/backups.execution.test.ts +++ b/app/server/modules/backups/__tests__/backups.execution.test.ts @@ -15,6 +15,8 @@ import { restic } from "~/server/core/restic"; import { NotFoundError, BadRequestError } from "http-errors-enhanced"; import { repositoriesService } from "~/server/modules/repositories/repositories.service"; import { repoMutex } from "~/server/core/repository-mutex"; +import { agentManager } from "~/server/modules/agents/agents-manager"; +import { fromAny } from "@total-typescript/shoehorn"; const setup = () => { const resticBackupMock = vi.fn((_: SafeSpawnParams) => @@ -22,6 +24,94 @@ const setup = () => { ); const resticForgetMock = vi.fn(() => Promise.resolve({ success: true, data: null })); const resticCopyMock = vi.fn(() => Promise.resolve({ success: true, output: "" })); + 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); + }, + } satisfies Partial), + ); + 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; + }); const refreshStatsMock = vi.fn(() => Promise.resolve({ total_size: 0, @@ -38,11 +128,15 @@ const setup = () => { vi.spyOn(restic, "copy").mockImplementation(resticCopyMock); vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock); vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID); + vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock); + vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock); return { resticBackupMock, resticForgetMock, resticCopyMock, + sendBackupMock, + cancelBackupMock, refreshStatsMock, }; }; diff --git a/app/server/modules/backups/__tests__/backups.service.test.ts b/app/server/modules/backups/__tests__/backups.service.test.ts index 357e5051..32b31d97 100644 --- a/app/server/modules/backups/__tests__/backups.service.test.ts +++ b/app/server/modules/backups/__tests__/backups.service.test.ts @@ -14,9 +14,99 @@ 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 { fromAny } from "@total-typescript/shoehorn"; const setup = () => { - const resticBackupMock = vi.fn(() => Promise.resolve({ exitCode: 0, summary: "", error: "" })); + const resticBackupMock = vi.fn((_: unknown) => Promise.resolve({ exitCode: 0, summary: "", error: "" })); + 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; + }); const refreshStatsMock = vi.fn(() => Promise.resolve({ total_size: 0, @@ -30,9 +120,13 @@ const setup = () => { vi.spyOn(spawnModule, "safeSpawn").mockImplementation(resticBackupMock); vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock); vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID); + vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock); + vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock); return { resticBackupMock, + sendBackupMock, + cancelBackupMock, refreshStatsMock, }; }; diff --git a/app/server/modules/backups/backup-executor.ts b/app/server/modules/backups/backup-executor.ts index fa45a472..e7984421 100644 --- a/app/server/modules/backups/backup-executor.ts +++ b/app/server/modules/backups/backup-executor.ts @@ -1,8 +1,11 @@ -import { restic } from "../../core/restic"; +import type { BackupRunPayload } from "@zerobyte/contracts/agent-protocol"; +import { agentManager } from "../agents/agents-manager"; +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 { decryptRepositoryConfig } from "../repositories/repository-config-secrets"; type BackupExecutionRequest = { scheduleId: number; @@ -36,52 +39,239 @@ export type BackupExecutionResult = message?: string; }; -const activeControllersByScheduleId = new Map(); +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 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 sourcePath = getVolumePath(volume); + const { signal: _ignoredSignal, ...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: { + tags: options.tags, + oneFileSystem: options.oneFileSystem, + exclude: options.exclude, + excludeIfPresent: options.excludeIfPresent, + includePaths: options.includePaths, + includePatterns: options.includePatterns, + customResticParams: options.customResticParams, + compressionMode: repository.compressionMode ?? "auto", + }, + runtime: { + password: resticPassword, + cacheDir: resticDeps.resticCacheDir, + passFile: resticDeps.resticPassFile, + defaultExcludes: resticDeps.defaultExcludes, + hostname: resticDeps.hostname, + }, + }; +}; + +agentManager.setBackupEventHandlers({ + onBackupProgress: ({ payload }) => { + const running = getActiveBackupByJobId(payload.jobId); + if (!running || running.activeBackup.scheduleShortId !== payload.scheduleId) { + return; + } + + running.activeBackup.onProgress?.(payload.progress); + }, + onBackupCompleted: ({ payload }) => { + const running = getActiveBackupByJobId(payload.jobId); + if (!running || running.activeBackup.scheduleShortId !== payload.scheduleId) { + return; + } + + activeScheduleIdsByJobId.delete(payload.jobId); + resolveActiveBackup(running.activeBackup, { + 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) { + return; + } + + activeScheduleIdsByJobId.delete(payload.jobId); + resolveActiveBackup(running.activeBackup, { + status: "failed", + error: payload.errorDetails ?? payload.error, + }); + }, + onBackupCancelled: ({ payload }) => { + const running = getActiveBackupByJobId(payload.jobId); + if (!running || running.activeBackup.scheduleShortId !== payload.scheduleId) { + 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, { + status: "cancelled", + message: payload.message, + }); + }, +}); export const backupExecutor = { track: (scheduleId: number) => { const abortController = new AbortController(); - activeControllersByScheduleId.set(scheduleId, abortController); + activeBackupsByScheduleId.set(scheduleId, { + abortController, + settled: false, + }); return abortController; }, untrack: (scheduleId: number, abortController: AbortController) => { - if (activeControllersByScheduleId.get(scheduleId) === abortController) { - activeControllersByScheduleId.delete(scheduleId); + const activeBackup = activeBackupsByScheduleId.get(scheduleId); + if (activeBackup?.abortController === abortController) { + if (activeBackup.jobId) { + activeScheduleIdsByJobId.delete(activeBackup.jobId); + } + activeBackupsByScheduleId.delete(scheduleId); } }, execute: async (params: BackupExecutionRequest): Promise => { - const { schedule, volume, repository, organizationId, signal, onProgress } = params; - try { - const volumePath = getVolumePath(volume); - const backupOptions = createBackupOptions(schedule, volumePath, signal); + const activeBackup = activeBackupsByScheduleId.get(params.scheduleId); + if (!activeBackup) { + throw new Error(`Backup ${params.scheduleId} is not tracked`); + } - const result = await restic.backup(repository.config, volumePath, { - ...backupOptions, - compressionMode: repository.compressionMode ?? "auto", - organizationId, - onProgress, + if (params.signal.aborted) { + throw getCancellationError(params.signal); + } + + const jobId = Bun.randomUUIDv7(); + const payload = await buildAgentBackupPayload(params, jobId); + + if (params.signal.aborted) { + throw getCancellationError(params.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 handleAbort = () => { + activeScheduleIdsByJobId.delete(jobId); + agentManager.cancelBackup(LOCAL_AGENT_ID, { + jobId, + scheduleId: params.schedule.shortId, }); + rejectActiveBackup(activeBackup, getCancellationError(params.signal)); + }; + params.signal.addEventListener("abort", handleAbort, { once: true }); + activeScheduleIdsByJobId.set(jobId, params.scheduleId); + + const dispatched = agentManager.sendBackup(LOCAL_AGENT_ID, payload); + if (!dispatched) { + params.signal.removeEventListener("abort", handleAbort); + activeScheduleIdsByJobId.delete(jobId); + resetPendingExecution(activeBackup); return { - status: "completed", - exitCode: result.exitCode, - result: result.result, - warningDetails: result.warningDetails, - } satisfies BackupExecutionResult; - } catch (error) { - return { - status: "failed", - error, + status: "unavailable", + error: new Error("Local backup agent is not connected"), } satisfies BackupExecutionResult; } + + try { + return await completion; + } finally { + params.signal.removeEventListener("abort", handleAbort); + } }, cancel: (scheduleId: number) => { - const abortController = activeControllersByScheduleId.get(scheduleId); - if (!abortController) { + const activeBackup = activeBackupsByScheduleId.get(scheduleId); + if (!activeBackup) { return false; } - abortController.abort(); + activeBackup.abortController.abort(new Error("Backup was stopped by the user")); return true; }, }; diff --git a/app/server/modules/lifecycle/bootstrap.ts b/app/server/modules/lifecycle/bootstrap.ts index f41fce78..0866be58 100644 --- a/app/server/modules/lifecycle/bootstrap.ts +++ b/app/server/modules/lifecycle/bootstrap.ts @@ -8,9 +8,9 @@ let bootstrapPromise: Promise | undefined; const runBootstrap = async () => { await runDbMigrations(); await runMigrations(); - await startup(); agentManager.start(); await spawnLocalAgent(); + await startup(); }; export const bootstrapApplication = async () => { diff --git a/apps/agent/src/controller-session.ts b/apps/agent/src/controller-session.ts index 50c8e1a3..f64462e3 100644 --- a/apps/agent/src/controller-session.ts +++ b/apps/agent/src/controller-session.ts @@ -1,11 +1,15 @@ -import { Effect, Fiber, Queue } from "effect"; +import { Effect, Fiber, Queue, Ref } from "effect"; import { createAgentMessage, parseControllerMessage, + type BackupRunPayload, type AgentWireMessage, + type BackupCancelPayload, type ControllerWireMessage, } from "@zerobyte/contracts/agent-protocol"; import { logger } from "@zerobyte/core/node"; +import { ResticError, type ResticDeps } from "@zerobyte/core/restic"; +import { createRestic } from "@zerobyte/core/restic/server"; const toMessage = (error: unknown) => { if (error instanceof Error) { @@ -15,6 +19,14 @@ const toMessage = (error: unknown) => { return String(error); }; +const toErrorDetails = (error: unknown) => { + if (error instanceof ResticError) { + return error.details || error.summary; + } + + return toMessage(error); +}; + export type ControllerSession = { onOpen: () => void; onMessage: (data: unknown) => void; @@ -24,6 +36,31 @@ export type ControllerSession = { export const createControllerSession = (ws: WebSocket): ControllerSession => { const outboundQueue = Effect.runSync(Queue.bounded(64)); const inboundQueue = Effect.runSync(Queue.bounded(64)); + const runningJobsRef = Effect.runSync( + Ref.make>(new Map()), + ); + + const getRunningJob = (jobId: string) => Effect.runSync(Ref.get(runningJobsRef)).get(jobId); + + const setRunningJob = (jobId: string, job: { scheduleId: string; abortController: AbortController }) => { + Effect.runSync( + Ref.update(runningJobsRef, (current) => { + const next = new Map(current); + next.set(jobId, job); + return next; + }), + ); + }; + + const deleteRunningJob = (jobId: string) => { + Effect.runSync( + Ref.update(runningJobsRef, (current) => { + const next = new Map(current); + next.delete(jobId); + return next; + }), + ); + }; const offerOutbound = (message: AgentWireMessage) => { void Effect.runPromise(Queue.offer(outboundQueue, message)).catch((error) => { @@ -70,14 +107,131 @@ export const createControllerSession = (ws: WebSocket): ControllerSession => { switch (parsed.data.type) { case "backup.run": { - logger.info(`Starting backup ${parsed.data.payload.jobId} for schedule ${parsed.data.payload.scheduleId}`); - yield* Queue.offer( - outboundQueue, - createAgentMessage("backup.started", { - jobId: parsed.data.payload.jobId, - scheduleId: parsed.data.payload.scheduleId, - }), - ); + const runBackup = async (payload: BackupRunPayload) => { + const existing = getRunningJob(payload.jobId); + if (existing) { + offerOutbound( + createAgentMessage("backup.failed", { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + error: "Backup job is already running", + }), + ); + return; + } + + logger.info(`Starting backup ${payload.jobId} for schedule ${payload.scheduleId}`); + const abortController = new AbortController(); + setRunningJob(payload.jobId, { scheduleId: payload.scheduleId, abortController }); + + offerOutbound( + createAgentMessage("backup.started", { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + }), + ); + + const deps: ResticDeps = { + resolveSecret: async (encrypted) => encrypted, + getOrganizationResticPassword: async () => payload.runtime.password, + resticCacheDir: payload.runtime.cacheDir, + resticPassFile: payload.runtime.passFile, + defaultExcludes: payload.runtime.defaultExcludes, + hostname: payload.runtime.hostname, + }; + + const restic = createRestic(deps); + + try { + const result = await restic.backup(payload.repositoryConfig, payload.sourcePath, { + organizationId: payload.organizationId, + tags: payload.options.tags, + oneFileSystem: payload.options.oneFileSystem, + exclude: payload.options.exclude, + excludeIfPresent: payload.options.excludeIfPresent, + includePaths: payload.options.includePaths, + includePatterns: payload.options.includePatterns, + customResticParams: payload.options.customResticParams, + compressionMode: payload.options.compressionMode, + signal: abortController.signal, + onProgress: (progress) => { + offerOutbound( + createAgentMessage("backup.progress", { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + progress, + }), + ); + }, + }); + + if (abortController.signal.aborted) { + offerOutbound( + createAgentMessage("backup.cancelled", { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + message: "Backup was cancelled", + }), + ); + return; + } + + offerOutbound( + createAgentMessage("backup.completed", { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + exitCode: result.exitCode, + result: result.result, + warningDetails: result.warningDetails ?? undefined, + }), + ); + } catch (error) { + if (abortController.signal.aborted) { + offerOutbound( + createAgentMessage("backup.cancelled", { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + message: "Backup was cancelled", + }), + ); + return; + } + + offerOutbound( + createAgentMessage("backup.failed", { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + error: toMessage(error), + errorDetails: toErrorDetails(error), + }), + ); + } finally { + deleteRunningJob(payload.jobId); + } + }; + + void runBackup(parsed.data.payload); + break; + } + case "backup.cancel": { + const cancelBackup = (payload: BackupCancelPayload) => { + const running = getRunningJob(payload.jobId); + if (!running) { + logger.warn(`Backup ${payload.jobId} is not running`); + return; + } + + if (running.scheduleId !== payload.scheduleId) { + logger.warn( + `Ignoring cancel for backup ${payload.jobId} due to schedule mismatch ${payload.scheduleId}`, + ); + return; + } + + running.abortController.abort(); + }; + + cancelBackup(parsed.data.payload); break; } case "heartbeat.ping": { @@ -107,6 +261,11 @@ export const createControllerSession = (ws: WebSocket): ControllerSession => { offerInbound(data as ControllerWireMessage); }, close: () => { + const runningJobs = Effect.runSync(Ref.get(runningJobsRef)); + for (const running of runningJobs.values()) { + running.abortController.abort(); + } + Effect.runSync(Ref.set(runningJobsRef, new Map())); void Effect.runPromise(Fiber.interrupt(writerFiber)).catch(() => {}); void Effect.runPromise(Fiber.interrupt(processorFiber)).catch(() => {}); void Effect.runPromise(Queue.shutdown(outboundQueue)).catch(() => {}); diff --git a/packages/contracts/src/agent-protocol.ts b/packages/contracts/src/agent-protocol.ts index 231b9e99..4e98a091 100644 --- a/packages/contracts/src/agent-protocol.ts +++ b/packages/contracts/src/agent-protocol.ts @@ -1,10 +1,58 @@ import { z } from "zod"; import { safeJsonParse } from "@zerobyte/core/utils"; +import { + repositoryConfigSchema, + resticBackupOutputSchema, + resticBackupProgressSchema, + type CompressionMode, +} from "@zerobyte/core/restic"; -const backupCommandSchema = z +const compressionModeSchema = z.enum(["off", "auto", "max"]) satisfies z.ZodType; + +const backupExecutionOptionsSchema = z + .object({ + tags: z.array(z.string()).optional(), + oneFileSystem: z.boolean().optional(), + exclude: z.array(z.string()).optional(), + excludeIfPresent: z.array(z.string()).optional(), + includePaths: z.array(z.string()).optional(), + includePatterns: z.array(z.string()).optional(), + customResticParams: z.array(z.string()).optional(), + compressionMode: compressionModeSchema.optional(), + }) + .strict(); + +const backupRuntimeSchema = z + .object({ + password: z.string(), + cacheDir: z.string(), + passFile: z.string(), + defaultExcludes: z.array(z.string()), + hostname: z.string().optional(), + }) + .strict(); + +const backupRunSchema = z .object({ type: z.literal("backup.run"), - payload: z.object({ jobId: z.string(), scheduleId: z.string() }), + payload: z + .object({ + jobId: z.string(), + scheduleId: z.string(), + organizationId: z.string(), + sourcePath: z.string(), + repositoryConfig: repositoryConfigSchema, + options: backupExecutionOptionsSchema, + runtime: backupRuntimeSchema, + }) + .strict(), + }) + .strict(); + +const backupCancelSchema = z + .object({ + type: z.literal("backup.cancel"), + payload: z.object({ jobId: z.string(), scheduleId: z.string() }).strict(), }) .strict(); @@ -29,6 +77,61 @@ const backupStartedSchema = z }) .strict(); +const backupProgressSchema = z + .object({ + type: z.literal("backup.progress"), + payload: z + .object({ + jobId: z.string(), + scheduleId: z.string(), + progress: resticBackupProgressSchema, + }) + .strict(), + }) + .strict(); + +const backupCompletedSchema = z + .object({ + type: z.literal("backup.completed"), + payload: z + .object({ + jobId: z.string(), + scheduleId: z.string(), + exitCode: z.number(), + result: resticBackupOutputSchema.nullable(), + warningDetails: z.string().optional(), + }) + .strict(), + }) + .strict(); + +const backupFailedSchema = z + .object({ + type: z.literal("backup.failed"), + payload: z + .object({ + jobId: z.string(), + scheduleId: z.string(), + error: z.string(), + errorDetails: z.string().optional(), + }) + .strict(), + }) + .strict(); + +const backupCancelledSchema = z + .object({ + type: z.literal("backup.cancelled"), + payload: z + .object({ + jobId: z.string(), + scheduleId: z.string(), + message: z.string().optional(), + }) + .strict(), + }) + .strict(); + const heartbeatPongSchema = z .object({ type: z.literal("heartbeat.pong"), @@ -36,10 +139,28 @@ const heartbeatPongSchema = z }) .strict(); -const controllerMessageSchema = z.discriminatedUnion("type", [backupCommandSchema, heartbeatPingSchema]); -const agentMessageSchema = z.discriminatedUnion("type", [agentReadySchema, backupStartedSchema, heartbeatPongSchema]); +const controllerMessageSchema = z.discriminatedUnion("type", [ + backupRunSchema, + backupCancelSchema, + heartbeatPingSchema, +]); +const agentMessageSchema = z.discriminatedUnion("type", [ + agentReadySchema, + backupStartedSchema, + backupProgressSchema, + backupCompletedSchema, + backupFailedSchema, + backupCancelledSchema, + heartbeatPongSchema, +]); -export type BackupCommandPayload = z.infer["payload"]; +export type BackupRunPayload = z.infer["payload"]; +export type BackupCancelPayload = z.infer["payload"]; +export type BackupStartedPayload = z.infer["payload"]; +export type BackupProgressPayload = z.infer["payload"]; +export type BackupCompletedPayload = z.infer["payload"]; +export type BackupFailedPayload = z.infer["payload"]; +export type BackupCancelledPayload = z.infer["payload"]; export type ControllerMessage = z.infer; export type AgentMessage = z.infer;