refactor: better effect error handling

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<AgentManagerRuntime> | 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 });
 

app/server/modules/agents/controller/session.ts

@@ -29,8 +29,9 @@ type SessionState = {
 	lastPongAt: number | null;
 };
 
-type InFlightVolumeCommand = {
+type PendingCommand = {
 	deferred: Deferred.Deferred<VolumeCommandResponsePayload, Error>;
+	description: string;
 };
 
 export type ControllerAgentSessionEvent =
@@ -56,7 +57,7 @@ export const createControllerAgentSession = (
 	Effect.gen(function* () {
 		let isClosed = false;
 		const outboundQueue = yield* Queue.bounded<ControllerWireMessage>(64);
-		const inFlightVolumeCommands = yield* Ref.make(new Map<string, InFlightVolumeCommand>());
+		const pendingCommands = yield* Ref.make(new Map<string, PendingCommand>());
 		const state = yield* Ref.make<SessionState>({
 			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) =>
+		const setPendingCommand = (commandId: string, pending: PendingCommand) =>
-			Ref.update(inFlightVolumeCommands, (current) => new Map(current).set(commandId, inFlight));
+			Ref.update(pendingCommands, (current) => new Map(current).set(commandId, pending));
 
-		const removeInFlightVolumeCommand = (commandId: string) =>
+		const removePendingCommand = (commandId: string) =>
-			Ref.modify(inFlightVolumeCommands, (current) => {
+			Ref.modify(pendingCommands, (current) => {
-				const inFlight = current.get(commandId) ?? null;
+				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 rejectPendingCommands = Effect.gen(function* () {
-			const inFlightCommands = yield* Ref.get(inFlightVolumeCommands);
+			const pendingCommandEntries = yield* Ref.get(pendingCommands);
-			yield* Ref.set(inFlightVolumeCommands, new Map());
+			yield* Ref.set(pendingCommands, new Map());
 
-			for (const [commandId, inFlight] of inFlightCommands) {
+			for (const pending of pendingCommandEntries.values()) {
 				yield* Deferred.fail(
-					inFlight.deferred,
+					pending.deferred,
-					new Error(`Agent session closed before volume command ${commandId} completed`),
+					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);
+				const pending = yield* removePendingCommand(payload.commandId);
-				if (!inFlight) {
+				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<VolumeCommandResponsePayload, Error>();
-					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)),

apps/agent/src/commands/volume.ts

@@ -1,10 +1,5 @@
-import { Effect } from "effect";
+import { Effect, Data } from "effect";
-import {
+import { createAgentMessage, type VolumeCommand, type VolumeCommandPayload } from "@zerobyte/contracts/agent-protocol";
-	createAgentMessage,
-	type VolumeCommand,
-	type VolumeCommandPayload,
-	type VolumeCommandResult,
-} 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<VolumeCommand, { volume: unknown }>;
 
+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<VolumeCommand, { volume: unknown }>,
 	operation: "mount" | "unmount" | "checkHealth",
-) => {
+) =>
-	const backend = createVolumeBackend(asVolume(command.volume));
+	Effect.tryPromise({
-	return backend[operation]();
+		try: () => {
-};
+			const backend = createVolumeBackend(asVolume(command.volume));
+			return backend[operation]();
+		},
+		catch: (error) => new VolumeCommandError({ cause: error }),
+	});
 
-const executeVolumeCommand = async (command: VolumeCommand): Promise<VolumeCommandResult> => {
+const executeVolumeCommand = (command: VolumeCommand) =>
-	switch (command.name) {
+	Effect.gen(function* () {
-		case "volume.mount":
+		switch (command.name) {
-			return { name: command.name, result: await runBackendOperation(command, "mount") };
+			case "volume.mount":
-		case "volume.unmount":
+				return { name: command.name, result: yield* runBackendOperation(command, "mount") };
-			return { name: command.name, result: await runBackendOperation(command, "unmount") };
+			case "volume.unmount":
-		case "volume.checkHealth":
+				return { name: command.name, result: yield* runBackendOperation(command, "unmount") };
-			return { name: command.name, result: await runBackendOperation(command, "checkHealth") };
+			case "volume.checkHealth":
-		case "volume.statfs":
+				return { name: command.name, result: yield* runBackendOperation(command, "checkHealth") };
-			return { name: command.name, result: await getStatFs(getVolumePath(asVolume(command.volume))) };
+			case "volume.statfs": {
-		case "volume.listFiles":
+				const result = yield* Effect.tryPromise({
-			return {
+					try: () => getStatFs(getVolumePath(asVolume(command.volume))),
-				name: command.name,
+					catch: (error) => new VolumeCommandError({ cause: error }),
-				result: await listVolumeFiles(asVolume(command.volume), command.subPath, command.offset, command.limit),
+				});
-			};
+				return { name: command.name, result };
-		case "volume.testConnection":
+			}
-			return { name: command.name, result: await testVolumeConnection(command.backendConfig as BackendConfig) };
+			case "volume.listFiles": {
-		case "filesystem.browse":
+				const result = yield* Effect.tryPromise({
-			return { name: command.name, result: await browseFilesystem(command.path) };
+					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 { name: command.name, result };
-	return Effect.promise(async () => {
+			}
-		try {
+			case "volume.testConnection": {
-			const command = await executeVolumeCommand(payload.command);
+				const result = yield* testVolumeConnection(command.backendConfig as BackendConfig);
-			await Effect.runPromise(
+				return { name: command.name, result };
-				context.offerOutbound(
+			}
-					createAgentMessage("volume.commandResult", {
+			case "filesystem.browse":
-						commandId: payload.commandId,
+				const result = yield* Effect.tryPromise({
-						status: "success",
+					try: () => browseFilesystem(command.path),
-						command,
+					catch: (error) => new VolumeCommandError({ cause: error }),
-					}),
+				});
-				),
+				return { name: command.name, result };
-			);
-		} catch (error) {
-			await Effect.runPromise(
-				context.offerOutbound(
-					createAgentMessage("volume.commandResult", {
-						commandId: payload.commandId,
-						status: "error",
-						error: toMessage(error),
-					}),
-				),
-			);
 		}
 	});
+
+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),
+				}),
+			);
+		}),
+	);
 };

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<typeof import("node:fs/promises")>();
+
+	return {
+		...actual,
+		access: vi.fn(actual.access),
+	};
+});
+
+vi.mock("../fs", async (importOriginal) => {
+	const actual = await importOriginal<typeof import("../fs")>();
+
+	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();
+	});
+});

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) => {
+class TempDirError extends Data.TaggedError("TempDirError")<{
-	const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-test-"));
+	cause: unknown;
-	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",
-		};
 
-		const backend = createVolumeBackend(mockVolume, tempDir);
+class CleanupError extends Data.TaggedError("CleanupError")<{
-		const { error } = await backend.mount();
+	cause: unknown;
+	tempDir: string;
+}> {}
 
-		await backend.unmount();
+class MountError extends Data.TaggedError("MountError")<{
+	cause: unknown;
+}> {}
 
-		return {
+const createTempDir = Effect.acquireRelease(
-			success: !error,
+	Effect.tryPromise({
-			message: error ? toMessage(error) : "Connection successful",
+		try: () => fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-test-")),
-		};
+		catch: (error) => new TempDirError({ cause: error }),
-	} finally {
+	}),
-		await fs.rm(tempDir, { recursive: true, force: true });
+	(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",
+			};
+		}),
+	);