refactor: move all volume operational code to agent app
This commit is contained in:
Nicolas Meienberger
parent
0ce3f79ff1
commit
a5d0cc986c
34 changed files with 1605 additions and 1206 deletions
|
|
@ -5,7 +5,7 @@ import { volumeService } from "../modules/volumes/volume.service";
|
|||
import { readMountInfo } from "../utils/mountinfo";
|
||||
import { getVolumePath } from "../modules/volumes/helpers";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import { executeUnmount } from "../modules/backends/utils/backend-utils";
|
||||
import { executeUnmount } from "../../../apps/agent/src/volume-host/backends/utils";
|
||||
import { toMessage } from "../utils/errors";
|
||||
import { VOLUME_MOUNT_BASE } from "../core/constants";
|
||||
import { db } from "../db/db";
|
||||
|
|
|
|||
|
|
@ -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 } from "@zerobyte/contracts/agent-protocol";
|
||||
import type { BackupRunPayload, VolumeCommand } from "@zerobyte/contracts/agent-protocol";
|
||||
|
||||
const setAgentRuntime = (agentManagerRuntime: Partial<AgentManagerRuntime> | null) => {
|
||||
(process as ProcessWithAgentRuntime).__zerobyteAgentRuntime = {
|
||||
|
|
@ -46,3 +46,30 @@ test("cancelBackup resolves a running backup when the cancel command cannot be d
|
|||
scheduleId: "schedule-1",
|
||||
});
|
||||
});
|
||||
|
||||
test("runVolumeCommand sends the command to the selected agent", async () => {
|
||||
const runVolumeCommand = vi.fn().mockResolvedValue({
|
||||
commandId: "command-1",
|
||||
status: "success",
|
||||
command: { name: "volume.mount", result: { status: "mounted" } },
|
||||
});
|
||||
setAgentRuntime({ runVolumeCommand });
|
||||
|
||||
const command = fromPartial<VolumeCommand>({ name: "volume.mount", volume: { agentId: "agent-1" } });
|
||||
|
||||
await expect(agentManager.runVolumeCommand("agent-1", command)).resolves.toEqual({
|
||||
name: "volume.mount",
|
||||
result: { status: "mounted" },
|
||||
});
|
||||
expect(runVolumeCommand).toHaveBeenCalledWith("agent-1", command);
|
||||
});
|
||||
|
||||
test("runVolumeCommand fails when the selected agent is unavailable", async () => {
|
||||
setAgentRuntime(null);
|
||||
|
||||
const command = fromPartial<VolumeCommand>({ name: "volume.mount", volume: { agentId: "agent-1" } });
|
||||
|
||||
await expect(agentManager.runVolumeCommand("agent-1", command)).rejects.toThrow(
|
||||
"Volume agent agent-1 is not connected",
|
||||
);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
import { logger } from "@zerobyte/core/node";
|
||||
import type { BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
|
||||
import type { BackupRunPayload, VolumeCommand, VolumeCommandResult } from "@zerobyte/contracts/agent-protocol";
|
||||
import { Effect } from "effect";
|
||||
import { config } from "../../core/config";
|
||||
import { createAgentManagerRuntime, type AgentManagerEvent } from "./controller/server";
|
||||
|
|
@ -266,6 +266,23 @@ export const agentManager = {
|
|||
cancelBackup: async (agentId: string, scheduleId: number) => {
|
||||
return requestBackupCancellation(agentId, scheduleId);
|
||||
},
|
||||
runVolumeCommand: async (agentId: string, command: VolumeCommand): Promise<VolumeCommandResult> => {
|
||||
const runtime = getAgentManagerRuntime();
|
||||
if (!runtime) {
|
||||
throw new Error(`Volume agent ${agentId} is not connected`);
|
||||
}
|
||||
|
||||
const response = await Effect.runPromise(runtime.runVolumeCommand(agentId, command));
|
||||
if (!response) {
|
||||
throw new Error(`Failed to send volume command ${command.name} to agent ${agentId}`);
|
||||
}
|
||||
|
||||
if (response.status === "error") {
|
||||
throw new Error(response.error);
|
||||
}
|
||||
|
||||
return response.command;
|
||||
},
|
||||
};
|
||||
|
||||
export const startLocalAgent = async () => {
|
||||
|
|
|
|||
|
|
@ -1,7 +1,13 @@
|
|||
import { Data, Effect, Exit, Fiber, Scope } from "effect";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import { toMessage } from "@zerobyte/core/utils";
|
||||
import type { AgentMessage, BackupCancelPayload, BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
|
||||
import type {
|
||||
AgentMessage,
|
||||
BackupCancelPayload,
|
||||
BackupRunPayload,
|
||||
VolumeCommand,
|
||||
VolumeCommandResponsePayload,
|
||||
} from "@zerobyte/contracts/agent-protocol";
|
||||
import {
|
||||
createControllerAgentSession,
|
||||
type AgentConnectionData,
|
||||
|
|
@ -319,10 +325,30 @@ export function createAgentManagerRuntime(onEvent: (event: AgentManagerEvent) =>
|
|||
logger.warn(`Cannot cancel backup command. Agent ${agentId} is no longer accepting commands.`);
|
||||
return false;
|
||||
}
|
||||
|
||||
logger.info(`Sent backup cancel for command ${payload.jobId} to agent ${agentId}`);
|
||||
return true;
|
||||
}),
|
||||
runVolumeCommand: (
|
||||
agentId: string,
|
||||
command: VolumeCommand,
|
||||
): Effect.Effect<VolumeCommandResponsePayload | null, Error> =>
|
||||
Effect.gen(function* () {
|
||||
const session = getSession(agentId);
|
||||
|
||||
if (!session) {
|
||||
yield* logger.effect.warn(`Cannot send volume command ${command.name}. Agent ${agentId} is not connected.`);
|
||||
return null;
|
||||
}
|
||||
|
||||
if (!(yield* session.isReady())) {
|
||||
yield* logger.effect.warn(`Cannot send volume command ${command.name}. Agent ${agentId} is not ready.`);
|
||||
return null;
|
||||
}
|
||||
|
||||
const result = yield* session.runVolumeCommand(command);
|
||||
yield* logger.effect.info(`Completed volume command ${command.name} on agent ${agentId}`);
|
||||
return result;
|
||||
}),
|
||||
stop,
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,6 +7,8 @@ import {
|
|||
type BackupCancelPayload,
|
||||
type BackupRunPayload,
|
||||
type ControllerWireMessage,
|
||||
type VolumeCommand,
|
||||
type VolumeCommandResponsePayload,
|
||||
} from "@zerobyte/contracts/agent-protocol";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import { toMessage } from "@zerobyte/core/utils";
|
||||
|
|
@ -27,13 +29,24 @@ type SessionState = {
|
|||
lastPongAt: number | null;
|
||||
};
|
||||
|
||||
export type ControllerAgentSessionEvent = AgentMessage | { type: "agent.disconnected" };
|
||||
type PendingVolumeCommand = {
|
||||
resolve: (payload: VolumeCommandResponsePayload) => void;
|
||||
reject: (error: Error) => void;
|
||||
timeout: ReturnType<typeof setTimeout>;
|
||||
};
|
||||
|
||||
export type ControllerAgentSessionEvent =
|
||||
| Exclude<AgentMessage, { type: "volume.commandResult" }>
|
||||
| {
|
||||
type: "agent.disconnected";
|
||||
};
|
||||
|
||||
export type ControllerAgentSession = {
|
||||
readonly connectionId: string;
|
||||
handleMessage: (data: string) => Effect.Effect<void>;
|
||||
sendBackup: (payload: BackupRunPayload) => Effect.Effect<boolean>;
|
||||
sendBackupCancel: (payload: BackupCancelPayload) => Effect.Effect<boolean>;
|
||||
runVolumeCommand: (command: VolumeCommand) => Effect.Effect<VolumeCommandResponsePayload, Error>;
|
||||
isReady: () => Effect.Effect<boolean>;
|
||||
run: Effect.Effect<void, never, Scope.Scope>;
|
||||
};
|
||||
|
|
@ -45,6 +58,7 @@ export const createControllerAgentSession = (
|
|||
Effect.gen(function* () {
|
||||
let isClosed = false;
|
||||
const outboundQueue = yield* Queue.bounded<ControllerWireMessage>(64);
|
||||
const pendingVolumeCommands = new Map<string, PendingVolumeCommand>();
|
||||
const state = yield* Ref.make<SessionState>({
|
||||
isReady: false,
|
||||
lastSeenAt: null,
|
||||
|
|
@ -63,9 +77,18 @@ export const createControllerAgentSession = (
|
|||
|
||||
const updateState = (update: (current: SessionState) => SessionState) => Ref.update(state, update);
|
||||
|
||||
const rejectPendingVolumeCommands = () => {
|
||||
for (const [commandId, pending] of pendingVolumeCommands) {
|
||||
clearTimeout(pending.timeout);
|
||||
pending.reject(new Error(`Agent session closed before volume command ${commandId} completed`));
|
||||
}
|
||||
pendingVolumeCommands.clear();
|
||||
};
|
||||
|
||||
const releaseSession = Effect.gen(function* () {
|
||||
const disconnectedAt = Date.now();
|
||||
yield* updateState((current) => ({ ...current, isReady: false, lastSeenAt: disconnectedAt }));
|
||||
yield* Effect.sync(rejectPendingVolumeCommands);
|
||||
yield* onEvent({ type: "agent.disconnected" });
|
||||
|
||||
yield* Queue.shutdown(outboundQueue);
|
||||
|
|
@ -129,20 +152,43 @@ export const createControllerAgentSession = (
|
|||
return yield* Effect.never;
|
||||
});
|
||||
|
||||
const handleVolumeCommandResult = (payload: VolumeCommandResponsePayload) => {
|
||||
const pending = pendingVolumeCommands.get(payload.commandId);
|
||||
if (!pending) {
|
||||
logger.warn(`Received response for unknown volume command ${payload.commandId}`);
|
||||
return;
|
||||
}
|
||||
|
||||
pendingVolumeCommands.delete(payload.commandId);
|
||||
clearTimeout(pending.timeout);
|
||||
pending.resolve(payload);
|
||||
};
|
||||
|
||||
const handleAgentMessage = (message: AgentMessage) =>
|
||||
Effect.gen(function* () {
|
||||
if (message.type === "agent.ready") {
|
||||
const readyAt = Date.now();
|
||||
yield* updateState((current) => ({ ...current, isReady: true, lastSeenAt: readyAt }));
|
||||
yield* logger.effect.info(`Agent "${socket.data.agentName}" (${socket.data.agentId}) is ready`);
|
||||
switch (message.type) {
|
||||
case "agent.ready": {
|
||||
const readyAt = Date.now();
|
||||
yield* updateState((current) => ({ ...current, isReady: true, lastSeenAt: readyAt }));
|
||||
yield* logger.effect.info(`Agent "${socket.data.agentName}" (${socket.data.agentId}) is ready`);
|
||||
yield* onEvent(message);
|
||||
break;
|
||||
}
|
||||
case "heartbeat.pong": {
|
||||
const seenAt = Date.now();
|
||||
yield* updateState((current) => ({ ...current, lastSeenAt: seenAt, lastPongAt: message.payload.sentAt }));
|
||||
yield* onEvent(message);
|
||||
break;
|
||||
}
|
||||
case "volume.commandResult": {
|
||||
yield* Effect.sync(() => handleVolumeCommandResult(message.payload));
|
||||
break;
|
||||
}
|
||||
default: {
|
||||
yield* onEvent(message);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (message.type === "heartbeat.pong") {
|
||||
const seenAt = Date.now();
|
||||
yield* updateState((current) => ({ ...current, lastSeenAt: seenAt, lastPongAt: message.payload.sentAt }));
|
||||
}
|
||||
|
||||
yield* onEvent(message);
|
||||
});
|
||||
|
||||
return {
|
||||
|
|
@ -166,6 +212,37 @@ export const createControllerAgentSession = (
|
|||
},
|
||||
sendBackup: (payload) => offerOutbound(createControllerMessage("backup.run", payload)),
|
||||
sendBackupCancel: (payload) => offerOutbound(createControllerMessage("backup.cancel", payload)),
|
||||
runVolumeCommand: (command) => {
|
||||
return Effect.tryPromise({
|
||||
try: async () => {
|
||||
const commandId = Bun.randomUUIDv7();
|
||||
const response = new Promise<VolumeCommandResponsePayload>((resolve, reject) => {
|
||||
const timeout = setTimeout(() => {
|
||||
pendingVolumeCommands.delete(commandId);
|
||||
reject(new Error(`Volume command ${command.name} timed out`));
|
||||
}, 60_000);
|
||||
|
||||
pendingVolumeCommands.set(commandId, { resolve, reject, timeout });
|
||||
});
|
||||
|
||||
const queued = await Effect.runPromise(
|
||||
offerOutbound(createControllerMessage("volume.command", { commandId, command })),
|
||||
);
|
||||
|
||||
if (!queued) {
|
||||
const pending = pendingVolumeCommands.get(commandId);
|
||||
if (pending) {
|
||||
clearTimeout(pending.timeout);
|
||||
pendingVolumeCommands.delete(commandId);
|
||||
}
|
||||
throw new Error(`Failed to queue volume command ${command.name}`);
|
||||
}
|
||||
|
||||
return response;
|
||||
},
|
||||
catch: (error) => (error instanceof Error ? error : new Error(toMessage(error))),
|
||||
});
|
||||
},
|
||||
isReady: () => Ref.get(state).pipe(Effect.map((current) => current.isReady)),
|
||||
run,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -1,46 +0,0 @@
|
|||
import type { BackendStatus } from "~/schemas/volumes";
|
||||
import type { Volume } from "../../db/schema";
|
||||
import { getVolumePath } from "../volumes/helpers";
|
||||
import { makeDirectoryBackend } from "./directory/directory-backend";
|
||||
import { makeNfsBackend } from "./nfs/nfs-backend";
|
||||
import { makeRcloneBackend } from "./rclone/rclone-backend";
|
||||
import { makeSmbBackend } from "./smb/smb-backend";
|
||||
import { makeWebdavBackend } from "./webdav/webdav-backend";
|
||||
import { makeSftpBackend } from "./sftp/sftp-backend";
|
||||
|
||||
type OperationResult = {
|
||||
error?: string;
|
||||
status: BackendStatus;
|
||||
};
|
||||
|
||||
export type VolumeBackend = {
|
||||
mount: () => Promise<OperationResult>;
|
||||
unmount: () => Promise<OperationResult>;
|
||||
checkHealth: () => Promise<OperationResult>;
|
||||
};
|
||||
|
||||
export const createVolumeBackend = (volume: Volume, mountPath = getVolumePath(volume)): VolumeBackend => {
|
||||
switch (volume.config.backend) {
|
||||
case "nfs": {
|
||||
return makeNfsBackend(volume.config, mountPath);
|
||||
}
|
||||
case "smb": {
|
||||
return makeSmbBackend(volume.config, mountPath);
|
||||
}
|
||||
case "directory": {
|
||||
return makeDirectoryBackend(volume.config, mountPath);
|
||||
}
|
||||
case "webdav": {
|
||||
return makeWebdavBackend(volume.config, mountPath);
|
||||
}
|
||||
case "rclone": {
|
||||
return makeRcloneBackend(volume.config, mountPath);
|
||||
}
|
||||
case "sftp": {
|
||||
return makeSftpBackend(volume.config, mountPath);
|
||||
}
|
||||
default: {
|
||||
throw new Error("Unsupported backend");
|
||||
}
|
||||
}
|
||||
};
|
||||
|
|
@ -1,135 +0,0 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import { BACKEND_STATUS, type BackendConfig } from "~/schemas/volumes";
|
||||
import { OPERATION_TIMEOUT } from "../../../core/constants";
|
||||
import { toMessage } from "../../../utils/errors";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import { getMountForPath } from "../../../utils/mountinfo";
|
||||
import { withTimeout } from "../../../utils/timeout";
|
||||
import type { VolumeBackend } from "../backend";
|
||||
import { assertMounted, executeMount, executeUnmount } from "../utils/backend-utils";
|
||||
|
||||
const mount = async (config: BackendConfig, path: string) => {
|
||||
logger.debug(`Mounting volume ${path}...`);
|
||||
|
||||
if (config.backend !== "nfs") {
|
||||
logger.error("Provided config is not for NFS backend");
|
||||
return {
|
||||
status: BACKEND_STATUS.error,
|
||||
error: "Provided config is not for NFS backend",
|
||||
};
|
||||
}
|
||||
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("NFS mounting is only supported on Linux hosts.");
|
||||
return {
|
||||
status: BACKEND_STATUS.error,
|
||||
error: "NFS mounting is only supported on Linux hosts.",
|
||||
};
|
||||
}
|
||||
|
||||
const { status } = await checkHealth(path);
|
||||
if (status === "mounted") {
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
}
|
||||
|
||||
if (status === "error") {
|
||||
logger.debug(`Trying to unmount any existing mounts at ${path} before mounting...`);
|
||||
await unmount(path);
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
await fs.mkdir(path, { recursive: true });
|
||||
|
||||
const source = `${config.server}:${config.exportPath}`;
|
||||
const options = [`vers=${config.version}`, `port=${config.port}`];
|
||||
if (config.version === "3") {
|
||||
options.push("nolock");
|
||||
}
|
||||
if (config.readOnly) {
|
||||
options.push("ro");
|
||||
}
|
||||
const args = ["-t", "nfs", "-o", options.join(","), source, path];
|
||||
|
||||
logger.debug(`Mounting volume ${path}...`);
|
||||
logger.info(`Executing mount: mount ${args.join(" ")}`);
|
||||
|
||||
try {
|
||||
await executeMount(args);
|
||||
} catch (error) {
|
||||
logger.warn(`Initial NFS mount failed, retrying with -i flag: ${toMessage(error)}`);
|
||||
// Fallback with -i flag if the first mount fails using the mount helper
|
||||
await executeMount(["-i", ...args]);
|
||||
}
|
||||
|
||||
logger.info(`NFS volume at ${path} mounted successfully.`);
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "NFS mount");
|
||||
} catch (err) {
|
||||
logger.error("Error mounting NFS volume", { error: toMessage(err) });
|
||||
return { status: BACKEND_STATUS.error, error: toMessage(err) };
|
||||
}
|
||||
};
|
||||
|
||||
const unmount = async (path: string) => {
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("NFS unmounting is only supported on Linux hosts.");
|
||||
return {
|
||||
status: BACKEND_STATUS.error,
|
||||
error: "NFS unmounting is only supported on Linux hosts.",
|
||||
};
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
const mount = await getMountForPath(path);
|
||||
if (!mount || mount.mountPoint !== path) {
|
||||
logger.debug(`Path ${path} is not a mount point. Skipping unmount.`);
|
||||
return { status: BACKEND_STATUS.unmounted };
|
||||
}
|
||||
|
||||
await executeUnmount(path);
|
||||
|
||||
await fs.rmdir(path).catch(() => {});
|
||||
|
||||
logger.info(`NFS volume at ${path} unmounted successfully.`);
|
||||
return { status: BACKEND_STATUS.unmounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "NFS unmount");
|
||||
} catch (err) {
|
||||
logger.error("Error unmounting NFS volume", {
|
||||
path,
|
||||
error: toMessage(err),
|
||||
});
|
||||
return { status: BACKEND_STATUS.error, error: toMessage(err) };
|
||||
}
|
||||
};
|
||||
|
||||
const checkHealth = async (path: string) => {
|
||||
const run = async () => {
|
||||
await assertMounted(path, (fstype) => fstype.startsWith("nfs"));
|
||||
|
||||
logger.debug(`NFS volume at ${path} is healthy and mounted.`);
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "NFS health check");
|
||||
} catch (error) {
|
||||
const message = toMessage(error);
|
||||
if (message !== "Volume is not mounted") {
|
||||
logger.error("NFS volume health check failed:", message);
|
||||
}
|
||||
return { status: BACKEND_STATUS.error, error: message };
|
||||
}
|
||||
};
|
||||
|
||||
export const makeNfsBackend = (config: BackendConfig, path: string): VolumeBackend => ({
|
||||
mount: () => mount(config, path),
|
||||
unmount: () => unmount(path),
|
||||
checkHealth: () => checkHealth(path),
|
||||
});
|
||||
|
|
@ -1,131 +0,0 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import { OPERATION_TIMEOUT, RCLONE_CONFIG_FILE } from "../../../core/constants";
|
||||
import { toMessage } from "../../../utils/errors";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import { getMountForPath } from "../../../utils/mountinfo";
|
||||
import { withTimeout } from "../../../utils/timeout";
|
||||
import type { VolumeBackend } from "../backend";
|
||||
import { assertMounted, executeUnmount } from "../utils/backend-utils";
|
||||
import { BACKEND_STATUS, type BackendConfig } from "~/schemas/volumes";
|
||||
import { safeExec } from "@zerobyte/core/node";
|
||||
import { config as zbConfig } from "~/server/core/config";
|
||||
|
||||
const mount = async (config: BackendConfig, path: string) => {
|
||||
logger.debug(`Mounting rclone volume ${path}...`);
|
||||
|
||||
if (config.backend !== "rclone") {
|
||||
logger.error("Provided config is not for rclone backend");
|
||||
return { status: BACKEND_STATUS.error, error: "Provided config is not for rclone backend" };
|
||||
}
|
||||
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("Rclone mounting is only supported on Linux hosts.");
|
||||
return { status: BACKEND_STATUS.error, error: "Rclone mounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const { status } = await checkHealth(path);
|
||||
if (status === "mounted") {
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
}
|
||||
|
||||
if (status === "error") {
|
||||
logger.debug(`Trying to unmount any existing mounts at ${path} before mounting...`);
|
||||
await unmount(path);
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
await fs.mkdir(path, { recursive: true });
|
||||
|
||||
const remotePath = `${config.remote}:${config.path}`;
|
||||
const args = ["mount", remotePath, path, "--daemon"];
|
||||
|
||||
if (config.readOnly) {
|
||||
args.push("--read-only");
|
||||
}
|
||||
|
||||
args.push("--vfs-cache-mode", "writes");
|
||||
args.push("--allow-non-empty");
|
||||
args.push("--allow-other");
|
||||
|
||||
logger.debug(`Mounting rclone volume ${path}...`);
|
||||
logger.info(`Executing rclone: rclone ${args.join(" ")}`);
|
||||
|
||||
const result = await safeExec({
|
||||
command: "rclone",
|
||||
args,
|
||||
env: { RCLONE_CONFIG: RCLONE_CONFIG_FILE },
|
||||
timeout: zbConfig.serverIdleTimeout * 1000,
|
||||
});
|
||||
|
||||
if (result.exitCode !== 0) {
|
||||
const errorMsg = result.stderr.toString() || result.stdout.toString() || "Unknown error";
|
||||
throw new Error(`Failed to mount rclone volume: ${errorMsg}`);
|
||||
}
|
||||
|
||||
logger.info(`Rclone volume at ${path} mounted successfully.`);
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), zbConfig.serverIdleTimeout * 1000, "Rclone mount");
|
||||
} catch (error) {
|
||||
const errorMsg = toMessage(error);
|
||||
|
||||
logger.error("Error mounting rclone volume", { error: errorMsg });
|
||||
return { status: BACKEND_STATUS.error, error: errorMsg };
|
||||
}
|
||||
};
|
||||
|
||||
const unmount = async (path: string) => {
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("Rclone unmounting is only supported on Linux hosts.");
|
||||
return { status: BACKEND_STATUS.error, error: "Rclone unmounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
const mount = await getMountForPath(path);
|
||||
if (!mount || mount.mountPoint !== path) {
|
||||
logger.debug(`Path ${path} is not a mount point. Skipping unmount.`);
|
||||
return { status: BACKEND_STATUS.unmounted };
|
||||
}
|
||||
|
||||
await executeUnmount(path);
|
||||
await fs.rmdir(path).catch(() => {});
|
||||
|
||||
logger.info(`Rclone volume at ${path} unmounted successfully.`);
|
||||
return { status: BACKEND_STATUS.unmounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "Rclone unmount");
|
||||
} catch (error) {
|
||||
logger.error("Error unmounting rclone volume", { path, error: toMessage(error) });
|
||||
return { status: BACKEND_STATUS.error, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
const checkHealth = async (path: string) => {
|
||||
const run = async () => {
|
||||
await assertMounted(path, (fstype) => fstype.includes("rclone"));
|
||||
|
||||
logger.debug(`Rclone volume at ${path} is healthy and mounted.`);
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "Rclone health check");
|
||||
} catch (error) {
|
||||
const message = toMessage(error);
|
||||
if (message !== "Volume is not mounted") {
|
||||
logger.error("Rclone volume health check failed:", message);
|
||||
}
|
||||
return { status: BACKEND_STATUS.error, error: message };
|
||||
}
|
||||
};
|
||||
|
||||
export const makeRcloneBackend = (config: BackendConfig, path: string): VolumeBackend => ({
|
||||
mount: () => mount(config, path),
|
||||
unmount: () => unmount(path),
|
||||
checkHealth: () => checkHealth(path),
|
||||
});
|
||||
|
|
@ -1,212 +0,0 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import * as path from "node:path";
|
||||
import { spawn } from "node:child_process";
|
||||
import { OPERATION_TIMEOUT, SSH_KEYS_DIR } from "../../../core/constants";
|
||||
import { cryptoUtils } from "../../../utils/crypto";
|
||||
import { toMessage } from "../../../utils/errors";
|
||||
import { logger, FILE_MODES, writeFileWithMode } from "@zerobyte/core/node";
|
||||
import { getMountForPath } from "../../../utils/mountinfo";
|
||||
import { withTimeout } from "../../../utils/timeout";
|
||||
import type { VolumeBackend } from "../backend";
|
||||
import { executeUnmount } from "../utils/backend-utils";
|
||||
import { BACKEND_STATUS, type BackendConfig } from "~/schemas/volumes";
|
||||
|
||||
const getPrivateKeyPath = (mountPath: string) => {
|
||||
const name = path.basename(mountPath);
|
||||
return path.join(SSH_KEYS_DIR, `${name}.key`);
|
||||
};
|
||||
|
||||
const getKnownHostsPath = (mountPath: string) => {
|
||||
const name = path.basename(mountPath);
|
||||
return path.join(SSH_KEYS_DIR, `${name}.known_hosts`);
|
||||
};
|
||||
|
||||
const mount = async (config: BackendConfig, mountPath: string) => {
|
||||
logger.debug(`Mounting SFTP volume ${mountPath}...`);
|
||||
|
||||
if (config.backend !== "sftp") {
|
||||
logger.error("Provided config is not for SFTP backend");
|
||||
return { status: BACKEND_STATUS.error, error: "Provided config is not for SFTP backend" };
|
||||
}
|
||||
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("SFTP mounting is only supported on Linux hosts.");
|
||||
return { status: BACKEND_STATUS.error, error: "SFTP mounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const { status } = await checkHealth(mountPath);
|
||||
if (status === "mounted") {
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
}
|
||||
|
||||
if (status === "error") {
|
||||
logger.debug(`Trying to unmount any existing mounts at ${mountPath} before mounting...`);
|
||||
await unmount(mountPath);
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
await fs.mkdir(mountPath, { recursive: true });
|
||||
await fs.mkdir(SSH_KEYS_DIR, { recursive: true });
|
||||
|
||||
const { uid, gid } = os.userInfo();
|
||||
const options = [
|
||||
"reconnect",
|
||||
"ServerAliveInterval=15",
|
||||
"ServerAliveCountMax=3",
|
||||
"allow_other",
|
||||
`uid=${uid}`,
|
||||
`gid=${gid}`,
|
||||
];
|
||||
|
||||
if (config.skipHostKeyCheck) {
|
||||
options.push("StrictHostKeyChecking=no", "UserKnownHostsFile=/dev/null");
|
||||
} else if (config.knownHosts) {
|
||||
const knownHostsPath = getKnownHostsPath(mountPath);
|
||||
await writeFileWithMode(knownHostsPath, config.knownHosts, FILE_MODES.ownerReadWrite);
|
||||
options.push(`UserKnownHostsFile=${knownHostsPath}`, "StrictHostKeyChecking=yes");
|
||||
} else {
|
||||
options.push("StrictHostKeyChecking=yes");
|
||||
}
|
||||
|
||||
if (config.readOnly) {
|
||||
options.push("ro");
|
||||
}
|
||||
|
||||
if (config.port) {
|
||||
options.push(`port=${config.port}`);
|
||||
}
|
||||
|
||||
const keyPath = getPrivateKeyPath(mountPath);
|
||||
if (config.privateKey) {
|
||||
const decryptedKey = await cryptoUtils.resolveSecret(config.privateKey);
|
||||
let normalizedKey = decryptedKey.replace(/\r\n/g, "\n");
|
||||
if (!normalizedKey.endsWith("\n")) {
|
||||
normalizedKey += "\n";
|
||||
}
|
||||
await writeFileWithMode(keyPath, normalizedKey, FILE_MODES.ownerReadWrite);
|
||||
options.push(`IdentityFile=${keyPath}`);
|
||||
}
|
||||
|
||||
const source = `${config.username}@${config.host}:${config.path || ""}`;
|
||||
const args = [source, mountPath, "-o", options.join(",")];
|
||||
|
||||
logger.debug(`Mounting SFTP volume ${mountPath}...`);
|
||||
|
||||
const runSshfs = async (mountArgs: string[], password?: string) => {
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
const child = spawn("sshfs", mountArgs, { stdio: ["pipe", "pipe", "pipe"] });
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
|
||||
child.stdout.setEncoding("utf8");
|
||||
child.stderr.setEncoding("utf8");
|
||||
|
||||
child.stdout.on("data", (data) => {
|
||||
stdout += data;
|
||||
});
|
||||
|
||||
child.stderr.on("data", (data) => {
|
||||
stderr += data;
|
||||
});
|
||||
|
||||
child.on("error", (error) => {
|
||||
reject(new Error(`Failed to start sshfs: ${error.message}`));
|
||||
});
|
||||
|
||||
child.on("close", (code) => {
|
||||
if (code === 0) {
|
||||
resolve();
|
||||
return;
|
||||
}
|
||||
|
||||
const errorMsg = stderr.trim() || stdout.trim() || "Unknown error";
|
||||
reject(new Error(`Failed to mount SFTP volume: ${errorMsg}`));
|
||||
});
|
||||
|
||||
if (password) {
|
||||
child.stdin.write(password);
|
||||
}
|
||||
child.stdin.end();
|
||||
});
|
||||
};
|
||||
|
||||
if (config.password) {
|
||||
const password = await cryptoUtils.resolveSecret(config.password);
|
||||
args.push("-o", "password_stdin");
|
||||
logger.info(`Executing sshfs: sshfs ${args.join(" ")}`);
|
||||
await runSshfs(args, password);
|
||||
} else {
|
||||
logger.info(`Executing sshfs: sshfs ${args.join(" ")}`);
|
||||
await runSshfs(args);
|
||||
}
|
||||
|
||||
logger.info(`SFTP volume at ${mountPath} mounted successfully.`);
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT * 2, "SFTP mount");
|
||||
} catch (error) {
|
||||
const errorMsg = toMessage(error);
|
||||
logger.error("Error mounting SFTP volume", { error: errorMsg });
|
||||
return { status: BACKEND_STATUS.error, error: errorMsg };
|
||||
}
|
||||
};
|
||||
|
||||
const unmount = async (mountPath: string) => {
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("SFTP unmounting is only supported on Linux hosts.");
|
||||
return { status: BACKEND_STATUS.error, error: "SFTP unmounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
const mount = await getMountForPath(mountPath);
|
||||
if (!mount || mount.mountPoint !== mountPath) {
|
||||
logger.debug(`Path ${mountPath} is not a mount point. Skipping unmount.`);
|
||||
} else {
|
||||
await executeUnmount(mountPath);
|
||||
}
|
||||
|
||||
const keyPath = getPrivateKeyPath(mountPath);
|
||||
await fs.unlink(keyPath).catch(() => {});
|
||||
|
||||
const knownHostsPath = getKnownHostsPath(mountPath);
|
||||
await fs.unlink(knownHostsPath).catch(() => {});
|
||||
|
||||
await fs.rmdir(mountPath).catch(() => {});
|
||||
|
||||
logger.info(`SFTP volume at ${mountPath} unmounted successfully.`);
|
||||
return { status: BACKEND_STATUS.unmounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "SFTP unmount");
|
||||
} catch (error) {
|
||||
logger.error("Error unmounting SFTP volume", { mountPath, error: toMessage(error) });
|
||||
return { status: BACKEND_STATUS.error, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
const checkHealth = async (mountPath: string) => {
|
||||
const mount = await getMountForPath(mountPath);
|
||||
|
||||
if (!mount || mount.mountPoint !== mountPath) {
|
||||
return { status: BACKEND_STATUS.unmounted };
|
||||
}
|
||||
|
||||
if (mount.fstype !== "fuse.sshfs") {
|
||||
return {
|
||||
status: BACKEND_STATUS.error,
|
||||
error: `Invalid filesystem type: ${mount.fstype} (expected fuse.sshfs)`,
|
||||
};
|
||||
}
|
||||
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
};
|
||||
|
||||
export const makeSftpBackend = (config: BackendConfig, mountPath: string): VolumeBackend => ({
|
||||
mount: () => mount(config, mountPath),
|
||||
unmount: () => unmount(mountPath),
|
||||
checkHealth: () => checkHealth(mountPath),
|
||||
});
|
||||
|
|
@ -1,141 +0,0 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import { OPERATION_TIMEOUT } from "../../../core/constants";
|
||||
import { cryptoUtils } from "../../../utils/crypto";
|
||||
import { toMessage } from "../../../utils/errors";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import { getMountForPath } from "../../../utils/mountinfo";
|
||||
import { withTimeout } from "../../../utils/timeout";
|
||||
import type { VolumeBackend } from "../backend";
|
||||
import { assertMounted, executeMount, executeUnmount } from "../utils/backend-utils";
|
||||
import { BACKEND_STATUS, type BackendConfig } from "~/schemas/volumes";
|
||||
|
||||
const mount = async (config: BackendConfig, path: string) => {
|
||||
logger.debug(`Mounting SMB volume ${path}...`);
|
||||
|
||||
if (config.backend !== "smb") {
|
||||
logger.error("Provided config is not for SMB backend");
|
||||
return { status: BACKEND_STATUS.error, error: "Provided config is not for SMB backend" };
|
||||
}
|
||||
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("SMB mounting is only supported on Linux hosts.");
|
||||
return { status: BACKEND_STATUS.error, error: "SMB mounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const { status } = await checkHealth(path);
|
||||
if (status === "mounted") {
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
}
|
||||
|
||||
if (status === "error") {
|
||||
logger.debug(`Trying to unmount any existing mounts at ${path} before mounting...`);
|
||||
await unmount(path);
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
await fs.mkdir(path, { recursive: true });
|
||||
|
||||
const source = `//${config.server}/${config.share}`;
|
||||
const { uid, gid } = os.userInfo();
|
||||
|
||||
const options = [`port=${config.port}`, `uid=${uid}`, `gid=${gid}`, "iocharset=utf8"];
|
||||
|
||||
if (config.guest) {
|
||||
options.push("username=guest", "password=");
|
||||
} else {
|
||||
const password = await cryptoUtils.resolveSecret(config.password ?? "");
|
||||
const safePassword = password.replace(/\\/g, "\\\\").replace(/,/g, "\\,");
|
||||
|
||||
options.push(`username=${config.username ?? "user"}`, `password=${safePassword}`);
|
||||
}
|
||||
|
||||
if (config.domain) {
|
||||
options.push(`domain=${config.domain}`);
|
||||
}
|
||||
|
||||
if (config.vers && config.vers !== "auto") {
|
||||
options.push(`vers=${config.vers}`);
|
||||
}
|
||||
|
||||
if (config.readOnly) {
|
||||
options.push("ro");
|
||||
}
|
||||
|
||||
const args = ["-t", "cifs", "-o", options.join(","), source, path];
|
||||
|
||||
logger.debug(`Mounting SMB volume ${path}...`);
|
||||
logger.info(`Executing SMB mount for ${source} at ${path}`);
|
||||
|
||||
try {
|
||||
await executeMount(args);
|
||||
} catch (error) {
|
||||
logger.error(`SMB mount failed: ${toMessage(error)}`);
|
||||
throw error;
|
||||
}
|
||||
|
||||
logger.info(`SMB volume at ${path} mounted successfully.`);
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "SMB mount");
|
||||
} catch (error) {
|
||||
logger.error("Error mounting SMB volume", { error: toMessage(error) });
|
||||
return { status: BACKEND_STATUS.error, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
const unmount = async (path: string) => {
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("SMB unmounting is only supported on Linux hosts.");
|
||||
return { status: BACKEND_STATUS.error, error: "SMB unmounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
const mount = await getMountForPath(path);
|
||||
if (!mount || mount.mountPoint !== path) {
|
||||
logger.debug(`Path ${path} is not a mount point. Skipping unmount.`);
|
||||
return { status: BACKEND_STATUS.unmounted };
|
||||
}
|
||||
|
||||
await executeUnmount(path);
|
||||
|
||||
await fs.rmdir(path).catch(() => {});
|
||||
|
||||
logger.info(`SMB volume at ${path} unmounted successfully.`);
|
||||
return { status: BACKEND_STATUS.unmounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "SMB unmount");
|
||||
} catch (error) {
|
||||
logger.error("Error unmounting SMB volume", { path, error: toMessage(error) });
|
||||
return { status: BACKEND_STATUS.error, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
const checkHealth = async (path: string) => {
|
||||
const run = async () => {
|
||||
await assertMounted(path, (fstype) => fstype === "cifs");
|
||||
|
||||
logger.debug(`SMB volume at ${path} is healthy and mounted.`);
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "SMB health check");
|
||||
} catch (error) {
|
||||
const message = toMessage(error);
|
||||
if (message !== "Volume is not mounted") {
|
||||
logger.error("SMB volume health check failed:", message);
|
||||
}
|
||||
return { status: BACKEND_STATUS.error, error: message };
|
||||
}
|
||||
};
|
||||
|
||||
export const makeSmbBackend = (config: BackendConfig, path: string): VolumeBackend => ({
|
||||
mount: () => mount(config, path),
|
||||
unmount: () => unmount(path),
|
||||
checkHealth: () => checkHealth(path),
|
||||
});
|
||||
|
|
@ -1,59 +0,0 @@
|
|||
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("../../../../utils/mountinfo", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("../../../../utils/mountinfo")>();
|
||||
|
||||
return {
|
||||
...actual,
|
||||
getMountForPath: vi.fn(actual.getMountForPath),
|
||||
};
|
||||
});
|
||||
|
||||
import * as fs from "node:fs/promises";
|
||||
import * as mountinfo from "../../../../utils/mountinfo";
|
||||
import { assertMounted } from "../backend-utils";
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
describe("assertMountedFilesystem", () => {
|
||||
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(mountinfo.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(mountinfo.getMountForPath).mockResolvedValueOnce({
|
||||
mountPoint: "/tmp/volume",
|
||||
fstype: "nfs4",
|
||||
});
|
||||
|
||||
await expect(assertMounted("/tmp/volume", (fstype) => fstype.startsWith("nfs"))).resolves.toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
|
@ -1,159 +0,0 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import { OPERATION_TIMEOUT } from "../../../core/constants";
|
||||
import { cryptoUtils } from "../../../utils/crypto";
|
||||
import { toMessage } from "../../../utils/errors";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import { getMountForPath } from "../../../utils/mountinfo";
|
||||
import { withTimeout } from "../../../utils/timeout";
|
||||
import type { VolumeBackend } from "../backend";
|
||||
import { assertMounted, executeMount, executeUnmount } from "../utils/backend-utils";
|
||||
import { BACKEND_STATUS, type BackendConfig } from "~/schemas/volumes";
|
||||
|
||||
const mount = async (config: BackendConfig, path: string) => {
|
||||
logger.debug(`Mounting WebDAV volume ${path}...`);
|
||||
|
||||
if (config.backend !== "webdav") {
|
||||
logger.error("Provided config is not for WebDAV backend");
|
||||
return { status: BACKEND_STATUS.error, error: "Provided config is not for WebDAV backend" };
|
||||
}
|
||||
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("WebDAV mounting is only supported on Linux hosts.");
|
||||
return { status: BACKEND_STATUS.error, error: "WebDAV mounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const { status } = await checkHealth(path);
|
||||
if (status === "mounted") {
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
}
|
||||
|
||||
if (status === "error") {
|
||||
logger.debug(`Trying to unmount any existing mounts at ${path} before mounting...`);
|
||||
await unmount(path);
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
await fs.mkdir(path, { recursive: true }).catch((err) => {
|
||||
logger.warn(`Failed to create directory ${path}: ${err.message}`);
|
||||
});
|
||||
|
||||
const protocol = config.ssl ? "https" : "http";
|
||||
const defaultPort = config.ssl ? 443 : 80;
|
||||
const port = config.port !== defaultPort ? `:${config.port}` : "";
|
||||
const source = `${protocol}://${config.server}${port}${config.path}`;
|
||||
|
||||
const { uid, gid } = os.userInfo();
|
||||
const options = config.readOnly
|
||||
? [`uid=${uid}`, `gid=${gid}`, "file_mode=0444", "dir_mode=0555", "ro"]
|
||||
: [`uid=${uid}`, `gid=${gid}`, "file_mode=0664", "dir_mode=0775"];
|
||||
|
||||
if (config.username && config.password) {
|
||||
const password = await cryptoUtils.resolveSecret(config.password);
|
||||
const secretsFile = "/etc/davfs2/secrets";
|
||||
const entry = [source, config.username, password].map((value) => value.replace(/[\r\n\t\s]+/g, " ")).join(" ");
|
||||
const secretsContent = `${entry}\n`;
|
||||
await fs.appendFile(secretsFile, secretsContent, { mode: 0o600 });
|
||||
}
|
||||
|
||||
logger.debug(`Mounting WebDAV volume ${path}...`);
|
||||
|
||||
const args = ["-t", "davfs", "-o", options.join(","), source, path];
|
||||
|
||||
try {
|
||||
await executeMount(args);
|
||||
} catch (error) {
|
||||
logger.warn(`Initial WebDAV mount failed, retrying with -i flag: ${toMessage(error)}`);
|
||||
// Fallback with -i flag if the first mount fails using the mount helper
|
||||
await executeMount(["-i", ...args]);
|
||||
}
|
||||
|
||||
logger.info(`WebDAV volume at ${path} mounted successfully.`);
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "WebDAV mount");
|
||||
} catch (error) {
|
||||
const errorMsg = toMessage(error);
|
||||
|
||||
if (errorMsg.includes("already mounted")) {
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
}
|
||||
|
||||
logger.error("Error mounting WebDAV volume", { error: errorMsg });
|
||||
|
||||
if (errorMsg.includes("option") && errorMsg.includes("requires argument")) {
|
||||
return {
|
||||
status: BACKEND_STATUS.error,
|
||||
error: "Invalid mount options. Please check your WebDAV server configuration.",
|
||||
};
|
||||
} else if (errorMsg.includes("connection refused") || errorMsg.includes("Connection refused")) {
|
||||
return {
|
||||
status: BACKEND_STATUS.error,
|
||||
error: "Cannot connect to WebDAV server. Please check the server address and port.",
|
||||
};
|
||||
} else if (errorMsg.includes("unauthorized") || errorMsg.includes("Unauthorized")) {
|
||||
return {
|
||||
status: BACKEND_STATUS.error,
|
||||
error: "Authentication failed. Please check your username and password.",
|
||||
};
|
||||
}
|
||||
|
||||
return { status: BACKEND_STATUS.error, error: errorMsg };
|
||||
}
|
||||
};
|
||||
|
||||
const unmount = async (path: string) => {
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("WebDAV unmounting is only supported on Linux hosts.");
|
||||
return { status: BACKEND_STATUS.error, error: "WebDAV unmounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
const mount = await getMountForPath(path);
|
||||
if (!mount || mount.mountPoint !== path) {
|
||||
logger.debug(`Path ${path} is not a mount point. Skipping unmount.`);
|
||||
return { status: BACKEND_STATUS.unmounted };
|
||||
}
|
||||
|
||||
await executeUnmount(path);
|
||||
|
||||
await fs.rmdir(path).catch(() => {});
|
||||
|
||||
logger.info(`WebDAV volume at ${path} unmounted successfully.`);
|
||||
return { status: BACKEND_STATUS.unmounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "WebDAV unmount");
|
||||
} catch (error) {
|
||||
logger.error("Error unmounting WebDAV volume", { path, error: toMessage(error) });
|
||||
return { status: BACKEND_STATUS.error, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
const checkHealth = async (path: string) => {
|
||||
const run = async () => {
|
||||
await assertMounted(path, (fstype) => fstype === "fuse" || fstype === "davfs");
|
||||
|
||||
logger.debug(`WebDAV volume at ${path} is healthy and mounted.`);
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "WebDAV health check");
|
||||
} catch (error) {
|
||||
const message = toMessage(error);
|
||||
if (message !== "Volume is not mounted") {
|
||||
logger.error("WebDAV volume health check failed:", message);
|
||||
}
|
||||
return { status: BACKEND_STATUS.error, error: message };
|
||||
}
|
||||
};
|
||||
|
||||
export const makeWebdavBackend = (config: BackendConfig, path: string): VolumeBackend => ({
|
||||
mount: () => mount(config, path),
|
||||
unmount: () => unmount(path),
|
||||
checkHealth: () => checkHealth(path),
|
||||
});
|
||||
|
|
@ -1,7 +1,7 @@
|
|||
import { afterEach, describe, expect, test, vi } from "vitest";
|
||||
import { Scheduler } from "../../../core/scheduler";
|
||||
import * as backendModule from "../../backends/backend";
|
||||
import type { VolumeBackend } from "../../backends/backend";
|
||||
import * as volumeHostModule from "../../../../../apps/agent/src/volume-host";
|
||||
import type { VolumeBackend } from "../../../../../apps/agent/src/volume-host";
|
||||
import * as bootstrapModule from "../bootstrap";
|
||||
import { createTestVolume } from "~/test/helpers/volume";
|
||||
|
||||
|
|
@ -40,7 +40,7 @@ describe("shutdown", () => {
|
|||
|
||||
vi.spyOn(Scheduler, "stop").mockImplementation(stopScheduler);
|
||||
vi.spyOn(bootstrapModule, "stopApplicationRuntime").mockImplementation(stopApplicationRuntime);
|
||||
vi.spyOn(backendModule, "createVolumeBackend").mockImplementation(
|
||||
vi.spyOn(volumeHostModule, "createVolumeBackend").mockImplementation(
|
||||
() =>
|
||||
({
|
||||
mount: async () => ({ status: "mounted" as const }),
|
||||
|
|
|
|||
|
|
@ -1,8 +1,13 @@
|
|||
import { Scheduler } from "../../core/scheduler";
|
||||
import { db } from "../../db/db";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import { createVolumeBackend } from "../backends/backend";
|
||||
import { stopApplicationRuntime } from "./bootstrap";
|
||||
import { decryptVolumeConfig } from "../volumes/volume-config-secrets";
|
||||
import {
|
||||
createVolumeBackend,
|
||||
type AgentVolume,
|
||||
type BackendConfig as HostBackendConfig,
|
||||
} from "../../../../apps/agent/src/volume-host";
|
||||
|
||||
export const shutdown = async () => {
|
||||
await Scheduler.stop();
|
||||
|
|
@ -13,7 +18,11 @@ export const shutdown = async () => {
|
|||
});
|
||||
|
||||
for (const volume of volumes) {
|
||||
const backend = createVolumeBackend(volume);
|
||||
const backend = createVolumeBackend({
|
||||
...volume,
|
||||
config: (await decryptVolumeConfig(volume.config)) as HostBackendConfig,
|
||||
provisioningId: volume.provisioningId ?? null,
|
||||
} satisfies AgentVolume);
|
||||
const { status, error } = await backend.unmount();
|
||||
|
||||
logger.info(`Volume ${volume.name} unmount status: ${status}${error ? `, error: ${error}` : ""}`);
|
||||
|
|
|
|||
|
|
@ -1,19 +1,26 @@
|
|||
import { afterEach, describe, expect, test, vi } from "vitest";
|
||||
const agentManagerMock = vi.hoisted(() => ({
|
||||
runVolumeCommand: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("../../agents/agents-manager", () => ({
|
||||
agentManager: agentManagerMock,
|
||||
}));
|
||||
|
||||
import { volumeService } from "../volume.service";
|
||||
import { db } from "~/server/db/db";
|
||||
import { volumesTable } from "~/server/db/schema";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import * as fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { createTestSession } from "~/test/helpers/auth";
|
||||
import { withContext } from "~/server/core/request-context";
|
||||
import { asShortId } from "~/server/utils/branded";
|
||||
import { createTestVolume } from "~/test/helpers/volume";
|
||||
import * as backendModule from "../../backends/backend";
|
||||
import { config } from "~/server/core/config";
|
||||
|
||||
afterEach(() => {
|
||||
config.flags.enableLocalAgent = false;
|
||||
vi.restoreAllMocks();
|
||||
agentManagerMock.runVolumeCommand.mockReset();
|
||||
});
|
||||
|
||||
describe("volumeService.getVolume", () => {
|
||||
|
|
@ -98,13 +105,7 @@ describe("volumeService.getVolume", () => {
|
|||
describe("volumeService.listFiles security", () => {
|
||||
test("should reject traversal outside the volume root in listFiles", async () => {
|
||||
const { organizationId, user } = await createTestSession();
|
||||
const tempRoot = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-vol-svc-"));
|
||||
const volumePath = path.join(tempRoot, "vol");
|
||||
const secretPath = path.join(tempRoot, "volume-secret");
|
||||
|
||||
await fs.mkdir(volumePath, { recursive: true });
|
||||
await fs.mkdir(secretPath, { recursive: true });
|
||||
await fs.writeFile(path.join(secretPath, "secret.txt"), "top secret", "utf-8");
|
||||
agentManagerMock.runVolumeCommand.mockRejectedValue(new Error("Invalid path"));
|
||||
|
||||
const [volume] = await db
|
||||
.insert(volumesTable)
|
||||
|
|
@ -113,44 +114,40 @@ describe("volumeService.listFiles security", () => {
|
|||
name: `test-vol-${randomUUID().slice(0, 8)}`,
|
||||
type: "directory",
|
||||
status: "mounted",
|
||||
config: { backend: "directory", path: volumePath },
|
||||
config: { backend: "directory", path: "/tmp/volume" },
|
||||
autoRemount: true,
|
||||
organizationId,
|
||||
})
|
||||
.returning();
|
||||
|
||||
try {
|
||||
await withContext({ organizationId, userId: user.id }, async () => {
|
||||
const traversalPath = `../${path.basename(secretPath)}`;
|
||||
|
||||
await expect(volumeService.listFiles(volume.shortId, traversalPath)).rejects.toThrow("Invalid path");
|
||||
});
|
||||
} finally {
|
||||
await fs.rm(tempRoot, { recursive: true, force: true });
|
||||
}
|
||||
await withContext({ organizationId, userId: user.id }, async () => {
|
||||
await expect(volumeService.listFiles(volume.shortId, "../volume-secret")).rejects.toThrow("Invalid path");
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("volumeService.mountVolume", () => {
|
||||
test("unmounts any existing mount before mounting", async () => {
|
||||
test("routes unmount and mount to the owning agent before updating state", async () => {
|
||||
const { organizationId, user } = await createTestSession();
|
||||
const volume = await createTestVolume({ organizationId, status: "mounted" });
|
||||
const unmount = vi.fn().mockResolvedValue({ status: "unmounted" });
|
||||
const mount = vi.fn().mockResolvedValue({ status: "mounted" });
|
||||
|
||||
vi.spyOn(backendModule, "createVolumeBackend").mockImplementation(() => ({
|
||||
mount,
|
||||
unmount,
|
||||
checkHealth: vi.fn().mockResolvedValue({ status: "mounted" }),
|
||||
}));
|
||||
const volume = await createTestVolume({ organizationId, status: "mounted", agentId: "agent-1" });
|
||||
agentManagerMock.runVolumeCommand
|
||||
.mockResolvedValueOnce({ name: "volume.unmount", result: { status: "unmounted" } })
|
||||
.mockResolvedValueOnce({ name: "volume.mount", result: { status: "mounted" } });
|
||||
|
||||
await withContext({ organizationId, userId: user.id }, async () => {
|
||||
const result = await volumeService.mountVolume(volume.shortId);
|
||||
|
||||
expect(result.status).toBe("mounted");
|
||||
expect(unmount).toHaveBeenCalledOnce();
|
||||
expect(mount).toHaveBeenCalledOnce();
|
||||
expect(unmount.mock.invocationCallOrder[0]).toBeLessThan(mount.mock.invocationCallOrder[0]);
|
||||
expect(agentManagerMock.runVolumeCommand).toHaveBeenNthCalledWith(
|
||||
1,
|
||||
volume.agentId,
|
||||
expect.objectContaining({ name: "volume.unmount", volume: expect.objectContaining({ id: volume.id }) }),
|
||||
);
|
||||
expect(agentManagerMock.runVolumeCommand).toHaveBeenNthCalledWith(
|
||||
2,
|
||||
volume.agentId,
|
||||
expect.objectContaining({ name: "volume.mount", volume: expect.objectContaining({ id: volume.id }) }),
|
||||
);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -158,15 +155,8 @@ describe("volumeService.mountVolume", () => {
|
|||
describe("volumeService.ensureHealthyVolume", () => {
|
||||
test("returns ready when the mounted volume passes its health check", async () => {
|
||||
const { organizationId, user } = await createTestSession();
|
||||
const volume = await createTestVolume({ organizationId, status: "mounted" });
|
||||
const mount = vi.fn().mockResolvedValue({ status: "mounted" });
|
||||
const checkHealth = vi.fn().mockResolvedValue({ status: "mounted" });
|
||||
|
||||
vi.spyOn(backendModule, "createVolumeBackend").mockImplementation(() => ({
|
||||
mount,
|
||||
unmount: vi.fn().mockResolvedValue({ status: "unmounted" }),
|
||||
checkHealth,
|
||||
}));
|
||||
const volume = await createTestVolume({ organizationId, status: "mounted", agentId: "agent-1" });
|
||||
agentManagerMock.runVolumeCommand.mockResolvedValue({ name: "volume.checkHealth", result: { status: "mounted" } });
|
||||
|
||||
await withContext({ organizationId, userId: user.id }, async () => {
|
||||
const result = await volumeService.ensureHealthyVolume(volume.shortId);
|
||||
|
|
@ -176,22 +166,21 @@ describe("volumeService.ensureHealthyVolume", () => {
|
|||
volume: expect.objectContaining({ id: volume.id, status: "mounted", lastError: null }),
|
||||
remounted: false,
|
||||
});
|
||||
expect(checkHealth).toHaveBeenCalledOnce();
|
||||
expect(mount).not.toHaveBeenCalled();
|
||||
expect(agentManagerMock.runVolumeCommand).toHaveBeenCalledOnce();
|
||||
expect(agentManagerMock.runVolumeCommand).toHaveBeenCalledWith(
|
||||
volume.agentId,
|
||||
expect.objectContaining({ name: "volume.checkHealth", volume: expect.objectContaining({ id: volume.id }) }),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
test("auto-remounts when the mounted volume fails its health check", async () => {
|
||||
const { organizationId, user } = await createTestSession();
|
||||
const volume = await createTestVolume({ organizationId, status: "mounted", autoRemount: true });
|
||||
const mount = vi.fn().mockResolvedValue({ status: "mounted" });
|
||||
const checkHealth = vi.fn().mockResolvedValue({ status: "error", error: "stale mount" });
|
||||
|
||||
vi.spyOn(backendModule, "createVolumeBackend").mockImplementation(() => ({
|
||||
mount,
|
||||
unmount: vi.fn().mockResolvedValue({ status: "unmounted" }),
|
||||
checkHealth,
|
||||
}));
|
||||
const volume = await createTestVolume({ organizationId, status: "mounted", autoRemount: true, agentId: "agent-1" });
|
||||
agentManagerMock.runVolumeCommand
|
||||
.mockResolvedValueOnce({ name: "volume.checkHealth", result: { status: "error", error: "stale mount" } })
|
||||
.mockResolvedValueOnce({ name: "volume.unmount", result: { status: "unmounted" } })
|
||||
.mockResolvedValueOnce({ name: "volume.mount", result: { status: "mounted" } });
|
||||
|
||||
await withContext({ organizationId, userId: user.id }, async () => {
|
||||
const result = await volumeService.ensureHealthyVolume(volume.shortId);
|
||||
|
|
@ -201,8 +190,7 @@ describe("volumeService.ensureHealthyVolume", () => {
|
|||
volume: expect.objectContaining({ id: volume.id, status: "mounted", lastError: null }),
|
||||
remounted: true,
|
||||
});
|
||||
expect(checkHealth).toHaveBeenCalledOnce();
|
||||
expect(mount).toHaveBeenCalledOnce();
|
||||
expect(agentManagerMock.runVolumeCommand).toHaveBeenCalledTimes(3);
|
||||
|
||||
const updatedVolume = await db.query.volumesTable.findFirst({ where: { id: volume.id } });
|
||||
expect(updatedVolume?.status).toBe("mounted");
|
||||
|
|
@ -212,15 +200,16 @@ describe("volumeService.ensureHealthyVolume", () => {
|
|||
|
||||
test("returns not ready when the health check fails and auto-remount is disabled", async () => {
|
||||
const { organizationId, user } = await createTestSession();
|
||||
const volume = await createTestVolume({ organizationId, status: "mounted", autoRemount: false });
|
||||
const mount = vi.fn().mockResolvedValue({ status: "mounted" });
|
||||
const checkHealth = vi.fn().mockResolvedValue({ status: "error", error: "stale mount" });
|
||||
|
||||
vi.spyOn(backendModule, "createVolumeBackend").mockImplementation(() => ({
|
||||
mount,
|
||||
unmount: vi.fn().mockResolvedValue({ status: "unmounted" }),
|
||||
checkHealth,
|
||||
}));
|
||||
const volume = await createTestVolume({
|
||||
organizationId,
|
||||
status: "mounted",
|
||||
autoRemount: false,
|
||||
agentId: "agent-1",
|
||||
});
|
||||
agentManagerMock.runVolumeCommand.mockResolvedValue({
|
||||
name: "volume.checkHealth",
|
||||
result: { status: "error", error: "stale mount" },
|
||||
});
|
||||
|
||||
await withContext({ organizationId, userId: user.id }, async () => {
|
||||
const result = await volumeService.ensureHealthyVolume(volume.shortId);
|
||||
|
|
@ -230,54 +219,17 @@ describe("volumeService.ensureHealthyVolume", () => {
|
|||
volume: expect.objectContaining({ id: volume.id, status: "error", lastError: "stale mount" }),
|
||||
reason: "stale mount",
|
||||
});
|
||||
expect(checkHealth).toHaveBeenCalledOnce();
|
||||
expect(mount).not.toHaveBeenCalled();
|
||||
expect(agentManagerMock.runVolumeCommand).toHaveBeenCalledOnce();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("volumeService.testConnection", () => {
|
||||
test("uses an isolated temp mount path for backend test connections", async () => {
|
||||
const mount = vi.fn().mockResolvedValue({ status: "mounted" });
|
||||
const unmount = vi.fn().mockResolvedValue({ status: "unmounted" });
|
||||
const createVolumeBackendSpy = vi.spyOn(backendModule, "createVolumeBackend").mockReturnValue({
|
||||
mount,
|
||||
unmount,
|
||||
checkHealth: vi.fn(),
|
||||
});
|
||||
|
||||
await volumeService.testConnection({
|
||||
backend: "nfs",
|
||||
server: "127.0.0.1",
|
||||
exportPath: "/exports/test",
|
||||
version: "4",
|
||||
port: 2049,
|
||||
readOnly: false,
|
||||
});
|
||||
|
||||
expect(createVolumeBackendSpy).toHaveBeenCalledOnce();
|
||||
const [, mountPath] = createVolumeBackendSpy.mock.calls[0];
|
||||
expect(mountPath).toEqual(expect.stringContaining(`${path.sep}zerobyte-test-`));
|
||||
await expect(fs.access(mountPath as string)).rejects.toThrow();
|
||||
expect(mount).toHaveBeenCalledOnce();
|
||||
expect(unmount).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
test("does not fail when backend unmount already removed the temp mount path", async () => {
|
||||
const mount = vi.fn().mockResolvedValue({ status: "mounted" });
|
||||
let mountPath: string | undefined;
|
||||
const unmount = vi.fn().mockImplementation(async () => {
|
||||
await fs.rm(mountPath!, { recursive: true, force: true });
|
||||
return { status: "unmounted" };
|
||||
});
|
||||
|
||||
vi.spyOn(backendModule, "createVolumeBackend").mockImplementation((_volume, tempPath) => {
|
||||
mountPath = tempPath;
|
||||
return {
|
||||
mount,
|
||||
unmount,
|
||||
checkHealth: vi.fn(),
|
||||
};
|
||||
test("routes test connections to the local agent", async () => {
|
||||
config.flags.enableLocalAgent = true;
|
||||
agentManagerMock.runVolumeCommand.mockResolvedValue({
|
||||
name: "volume.testConnection",
|
||||
result: { success: true, message: "Connection successful" },
|
||||
});
|
||||
|
||||
await expect(
|
||||
|
|
@ -294,9 +246,9 @@ describe("volumeService.testConnection", () => {
|
|||
message: "Connection successful",
|
||||
});
|
||||
|
||||
expect(mountPath).toEqual(expect.stringContaining(`${path.sep}zerobyte-test-`));
|
||||
await expect(fs.access(mountPath as string)).rejects.toThrow();
|
||||
expect(mount).toHaveBeenCalledOnce();
|
||||
expect(unmount).toHaveBeenCalledOnce();
|
||||
expect(agentManagerMock.runVolumeCommand).toHaveBeenCalledWith(
|
||||
"local",
|
||||
expect.objectContaining({ name: "volume.testConnection" }),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -32,3 +32,7 @@ export const mapVolumeConfigSecrets = async (
|
|||
export const encryptVolumeConfig = async (config: BackendConfig): Promise<BackendConfig> => {
|
||||
return await mapVolumeConfigSecrets(config, cryptoUtils.sealSecret);
|
||||
};
|
||||
|
||||
export const decryptVolumeConfig = async (config: BackendConfig): Promise<BackendConfig> => {
|
||||
return await mapVolumeConfigSecrets(config, cryptoUtils.resolveSecret);
|
||||
};
|
||||
|
|
|
|||
|
|
@ -1,26 +1,35 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import * as path from "node:path";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { BadRequestError, InternalServerError, NotFoundError } from "http-errors-enhanced";
|
||||
import { db } from "../../db/db";
|
||||
import { volumesTable } from "../../db/schema";
|
||||
import { toMessage } from "../../utils/errors";
|
||||
import { generateShortId } from "../../utils/id";
|
||||
import { getStatFs, type StatFs } from "../../utils/mountinfo";
|
||||
import type { StatFs } from "../../utils/mountinfo";
|
||||
import { withTimeout } from "../../utils/timeout";
|
||||
import { createVolumeBackend } from "../backends/backend";
|
||||
import { config } from "../../core/config";
|
||||
import { LOCAL_AGENT_ID } from "../agents/constants";
|
||||
import { agentManager } from "../agents/agents-manager";
|
||||
import type { UpdateVolumeBody } from "./volume.dto";
|
||||
import { getVolumePath } from "./helpers";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import { serverEvents } from "../../core/events";
|
||||
import type { Volume } from "../../db/schema";
|
||||
import { volumeConfigSchema, type BackendConfig } from "~/schemas/volumes";
|
||||
import { getOrganizationId } from "~/server/core/request-context";
|
||||
import { isNodeJSErrnoException } from "~/server/utils/fs";
|
||||
import { asShortId, type ShortId } from "~/server/utils/branded";
|
||||
import { encryptVolumeConfig } from "./volume-config-secrets";
|
||||
import { type ShortId } from "~/server/utils/branded";
|
||||
import { decryptVolumeConfig, encryptVolumeConfig } from "./volume-config-secrets";
|
||||
import type { VolumeCommand, VolumeCommandResult } from "@zerobyte/contracts/agent-protocol";
|
||||
import {
|
||||
createVolumeBackend,
|
||||
getStatFs,
|
||||
getVolumePath,
|
||||
type AgentVolume,
|
||||
type BackendConfig as HostBackendConfig,
|
||||
} from "../../../../apps/agent/src/volume-host";
|
||||
import {
|
||||
browseFilesystem as browseHostFilesystem,
|
||||
listVolumeFiles,
|
||||
testVolumeConnection,
|
||||
} from "../../../../apps/agent/src/volume-host/operations";
|
||||
|
||||
type EnsureHealthyVolumeResult =
|
||||
| {
|
||||
|
|
@ -34,6 +43,9 @@ type EnsureHealthyVolumeResult =
|
|||
reason: string;
|
||||
};
|
||||
|
||||
type VolumeBackendCommandName = "volume.mount" | "volume.unmount" | "volume.checkHealth";
|
||||
type VolumeBackendCommandResult = Extract<VolumeCommandResult, { name: VolumeBackendCommandName }>["result"];
|
||||
|
||||
const listVolumes = async () => {
|
||||
const organizationId = getOrganizationId();
|
||||
const volumes = await db.query.volumesTable.findMany({
|
||||
|
|
@ -53,6 +65,64 @@ const findVolume = async (shortId: ShortId) => {
|
|||
});
|
||||
};
|
||||
|
||||
const runVolumeCommand = async <TCommand extends VolumeCommand>(agentId: string, command: TCommand) => {
|
||||
const result = await agentManager.runVolumeCommand(agentId, command);
|
||||
if (result.name !== command.name) {
|
||||
throw new InternalServerError(`Unexpected agent response for ${command.name}`);
|
||||
}
|
||||
|
||||
return result as Extract<VolumeCommandResult, { name: TCommand["name"] }>;
|
||||
};
|
||||
|
||||
const volumeForAgent = async (volume: Volume): Promise<Volume> => ({
|
||||
...volume,
|
||||
config: await decryptVolumeConfig(volume.config),
|
||||
});
|
||||
|
||||
const volumeForHost = async (volume: Volume): Promise<AgentVolume> => ({
|
||||
...volume,
|
||||
shortId: volume.shortId,
|
||||
config: (await decryptVolumeConfig(volume.config)) as HostBackendConfig,
|
||||
provisioningId: volume.provisioningId ?? null,
|
||||
});
|
||||
|
||||
// TODO(agent-rollout): Remove the local host execution branch once all installs run volume operations through agents.
|
||||
const shouldRunViaAgent = (volume: Volume) => volume.agentId !== LOCAL_AGENT_ID || config.flags.enableLocalAgent;
|
||||
|
||||
const runVolumeBackendCommand = async (
|
||||
volume: Volume,
|
||||
name: "volume.mount" | "volume.unmount" | "volume.checkHealth",
|
||||
) => {
|
||||
if (!shouldRunViaAgent(volume)) {
|
||||
const backend = createVolumeBackend(await volumeForHost(volume));
|
||||
switch (name) {
|
||||
case "volume.mount":
|
||||
return backend.mount();
|
||||
case "volume.unmount":
|
||||
return backend.unmount();
|
||||
case "volume.checkHealth":
|
||||
return backend.checkHealth();
|
||||
}
|
||||
}
|
||||
|
||||
const command = await runVolumeCommand(volume.agentId, {
|
||||
name,
|
||||
volume: await volumeForAgent(volume),
|
||||
} as VolumeCommand);
|
||||
return command.result as VolumeBackendCommandResult;
|
||||
};
|
||||
|
||||
const mapAgentFileError = (error: unknown) => {
|
||||
const message = toMessage(error);
|
||||
if (message === "Invalid path") {
|
||||
throw new BadRequestError("Invalid path");
|
||||
}
|
||||
if (message === "Directory not found") {
|
||||
throw new NotFoundError("Directory not found");
|
||||
}
|
||||
throw error;
|
||||
};
|
||||
|
||||
const createVolume = async (name: string, backendConfig: BackendConfig) => {
|
||||
const organizationId = getOrganizationId();
|
||||
const trimmedName = name.trim();
|
||||
|
|
@ -80,8 +150,7 @@ const createVolume = async (name: string, backendConfig: BackendConfig) => {
|
|||
throw new InternalServerError("Failed to create volume");
|
||||
}
|
||||
|
||||
const backend = createVolumeBackend(created);
|
||||
const { error, status } = await backend.mount();
|
||||
const { error, status } = await runVolumeBackendCommand(created, "volume.mount");
|
||||
|
||||
await db
|
||||
.update(volumesTable)
|
||||
|
|
@ -99,8 +168,7 @@ const deleteVolume = async (shortId: ShortId) => {
|
|||
throw new NotFoundError("Volume not found");
|
||||
}
|
||||
|
||||
const backend = createVolumeBackend(volume);
|
||||
await backend.unmount();
|
||||
await runVolumeBackendCommand(volume, "volume.unmount");
|
||||
await db
|
||||
.delete(volumesTable)
|
||||
.where(and(eq(volumesTable.id, volume.id), eq(volumesTable.organizationId, organizationId)));
|
||||
|
|
@ -114,9 +182,8 @@ const mountVolume = async (shortId: ShortId) => {
|
|||
throw new NotFoundError("Volume not found");
|
||||
}
|
||||
|
||||
const backend = createVolumeBackend(volume);
|
||||
await backend.unmount();
|
||||
const { error, status } = await backend.mount();
|
||||
await runVolumeBackendCommand(volume, "volume.unmount");
|
||||
const { error, status } = await runVolumeBackendCommand(volume, "volume.mount");
|
||||
|
||||
await db
|
||||
.update(volumesTable)
|
||||
|
|
@ -138,8 +205,7 @@ const unmountVolume = async (shortId: ShortId) => {
|
|||
throw new NotFoundError("Volume not found");
|
||||
}
|
||||
|
||||
const backend = createVolumeBackend(volume);
|
||||
const { status, error } = await backend.unmount();
|
||||
const { status, error } = await runVolumeBackendCommand(volume, "volume.unmount");
|
||||
|
||||
await db
|
||||
.update(volumesTable)
|
||||
|
|
@ -162,7 +228,15 @@ const getVolume = async (shortId: ShortId) => {
|
|||
|
||||
let statfs: Partial<StatFs> = {};
|
||||
if (volume.status === "mounted") {
|
||||
statfs = await withTimeout(getStatFs(getVolumePath(volume)), 1000, "getStatFs").catch((error) => {
|
||||
statfs = await withTimeout(
|
||||
shouldRunViaAgent(volume)
|
||||
? runVolumeCommand(volume.agentId, { name: "volume.statfs", volume: await volumeForAgent(volume) }).then(
|
||||
(command) => command.result,
|
||||
)
|
||||
: volumeForHost(volume).then((hostVolume) => getStatFs(getVolumePath(hostVolume))),
|
||||
1000,
|
||||
"volume.statfs",
|
||||
).catch((error) => {
|
||||
logger.warn(`Failed to get statfs for volume ${volume.name}: ${toMessage(error)}`);
|
||||
return {};
|
||||
});
|
||||
|
|
@ -190,8 +264,7 @@ const updateVolume = async (shortId: ShortId, volumeData: UpdateVolumeBody) => {
|
|||
|
||||
if (configChanged) {
|
||||
logger.debug("Unmounting existing volume before applying new config");
|
||||
const backend = createVolumeBackend(existing);
|
||||
await backend.unmount();
|
||||
await runVolumeBackendCommand(existing, "volume.unmount");
|
||||
}
|
||||
|
||||
const newConfigResult = volumeConfigSchema.safeParse(volumeData.config || existing.config);
|
||||
|
|
@ -219,8 +292,7 @@ const updateVolume = async (shortId: ShortId, volumeData: UpdateVolumeBody) => {
|
|||
}
|
||||
|
||||
if (configChanged) {
|
||||
const backend = createVolumeBackend(updated);
|
||||
const { error, status } = await backend.mount();
|
||||
const { error, status } = await runVolumeBackendCommand(updated, "volume.mount");
|
||||
await db
|
||||
.update(volumesTable)
|
||||
.set({ status, lastError: error ?? null, lastHealthCheck: Date.now() })
|
||||
|
|
@ -233,40 +305,12 @@ const updateVolume = async (shortId: ShortId, volumeData: UpdateVolumeBody) => {
|
|||
};
|
||||
|
||||
const testConnection = async (backendConfig: BackendConfig) => {
|
||||
const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-test-"));
|
||||
try {
|
||||
const encryptedConfig = await encryptVolumeConfig(backendConfig);
|
||||
|
||||
const mockVolume = {
|
||||
id: 0,
|
||||
shortId: asShortId("test"),
|
||||
name: "test-connection",
|
||||
path: tempDir,
|
||||
config: encryptedConfig,
|
||||
createdAt: Date.now(),
|
||||
updatedAt: Date.now(),
|
||||
lastHealthCheck: Date.now(),
|
||||
type: encryptedConfig.backend,
|
||||
status: "unmounted" as const,
|
||||
lastError: null,
|
||||
provisioningId: null,
|
||||
autoRemount: true,
|
||||
agentId: LOCAL_AGENT_ID,
|
||||
organizationId: "test-org",
|
||||
};
|
||||
|
||||
const backend = createVolumeBackend(mockVolume, tempDir);
|
||||
const { error } = await backend.mount();
|
||||
|
||||
await backend.unmount();
|
||||
|
||||
return {
|
||||
success: !error,
|
||||
message: error ? toMessage(error) : "Connection successful",
|
||||
};
|
||||
} finally {
|
||||
await fs.rm(tempDir, { recursive: true, force: true });
|
||||
if (!config.flags.enableLocalAgent) {
|
||||
return await testVolumeConnection(backendConfig as HostBackendConfig);
|
||||
}
|
||||
|
||||
const command = await runVolumeCommand(LOCAL_AGENT_ID, { name: "volume.testConnection", backendConfig });
|
||||
return command.result;
|
||||
};
|
||||
|
||||
const checkHealth = async (shortId: ShortId) => {
|
||||
|
|
@ -277,8 +321,7 @@ const checkHealth = async (shortId: ShortId) => {
|
|||
throw new NotFoundError("Volume not found");
|
||||
}
|
||||
|
||||
const backend = createVolumeBackend(volume);
|
||||
const { error, status } = await backend.checkHealth();
|
||||
const { error, status } = await runVolumeBackendCommand(volume, "volume.checkHealth");
|
||||
|
||||
if (status !== volume.status) {
|
||||
serverEvents.emit("volume:status_changed", { organizationId, volumeName: volume.name, status });
|
||||
|
|
@ -346,7 +389,6 @@ const ensureHealthyVolume = async (shortId: ShortId): Promise<EnsureHealthyVolum
|
|||
};
|
||||
|
||||
const DEFAULT_PAGE_SIZE = 500;
|
||||
const MAX_PAGE_SIZE = 500;
|
||||
|
||||
const listFiles = async (shortId: ShortId, subPath?: string, offset: number = 0, limit: number = DEFAULT_PAGE_SIZE) => {
|
||||
const volume = await findVolume(shortId);
|
||||
|
|
@ -359,110 +401,33 @@ const listFiles = async (shortId: ShortId, subPath?: string, offset: number = 0,
|
|||
throw new InternalServerError("Volume is not mounted");
|
||||
}
|
||||
|
||||
const volumePath = getVolumePath(volume);
|
||||
const requestedPath = subPath ? path.join(volumePath, subPath) : volumePath;
|
||||
const normalizedPath = path.normalize(requestedPath);
|
||||
const relative = path.relative(volumePath, normalizedPath);
|
||||
|
||||
if (relative.startsWith("..") || path.isAbsolute(relative)) {
|
||||
throw new BadRequestError("Invalid path");
|
||||
}
|
||||
|
||||
const pageSize = Math.min(Math.max(limit, 1), MAX_PAGE_SIZE);
|
||||
const startOffset = Math.max(offset, 0);
|
||||
|
||||
try {
|
||||
const dirents = await fs.readdir(normalizedPath, { withFileTypes: true });
|
||||
|
||||
dirents.sort((a, b) => {
|
||||
const aIsDir = a.isDirectory();
|
||||
const bIsDir = b.isDirectory();
|
||||
|
||||
if (aIsDir === bIsDir) {
|
||||
return a.name.localeCompare(b.name);
|
||||
}
|
||||
return aIsDir ? -1 : 1;
|
||||
});
|
||||
|
||||
const total = dirents.length;
|
||||
const paginatedDirents = dirents.slice(startOffset, startOffset + pageSize);
|
||||
|
||||
const entries = (
|
||||
await Promise.all(
|
||||
paginatedDirents.map(async (dirent) => {
|
||||
const fullPath = path.join(normalizedPath, dirent.name);
|
||||
|
||||
try {
|
||||
const stats = await fs.stat(fullPath);
|
||||
const relativePath = path.relative(volumePath, fullPath);
|
||||
|
||||
return {
|
||||
name: dirent.name,
|
||||
path: `/${relativePath}`,
|
||||
type: dirent.isDirectory() ? ("directory" as const) : ("file" as const),
|
||||
size: dirent.isFile() ? stats.size : undefined,
|
||||
modifiedAt: stats.mtimeMs,
|
||||
};
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}),
|
||||
)
|
||||
).filter((e) => e !== null);
|
||||
|
||||
return {
|
||||
files: entries,
|
||||
path: subPath || "/",
|
||||
offset: startOffset,
|
||||
limit: pageSize,
|
||||
total,
|
||||
hasMore: startOffset + pageSize < total,
|
||||
};
|
||||
} catch (error) {
|
||||
if (isNodeJSErrnoException(error) && error.code === "ENOENT") {
|
||||
throw new NotFoundError("Directory not found");
|
||||
if (!shouldRunViaAgent(volume)) {
|
||||
return await listVolumeFiles(await volumeForHost(volume), subPath, offset, limit);
|
||||
}
|
||||
|
||||
const command = await runVolumeCommand(volume.agentId, {
|
||||
name: "volume.listFiles",
|
||||
volume: await volumeForAgent(volume),
|
||||
subPath,
|
||||
offset,
|
||||
limit,
|
||||
});
|
||||
return command.result;
|
||||
} catch (error) {
|
||||
mapAgentFileError(error);
|
||||
throw new InternalServerError(`Failed to list files: ${toMessage(error)}`);
|
||||
}
|
||||
};
|
||||
|
||||
const browseFilesystem = async (browsePath: string) => {
|
||||
const normalizedPath = path.normalize(browsePath);
|
||||
|
||||
try {
|
||||
const entries = await fs.readdir(normalizedPath, { withFileTypes: true });
|
||||
if (!config.flags.enableLocalAgent) {
|
||||
return await browseHostFilesystem(browsePath);
|
||||
}
|
||||
|
||||
const directories = await Promise.all(
|
||||
entries
|
||||
.filter((entry) => entry.isDirectory())
|
||||
.map(async (entry) => {
|
||||
const fullPath = path.join(normalizedPath, entry.name);
|
||||
|
||||
try {
|
||||
const stats = await fs.stat(fullPath);
|
||||
return {
|
||||
name: entry.name,
|
||||
path: fullPath,
|
||||
type: "directory" as const,
|
||||
size: undefined,
|
||||
modifiedAt: stats.mtimeMs,
|
||||
};
|
||||
} catch {
|
||||
return {
|
||||
name: entry.name,
|
||||
path: fullPath,
|
||||
type: "directory" as const,
|
||||
size: undefined,
|
||||
modifiedAt: undefined,
|
||||
};
|
||||
}
|
||||
}),
|
||||
);
|
||||
|
||||
return {
|
||||
directories: directories.sort((a, b) => a.name.localeCompare(b.name)),
|
||||
path: normalizedPath,
|
||||
};
|
||||
const command = await runVolumeCommand(LOCAL_AGENT_ID, { name: "filesystem.browse", path: browsePath });
|
||||
return command.result;
|
||||
} catch (error) {
|
||||
throw new InternalServerError(`Failed to browse filesystem: ${toMessage(error)}`);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ import { handleBackupCancelCommand } from "./backup-cancel";
|
|||
import { handleBackupRunCommand } from "./backup-run";
|
||||
import type { ControllerCommandContext } from "../context";
|
||||
import { handleHeartbeatPingCommand } from "./heartbeat-ping";
|
||||
import { handleVolumeCommand } from "./volume";
|
||||
|
||||
export const handleControllerCommand = (context: ControllerCommandContext, message: ControllerMessage) => {
|
||||
switch (message.type) {
|
||||
|
|
@ -12,6 +13,9 @@ export const handleControllerCommand = (context: ControllerCommandContext, messa
|
|||
case "backup.cancel": {
|
||||
return handleBackupCancelCommand(context, message.payload);
|
||||
}
|
||||
case "volume.command": {
|
||||
return handleVolumeCommand(context, message.payload);
|
||||
}
|
||||
case "heartbeat.ping": {
|
||||
return handleHeartbeatPingCommand(context, message.payload);
|
||||
}
|
||||
|
|
|
|||
76
apps/agent/src/commands/volume.ts
Normal file
76
apps/agent/src/commands/volume.ts
Normal file
|
|
@ -0,0 +1,76 @@
|
|||
import { Effect } from "effect";
|
||||
import {
|
||||
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";
|
||||
import type { ControllerCommandContext } from "../context";
|
||||
|
||||
type VolumeBackedCommand = Extract<VolumeCommand, { volume: unknown }>;
|
||||
|
||||
const asVolume = (volume: VolumeBackedCommand["volume"]): AgentVolume => ({
|
||||
...volume,
|
||||
config: volume.config as BackendConfig,
|
||||
provisioningId: volume.provisioningId ?? null,
|
||||
});
|
||||
|
||||
const runBackendOperation = async (
|
||||
command: Extract<VolumeCommand, { volume: unknown }>,
|
||||
operation: "mount" | "unmount" | "checkHealth",
|
||||
) => {
|
||||
const backend = createVolumeBackend(asVolume(command.volume));
|
||||
return backend[operation]();
|
||||
};
|
||||
|
||||
const executeVolumeCommand = async (command: VolumeCommand): Promise<VolumeCommandResult> => {
|
||||
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) };
|
||||
}
|
||||
};
|
||||
|
||||
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),
|
||||
}),
|
||||
),
|
||||
);
|
||||
}
|
||||
});
|
||||
};
|
||||
|
|
@ -1,12 +1,11 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import { toMessage } from "../../../utils/errors";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import type { VolumeBackend } from "../backend";
|
||||
import { BACKEND_STATUS, type BackendConfig } from "~/schemas/volumes";
|
||||
import { toMessage } from "@zerobyte/core/utils";
|
||||
import type { BackendConfig, VolumeBackend } from "../types";
|
||||
|
||||
const mount = async (config: BackendConfig, _volumePath: string) => {
|
||||
const mount = async (config: BackendConfig, volumePath: string) => {
|
||||
if (config.backend !== "directory") {
|
||||
return { status: BACKEND_STATUS.error, error: "Invalid backend type" };
|
||||
return { status: "error" as const, error: "Invalid backend type" };
|
||||
}
|
||||
|
||||
logger.info("Mounting directory volume from:", config.path);
|
||||
|
|
@ -16,33 +15,33 @@ const mount = async (config: BackendConfig, _volumePath: string) => {
|
|||
const stats = await fs.stat(config.path);
|
||||
|
||||
if (!stats.isDirectory()) {
|
||||
return { status: BACKEND_STATUS.error, error: "Path is not a directory" };
|
||||
return { status: "error" as const, error: "Path is not a directory" };
|
||||
}
|
||||
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
return { status: "mounted" as const };
|
||||
} catch (error) {
|
||||
logger.error("Failed to mount directory volume:", error);
|
||||
return { status: BACKEND_STATUS.error, error: toMessage(error) };
|
||||
return { status: "error" as const, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
const unmount = async () => {
|
||||
logger.info("Cannot unmount directory volume.");
|
||||
return { status: BACKEND_STATUS.unmounted };
|
||||
return { status: "unmounted" as const };
|
||||
};
|
||||
|
||||
const checkHealth = async (config: BackendConfig) => {
|
||||
if (config.backend !== "directory") {
|
||||
return { status: BACKEND_STATUS.error, error: "Invalid backend type" };
|
||||
return { status: "error" as const, error: "Invalid backend type" };
|
||||
}
|
||||
|
||||
try {
|
||||
await fs.access(config.path);
|
||||
|
||||
return { status: BACKEND_STATUS.mounted };
|
||||
return { status: "mounted" as const };
|
||||
} catch (error) {
|
||||
logger.error("Directory health check failed:", error);
|
||||
return { status: BACKEND_STATUS.error, error: toMessage(error) };
|
||||
return { status: "error" as const, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
112
apps/agent/src/volume-host/backends/nfs.ts
Normal file
112
apps/agent/src/volume-host/backends/nfs.ts
Normal file
|
|
@ -0,0 +1,112 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import { toMessage } from "@zerobyte/core/utils";
|
||||
import { OPERATION_TIMEOUT } from "../constants";
|
||||
import { withTimeout } from "../timeout";
|
||||
import { getMountForPath } from "../fs";
|
||||
import type { BackendConfig, VolumeBackend } from "../types";
|
||||
import { assertMounted, executeMount, executeUnmount } from "./utils";
|
||||
|
||||
const checkHealth = async (mountPath: string) => {
|
||||
const run = async () => {
|
||||
await assertMounted(mountPath, (fstype) => fstype.startsWith("nfs"));
|
||||
|
||||
logger.debug(`NFS volume at ${mountPath} is healthy and mounted.`);
|
||||
return { status: "mounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "NFS health check");
|
||||
} catch (error) {
|
||||
const message = toMessage(error);
|
||||
if (message !== "Volume is not mounted") {
|
||||
logger.error("NFS volume health check failed:", message);
|
||||
}
|
||||
return { status: "error" as const, error: message };
|
||||
}
|
||||
};
|
||||
|
||||
const unmount = async (mountPath: string) => {
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("NFS unmounting is only supported on Linux hosts.");
|
||||
return { status: "error" as const, error: "NFS unmounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
const mount = await getMountForPath(mountPath);
|
||||
if (!mount || mount.mountPoint !== mountPath) {
|
||||
logger.debug(`Path ${mountPath} is not a mount point. Skipping unmount.`);
|
||||
return { status: "unmounted" as const };
|
||||
}
|
||||
|
||||
await executeUnmount(mountPath);
|
||||
await fs.rmdir(mountPath).catch(() => {});
|
||||
|
||||
logger.info(`NFS volume at ${mountPath} unmounted successfully.`);
|
||||
return { status: "unmounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "NFS unmount");
|
||||
} catch (error) {
|
||||
logger.error("Error unmounting NFS volume", { mountPath, error: toMessage(error) });
|
||||
return { status: "error" as const, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
const mount = async (config: BackendConfig, mountPath: string) => {
|
||||
logger.debug(`Mounting volume ${mountPath}...`);
|
||||
|
||||
if (config.backend !== "nfs") {
|
||||
logger.error("Provided config is not for NFS backend");
|
||||
return { status: "error" as const, error: "Provided config is not for NFS backend" };
|
||||
}
|
||||
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("NFS mounting is only supported on Linux hosts.");
|
||||
return { status: "error" as const, error: "NFS mounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const { status } = await checkHealth(mountPath);
|
||||
if (status === "mounted") return { status: "mounted" as const };
|
||||
|
||||
if (status === "error") {
|
||||
logger.debug(`Trying to unmount any existing mounts at ${mountPath} before mounting...`);
|
||||
await unmount(mountPath);
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
await fs.mkdir(mountPath, { recursive: true });
|
||||
const options = [`vers=${config.version}`, `port=${config.port}`];
|
||||
if (config.version === "3") options.push("nolock");
|
||||
if (config.readOnly) options.push("ro");
|
||||
const args = ["-t", "nfs", "-o", options.join(","), `${config.server}:${config.exportPath}`, mountPath];
|
||||
|
||||
logger.debug(`Mounting volume ${mountPath}...`);
|
||||
logger.info(`Executing mount: mount ${args.join(" ")}`);
|
||||
|
||||
try {
|
||||
await executeMount(args);
|
||||
} catch (error) {
|
||||
logger.warn(`Initial NFS mount failed, retrying with -i flag: ${toMessage(error)}`);
|
||||
await executeMount(["-i", ...args]);
|
||||
}
|
||||
|
||||
logger.info(`NFS volume at ${mountPath} mounted successfully.`);
|
||||
return { status: "mounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "NFS mount");
|
||||
} catch (error) {
|
||||
logger.error("Error mounting NFS volume", { error: toMessage(error) });
|
||||
return { status: "error" as const, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
export const makeNfsBackend = (config: BackendConfig, mountPath: string): VolumeBackend => ({
|
||||
mount: () => mount(config, mountPath),
|
||||
unmount: () => unmount(mountPath),
|
||||
checkHealth: () => checkHealth(mountPath),
|
||||
});
|
||||
124
apps/agent/src/volume-host/backends/rclone.ts
Normal file
124
apps/agent/src/volume-host/backends/rclone.ts
Normal file
|
|
@ -0,0 +1,124 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import { logger, safeExec } from "@zerobyte/core/node";
|
||||
import { toMessage } from "@zerobyte/core/utils";
|
||||
import { OPERATION_TIMEOUT, RCLONE_CONFIG_FILE, RCLONE_TIMEOUT } from "../constants";
|
||||
import { withTimeout } from "../timeout";
|
||||
import { getMountForPath } from "../fs";
|
||||
import type { BackendConfig, VolumeBackend } from "../types";
|
||||
import { assertMounted, executeUnmount } from "./utils";
|
||||
|
||||
const checkHealth = async (mountPath: string) => {
|
||||
const run = async () => {
|
||||
await assertMounted(mountPath, (fstype) => fstype.includes("rclone"));
|
||||
|
||||
logger.debug(`Rclone volume at ${mountPath} is healthy and mounted.`);
|
||||
return { status: "mounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "Rclone health check");
|
||||
} catch (error) {
|
||||
const message = toMessage(error);
|
||||
if (message !== "Volume is not mounted") {
|
||||
logger.error("Rclone volume health check failed:", message);
|
||||
}
|
||||
return { status: "error" as const, error: message };
|
||||
}
|
||||
};
|
||||
|
||||
const unmount = async (mountPath: string) => {
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("Rclone unmounting is only supported on Linux hosts.");
|
||||
return { status: "error" as const, error: "Rclone unmounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
const mount = await getMountForPath(mountPath);
|
||||
if (!mount || mount.mountPoint !== mountPath) {
|
||||
logger.debug(`Path ${mountPath} is not a mount point. Skipping unmount.`);
|
||||
return { status: "unmounted" as const };
|
||||
}
|
||||
|
||||
await executeUnmount(mountPath);
|
||||
await fs.rmdir(mountPath).catch(() => {});
|
||||
|
||||
logger.info(`Rclone volume at ${mountPath} unmounted successfully.`);
|
||||
return { status: "unmounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "Rclone unmount");
|
||||
} catch (error) {
|
||||
logger.error("Error unmounting rclone volume", { mountPath, error: toMessage(error) });
|
||||
return { status: "error" as const, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
const mount = async (config: BackendConfig, mountPath: string) => {
|
||||
logger.debug(`Mounting rclone volume ${mountPath}...`);
|
||||
|
||||
if (config.backend !== "rclone") {
|
||||
logger.error("Provided config is not for rclone backend");
|
||||
return { status: "error" as const, error: "Provided config is not for rclone backend" };
|
||||
}
|
||||
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("Rclone mounting is only supported on Linux hosts.");
|
||||
return { status: "error" as const, error: "Rclone mounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const { status } = await checkHealth(mountPath);
|
||||
if (status === "mounted") return { status: "mounted" as const };
|
||||
|
||||
if (status === "error") {
|
||||
logger.debug(`Trying to unmount any existing mounts at ${mountPath} before mounting...`);
|
||||
await unmount(mountPath);
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
await fs.mkdir(mountPath, { recursive: true });
|
||||
const args = [
|
||||
"mount",
|
||||
`${config.remote}:${config.path}`,
|
||||
mountPath,
|
||||
"--daemon",
|
||||
"--vfs-cache-mode",
|
||||
"writes",
|
||||
"--allow-non-empty",
|
||||
"--allow-other",
|
||||
];
|
||||
if (config.readOnly) args.push("--read-only");
|
||||
|
||||
logger.debug(`Mounting rclone volume ${mountPath}...`);
|
||||
logger.info(`Executing rclone: rclone ${args.join(" ")}`);
|
||||
|
||||
const result = await safeExec({
|
||||
command: "rclone",
|
||||
args,
|
||||
env: { RCLONE_CONFIG: RCLONE_CONFIG_FILE },
|
||||
timeout: RCLONE_TIMEOUT,
|
||||
});
|
||||
if (result.exitCode !== 0) {
|
||||
const errorMsg = result.stderr.toString() || result.stdout.toString() || "Unknown error";
|
||||
throw new Error(`Failed to mount rclone volume: ${errorMsg}`);
|
||||
}
|
||||
|
||||
logger.info(`Rclone volume at ${mountPath} mounted successfully.`);
|
||||
return { status: "mounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), RCLONE_TIMEOUT, "Rclone mount");
|
||||
} catch (error) {
|
||||
const errorMsg = toMessage(error);
|
||||
logger.error("Error mounting rclone volume", { error: errorMsg });
|
||||
return { status: "error" as const, error: errorMsg };
|
||||
}
|
||||
};
|
||||
|
||||
export const makeRcloneBackend = (config: BackendConfig, mountPath: string): VolumeBackend => ({
|
||||
mount: () => mount(config, mountPath),
|
||||
unmount: () => unmount(mountPath),
|
||||
checkHealth: () => checkHealth(mountPath),
|
||||
});
|
||||
161
apps/agent/src/volume-host/backends/sftp.ts
Normal file
161
apps/agent/src/volume-host/backends/sftp.ts
Normal file
|
|
@ -0,0 +1,161 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import * as path from "node:path";
|
||||
import { spawn } from "node:child_process";
|
||||
import { FILE_MODES, logger, writeFileWithMode } from "@zerobyte/core/node";
|
||||
import { toMessage } from "@zerobyte/core/utils";
|
||||
import { OPERATION_TIMEOUT, SSH_KEYS_DIR } from "../constants";
|
||||
import { getMountForPath } from "../fs";
|
||||
import { withTimeout } from "../timeout";
|
||||
import type { BackendConfig, VolumeBackend } from "../types";
|
||||
import { executeUnmount } from "./utils";
|
||||
|
||||
const getPrivateKeyPath = (mountPath: string) => path.join(SSH_KEYS_DIR, `${path.basename(mountPath)}.key`);
|
||||
const getKnownHostsPath = (mountPath: string) => path.join(SSH_KEYS_DIR, `${path.basename(mountPath)}.known_hosts`);
|
||||
|
||||
const runSshfs = async (args: string[], password?: string) =>
|
||||
new Promise<void>((resolve, reject) => {
|
||||
const child = spawn("sshfs", args, { stdio: ["pipe", "pipe", "pipe"] });
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
|
||||
child.stdout.setEncoding("utf8");
|
||||
child.stderr.setEncoding("utf8");
|
||||
|
||||
child.stdout.on("data", (data) => {
|
||||
stdout += data;
|
||||
});
|
||||
child.stderr.on("data", (data) => {
|
||||
stderr += data;
|
||||
});
|
||||
child.on("error", (error) => {
|
||||
reject(new Error(`Failed to start sshfs: ${error.message}`));
|
||||
});
|
||||
child.on("close", (code) => {
|
||||
if (code === 0) {
|
||||
resolve();
|
||||
return;
|
||||
}
|
||||
|
||||
const errorMsg = stderr.trim() || stdout.trim() || "Unknown error";
|
||||
reject(new Error(`Failed to mount SFTP volume: ${errorMsg}`));
|
||||
});
|
||||
|
||||
if (password) child.stdin.write(password);
|
||||
child.stdin.end();
|
||||
});
|
||||
|
||||
const checkHealth = async (mountPath: string) => {
|
||||
const mount = await getMountForPath(mountPath);
|
||||
if (!mount || mount.mountPoint !== mountPath) return { status: "unmounted" as const };
|
||||
if (mount.fstype !== "fuse.sshfs") {
|
||||
return { status: "error" as const, error: `Invalid filesystem type: ${mount.fstype} (expected fuse.sshfs)` };
|
||||
}
|
||||
return { status: "mounted" as const };
|
||||
};
|
||||
|
||||
const unmount = async (mountPath: string) => {
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("SFTP unmounting is only supported on Linux hosts.");
|
||||
return { status: "error" as const, error: "SFTP unmounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
const mount = await getMountForPath(mountPath);
|
||||
if (!mount || mount.mountPoint !== mountPath) {
|
||||
logger.debug(`Path ${mountPath} is not a mount point. Skipping unmount.`);
|
||||
} else {
|
||||
await executeUnmount(mountPath);
|
||||
}
|
||||
|
||||
await fs.unlink(getPrivateKeyPath(mountPath)).catch(() => {});
|
||||
await fs.unlink(getKnownHostsPath(mountPath)).catch(() => {});
|
||||
await fs.rmdir(mountPath).catch(() => {});
|
||||
|
||||
logger.info(`SFTP volume at ${mountPath} unmounted successfully.`);
|
||||
return { status: "unmounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "SFTP unmount");
|
||||
} catch (error) {
|
||||
logger.error("Error unmounting SFTP volume", { mountPath, error: toMessage(error) });
|
||||
return { status: "error" as const, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
const mount = async (config: BackendConfig, mountPath: string) => {
|
||||
logger.debug(`Mounting SFTP volume ${mountPath}...`);
|
||||
|
||||
if (config.backend !== "sftp") {
|
||||
logger.error("Provided config is not for SFTP backend");
|
||||
return { status: "error" as const, error: "Provided config is not for SFTP backend" };
|
||||
}
|
||||
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("SFTP mounting is only supported on Linux hosts.");
|
||||
return { status: "error" as const, error: "SFTP mounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const { status } = await checkHealth(mountPath);
|
||||
if (status === "mounted") return { status: "mounted" as const };
|
||||
|
||||
if (status === "error") {
|
||||
logger.debug(`Trying to unmount any existing mounts at ${mountPath} before mounting...`);
|
||||
await unmount(mountPath);
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
await fs.mkdir(mountPath, { recursive: true });
|
||||
await fs.mkdir(SSH_KEYS_DIR, { recursive: true });
|
||||
const { uid, gid } = os.userInfo();
|
||||
const options = [
|
||||
"reconnect",
|
||||
"ServerAliveInterval=15",
|
||||
"ServerAliveCountMax=3",
|
||||
"allow_other",
|
||||
`uid=${uid}`,
|
||||
`gid=${gid}`,
|
||||
];
|
||||
|
||||
if (config.skipHostKeyCheck) {
|
||||
options.push("StrictHostKeyChecking=no", "UserKnownHostsFile=/dev/null");
|
||||
} else if (config.knownHosts) {
|
||||
await writeFileWithMode(getKnownHostsPath(mountPath), config.knownHosts, FILE_MODES.ownerReadWrite);
|
||||
options.push(`UserKnownHostsFile=${getKnownHostsPath(mountPath)}`, "StrictHostKeyChecking=yes");
|
||||
} else {
|
||||
options.push("StrictHostKeyChecking=yes");
|
||||
}
|
||||
|
||||
if (config.readOnly) options.push("ro");
|
||||
if (config.port) options.push(`port=${config.port}`);
|
||||
if (config.privateKey) {
|
||||
let key = config.privateKey.replace(/\r\n/g, "\n");
|
||||
if (!key.endsWith("\n")) key += "\n";
|
||||
await writeFileWithMode(getPrivateKeyPath(mountPath), key, FILE_MODES.ownerReadWrite);
|
||||
options.push(`IdentityFile=${getPrivateKeyPath(mountPath)}`);
|
||||
}
|
||||
|
||||
const args = [`${config.username}@${config.host}:${config.path || ""}`, mountPath, "-o", options.join(",")];
|
||||
if (config.password) args.push("-o", "password_stdin");
|
||||
|
||||
logger.info(`Executing sshfs: sshfs ${args.join(" ")}`);
|
||||
await runSshfs(args, config.password);
|
||||
|
||||
logger.info(`SFTP volume at ${mountPath} mounted successfully.`);
|
||||
return { status: "mounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT * 2, "SFTP mount");
|
||||
} catch (error) {
|
||||
logger.error("Error mounting SFTP volume", { error: toMessage(error) });
|
||||
return { status: "error" as const, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
export const makeSftpBackend = (config: BackendConfig, mountPath: string): VolumeBackend => ({
|
||||
mount: () => mount(config, mountPath),
|
||||
unmount: () => unmount(mountPath),
|
||||
checkHealth: () => checkHealth(mountPath),
|
||||
});
|
||||
124
apps/agent/src/volume-host/backends/smb.ts
Normal file
124
apps/agent/src/volume-host/backends/smb.ts
Normal file
|
|
@ -0,0 +1,124 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import { toMessage } from "@zerobyte/core/utils";
|
||||
import { OPERATION_TIMEOUT } from "../constants";
|
||||
import { withTimeout } from "../timeout";
|
||||
import { getMountForPath } from "../fs";
|
||||
import type { BackendConfig, VolumeBackend } from "../types";
|
||||
import { assertMounted, executeMount, executeUnmount } from "./utils";
|
||||
|
||||
const checkHealth = async (mountPath: string) => {
|
||||
const run = async () => {
|
||||
await assertMounted(mountPath, (fstype) => fstype === "cifs");
|
||||
|
||||
logger.debug(`SMB volume at ${mountPath} is healthy and mounted.`);
|
||||
return { status: "mounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "SMB health check");
|
||||
} catch (error) {
|
||||
const message = toMessage(error);
|
||||
if (message !== "Volume is not mounted") {
|
||||
logger.error("SMB volume health check failed:", message);
|
||||
}
|
||||
return { status: "error" as const, error: message };
|
||||
}
|
||||
};
|
||||
|
||||
const unmount = async (mountPath: string) => {
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("SMB unmounting is only supported on Linux hosts.");
|
||||
return { status: "error" as const, error: "SMB unmounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
const mount = await getMountForPath(mountPath);
|
||||
if (!mount || mount.mountPoint !== mountPath) {
|
||||
logger.debug(`Path ${mountPath} is not a mount point. Skipping unmount.`);
|
||||
return { status: "unmounted" as const };
|
||||
}
|
||||
|
||||
await executeUnmount(mountPath);
|
||||
await fs.rmdir(mountPath).catch(() => {});
|
||||
|
||||
logger.info(`SMB volume at ${mountPath} unmounted successfully.`);
|
||||
return { status: "unmounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "SMB unmount");
|
||||
} catch (error) {
|
||||
logger.error("Error unmounting SMB volume", { mountPath, error: toMessage(error) });
|
||||
return { status: "error" as const, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
const mount = async (config: BackendConfig, mountPath: string) => {
|
||||
logger.debug(`Mounting SMB volume ${mountPath}...`);
|
||||
|
||||
if (config.backend !== "smb") {
|
||||
logger.error("Provided config is not for SMB backend");
|
||||
return { status: "error" as const, error: "Provided config is not for SMB backend" };
|
||||
}
|
||||
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("SMB mounting is only supported on Linux hosts.");
|
||||
return { status: "error" as const, error: "SMB mounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const { status } = await checkHealth(mountPath);
|
||||
if (status === "mounted") return { status: "mounted" as const };
|
||||
|
||||
if (status === "error") {
|
||||
logger.debug(`Trying to unmount any existing mounts at ${mountPath} before mounting...`);
|
||||
await unmount(mountPath);
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
await fs.mkdir(mountPath, { recursive: true });
|
||||
const { uid, gid } = os.userInfo();
|
||||
const options = [`port=${config.port}`, `uid=${uid}`, `gid=${gid}`, "iocharset=utf8"];
|
||||
|
||||
if (config.guest) {
|
||||
options.push("username=guest", "password=");
|
||||
} else {
|
||||
const safePassword = (config.password ?? "").replace(/\\/g, "\\\\").replace(/,/g, "\\,");
|
||||
options.push(`username=${config.username ?? "user"}`, `password=${safePassword}`);
|
||||
}
|
||||
|
||||
if (config.domain) options.push(`domain=${config.domain}`);
|
||||
if (config.vers && config.vers !== "auto") options.push(`vers=${config.vers}`);
|
||||
if (config.readOnly) options.push("ro");
|
||||
|
||||
const source = `//${config.server}/${config.share}`;
|
||||
const args = ["-t", "cifs", "-o", options.join(","), source, mountPath];
|
||||
|
||||
logger.debug(`Mounting SMB volume ${mountPath}...`);
|
||||
logger.info(`Executing SMB mount for ${source} at ${mountPath}`);
|
||||
|
||||
try {
|
||||
await executeMount(args);
|
||||
} catch (error) {
|
||||
logger.error(`SMB mount failed: ${toMessage(error)}`);
|
||||
throw error;
|
||||
}
|
||||
|
||||
logger.info(`SMB volume at ${mountPath} mounted successfully.`);
|
||||
return { status: "mounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "SMB mount");
|
||||
} catch (error) {
|
||||
logger.error("Error mounting SMB volume", { error: toMessage(error) });
|
||||
return { status: "error" as const, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
export const makeSmbBackend = (config: BackendConfig, mountPath: string): VolumeBackend => ({
|
||||
mount: () => mount(config, mountPath),
|
||||
unmount: () => unmount(mountPath),
|
||||
checkHealth: () => checkHealth(mountPath),
|
||||
});
|
||||
|
|
@ -1,7 +1,6 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import { safeExec } from "@zerobyte/core/node";
|
||||
import { getMountForPath } from "../../../utils/mountinfo";
|
||||
import { logger, safeExec } from "@zerobyte/core/node";
|
||||
import { getMountForPath } from "../fs";
|
||||
|
||||
export const executeMount = async (args: string[]): Promise<void> => {
|
||||
const shouldBeVerbose = process.env.LOG_LEVEL === "debug" || process.env.NODE_ENV !== "production";
|
||||
|
|
@ -10,7 +9,6 @@ export const executeMount = async (args: string[]): Promise<void> => {
|
|||
|
||||
logger.debug(`Executing mount ${effectiveArgs.join(" ")}`);
|
||||
const result = await safeExec({ command: "mount", args: effectiveArgs, timeout: 10000 });
|
||||
|
||||
const stdout = result.stdout.toString().trim();
|
||||
const stderr = result.stderr.toString().trim();
|
||||
|
||||
|
|
@ -26,37 +24,29 @@ export const executeMount = async (args: string[]): Promise<void> => {
|
|||
throw new Error(`Mount command failed with exit code ${result.exitCode}: ${stderr || stdout || "unknown error"}`);
|
||||
};
|
||||
|
||||
export const executeUnmount = async (path: string): Promise<void> => {
|
||||
let stderr: string | undefined;
|
||||
|
||||
logger.debug(`Executing umount -l ${path}`);
|
||||
const result = await safeExec({ command: "umount", args: ["-l", path], timeout: 10000 });
|
||||
|
||||
stderr = result.stderr.toString();
|
||||
|
||||
if (stderr?.trim()) {
|
||||
logger.warn(stderr.trim());
|
||||
}
|
||||
export const executeUnmount = async (mountPath: string): Promise<void> => {
|
||||
logger.debug(`Executing umount -l ${mountPath}`);
|
||||
const result = await safeExec({ command: "umount", args: ["-l", mountPath], timeout: 10000 });
|
||||
const stderr = result.stderr.toString();
|
||||
|
||||
if (stderr.trim()) logger.warn(stderr.trim());
|
||||
if (result.exitCode !== 0) {
|
||||
throw new Error(`Mount command failed with exit code ${result.exitCode}: ${stderr?.trim()}`);
|
||||
throw new Error(`Mount command failed with exit code ${result.exitCode}: ${stderr.trim()}`);
|
||||
}
|
||||
};
|
||||
|
||||
export const assertMounted = async (path: string, isExpectedFilesystem: (fstype: string) => boolean) => {
|
||||
export const assertMounted = async (mountPath: string, isExpectedFilesystem: (fstype: string) => boolean) => {
|
||||
try {
|
||||
await fs.access(path);
|
||||
await fs.access(mountPath);
|
||||
} catch {
|
||||
throw new Error("Volume is not mounted");
|
||||
}
|
||||
|
||||
const mount = await getMountForPath(path);
|
||||
|
||||
if (!mount || mount.mountPoint !== path) {
|
||||
const mount = await getMountForPath(mountPath);
|
||||
if (!mount || mount.mountPoint !== mountPath) {
|
||||
throw new Error("Volume is not mounted");
|
||||
}
|
||||
|
||||
if (!isExpectedFilesystem(mount.fstype)) {
|
||||
throw new Error(`Path ${path} is not mounted as correct fstype (found ${mount.fstype}).`);
|
||||
throw new Error(`Path ${mountPath} is not mounted as correct fstype (found ${mount.fstype}).`);
|
||||
}
|
||||
};
|
||||
144
apps/agent/src/volume-host/backends/webdav.ts
Normal file
144
apps/agent/src/volume-host/backends/webdav.ts
Normal file
|
|
@ -0,0 +1,144 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import { logger } from "@zerobyte/core/node";
|
||||
import { toMessage } from "@zerobyte/core/utils";
|
||||
import { OPERATION_TIMEOUT } from "../constants";
|
||||
import { withTimeout } from "../timeout";
|
||||
import { getMountForPath } from "../fs";
|
||||
import type { BackendConfig, VolumeBackend } from "../types";
|
||||
import { assertMounted, executeMount, executeUnmount } from "./utils";
|
||||
|
||||
const checkHealth = async (mountPath: string) => {
|
||||
const run = async () => {
|
||||
await assertMounted(mountPath, (fstype) => fstype === "fuse" || fstype === "davfs");
|
||||
|
||||
logger.debug(`WebDAV volume at ${mountPath} is healthy and mounted.`);
|
||||
return { status: "mounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "WebDAV health check");
|
||||
} catch (error) {
|
||||
const message = toMessage(error);
|
||||
if (message !== "Volume is not mounted") {
|
||||
logger.error("WebDAV volume health check failed:", message);
|
||||
}
|
||||
return { status: "error" as const, error: message };
|
||||
}
|
||||
};
|
||||
|
||||
const unmount = async (mountPath: string) => {
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("WebDAV unmounting is only supported on Linux hosts.");
|
||||
return { status: "error" as const, error: "WebDAV unmounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
const mount = await getMountForPath(mountPath);
|
||||
if (!mount || mount.mountPoint !== mountPath) {
|
||||
logger.debug(`Path ${mountPath} is not a mount point. Skipping unmount.`);
|
||||
return { status: "unmounted" as const };
|
||||
}
|
||||
|
||||
await executeUnmount(mountPath);
|
||||
await fs.rmdir(mountPath).catch(() => {});
|
||||
|
||||
logger.info(`WebDAV volume at ${mountPath} unmounted successfully.`);
|
||||
return { status: "unmounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "WebDAV unmount");
|
||||
} catch (error) {
|
||||
logger.error("Error unmounting WebDAV volume", { mountPath, error: toMessage(error) });
|
||||
return { status: "error" as const, error: toMessage(error) };
|
||||
}
|
||||
};
|
||||
|
||||
const mount = async (config: BackendConfig, mountPath: string) => {
|
||||
logger.debug(`Mounting WebDAV volume ${mountPath}...`);
|
||||
|
||||
if (config.backend !== "webdav") {
|
||||
logger.error("Provided config is not for WebDAV backend");
|
||||
return { status: "error" as const, error: "Provided config is not for WebDAV backend" };
|
||||
}
|
||||
|
||||
if (os.platform() !== "linux") {
|
||||
logger.error("WebDAV mounting is only supported on Linux hosts.");
|
||||
return { status: "error" as const, error: "WebDAV mounting is only supported on Linux hosts." };
|
||||
}
|
||||
|
||||
const { status } = await checkHealth(mountPath);
|
||||
if (status === "mounted") return { status: "mounted" as const };
|
||||
|
||||
if (status === "error") {
|
||||
logger.debug(`Trying to unmount any existing mounts at ${mountPath} before mounting...`);
|
||||
await unmount(mountPath);
|
||||
}
|
||||
|
||||
const run = async () => {
|
||||
await fs.mkdir(mountPath, { recursive: true }).catch((error) => {
|
||||
logger.warn(`Failed to create directory ${mountPath}: ${toMessage(error)}`);
|
||||
});
|
||||
const protocol = config.ssl ? "https" : "http";
|
||||
const defaultPort = config.ssl ? 443 : 80;
|
||||
const source = `${protocol}://${config.server}${config.port !== defaultPort ? `:${config.port}` : ""}${config.path}`;
|
||||
const { uid, gid } = os.userInfo();
|
||||
const options = config.readOnly
|
||||
? [`uid=${uid}`, `gid=${gid}`, "file_mode=0444", "dir_mode=0555", "ro"]
|
||||
: [`uid=${uid}`, `gid=${gid}`, "file_mode=0664", "dir_mode=0775"];
|
||||
|
||||
if (config.username && config.password) {
|
||||
const entry = [source, config.username, config.password]
|
||||
.map((value) => value.replace(/[\r\n\t\s]+/g, " "))
|
||||
.join(" ");
|
||||
await fs.appendFile("/etc/davfs2/secrets", `${entry}\n`, { mode: 0o600 });
|
||||
}
|
||||
|
||||
logger.debug(`Mounting WebDAV volume ${mountPath}...`);
|
||||
|
||||
const args = ["-t", "davfs", "-o", options.join(","), source, mountPath];
|
||||
try {
|
||||
await executeMount(args);
|
||||
} catch (error) {
|
||||
logger.warn(`Initial WebDAV mount failed, retrying with -i flag: ${toMessage(error)}`);
|
||||
await executeMount(["-i", ...args]);
|
||||
}
|
||||
|
||||
logger.info(`WebDAV volume at ${mountPath} mounted successfully.`);
|
||||
return { status: "mounted" as const };
|
||||
};
|
||||
|
||||
try {
|
||||
return await withTimeout(run(), OPERATION_TIMEOUT, "WebDAV mount");
|
||||
} catch (error) {
|
||||
const message = toMessage(error);
|
||||
if (message.includes("already mounted")) return { status: "mounted" as const };
|
||||
|
||||
logger.error("Error mounting WebDAV volume", { error: message });
|
||||
|
||||
if (message.includes("option") && message.includes("requires argument")) {
|
||||
return {
|
||||
status: "error" as const,
|
||||
error: "Invalid mount options. Please check your WebDAV server configuration.",
|
||||
};
|
||||
}
|
||||
if (message.includes("connection refused") || message.includes("Connection refused")) {
|
||||
return {
|
||||
status: "error" as const,
|
||||
error: "Cannot connect to WebDAV server. Please check the server address and port.",
|
||||
};
|
||||
}
|
||||
if (message.includes("unauthorized") || message.includes("Unauthorized")) {
|
||||
return { status: "error" as const, error: "Authentication failed. Please check your username and password." };
|
||||
}
|
||||
|
||||
return { status: "error" as const, error: message };
|
||||
}
|
||||
};
|
||||
|
||||
export const makeWebdavBackend = (config: BackendConfig, mountPath: string): VolumeBackend => ({
|
||||
mount: () => mount(config, mountPath),
|
||||
unmount: () => unmount(mountPath),
|
||||
checkHealth: () => checkHealth(mountPath),
|
||||
});
|
||||
9
apps/agent/src/volume-host/constants.ts
Normal file
9
apps/agent/src/volume-host/constants.ts
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
import * as path from "node:path";
|
||||
|
||||
export const OPERATION_TIMEOUT = 5000;
|
||||
export const VOLUME_MOUNT_BASE = process.env.ZEROBYTE_VOLUMES_DIR || "/var/lib/zerobyte/volumes";
|
||||
export const SSH_KEYS_DIR = "/var/lib/zerobyte/ssh";
|
||||
export const RCLONE_CONFIG_DIR = process.env.RCLONE_CONFIG_DIR || "/root/.config/rclone";
|
||||
export const RCLONE_CONFIG_FILE = path.join(RCLONE_CONFIG_DIR, "rclone.conf");
|
||||
const serverIdleTimeout = Number(process.env.SERVER_IDLE_TIMEOUT ?? 60);
|
||||
export const RCLONE_TIMEOUT = (Number.isFinite(serverIdleTimeout) ? serverIdleTimeout : 60) * 1000;
|
||||
66
apps/agent/src/volume-host/fs.ts
Normal file
66
apps/agent/src/volume-host/fs.ts
Normal file
|
|
@ -0,0 +1,66 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import { isPathWithin } from "@zerobyte/core/utils";
|
||||
|
||||
type MountInfo = {
|
||||
mountPoint: string;
|
||||
fstype: string;
|
||||
};
|
||||
|
||||
const unescapeMount = (value: string) =>
|
||||
value.replace(/\\([0-7]{3})/g, (_, oct) => String.fromCharCode(parseInt(oct, 8)));
|
||||
|
||||
export const readMountInfo = async (): Promise<MountInfo[]> => {
|
||||
const text = await fs.readFile("/proc/self/mountinfo", "utf-8");
|
||||
const result: MountInfo[] = [];
|
||||
|
||||
for (const line of text.split("\n")) {
|
||||
if (!line) continue;
|
||||
const sep = line.indexOf(" - ");
|
||||
if (sep === -1) continue;
|
||||
|
||||
const left = line.slice(0, sep).split(" ");
|
||||
const right = line.slice(sep + 3).split(" ");
|
||||
const mpRaw = left[4];
|
||||
const fstype = right[0];
|
||||
|
||||
if (!mpRaw || !fstype) continue;
|
||||
result.push({ mountPoint: unescapeMount(mpRaw), fstype });
|
||||
}
|
||||
|
||||
return result;
|
||||
};
|
||||
|
||||
export const getMountForPath = async (targetPath: string): Promise<MountInfo | undefined> => {
|
||||
const mounts = await readMountInfo();
|
||||
let best: MountInfo | undefined;
|
||||
|
||||
for (const mount of mounts) {
|
||||
if (!isPathWithin(mount.mountPoint, targetPath)) continue;
|
||||
if (!best || mount.mountPoint.length > best.mountPoint.length) {
|
||||
best = mount;
|
||||
}
|
||||
}
|
||||
|
||||
return best;
|
||||
};
|
||||
|
||||
export const getStatFs = async (mountPoint: string) => {
|
||||
const stat = await fs.statfs(mountPoint, { bigint: true });
|
||||
const unit = stat.bsize > 0n ? stat.bsize : 1n;
|
||||
const blocks = stat.blocks > 0n ? stat.blocks : 0n;
|
||||
let bfree = stat.bfree > 0n ? stat.bfree : 0n;
|
||||
if (bfree > blocks) bfree = blocks;
|
||||
const bavail = stat.bavail > 0n ? stat.bavail : 0n;
|
||||
const max = BigInt(Number.MAX_SAFE_INTEGER);
|
||||
const toNumber = (value: bigint) => (value > max ? Number.MAX_SAFE_INTEGER : Number(value));
|
||||
|
||||
return {
|
||||
total: toNumber(blocks * unit),
|
||||
used: toNumber((blocks - bfree) * unit),
|
||||
free: toNumber(bavail * unit),
|
||||
};
|
||||
};
|
||||
|
||||
export const isNodeJSErrnoException = (error: unknown): error is NodeJS.ErrnoException => {
|
||||
return error instanceof Error && "code" in error;
|
||||
};
|
||||
31
apps/agent/src/volume-host/index.ts
Normal file
31
apps/agent/src/volume-host/index.ts
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
import { makeDirectoryBackend } from "./backends/directory";
|
||||
import { makeNfsBackend } from "./backends/nfs";
|
||||
import { makeRcloneBackend } from "./backends/rclone";
|
||||
import { makeSftpBackend } from "./backends/sftp";
|
||||
import { makeSmbBackend } from "./backends/smb";
|
||||
import { makeWebdavBackend } from "./backends/webdav";
|
||||
import { getVolumePath } from "./paths";
|
||||
import type { AgentVolume, VolumeBackend } from "./types";
|
||||
|
||||
export { getStatFs, isNodeJSErrnoException } from "./fs";
|
||||
export { getVolumePath } from "./paths";
|
||||
export type { AgentVolume, BackendConfig, VolumeBackend } from "./types";
|
||||
|
||||
export const createVolumeBackend = (volume: AgentVolume, mountPath = getVolumePath(volume)): VolumeBackend => {
|
||||
switch (volume.config.backend) {
|
||||
case "directory":
|
||||
return makeDirectoryBackend(volume.config, mountPath);
|
||||
case "nfs":
|
||||
return makeNfsBackend(volume.config, mountPath);
|
||||
case "smb":
|
||||
return makeSmbBackend(volume.config, mountPath);
|
||||
case "webdav":
|
||||
return makeWebdavBackend(volume.config, mountPath);
|
||||
case "rclone":
|
||||
return makeRcloneBackend(volume.config, mountPath);
|
||||
case "sftp":
|
||||
return makeSftpBackend(volume.config, mountPath);
|
||||
}
|
||||
|
||||
throw new Error("Unsupported backend");
|
||||
};
|
||||
156
apps/agent/src/volume-host/operations.ts
Normal file
156
apps/agent/src/volume-host/operations.ts
Normal file
|
|
@ -0,0 +1,156 @@
|
|||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import * as path from "node:path";
|
||||
import { toMessage } from "@zerobyte/core/utils";
|
||||
import { createVolumeBackend, getVolumePath, isNodeJSErrnoException } from ".";
|
||||
import type { AgentVolume, BackendConfig } from "./types";
|
||||
|
||||
const DEFAULT_PAGE_SIZE = 500;
|
||||
const MAX_PAGE_SIZE = 500;
|
||||
|
||||
export const listVolumeFiles = async (
|
||||
volume: AgentVolume,
|
||||
subPath?: string,
|
||||
offset: number = 0,
|
||||
limit: number = DEFAULT_PAGE_SIZE,
|
||||
) => {
|
||||
const volumePath = getVolumePath(volume);
|
||||
const requestedPath = subPath ? path.join(volumePath, subPath) : volumePath;
|
||||
const normalizedPath = path.normalize(requestedPath);
|
||||
const relative = path.relative(volumePath, normalizedPath);
|
||||
|
||||
if (relative.startsWith("..") || path.isAbsolute(relative)) {
|
||||
throw new Error("Invalid path");
|
||||
}
|
||||
|
||||
const pageSize = Math.min(Math.max(limit, 1), MAX_PAGE_SIZE);
|
||||
const startOffset = Math.max(offset, 0);
|
||||
|
||||
try {
|
||||
const dirents = await fs.readdir(normalizedPath, { withFileTypes: true });
|
||||
|
||||
dirents.sort((a, b) => {
|
||||
const aIsDir = a.isDirectory();
|
||||
const bIsDir = b.isDirectory();
|
||||
|
||||
if (aIsDir === bIsDir) {
|
||||
return a.name.localeCompare(b.name);
|
||||
}
|
||||
return aIsDir ? -1 : 1;
|
||||
});
|
||||
|
||||
const total = dirents.length;
|
||||
const paginatedDirents = dirents.slice(startOffset, startOffset + pageSize);
|
||||
|
||||
const entries = (
|
||||
await Promise.all(
|
||||
paginatedDirents.map(async (dirent) => {
|
||||
const fullPath = path.join(normalizedPath, dirent.name);
|
||||
|
||||
try {
|
||||
const stats = await fs.stat(fullPath);
|
||||
const relativePath = path.relative(volumePath, fullPath);
|
||||
|
||||
return {
|
||||
name: dirent.name,
|
||||
path: `/${relativePath}`,
|
||||
type: dirent.isDirectory() ? ("directory" as const) : ("file" as const),
|
||||
size: dirent.isFile() ? stats.size : undefined,
|
||||
modifiedAt: stats.mtimeMs,
|
||||
};
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}),
|
||||
)
|
||||
).filter((entry) => entry !== null);
|
||||
|
||||
return {
|
||||
files: entries,
|
||||
path: subPath || "/",
|
||||
offset: startOffset,
|
||||
limit: pageSize,
|
||||
total,
|
||||
hasMore: startOffset + pageSize < total,
|
||||
};
|
||||
} catch (error) {
|
||||
if (isNodeJSErrnoException(error) && error.code === "ENOENT") {
|
||||
throw new Error("Directory not found");
|
||||
}
|
||||
if (toMessage(error) === "Invalid path") {
|
||||
throw error;
|
||||
}
|
||||
throw new Error(`Failed to list files: ${toMessage(error)}`);
|
||||
}
|
||||
};
|
||||
|
||||
export const browseFilesystem = async (browsePath: string) => {
|
||||
const normalizedPath = path.normalize(browsePath);
|
||||
const entries = await fs.readdir(normalizedPath, { withFileTypes: true });
|
||||
|
||||
const directories = await Promise.all(
|
||||
entries
|
||||
.filter((entry) => entry.isDirectory())
|
||||
.map(async (entry) => {
|
||||
const fullPath = path.join(normalizedPath, entry.name);
|
||||
|
||||
try {
|
||||
const stats = await fs.stat(fullPath);
|
||||
return {
|
||||
name: entry.name,
|
||||
path: fullPath,
|
||||
type: "directory" as const,
|
||||
size: undefined,
|
||||
modifiedAt: stats.mtimeMs,
|
||||
};
|
||||
} catch {
|
||||
return {
|
||||
name: entry.name,
|
||||
path: fullPath,
|
||||
type: "directory" as const,
|
||||
size: undefined,
|
||||
modifiedAt: undefined,
|
||||
};
|
||||
}
|
||||
}),
|
||||
);
|
||||
|
||||
return {
|
||||
directories: directories.sort((a, b) => a.name.localeCompare(b.name)),
|
||||
path: normalizedPath,
|
||||
};
|
||||
};
|
||||
|
||||
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",
|
||||
};
|
||||
|
||||
const backend = createVolumeBackend(mockVolume, tempDir);
|
||||
const { error } = await backend.mount();
|
||||
|
||||
await backend.unmount();
|
||||
|
||||
return {
|
||||
success: !error,
|
||||
message: error ? toMessage(error) : "Connection successful",
|
||||
};
|
||||
} finally {
|
||||
await fs.rm(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
};
|
||||
10
apps/agent/src/volume-host/paths.ts
Normal file
10
apps/agent/src/volume-host/paths.ts
Normal file
|
|
@ -0,0 +1,10 @@
|
|||
import { VOLUME_MOUNT_BASE } from "./constants";
|
||||
import type { AgentVolume } from "./types";
|
||||
|
||||
export const getVolumePath = (volume: AgentVolume) => {
|
||||
if (volume.config.backend === "directory") {
|
||||
return volume.config.path;
|
||||
}
|
||||
|
||||
return `${VOLUME_MOUNT_BASE}/${volume.shortId}/_data`;
|
||||
};
|
||||
21
apps/agent/src/volume-host/timeout.ts
Normal file
21
apps/agent/src/volume-host/timeout.ts
Normal file
|
|
@ -0,0 +1,21 @@
|
|||
class TimeoutError extends Error {
|
||||
code = "ETIMEOUT";
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "TimeoutError";
|
||||
}
|
||||
}
|
||||
|
||||
export const withTimeout = async <T>(promise: Promise<T>, ms: number, label: string): Promise<T> => {
|
||||
let timeout: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
return await Promise.race([
|
||||
promise,
|
||||
new Promise<never>((_, reject) => {
|
||||
timeout = setTimeout(() => reject(new TimeoutError(`${label} timed out after ${ms}ms`)), ms);
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
if (timeout) clearTimeout(timeout);
|
||||
}
|
||||
};
|
||||
67
apps/agent/src/volume-host/types.ts
Normal file
67
apps/agent/src/volume-host/types.ts
Normal file
|
|
@ -0,0 +1,67 @@
|
|||
export type BackendStatus = "mounted" | "unmounted" | "error";
|
||||
|
||||
type BaseConfig = { backend: string; readOnly?: boolean };
|
||||
|
||||
export type BackendConfig =
|
||||
| (BaseConfig & { backend: "directory"; path: string })
|
||||
| (BaseConfig & { backend: "nfs"; server: string; exportPath: string; port: number; version: "3" | "4" | "4.1" })
|
||||
| (BaseConfig & {
|
||||
backend: "smb";
|
||||
server: string;
|
||||
share: string;
|
||||
username?: string;
|
||||
password?: string;
|
||||
guest?: boolean;
|
||||
vers?: "1.0" | "2.0" | "2.1" | "3.0" | "auto";
|
||||
domain?: string;
|
||||
port: number;
|
||||
})
|
||||
| (BaseConfig & {
|
||||
backend: "webdav";
|
||||
server: string;
|
||||
path: string;
|
||||
username?: string;
|
||||
password?: string;
|
||||
port: number;
|
||||
ssl?: boolean;
|
||||
})
|
||||
| (BaseConfig & { backend: "rclone"; remote: string; path: string })
|
||||
| (BaseConfig & {
|
||||
backend: "sftp";
|
||||
host: string;
|
||||
port: number;
|
||||
username: string;
|
||||
password?: string;
|
||||
privateKey?: string;
|
||||
path: string;
|
||||
skipHostKeyCheck?: boolean;
|
||||
knownHosts?: string;
|
||||
});
|
||||
|
||||
export type AgentVolume = {
|
||||
id: number;
|
||||
shortId: string;
|
||||
name: string;
|
||||
config: BackendConfig;
|
||||
createdAt: number;
|
||||
updatedAt: number;
|
||||
lastHealthCheck: number;
|
||||
type: string;
|
||||
status: BackendStatus;
|
||||
lastError: string | null;
|
||||
provisioningId?: string | null;
|
||||
autoRemount: boolean;
|
||||
agentId: string;
|
||||
organizationId: string;
|
||||
};
|
||||
|
||||
export type OperationResult = {
|
||||
status: BackendStatus;
|
||||
error?: string;
|
||||
};
|
||||
|
||||
export type VolumeBackend = {
|
||||
mount: () => Promise<OperationResult>;
|
||||
unmount: () => Promise<OperationResult>;
|
||||
checkHealth: () => Promise<OperationResult>;
|
||||
};
|
||||
|
|
@ -51,6 +51,111 @@ const backupCancelSchema = z.object({
|
|||
payload: z.object({ jobId: z.string(), scheduleId: z.string() }),
|
||||
});
|
||||
|
||||
const backendStatusSchema = z.enum(["mounted", "unmounted", "error"]);
|
||||
|
||||
const volumeSchema = z.object({
|
||||
id: z.number(),
|
||||
shortId: z.string(),
|
||||
name: z.string(),
|
||||
path: z.string().nullable().optional(),
|
||||
config: z.record(z.string(), z.unknown()).and(z.object({ backend: z.string() })),
|
||||
createdAt: z.number(),
|
||||
updatedAt: z.number(),
|
||||
lastHealthCheck: z.number(),
|
||||
type: z.string(),
|
||||
status: backendStatusSchema,
|
||||
lastError: z.string().nullable(),
|
||||
provisioningId: z.string().nullable().optional(),
|
||||
autoRemount: z.boolean(),
|
||||
agentId: z.string(),
|
||||
organizationId: z.string(),
|
||||
});
|
||||
|
||||
const volumeOperationResultSchema = z.object({
|
||||
status: backendStatusSchema,
|
||||
error: z.string().optional(),
|
||||
});
|
||||
|
||||
const statfsSchema = z.object({
|
||||
total: z.number().optional(),
|
||||
used: z.number().optional(),
|
||||
free: z.number().optional(),
|
||||
});
|
||||
|
||||
const fileEntrySchema = z.object({
|
||||
name: z.string(),
|
||||
path: z.string(),
|
||||
type: z.enum(["directory", "file"]),
|
||||
size: z.number().optional(),
|
||||
modifiedAt: z.number().optional(),
|
||||
});
|
||||
|
||||
const directoryEntrySchema = z.object({
|
||||
name: z.string(),
|
||||
path: z.string(),
|
||||
type: z.literal("directory"),
|
||||
size: z.undefined().optional(),
|
||||
modifiedAt: z.number().optional(),
|
||||
});
|
||||
|
||||
const volumeCommandSchema = z.discriminatedUnion("name", [
|
||||
z.object({ name: z.literal("volume.mount"), volume: volumeSchema }),
|
||||
z.object({ name: z.literal("volume.unmount"), volume: volumeSchema }),
|
||||
z.object({ name: z.literal("volume.checkHealth"), volume: volumeSchema }),
|
||||
z.object({ name: z.literal("volume.statfs"), volume: volumeSchema }),
|
||||
z.object({
|
||||
name: z.literal("volume.listFiles"),
|
||||
volume: volumeSchema,
|
||||
subPath: z.string().optional(),
|
||||
offset: z.number(),
|
||||
limit: z.number(),
|
||||
}),
|
||||
z.object({ name: z.literal("volume.testConnection"), backendConfig: z.record(z.string(), z.unknown()) }),
|
||||
z.object({ name: z.literal("filesystem.browse"), path: z.string() }),
|
||||
]);
|
||||
|
||||
const volumeCommandRequestSchema = z.object({
|
||||
type: z.literal("volume.command"),
|
||||
payload: z.object({
|
||||
commandId: z.string(),
|
||||
command: volumeCommandSchema,
|
||||
}),
|
||||
});
|
||||
|
||||
const volumeCommandResultSchema = z.discriminatedUnion("name", [
|
||||
z.object({ name: z.literal("volume.mount"), result: volumeOperationResultSchema }),
|
||||
z.object({ name: z.literal("volume.unmount"), result: volumeOperationResultSchema }),
|
||||
z.object({ name: z.literal("volume.checkHealth"), result: volumeOperationResultSchema }),
|
||||
z.object({ name: z.literal("volume.statfs"), result: statfsSchema }),
|
||||
z.object({
|
||||
name: z.literal("volume.listFiles"),
|
||||
result: z.object({
|
||||
files: z.array(fileEntrySchema),
|
||||
path: z.string(),
|
||||
offset: z.number(),
|
||||
limit: z.number(),
|
||||
total: z.number(),
|
||||
hasMore: z.boolean(),
|
||||
}),
|
||||
}),
|
||||
z.object({
|
||||
name: z.literal("volume.testConnection"),
|
||||
result: z.object({ success: z.boolean(), message: z.string() }),
|
||||
}),
|
||||
z.object({
|
||||
name: z.literal("filesystem.browse"),
|
||||
result: z.object({ directories: z.array(directoryEntrySchema), path: z.string() }),
|
||||
}),
|
||||
]);
|
||||
|
||||
const volumeCommandResponseSchema = z.object({
|
||||
type: z.literal("volume.commandResult"),
|
||||
payload: z.discriminatedUnion("status", [
|
||||
z.object({ commandId: z.string(), status: z.literal("success"), command: volumeCommandResultSchema }),
|
||||
z.object({ commandId: z.string(), status: z.literal("error"), error: z.string() }),
|
||||
]),
|
||||
});
|
||||
|
||||
const heartbeatPingSchema = z.object({
|
||||
type: z.literal("heartbeat.ping"),
|
||||
payload: z.object({ sentAt: z.number() }),
|
||||
|
|
@ -113,6 +218,7 @@ const heartbeatPongSchema = z.object({
|
|||
const controllerMessageSchema = z.discriminatedUnion("type", [
|
||||
backupRunSchema,
|
||||
backupCancelSchema,
|
||||
volumeCommandRequestSchema,
|
||||
heartbeatPingSchema,
|
||||
]);
|
||||
const agentMessageSchema = z.discriminatedUnion("type", [
|
||||
|
|
@ -122,6 +228,7 @@ const agentMessageSchema = z.discriminatedUnion("type", [
|
|||
backupCompletedSchema,
|
||||
backupFailedSchema,
|
||||
backupCancelledSchema,
|
||||
volumeCommandResponseSchema,
|
||||
heartbeatPongSchema,
|
||||
]);
|
||||
|
||||
|
|
@ -132,6 +239,10 @@ export type BackupProgressPayload = z.infer<typeof backupProgressSchema>["payloa
|
|||
export type BackupCompletedPayload = z.infer<typeof backupCompletedSchema>["payload"];
|
||||
export type BackupFailedPayload = z.infer<typeof backupFailedSchema>["payload"];
|
||||
export type BackupCancelledPayload = z.infer<typeof backupCancelledSchema>["payload"];
|
||||
export type VolumeCommandPayload = z.infer<typeof volumeCommandRequestSchema>["payload"];
|
||||
export type VolumeCommand = z.infer<typeof volumeCommandSchema>;
|
||||
export type VolumeCommandResult = z.infer<typeof volumeCommandResultSchema>;
|
||||
export type VolumeCommandResponsePayload = z.infer<typeof volumeCommandResponseSchema>["payload"];
|
||||
export type ControllerMessage = z.infer<typeof controllerMessageSchema>;
|
||||
export type AgentMessage = z.infer<typeof agentMessageSchema>;
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue