diff --git a/app/server/modules/backups/backup-executor.ts b/app/server/modules/backups/backup-executor.ts index 4aea9aa3..60c9918c 100644 --- a/app/server/modules/backups/backup-executor.ts +++ b/app/server/modules/backups/backup-executor.ts @@ -56,26 +56,28 @@ export const backupExecutor = { const volumePath = getVolumePath(volume); const backupOptions = createBackupOptions(schedule, volumePath, signal); - const result = await Effect.runPromise( - restic.backup(repository.config, volumePath, { - ...backupOptions, - compressionMode: repository.compressionMode ?? "auto", - organizationId, - onProgress, - }), + 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 })), + ), ); - return { - status: "completed", - exitCode: result.exitCode, - result: result.result, - warningDetails: result.warningDetails, - } satisfies BackupExecutionResult; + if (!execution.success) { + throw execution.error; + } + + const { exitCode, result, warningDetails } = execution.result; + return { status: "completed", exitCode, result, warningDetails }; } catch (error) { - return { - status: "failed", - error, - } satisfies BackupExecutionResult; + return { status: "failed", error }; } }, cancel: (scheduleId: number) => { diff --git a/apps/agent/package.json b/apps/agent/package.json index 40361ac1..c3bfdbb2 100644 --- a/apps/agent/package.json +++ b/apps/agent/package.json @@ -2,7 +2,7 @@ "name": "agent", "private": true, "type": "module", - "module": "index.ts", + "module": "src/index.ts", "scripts": { "tsc": "tsc --noEmit" }, @@ -15,6 +15,6 @@ "@types/bun": "latest" }, "peerDependencies": { - "typescript": "^5" + "typescript": "^6" } } diff --git a/apps/agent/src/commands/backup-cancel.ts b/apps/agent/src/commands/backup-cancel.ts index c36c21e9..ec09d033 100644 --- a/apps/agent/src/commands/backup-cancel.ts +++ b/apps/agent/src/commands/backup-cancel.ts @@ -3,8 +3,8 @@ import { type BackupCancelPayload } from "@zerobyte/contracts/agent-protocol"; import { logger } from "@zerobyte/core/node"; import type { ControllerCommandContext } from "../context"; -export const handleBackupCancelCommand = (context: ControllerCommandContext, payload: BackupCancelPayload) => - Effect.gen(function* () { +export const handleBackupCancelCommand = (context: ControllerCommandContext, payload: BackupCancelPayload) => { + return Effect.gen(function* () { const running = yield* context.getRunningJob(payload.jobId); if (!running) { logger.warn(`Backup ${payload.jobId} is not running`); @@ -18,3 +18,4 @@ export const handleBackupCancelCommand = (context: ControllerCommandContext, pay running.abortController.abort(); }); +}; diff --git a/apps/agent/src/commands/backup-run.ts b/apps/agent/src/commands/backup-run.ts index 0719cd4c..7af0cd62 100644 --- a/apps/agent/src/commands/backup-run.ts +++ b/apps/agent/src/commands/backup-run.ts @@ -6,8 +6,8 @@ import { createRestic } from "@zerobyte/core/restic/server"; import { toErrorDetails, toMessage } from "@zerobyte/core/utils"; import type { ControllerCommandContext } from "../context"; -export const handleBackupRunCommand = (context: ControllerCommandContext, payload: BackupRunPayload) => - Effect.fork( +export const handleBackupRunCommand = (context: ControllerCommandContext, payload: BackupRunPayload) => { + return Effect.fork( Effect.gen(function* () { const existing = yield* context.getRunningJob(payload.jobId); if (existing) { @@ -110,3 +110,4 @@ export const handleBackupRunCommand = (context: ControllerCommandContext, payloa ); }), ).pipe(Effect.asVoid); +}; diff --git a/apps/agent/src/commands/heartbeat-ping.ts b/apps/agent/src/commands/heartbeat-ping.ts index 7012141f..2030ffe5 100644 --- a/apps/agent/src/commands/heartbeat-ping.ts +++ b/apps/agent/src/commands/heartbeat-ping.ts @@ -3,9 +3,10 @@ import type { ControllerCommandContext } from "../context"; type HeartbeatPingPayload = Extract["payload"]; -export const handleHeartbeatPingCommand = (context: ControllerCommandContext, payload: HeartbeatPingPayload) => - context.offerOutbound( +export const handleHeartbeatPingCommand = (context: ControllerCommandContext, payload: HeartbeatPingPayload) => { + return context.offerOutbound( createAgentMessage("heartbeat.pong", { sentAt: payload.sentAt, }), ); +}; diff --git a/apps/agent/src/controller-session.ts b/apps/agent/src/controller-session.ts index c1b45b9d..9948d00e 100644 --- a/apps/agent/src/controller-session.ts +++ b/apps/agent/src/controller-session.ts @@ -106,11 +106,6 @@ export const createControllerSession = (ws: WebSocket): ControllerSession => { }); }, onMessage: (data) => { - if (typeof data !== "string") { - logger.warn("Agent received a non-text message"); - return; - } - void Effect.runPromise(offerInbound(data as ControllerWireMessage)).catch((error) => { logger.error(`Failed to queue inbound message: ${toMessage(error)}`); }); diff --git a/apps/agent/src/index.ts b/apps/agent/src/index.ts index 1e5b004d..1b545406 100644 --- a/apps/agent/src/index.ts +++ b/apps/agent/src/index.ts @@ -3,12 +3,29 @@ import { createControllerSession, type ControllerSession } from "./controller-se const controllerUrl = process.env.ZEROBYTE_CONTROLLER_URL; const agentToken = process.env.ZEROBYTE_AGENT_TOKEN; +const RECONNECT_DELAY_MS = 1000; class Agent { private ws: WebSocket | null = null; private controllerSession: ControllerSession | null = null; + private reconnectTimeout: ReturnType | null = null; + + private scheduleReconnect() { + if (this.reconnectTimeout) { + return; + } + + this.reconnectTimeout = setTimeout(() => { + this.reconnectTimeout = null; + this.connect(); + }, RECONNECT_DELAY_MS); + } connect() { + if (this.ws) { + return; + } + if (!controllerUrl) { throw new Error("Env variable ZEROBYTE_CONTROLLER_URL is not set"); } @@ -36,6 +53,7 @@ class Agent { this.controllerSession = null; this.ws = null; logger.info("Agent disconnected from controller"); + this.scheduleReconnect(); }; this.ws.onerror = (error) => { logger.error("Agent encountered an error:", error); @@ -43,5 +61,7 @@ class Agent { } } -const agent = new Agent(); -agent.connect(); +if (import.meta.main) { + const agent = new Agent(); + agent.connect(); +} diff --git a/bun.lock b/bun.lock index 410a1230..717d99ff 100644 --- a/bun.lock +++ b/bun.lock @@ -128,7 +128,7 @@ "@types/bun": "latest", }, "peerDependencies": { - "typescript": "^5", + "typescript": "^6", }, }, "packages/contracts": { diff --git a/packages/core/src/restic/commands/backup.ts b/packages/core/src/restic/commands/backup.ts index 88949707..e6c03119 100644 --- a/packages/core/src/restic/commands/backup.ts +++ b/packages/core/src/restic/commands/backup.ts @@ -1,7 +1,7 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import { Effect } from "effect"; +import { Data, Effect } from "effect"; import { throttle } from "es-toolkit"; import type { CompressionMode, RepositoryConfig } from "../schemas"; import { type ResticBackupProgressDto, resticBackupOutputSchema, resticBackupProgressSchema } from "../restic-dto"; @@ -13,6 +13,12 @@ import { validateCustomResticParams } from "../helpers/validate-custom-params"; import { ResticError } from "../error"; import { logger, safeSpawn } from "../../node"; import type { ResticDeps } from "../types"; +import { toMessage } from "../../utils"; + +class ResticBackupCommandError extends Data.TaggedError("ResticBackupCommandError")<{ + cause: unknown; + message: string; +}> {} export const backup = ( config: RepositoryConfig, @@ -31,8 +37,8 @@ export const backup = ( customResticParams?: string[]; }, deps: ResticDeps, -) => - Effect.tryPromise({ +) => { + return Effect.tryPromise({ try: async () => { const repoUrl = buildRepoUrl(config); const env = await buildEnv(config, options.organizationId, deps); @@ -214,5 +220,15 @@ export const backup = ( warningDetails: stderrLines.length > 0 ? stderrLines.join("\n") : null, }; }, - catch: (error) => (error instanceof Error ? error : new Error(String(error))), + catch: (error) => { + if (error instanceof ResticError) { + return error; + } + + return new ResticBackupCommandError({ + cause: error, + message: toMessage(error), + }); + }, }); +};