From e2498ff6381c03643c43da2d118ea2bdf69111e5 Mon Sep 17 00:00:00 2001 From: Nicolas Meienberger Date: Sat, 9 May 2026 11:00:27 +0200 Subject: [PATCH] refactor(agent): create dedicated jobs for recurring tasks --- app/server/modules/volumes/volume.service.ts | 5 ++--- apps/agent/src/index.ts | 21 ++++++------------ apps/agent/src/jobs/cleanup-dangling.ts | 21 ++++++++++++++++++ apps/agent/src/jobs/index.ts | 23 ++++++++++++++++++++ apps/agent/src/jobs/types.ts | 7 ++++++ 5 files changed, 60 insertions(+), 17 deletions(-) create mode 100644 apps/agent/src/jobs/cleanup-dangling.ts create mode 100644 apps/agent/src/jobs/index.ts create mode 100644 apps/agent/src/jobs/types.ts diff --git a/app/server/modules/volumes/volume.service.ts b/app/server/modules/volumes/volume.service.ts index 44cc3e60..7eaf091b 100644 --- a/app/server/modules/volumes/volume.service.ts +++ b/app/server/modules/volumes/volume.service.ts @@ -23,7 +23,6 @@ import { getStatFs, getVolumePath, type AgentVolume, - type BackendConfig as HostBackendConfig, } from "../../../../apps/agent/src/volume-host"; import { browseFilesystem as browseHostFilesystem, @@ -72,7 +71,7 @@ const volumeForAgent = async (volume: Volume): Promise => ({ const volumeForHost = async (volume: Volume): Promise => ({ ...volume, shortId: volume.shortId, - config: (await decryptVolumeConfig(volume.config)) as HostBackendConfig, + config: await decryptVolumeConfig(volume.config), provisioningId: volume.provisioningId ?? null, }); @@ -290,7 +289,7 @@ const updateVolume = async (shortId: ShortId, volumeData: UpdateVolumeBody) => { const testConnection = async (backendConfig: BackendConfig) => { if (!config.flags.enableLocalAgent) { - return Effect.runPromise(testVolumeConnection(backendConfig as HostBackendConfig)); + return Effect.runPromise(testVolumeConnection(backendConfig)); } const command = await runVolumeCommand(LOCAL_AGENT_ID, { name: "volume.testConnection", backendConfig }); diff --git a/apps/agent/src/index.ts b/apps/agent/src/index.ts index 4f6b210e..9196337a 100644 --- a/apps/agent/src/index.ts +++ b/apps/agent/src/index.ts @@ -1,31 +1,24 @@ import { logger } from "@zerobyte/core/node"; +import { Fiber } from "effect"; import { createControllerSession, type ControllerSession } from "./controller-session"; -import { cleanupDanglingVolumeMountDirectories } from "./volume-host/cleanup"; +import { startAgentJobs } from "./jobs"; const controllerUrl = process.env.ZEROBYTE_CONTROLLER_URL; const agentToken = process.env.ZEROBYTE_AGENT_TOKEN; const RECONNECT_DELAY_MS = 1000; -const VOLUME_CLEANUP_INTERVAL_MS = 60 * 60 * 1000; export class Agent { private ws: WebSocket | null = null; private controllerSession: ControllerSession | null = null; private reconnectTimeout: ReturnType | null = null; - private volumeCleanupInterval: ReturnType | null = null; + private jobFibers: Fiber.RuntimeFiber[] | null = null; - private runVolumeCleanup() { - void cleanupDanglingVolumeMountDirectories().catch((error) => { - logger.warn(`Agent volume cleanup failed: ${error instanceof Error ? error.message : String(error)}`); - }); - } - - private startVolumeCleanup() { - if (this.volumeCleanupInterval) { + private startJobs() { + if (this.jobFibers) { return; } - this.runVolumeCleanup(); - this.volumeCleanupInterval = setInterval(() => this.runVolumeCleanup(), VOLUME_CLEANUP_INTERVAL_MS); + this.jobFibers = startAgentJobs(); } private scheduleReconnect() { @@ -40,7 +33,7 @@ export class Agent { } connect() { - this.startVolumeCleanup(); + this.startJobs(); if (this.reconnectTimeout) { clearTimeout(this.reconnectTimeout); diff --git a/apps/agent/src/jobs/cleanup-dangling.ts b/apps/agent/src/jobs/cleanup-dangling.ts new file mode 100644 index 00000000..9b478451 --- /dev/null +++ b/apps/agent/src/jobs/cleanup-dangling.ts @@ -0,0 +1,21 @@ +import { logger } from "@zerobyte/core/node"; +import { toMessage } from "@zerobyte/core/utils"; +import { Effect } from "effect"; +import { cleanupDanglingVolumeMountDirectories } from "../volume-host/cleanup"; +import type { AgentJob } from "./types"; + +const VOLUME_CLEANUP_INTERVAL_MS = 60 * 60 * 1000; + +export const CleanupDanglingVolumeMountsJob: AgentJob = { + name: "cleanup-dangling-volume-mounts", + intervalMs: VOLUME_CLEANUP_INTERVAL_MS, + run: () => + Effect.tryPromise({ + try: cleanupDanglingVolumeMountDirectories, + catch: (error) => error, + }).pipe( + Effect.catchAll((error) => + Effect.sync(() => logger.warn(`Agent volume cleanup failed: ${toMessage(error)}`)), + ), + ), +}; diff --git a/apps/agent/src/jobs/index.ts b/apps/agent/src/jobs/index.ts new file mode 100644 index 00000000..d40c868d --- /dev/null +++ b/apps/agent/src/jobs/index.ts @@ -0,0 +1,23 @@ +import { logger } from "@zerobyte/core/node"; +import { Effect, Fiber } from "effect"; +import { CleanupDanglingVolumeMountsJob } from "./cleanup-dangling"; +import type { AgentJob } from "./types"; + +const agentJobs = [CleanupDanglingVolumeMountsJob]; + +export const startAgentJobs = (jobs: readonly AgentJob[] = agentJobs) => { + return jobs.map((job) => + Effect.runFork( + Effect.forever( + Effect.gen(function* () { + yield* job.run(); + yield* Effect.sleep(job.intervalMs); + }), + ).pipe(Effect.ensuring(logger.effect.debug(`Agent job stopped: ${job.name}`))), + ), + ); +}; + +export const stopAgentJobs = (fibers: readonly Fiber.RuntimeFiber[]) => { + return Effect.forEach(fibers, (fiber) => Fiber.interrupt(fiber), { discard: true }); +}; diff --git a/apps/agent/src/jobs/types.ts b/apps/agent/src/jobs/types.ts new file mode 100644 index 00000000..6421b35c --- /dev/null +++ b/apps/agent/src/jobs/types.ts @@ -0,0 +1,7 @@ +import type { Effect } from "effect"; + +export type AgentJob = { + name: string; + intervalMs: number; + run: () => Effect.Effect; +};