danvics/zerobyte
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix: correctly propagate agent restic error

Browse source
This commit is contained in:
Nicolas Meienberger 2026-03-30 20:39:47 +02:00
parent d291bb0382
commit 3162cba8b2
4 changed files with 319 additions and 230 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

71
apps/agent/src/__tests__/controller-session.test.ts Normal file
View file

@ -0,0 +1,71 @@
import { afterEach, expect, mock, spyOn, test } from "bun:test";
import { Effect } from "effect";
import waitForExpect from "wait-for-expect";
import { fromAny } from "@total-typescript/shoehorn";
import { createControllerMessage, parseAgentMessage } from "@zerobyte/contracts/agent-protocol";
import * as resticServer from "@zerobyte/core/restic/server";
import { createControllerSession } from "../controller-session";
afterEach(() => {
mock.restore();
});
test("emits backup.failed when a backup command hits a restic error", async () => {
spyOn(resticServer, "createRestic").mockReturnValue(
fromAny({
backup: () => Effect.fail(new Error("source path missing")),
}),
);
const outboundMessages: string[] = [];
const session = createControllerSession(
fromAny({
send: (message: string) => {
outboundMessages.push(message);
},
}),
);
try {
session.onOpen();
session.onMessage(
createControllerMessage("backup.run", {
jobId: "job-1",
scheduleId: "schedule-1",
organizationId: "org-1",
sourcePath: "/tmp/missing-source",
repositoryConfig: {
backend: "local",
path: "/tmp/test-repository",
},
options: {},
runtime: {
password: "password",
cacheDir: "/tmp/restic-cache",
passFile: "/tmp/restic-pass",
defaultExcludes: [],
},
}),
);
await waitForExpect(() => {
const failedMessage = outboundMessages
.map((message) => parseAgentMessage(message))
.find((message) => message?.success && message.data.type === "backup.failed");
expect(failedMessage?.success).toBe(true);
if (!failedMessage || !failedMessage.success || failedMessage.data.type !== "backup.failed") {
return;
}
expect(failedMessage.data.payload).toEqual({
jobId: "job-1",
scheduleId: "schedule-1",
error: "source path missing",
errorDetails: "source path missing",
});
});
} finally {
session.close();
}
});

124
apps/agent/src/commands/backup-run.ts
View file

@ -43,66 +43,76 @@ export const handleBackupRunCommand = (context: ControllerCommandContext, payloa
const restic = createRestic(deps);
try {
const result = yield* Effect.tryPromise(() =>
restic.backup(payload.repositoryConfig, payload.sourcePath, {
organizationId: payload.organizationId,
...payload.options,
signal: abortController.signal,
onProgress: (progress) => {
context.offerOutbound(
createAgentMessage("backup.progress", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
progress,
}),
);
yield* restic
.backup(payload.repositoryConfig, payload.sourcePath, {
organizationId: payload.organizationId,
...payload.options,
signal: abortController.signal,
onProgress: (progress) => {
context.offerOutbound(
createAgentMessage("backup.progress", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
progress,
}),
);
},
})
.pipe(
Effect.matchEffect({
onSuccess: (result) => {
return Effect.sync(() => {
if (abortController.signal.aborted) {
context.offerOutbound(
createAgentMessage("backup.cancelled", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
message: "Backup was cancelled",
}),
);
return;
}
context.offerOutbound(
createAgentMessage("backup.completed", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
exitCode: result.exitCode,
result: result.result,
warningDetails: result.warningDetails ?? undefined,
}),
);
});
},
onFailure: (error) => {
return Effect.sync(() => {
if (abortController.signal.aborted) {
context.offerOutbound(
createAgentMessage("backup.cancelled", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
message: "Backup was cancelled",
}),
);
return;
}
context.offerOutbound(
createAgentMessage("backup.failed", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
error: toMessage(error),
errorDetails: toErrorDetails(error),
}),
);
});
},
}),
);
if (abortController.signal.aborted) {
context.offerOutbound(
createAgentMessage("backup.cancelled", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
message: "Backup was cancelled",
Effect.ensuring(
Effect.sync(() => {
context.deleteRunningJob(payload.jobId);
}),
);
return;
}
context.offerOutbound(
createAgentMessage("backup.completed", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
exitCode: result.exitCode,
result: result.result,
warningDetails: result.warningDetails ?? undefined,
}),
),
);
} catch (error) {
if (abortController.signal.aborted) {
context.offerOutbound(
createAgentMessage("backup.cancelled", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
message: "Backup was cancelled",
}),
);
return;
}
context.offerOutbound(
createAgentMessage("backup.failed", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
error: toMessage(error),
errorDetails: toErrorDetails(error),
}),
);
} finally {
context.deleteRunningJob(payload.jobId);
}
}),
).pipe(Effect.asVoid);

