diff --git a/app/server/modules/agents/__tests__/agents-manager.backups.test.ts b/app/server/modules/agents/__tests__/agents-manager.backups.test.ts index 053dfed2..eaab3870 100644 --- a/app/server/modules/agents/__tests__/agents-manager.backups.test.ts +++ b/app/server/modules/agents/__tests__/agents-manager.backups.test.ts @@ -4,7 +4,7 @@ import { fromAny, fromPartial } from "@total-typescript/shoehorn"; import { Effect } from "effect"; import { agentManager, type ProcessWithAgentRuntime } from "../agents-manager"; import type { AgentManagerRuntime } from "../controller/server"; -import type { BackupRunPayload, VolumeCommand } from "@zerobyte/contracts/agent-protocol"; +import type { BackupRunPayload, VolumeCommand, VolumeCommandResponsePayload } from "@zerobyte/contracts/agent-protocol"; const setAgentRuntime = (agentManagerRuntime: Partial | null) => { (process as ProcessWithAgentRuntime).__zerobyteAgentRuntime = { @@ -53,7 +53,7 @@ test("runVolumeCommand sends the command to the selected agent", async () => { commandId: "command-1", status: "success", command: { name: "volume.mount", result: { status: "mounted" } }, - }), + } satisfies VolumeCommandResponsePayload), ); setAgentRuntime({ runVolumeCommand }); diff --git a/app/server/modules/agents/controller/session.ts b/app/server/modules/agents/controller/session.ts index 0187d112..5a68402e 100644 --- a/app/server/modules/agents/controller/session.ts +++ b/app/server/modules/agents/controller/session.ts @@ -29,8 +29,9 @@ type SessionState = { lastPongAt: number | null; }; -type InFlightVolumeCommand = { +type PendingCommand = { deferred: Deferred.Deferred; + description: string; }; export type ControllerAgentSessionEvent = @@ -56,7 +57,7 @@ export const createControllerAgentSession = ( Effect.gen(function* () { let isClosed = false; const outboundQueue = yield* Queue.bounded(64); - const inFlightVolumeCommands = yield* Ref.make(new Map()); + const pendingCommands = yield* Ref.make(new Map()); const state = yield* Ref.make({ isReady: false, lastSeenAt: null, @@ -75,25 +76,25 @@ export const createControllerAgentSession = ( const updateState = (update: (current: SessionState) => SessionState) => Ref.update(state, update); - const setInFlightVolumeCommand = (commandId: string, inFlight: InFlightVolumeCommand) => - Ref.update(inFlightVolumeCommands, (current) => new Map(current).set(commandId, inFlight)); + const setPendingCommand = (commandId: string, pending: PendingCommand) => + Ref.update(pendingCommands, (current) => new Map(current).set(commandId, pending)); - const removeInFlightVolumeCommand = (commandId: string) => - Ref.modify(inFlightVolumeCommands, (current) => { - const inFlight = current.get(commandId) ?? null; + const removePendingCommand = (commandId: string) => + Ref.modify(pendingCommands, (current) => { + const pending = current.get(commandId) ?? null; const next = new Map(current); next.delete(commandId); - return [inFlight, next]; + return [pending, next]; }); - const rejectInFlightVolumeCommands = Effect.gen(function* () { - const inFlightCommands = yield* Ref.get(inFlightVolumeCommands); - yield* Ref.set(inFlightVolumeCommands, new Map()); + const rejectPendingCommands = Effect.gen(function* () { + const pendingCommandEntries = yield* Ref.get(pendingCommands); + yield* Ref.set(pendingCommands, new Map()); - for (const [commandId, inFlight] of inFlightCommands) { + for (const pending of pendingCommandEntries.values()) { yield* Deferred.fail( - inFlight.deferred, - new Error(`Agent session closed before volume command ${commandId} completed`), + pending.deferred, + new Error(`Agent session closed before ${pending.description} completed`), ); } }); @@ -101,7 +102,7 @@ export const createControllerAgentSession = ( const releaseSession = Effect.gen(function* () { const disconnectedAt = Date.now(); yield* updateState((current) => ({ ...current, isReady: false, lastSeenAt: disconnectedAt })); - yield* rejectInFlightVolumeCommands; + yield* rejectPendingCommands; yield* onEvent({ type: "agent.disconnected" }); yield* Queue.shutdown(outboundQueue); @@ -167,13 +168,13 @@ export const createControllerAgentSession = ( const handleVolumeCommandResult = (payload: VolumeCommandResponsePayload) => Effect.gen(function* () { - const inFlight = yield* removeInFlightVolumeCommand(payload.commandId); - if (!inFlight) { + const pending = yield* removePendingCommand(payload.commandId); + if (!pending) { yield* logger.effect.warn(`Received response for unknown volume command ${payload.commandId}`); return; } - yield* Deferred.succeed(inFlight.deferred, payload); + yield* Deferred.succeed(pending.deferred, payload); }); const handleAgentMessage = (message: AgentMessage) => @@ -227,12 +228,13 @@ export const createControllerAgentSession = ( runVolumeCommand: (command) => Effect.gen(function* () { const commandId = Bun.randomUUIDv7(); + const description = `volume command ${command.name}`; const deferred = yield* Deferred.make(); - yield* setInFlightVolumeCommand(commandId, { deferred }); + yield* setPendingCommand(commandId, { deferred, description }); const queued = yield* offerOutbound(createControllerMessage("volume.command", { commandId, command })); if (!queued) { - yield* removeInFlightVolumeCommand(commandId); + yield* removePendingCommand(commandId); return yield* Effect.fail(new Error(`Failed to queue volume command ${command.name}`)); } @@ -241,7 +243,7 @@ export const createControllerAgentSession = ( duration: "60 seconds", onTimeout: () => new Error(`Volume command ${command.name} timed out`), }), - Effect.ensuring(removeInFlightVolumeCommand(commandId)), + Effect.ensuring(removePendingCommand(commandId)), ); }), isReady: () => Ref.get(state).pipe(Effect.map((current) => current.isReady)), diff --git a/apps/agent/src/commands/volume.ts b/apps/agent/src/commands/volume.ts index a2243097..55b3c1a5 100644 --- a/apps/agent/src/commands/volume.ts +++ b/apps/agent/src/commands/volume.ts @@ -1,10 +1,5 @@ -import { Effect } from "effect"; -import { - createAgentMessage, - type VolumeCommand, - type VolumeCommandPayload, - type VolumeCommandResult, -} from "@zerobyte/contracts/agent-protocol"; +import { Effect, Data } from "effect"; +import { createAgentMessage, type VolumeCommand, type VolumeCommandPayload } from "@zerobyte/contracts/agent-protocol"; import { toMessage } from "@zerobyte/core/utils"; import { createVolumeBackend, getStatFs, getVolumePath, type AgentVolume, type BackendConfig } from "../volume-host"; import { browseFilesystem, listVolumeFiles, testVolumeConnection } from "../volume-host/operations"; @@ -12,65 +7,87 @@ import type { ControllerCommandContext } from "../context"; type VolumeBackedCommand = Extract; +class VolumeCommandError extends Data.TaggedError("StopAgentManagerServerError")<{ + cause: unknown; +}> {} + const asVolume = (volume: VolumeBackedCommand["volume"]): AgentVolume => ({ ...volume, config: volume.config as BackendConfig, provisioningId: volume.provisioningId ?? null, }); -const runBackendOperation = async ( +const runBackendOperation = ( command: Extract, operation: "mount" | "unmount" | "checkHealth", -) => { - const backend = createVolumeBackend(asVolume(command.volume)); - return backend[operation](); -}; +) => + Effect.tryPromise({ + try: () => { + const backend = createVolumeBackend(asVolume(command.volume)); + return backend[operation](); + }, + catch: (error) => new VolumeCommandError({ cause: error }), + }); -const executeVolumeCommand = async (command: VolumeCommand): Promise => { - switch (command.name) { - case "volume.mount": - return { name: command.name, result: await runBackendOperation(command, "mount") }; - case "volume.unmount": - return { name: command.name, result: await runBackendOperation(command, "unmount") }; - case "volume.checkHealth": - return { name: command.name, result: await runBackendOperation(command, "checkHealth") }; - case "volume.statfs": - return { name: command.name, result: await getStatFs(getVolumePath(asVolume(command.volume))) }; - case "volume.listFiles": - return { - name: command.name, - result: await listVolumeFiles(asVolume(command.volume), command.subPath, command.offset, command.limit), - }; - case "volume.testConnection": - return { name: command.name, result: await testVolumeConnection(command.backendConfig as BackendConfig) }; - case "filesystem.browse": - return { name: command.name, result: await browseFilesystem(command.path) }; - } -}; +const executeVolumeCommand = (command: VolumeCommand) => + Effect.gen(function* () { + switch (command.name) { + case "volume.mount": + return { name: command.name, result: yield* runBackendOperation(command, "mount") }; + case "volume.unmount": + return { name: command.name, result: yield* runBackendOperation(command, "unmount") }; + case "volume.checkHealth": + return { name: command.name, result: yield* runBackendOperation(command, "checkHealth") }; + case "volume.statfs": { + const result = yield* Effect.tryPromise({ + try: () => getStatFs(getVolumePath(asVolume(command.volume))), + catch: (error) => new VolumeCommandError({ cause: error }), + }); + return { name: command.name, result }; + } + case "volume.listFiles": { + const result = yield* Effect.tryPromise({ + try: () => listVolumeFiles(asVolume(command.volume), command.subPath, command.offset, command.limit), + catch: (error) => new VolumeCommandError({ cause: error }), + }); -export const handleVolumeCommand = (context: ControllerCommandContext, payload: VolumeCommandPayload) => { - return Effect.promise(async () => { - try { - const command = await executeVolumeCommand(payload.command); - await Effect.runPromise( - context.offerOutbound( - createAgentMessage("volume.commandResult", { - commandId: payload.commandId, - status: "success", - command, - }), - ), - ); - } catch (error) { - await Effect.runPromise( - context.offerOutbound( - createAgentMessage("volume.commandResult", { - commandId: payload.commandId, - status: "error", - error: toMessage(error), - }), - ), - ); + return { name: command.name, result }; + } + case "volume.testConnection": { + const result = yield* testVolumeConnection(command.backendConfig as BackendConfig); + return { name: command.name, result }; + } + case "filesystem.browse": + const result = yield* Effect.tryPromise({ + try: () => browseFilesystem(command.path), + catch: (error) => new VolumeCommandError({ cause: error }), + }); + return { name: command.name, result }; } }); + +export const handleVolumeCommand = (context: ControllerCommandContext, payload: VolumeCommandPayload) => { + return Effect.gen(function* () { + const command = yield* executeVolumeCommand(payload.command); + + yield* context.offerOutbound( + createAgentMessage("volume.commandResult", { + commandId: payload.commandId, + status: "success", + command, + }), + ); + + return command; + }).pipe( + Effect.tapError((error) => { + return context.offerOutbound( + createAgentMessage("volume.commandResult", { + commandId: payload.commandId, + status: "error", + error: toMessage(error?.cause), + }), + ); + }), + ); }; diff --git a/apps/agent/src/volume-host/backends/utils.test.ts b/apps/agent/src/volume-host/backends/utils.test.ts new file mode 100644 index 00000000..7ca06b93 --- /dev/null +++ b/apps/agent/src/volume-host/backends/utils.test.ts @@ -0,0 +1,59 @@ +import { afterEach, describe, expect, test, vi } from "vitest"; + +vi.mock("node:fs/promises", async (importOriginal) => { + const actual = await importOriginal(); + + return { + ...actual, + access: vi.fn(actual.access), + }; +}); + +vi.mock("../fs", async (importOriginal) => { + const actual = await importOriginal(); + + return { + ...actual, + getMountForPath: vi.fn(actual.getMountForPath), + }; +}); + +import * as fs from "node:fs/promises"; +import * as volumeFs from "../fs"; +import { assertMounted } from "./utils"; + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("assertMounted", () => { + test("throws when the path is not accessible", async () => { + vi.mocked(fs.access).mockRejectedValueOnce(new Error("missing")); + + await expect(assertMounted("/tmp/volume", (fstype) => fstype.startsWith("nfs"))).rejects.toThrow( + "Volume is not mounted", + ); + }); + + test("throws when the mount filesystem does not match", async () => { + vi.mocked(fs.access).mockResolvedValueOnce(undefined); + vi.mocked(volumeFs.getMountForPath).mockResolvedValueOnce({ + mountPoint: "/tmp/volume", + fstype: "cifs", + }); + + await expect(assertMounted("/tmp/volume", (fstype) => fstype.startsWith("nfs"))).rejects.toThrow( + "Path /tmp/volume is not mounted as correct fstype (found cifs).", + ); + }); + + test("accepts a matching mounted filesystem", async () => { + vi.mocked(fs.access).mockResolvedValueOnce(undefined); + vi.mocked(volumeFs.getMountForPath).mockResolvedValueOnce({ + mountPoint: "/tmp/volume", + fstype: "nfs4", + }); + + await expect(assertMounted("/tmp/volume", (fstype) => fstype.startsWith("nfs"))).resolves.toBeUndefined(); + }); +}); diff --git a/apps/agent/src/volume-host/operations.ts b/apps/agent/src/volume-host/operations.ts index ac8adaf2..527bee97 100644 --- a/apps/agent/src/volume-host/operations.ts +++ b/apps/agent/src/volume-host/operations.ts @@ -4,6 +4,7 @@ import * as path from "node:path"; import { toMessage } from "@zerobyte/core/utils"; import { createVolumeBackend, getVolumePath, isNodeJSErrnoException } from "."; import type { AgentVolume, BackendConfig } from "./types"; +import { Data, Effect } from "effect"; const DEFAULT_PAGE_SIZE = 500; const MAX_PAGE_SIZE = 500; @@ -122,36 +123,68 @@ export const browseFilesystem = async (browsePath: string) => { }; }; -export const testVolumeConnection = async (backendConfig: BackendConfig) => { - const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-test-")); - try { - const mockVolume: AgentVolume = { - id: 0, - shortId: "test", - name: "test-connection", - config: backendConfig, - createdAt: Date.now(), - updatedAt: Date.now(), - lastHealthCheck: Date.now(), - type: backendConfig.backend, - status: "unmounted", - lastError: null, - provisioningId: null, - autoRemount: true, - agentId: "local", - organizationId: "test-org", - }; +class TempDirError extends Data.TaggedError("TempDirError")<{ + cause: unknown; +}> {} - const backend = createVolumeBackend(mockVolume, tempDir); - const { error } = await backend.mount(); +class CleanupError extends Data.TaggedError("CleanupError")<{ + cause: unknown; + tempDir: string; +}> {} - await backend.unmount(); +class MountError extends Data.TaggedError("MountError")<{ + cause: unknown; +}> {} - return { - success: !error, - message: error ? toMessage(error) : "Connection successful", - }; - } finally { - await fs.rm(tempDir, { recursive: true, force: true }); - } -}; +const createTempDir = Effect.acquireRelease( + Effect.tryPromise({ + try: () => fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-test-")), + catch: (error) => new TempDirError({ cause: error }), + }), + (tempDir) => + Effect.tryPromise({ + try: () => fs.rm(tempDir, { recursive: true, force: true }), + catch: (error) => new CleanupError({ cause: error, tempDir }), + }).pipe(Effect.orDie), +); + +export const testVolumeConnection = (backendConfig: BackendConfig) => + Effect.scoped( + Effect.gen(function* () { + const tempDir = yield* createTempDir; + + const mockVolume: AgentVolume = { + id: 0, + shortId: "test", + name: "test-connection", + config: backendConfig, + createdAt: Date.now(), + updatedAt: Date.now(), + lastHealthCheck: Date.now(), + type: backendConfig.backend, + status: "unmounted", + lastError: null, + provisioningId: null, + autoRemount: true, + agentId: "local", + organizationId: "test-org", + }; + + const backend = createVolumeBackend(mockVolume, tempDir); + + const mountResult = yield* Effect.tryPromise({ + try: () => backend.mount(), + catch: (error) => new MountError({ cause: error }), + }); + + yield* Effect.tryPromise({ + try: () => backend.unmount(), + catch: () => undefined, + }); + + return { + success: !mountResult.error, + message: mountResult.error ? toMessage(mountResult.error) : "Connection successful", + }; + }), + );