41
packages/core/src/restic/commands/__tests__/backup.test.ts
View file

@ -1,4 +1,5 @@
import { afterEach, describe, expect, test, vi } from "vitest";
import { Effect } from "effect";
import * as cleanupModule from "../../helpers/cleanup-temporary-keys";
import * as spawnModule from "../../../utils/spawn";
import { ResticError } from "../../error";
@ -62,7 +63,7 @@ const setup = ({ spawnResult = {}, onSpawnCall }: SetupOptions = {}) => {
let capturedArgs: string[] = [];
vi.spyOn(cleanupModule, "cleanupTemporaryKeys").mockImplementation(() => Promise.resolve());
vi.spyOn(spawnModule, "safeSpawn").mockImplementation((params) => {
vi.spyOn(spawnModule, "safeSpawn").mockImplementation((params: SafeSpawnParams) => {
capturedArgs = params.args;
return Promise.resolve(onSpawnCall?.(params)).then(() => ({
exitCode: 0,
@ -87,6 +88,9 @@ const setup = ({ spawnResult = {}, onSpawnCall }: SetupOptions = {}) => {
};
};
const runBackup = (...args: Parameters<typeof backup>) => Effect.runPromise(backup(...args));
const runBackupError = (...args: Parameters<typeof backup>) => Effect.runPromise(Effect.flip(backup(...args)));
afterEach(() => {
vi.restoreAllMocks();
});
@ -95,7 +99,7 @@ describe("backup command", () => {
describe("argument construction", () => {
test("passes source path as positional arg when no include list is given", async () => {
const { getArgs, hasFlag } = setup();
await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(getArgs()).toContain("/mnt/data");
expect(hasFlag("--files-from")).toBe(false);
@ -105,7 +109,7 @@ describe("backup command", () => {
const { getArgs } = setup();
const source = "--help";
await backup(config, source, { organizationId: "org-1" }, mockDeps);
await runBackup(config, source, { organizationId: "org-1" }, mockDeps);
const separatorIndex = getArgs().indexOf("--");
expect(separatorIndex).toBeGreaterThan(-1);
@ -132,7 +136,7 @@ describe("backup command", () => {
},
});
await backup(
await runBackup(
config,
"/mnt/data",
{
@ -152,7 +156,7 @@ describe("backup command", () => {
test("always includes DEFAULT_EXCLUDES as --exclude args", async () => {
const { getOptionValues } = setup();
await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(getOptionValues("--exclude").length).toBeGreaterThan(0);
});
@ -161,7 +165,7 @@ describe("backup command", () => {
describe("exit code handling", () => {
test("returns parsed result on exit code 0", async () => {
setup();
const { result, exitCode } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
const { result, exitCode } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(exitCode).toBe(0);
expect(result?.snapshot_id).toBe("abcd1234");
@ -169,7 +173,7 @@ describe("backup command", () => {
test("returns result without throwing on exit code 3 (partial read errors)", async () => {
setup({ spawnResult: { exitCode: 3 } });
const { result, exitCode } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
const { result, exitCode } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(exitCode).toBe(3);
expect(result).not.toBeNull();
@ -178,15 +182,14 @@ describe("backup command", () => {
test("throws ResticError on non-zero, non-3 exit codes", async () => {
setup({ spawnResult: { exitCode: 1, summary: "", error: "fatal error" } });
await expect(backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps)).rejects.toBeInstanceOf(
ResticError,
);
const error = await runBackupError(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(error).toBeInstanceOf(ResticError);
});
test("preserves the exit code inside the thrown ResticError", async () => {
setup({ spawnResult: { exitCode: 12, summary: "", error: "wrong password" } });
const error = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps).catch((e) => e);
const error = await runBackupError(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(error).toBeInstanceOf(ResticError);
expect((error as ResticError).code).toBe(12);
});
@ -204,7 +207,7 @@ describe("backup command", () => {
},
});
const error = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps).catch((e) => e);
const error = await runBackupError(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(error).toBeInstanceOf(ResticError);
expect((error as ResticError).summary).toBe("Command failed: An error occurred while executing the command.");
expect((error as ResticError).details).toBe(
@ -219,7 +222,7 @@ describe("backup command", () => {
spawnResult: { exitCode: 130, summary: "", error: "" },
});
const { result, exitCode, warningDetails } = await backup(
const { result, exitCode, warningDetails } = await runBackup(
config,
"/mnt/data",
{
@ -238,7 +241,7 @@ describe("backup command", () => {
describe("output parsing", () => {
test("returns a fully parsed summary object on valid output", async () => {
setup();
const { result } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
const { result } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(result).toMatchObject({
message_type: "summary",
@ -249,14 +252,14 @@ describe("backup command", () => {
test("returns { result: null } when summary line is not valid JSON", async () => {
setup({ spawnResult: { summary: "not-json" } });
const { result } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
const { result } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(result).toBeNull();
});
test("returns { result: null } when summary JSON does not satisfy the schema", async () => {
setup({ spawnResult: { summary: JSON.stringify({ message_type: "summary" }) } });
const { result } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
const { result } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(result).toBeNull();
});
@ -267,7 +270,7 @@ describe("backup command", () => {
const progressUpdates: unknown[] = [];
setup({ onSpawnCall: (params) => params.onStdout?.(VALID_PROGRESS_LINE) });
await backup(
await runBackup(
config,
"/mnt/data",
{
@ -294,7 +297,7 @@ describe("backup command", () => {
});
await expect(
backup(config, "/mnt/data", { organizationId: "org-1", onProgress: () => {} }, mockDeps),
runBackup(config, "/mnt/data", { organizationId: "org-1", onProgress: () => {} }, mockDeps),
).resolves.toBeDefined();
});
@ -304,7 +307,7 @@ describe("backup command", () => {
onSpawnCall: (params) => params.onStdout?.(JSON.stringify({ message_type: "verbose_status", action: "scan" })),
});
await backup(
await runBackup(
config,
"/mnt/data",
{

313
packages/core/src/restic/commands/backup.ts
View file

@ -1,6 +1,7 @@
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { Effect } from "effect";
import { throttle } from "es-toolkit";
import type { CompressionMode, RepositoryConfig } from "../schemas";
import { type ResticBackupProgressDto, resticBackupOutputSchema, resticBackupProgressSchema } from "../restic-dto";
@ -13,7 +14,7 @@ import { ResticError } from "../error";
import { logger, safeSpawn } from "../../node";
import type { ResticDeps } from "../types";
export const backup = async (
export const backup = (
config: RepositoryConfig,
source: string,
options: {
@ -30,184 +31,188 @@ export const backup = async (
customResticParams?: string[];
},
deps: ResticDeps,
) => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, options.organizationId, deps);
) =>
Effect.tryPromise({
try: async () => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, options.organizationId, deps);
const args: string[] = ["--repo", repoUrl, "backup", "--compression", options.compressionMode ?? "auto"];
const args: string[] = ["--repo", repoUrl, "backup", "--compression", options.compressionMode ?? "auto"];
if (options.oneFileSystem) {
args.push("--one-file-system");
}
if (options.oneFileSystem) {
args.push("--one-file-system");
}
if (deps.hostname) {
args.push("--host", deps.hostname);
}
if (deps.hostname) {
args.push("--host", deps.hostname);
}
if (options.tags && options.tags.length > 0) {
for (const tag of options.tags) {
args.push("--tag", tag);
}
}
if (options.tags && options.tags.length > 0) {
for (const tag of options.tags) {
args.push("--tag", tag);
}
}
let includeFile: string | null = null;
let rawIncludeFile: string | null = null;
const usesSourceArg =
(!options.includePaths || options.includePaths.length === 0) &&
(!options.includePatterns || options.includePatterns.length === 0);
let includeFile: string | null = null;
let rawIncludeFile: string | null = null;
const usesSourceArg =
(!options.includePaths || options.includePaths.length === 0) &&
(!options.includePatterns || options.includePatterns.length === 0);
if (options.includePatterns?.length) {
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-include-"));
includeFile = path.join(tmp, "include.txt");
if (options.includePatterns?.length) {
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-include-"));
includeFile = path.join(tmp, "include.txt");
await fs.writeFile(includeFile, options.includePatterns.join("\n"), "utf-8");
await fs.writeFile(includeFile, options.includePatterns.join("\n"), "utf-8");
args.push("--files-from", includeFile);
}
args.push("--files-from", includeFile);
}
if (options.includePaths?.length) {
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-include-raw-"));
rawIncludeFile = path.join(tmp, "include.raw");
if (options.includePaths?.length) {
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-include-raw-"));
rawIncludeFile = path.join(tmp, "include.raw");
await fs.writeFile(rawIncludeFile, Buffer.from(`${options.includePaths.join("\0")}\0`, "utf-8"));
await fs.writeFile(rawIncludeFile, Buffer.from(`${options.includePaths.join("\0")}\0`, "utf-8"));
args.push("--files-from-raw", rawIncludeFile);
}
args.push("--files-from-raw", rawIncludeFile);
}
for (const exclude of deps.defaultExcludes) {
args.push("--exclude", exclude);
}
for (const exclude of deps.defaultExcludes) {
args.push("--exclude", exclude);
}
let excludeFile: string | null = null;
if (options.exclude?.length) {
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-exclude-"));
excludeFile = path.join(tmp, "exclude.txt");
let excludeFile: string | null = null;
if (options.exclude?.length) {
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-exclude-"));
excludeFile = path.join(tmp, "exclude.txt");
await fs.writeFile(excludeFile, options.exclude.join("\n"), "utf-8");
await fs.writeFile(excludeFile, options.exclude.join("\n"), "utf-8");
args.push("--exclude-file", excludeFile);
}
args.push("--exclude-file", excludeFile);
}
if (options.excludeIfPresent?.length) {
for (const filename of options.excludeIfPresent) {
args.push("--exclude-if-present", filename);
}
}
if (options.excludeIfPresent?.length) {
for (const filename of options.excludeIfPresent) {
args.push("--exclude-if-present", filename);
}
}
if (options.customResticParams?.length) {
const validationError = validateCustomResticParams(options.customResticParams);
if (validationError) {
throw new Error(`Invalid customResticParams: ${validationError}`);
}
for (const param of options.customResticParams) {
const tokens = param.trim().split(/\s+/).filter(Boolean);
args.push(...tokens);
}
}
if (options.customResticParams?.length) {
const validationError = validateCustomResticParams(options.customResticParams);
if (validationError) {
throw new Error(`Invalid customResticParams: ${validationError}`);
}
for (const param of options.customResticParams) {
const tokens = param.trim().split(/\s+/).filter(Boolean);
args.push(...tokens);
}
}
addCommonArgs(args, env, config);
addCommonArgs(args, env, config);
if (usesSourceArg) {
args.push("--", source);
}
if (usesSourceArg) {
args.push("--", source);
}
const logData = throttle((data: string) => {
logger.info(data.trim());
}, 5000);
const stderrLines: string[] = [];
const logData = throttle((data: string) => {
logger.info(data.trim());
}, 5000);
const stderrLines: string[] = [];
const streamProgress = throttle((data: string) => {
if (options.onProgress) {
const streamProgress = throttle((data: string) => {
if (options.onProgress) {
try {
const jsonData = JSON.parse(data);
if (jsonData.message_type !== "status") {
return;
}
const progressResult = resticBackupProgressSchema.safeParse(jsonData);
if (progressResult.success) {
options.onProgress(progressResult.data);
} else {
logger.error(progressResult.error.message);
}
} catch {
// Ignore JSON parse errors for non-JSON lines
}
}
}, 1000);
logger.debug(`Executing: restic ${args.join(" ")}`);
const res = await safeSpawn({
command: "restic",
args,
env,
signal: options.signal,
onStdout: (data) => {
logData(data);
if (options.onProgress) {
streamProgress(data);
}
},
onStderr: (error) => {
const line = error.trim();
if (line.length > 0) {
stderrLines.push(line);
logger.error(`restic stderr: ${line}`);
}
},
});
if (includeFile) {
await fs.unlink(includeFile).catch(() => {});
}
if (rawIncludeFile) {
await fs.unlink(rawIncludeFile).catch(() => {});
}
if (excludeFile) {
await fs.unlink(excludeFile).catch(() => {});
}
await cleanupTemporaryKeys(env, deps);
if (options.signal?.aborted) {
logger.warn("Restic backup was aborted by signal.");
return { result: null, exitCode: res.exitCode, warningDetails: "Backup was stopped by the user" };
}
if (res.exitCode === 3) {
logger.error(`Restic backup encountered read errors: ${res.error}`);
}
if (res.exitCode !== 0 && res.exitCode !== 3) {
logger.error(`Restic backup failed: ${res.error}`);
logger.error(`Command executed: restic ${args.join(" ")}`);
throw new ResticError(res.exitCode, stderrLines.join("\n") || res.stderr || res.error);
}
const lastLine = res.summary.trim();
let summaryLine: unknown = {};
try {
const jsonData = JSON.parse(data);
if (jsonData.message_type !== "status") {
return;
}
const progressResult = resticBackupProgressSchema.safeParse(jsonData);
if (progressResult.success) {
options.onProgress(progressResult.data);
} else {
logger.error(progressResult.error.message);
}
summaryLine = JSON.parse(lastLine ?? "{}");
} catch {
// Ignore JSON parse errors for non-JSON lines
logger.warn("Failed to parse restic backup output JSON summary.", lastLine);
summaryLine = {};
}
}
}, 1000);
logger.debug(`Executing: restic ${args.join(" ")}`);
const res = await safeSpawn({
command: "restic",
args,
env,
signal: options.signal,
onStdout: (data) => {
logData(data);
if (options.onProgress) {
streamProgress(data);
}
},
onStderr: (error) => {
const line = error.trim();
if (line.length > 0) {
stderrLines.push(line);
logger.error(`restic stderr: ${line}`);
logger.debug(`Restic backup output last line: ${JSON.stringify(summaryLine)}`);
const result = resticBackupOutputSchema.safeParse(summaryLine);
if (!result.success) {
logger.error(`Restic backup output validation failed: ${result.error.message}`);
return {
result: null,
exitCode: res.exitCode,
warningDetails: stderrLines.length > 0 ? stderrLines.join("\n") : null,
};
}
return {
result: result.data,
exitCode: res.exitCode,
warningDetails: stderrLines.length > 0 ? stderrLines.join("\n") : null,
};
},
catch: (error) => (error instanceof Error ? error : new Error(String(error))),
});
if (includeFile) {
await fs.unlink(includeFile).catch(() => {});
}
if (rawIncludeFile) {
await fs.unlink(rawIncludeFile).catch(() => {});
}
if (excludeFile) {
await fs.unlink(excludeFile).catch(() => {});
}
await cleanupTemporaryKeys(env, deps);
if (options.signal?.aborted) {
logger.warn("Restic backup was aborted by signal.");
return { result: null, exitCode: res.exitCode, warningDetails: "Backup was stopped by the user" };
}
if (res.exitCode === 3) {
logger.error(`Restic backup encountered read errors: ${res.error}`);
}
if (res.exitCode !== 0 && res.exitCode !== 3) {
logger.error(`Restic backup failed: ${res.error}`);
logger.error(`Command executed: restic ${args.join(" ")}`);
throw new ResticError(res.exitCode, stderrLines.join("\n") || res.stderr || res.error);
}
const lastLine = res.summary.trim();
let summaryLine: unknown = {};
try {
summaryLine = JSON.parse(lastLine ?? "{}");
} catch {
logger.warn("Failed to parse restic backup output JSON summary.", lastLine);
summaryLine = {};
}
logger.debug(`Restic backup output last line: ${JSON.stringify(summaryLine)}`);
const result = resticBackupOutputSchema.safeParse(summaryLine);
if (!result.success) {
logger.error(`Restic backup output validation failed: ${result.error.message}`);
return {
result: null,
exitCode: res.exitCode,
warningDetails: stderrLines.length > 0 ? stderrLines.join("\n") : null,
};
}
return {
result: result.data,
exitCode: res.exitCode,
warningDetails: stderrLines.length > 0 ? stderrLines.join("\n") : null,
};
};