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

fix(restic): preserve effect failure errors

Browse source
This commit is contained in:
Nicolas Meienberger 2026-05-30 10:03:21 +02:00
parent fa8bff1ece
commit cdb2b3f0d8
No known key found for this signature in database
8 changed files with 52 additions and 44 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
app/server/modules/backups/backup-executor.ts
View file

@ -1,4 +1,3 @@
import { Effect } from "effect";
import { runBackupLifecycle } from "@zerobyte/core/backup-hooks"; import { runBackupLifecycle } from "@zerobyte/core/backup-hooks";
import type { BackupSchedule, Volume, Repository } from "../../db/schema"; import type { BackupSchedule, Volume, Repository } from "../../db/schema";
import { config } from "../../core/config"; import { config } from "../../core/config";
@ -10,7 +9,7 @@ import { getVolumePath } from "../volumes/helpers";
import { decryptVolumeConfig } from "../volumes/volume-config-secrets"; import { decryptVolumeConfig } from "../volumes/volume-config-secrets";
import { decryptRepositoryConfig } from "../repositories/repository-config-secrets"; import { decryptRepositoryConfig } from "../repositories/repository-config-secrets";
import { createBackupOptions } from "./backup.helpers"; import { createBackupOptions } from "./backup.helpers";
import { toErrorDetails } from "../../utils/errors"; import { runEffectPromise, toErrorDetails } from "../../utils/errors";
import { BadRequestError } from "http-errors-enhanced"; import { BadRequestError } from "http-errors-enhanced";
const FUSE_VOLUME_BACKENDS = new Set<Volume["type"]>(["rclone", "sftp", "webdav"]); const FUSE_VOLUME_BACKENDS = new Set<Volume["type"]>(["rclone", "sftp", "webdav"]);
@ -90,7 +89,7 @@ const executeBackupWithoutAgent = async (
compressionMode: payload.options.compressionMode, compressionMode: payload.options.compressionMode,
}; };
return Effect.runPromise( return runEffectPromise(
runBackupLifecycle({ runBackupLifecycle({
restic, restic,
repositoryConfig: payload.repositoryConfig, repositoryConfig: payload.repositoryConfig,

4
app/server/modules/backups/backups.service.ts
View file

@ -28,7 +28,7 @@ import { getScheduleByIdOrShortId } from "./helpers/backup-schedule-lookups";
import { copyToMirrors, runForget, syncSnapshotsToMirror } from "./helpers/backup-maintenance"; import { copyToMirrors, runForget, syncSnapshotsToMirror } from "./helpers/backup-maintenance";
import { restic } from "../../core/restic"; import { restic } from "../../core/restic";
import { mirrorQueries } from "./backups.queries"; import { mirrorQueries } from "./backups.queries";
import { toMessage } from "../../utils/errors"; import { runEffectPromise, toMessage } from "../../utils/errors";
import { Effect } from "effect"; import { Effect } from "effect";
const listSchedules = async () => { const listSchedules = async () => {
const organizationId = getOrganizationId(); const organizationId = getOrganizationId();
@ -514,7 +514,7 @@ const getMirrorSyncStatus = async (scheduleIdOrShortId: number | string, mirrorS
throw new NotFoundError("Mirror not found for this schedule"); throw new NotFoundError("Mirror not found for this schedule");
} }
const [sourceSnapshots, mirrorSnapshots] = await Effect.runPromise( const [sourceSnapshots, mirrorSnapshots] = await runEffectPromise(
Effect.all( Effect.all(
[ [
restic.snapshots(schedule.repository.config, { tags: [schedule.shortId], organizationId }), restic.snapshots(schedule.repository.config, { tags: [schedule.shortId], organizationId }),

9
app/server/modules/backups/helpers/backup-maintenance.ts
View file

@ -5,10 +5,9 @@ import { restic } from "../../../core/restic";
import { repoMutex } from "../../../core/repository-mutex"; import { repoMutex } from "../../../core/repository-mutex";
import { serverEvents } from "../../../core/events"; import { serverEvents } from "../../../core/events";
import { cache, cacheKeys } from "../../../utils/cache"; import { cache, cacheKeys } from "../../../utils/cache";
import { toMessage } from "../../../utils/errors"; import { runEffectPromise, toMessage } from "../../../utils/errors";
import { getOrganizationId } from "~/server/core/request-context"; import { getOrganizationId } from "~/server/core/request-context";
import { mirrorQueries, repositoryQueries, scheduleQueries } from "../backups.queries"; import { mirrorQueries, repositoryQueries, scheduleQueries } from "../backups.queries";
import { Effect } from "effect";
export async function runForget(scheduleId: number, repositoryId?: string, organizationIdOverride?: string) { export async function runForget(scheduleId: number, repositoryId?: string, organizationIdOverride?: string) {
const organizationId = organizationIdOverride ?? getOrganizationId(); const organizationId = organizationIdOverride ?? getOrganizationId();
@ -32,7 +31,7 @@ export async function runForget(scheduleId: number, repositoryId?: string, organ
const releaseLock = await repoMutex.acquireExclusive(repository.id, `forget:${scheduleId}`); const releaseLock = await repoMutex.acquireExclusive(repository.id, `forget:${scheduleId}`);
try { try {
await Effect.runPromise( await runEffectPromise(
restic.forget(repository.config, schedule.retentionPolicy, { tag: schedule.shortId, organizationId }), restic.forget(repository.config, schedule.retentionPolicy, { tag: schedule.shortId, organizationId }),
); );
cache.delByPrefix(cacheKeys.repository.all(repository.id)); cache.delByPrefix(cacheKeys.repository.all(repository.id));
@ -121,7 +120,7 @@ export async function syncSnapshotsToMirror(
]); ]);
try { try {
await Effect.runPromise( await runEffectPromise(
restic.copy(sourceRepository.config, mirrorRepository.config, { restic.copy(sourceRepository.config, mirrorRepository.config, {
tag: schedule.shortId, tag: schedule.shortId,
organizationId, organizationId,
@ -211,7 +210,7 @@ async function copyToSingleMirror(
]); ]);
try { try {
await Effect.runPromise( await runEffectPromise(
restic.copy(sourceRepository.config, mirror.repository.config, { restic.copy(sourceRepository.config, mirror.repository.config, {
tag: schedule.shortId, tag: schedule.shortId,
organizationId, organizationId,

5
app/server/modules/provisioning/provisioning.ts
View file

@ -18,9 +18,8 @@ import { mapRepositoryConfigSecrets } from "~/server/modules/repositories/reposi
import { mapVolumeConfigSecrets } from "~/server/modules/volumes/volume-config-secrets"; import { mapVolumeConfigSecrets } from "~/server/modules/volumes/volume-config-secrets";
import { BACKEND_TYPES, volumeConfigSchema, type BackendConfig } from "@zerobyte/contracts/volumes"; import { BACKEND_TYPES, volumeConfigSchema, type BackendConfig } from "@zerobyte/contracts/volumes";
import { cryptoUtils } from "~/server/utils/crypto"; import { cryptoUtils } from "~/server/utils/crypto";
import { toMessage } from "~/server/utils/errors"; import { runEffectPromise, toMessage } from "~/server/utils/errors";
import { generateShortId } from "~/server/utils/id"; import { generateShortId } from "~/server/utils/id";
import { Effect } from "effect";
const envSecretPrefix = "env://"; const envSecretPrefix = "env://";
const fileSecretPrefix = "file://"; const fileSecretPrefix = "file://";
@ -172,7 +171,7 @@ const syncProvisionedRepositories = async (repositories: ProvisionedRepository[]
}); });
if (!repository.config.isExistingRepository) { if (!repository.config.isExistingRepository) {
const result = await Effect.runPromise( const result = await runEffectPromise(
restic.init(encryptedConfig, { restic.init(encryptedConfig, {
organizationId: repository.organizationId, organizationId: repository.organizationId,
timeoutMs: appConfig.serverIdleTimeout * 1000, timeoutMs: appConfig.serverIdleTimeout * 1000,

19
app/server/modules/repositories/__tests__/repositories.controller.test.ts
View file

@ -328,17 +328,17 @@ describe("repositories updates", () => {
}); });
describe("delete snapshot", () => { describe("delete snapshot", () => {
test("should return 500 when restic deleteSnapshot throws ResticError", async () => { test("should return ResticError details when restic deleteSnapshot fails", async () => {
const repository = await createRepositoryRecord(session.organizationId); const repository = await createRepositoryRecord(session.organizationId);
const { restic } = await import("~/server/core/restic"); const { restic } = await import("~/server/core/restic");
const { ResticError } = await import("@zerobyte/core/restic"); const { ResticError } = await import("@zerobyte/core/restic");
const deleteSnapshotSpy = vi.spyOn(restic, "deleteSnapshot").mockImplementation(() => const deleteSnapshotSpy = vi
Effect.sync(() => { .spyOn(restic, "deleteSnapshot")
throw new ResticError(1, "Fatal: unexpected HTTP response (403): 403 Forbidden"); .mockImplementation(() =>
}), Effect.fail(new ResticError(1, "Fatal: unexpected HTTP response (403): 403 Forbidden")),
); );
try { try {
const res = await app.request(`/api/v1/repositories/${repository.shortId}/snapshots/snap123`, { const res = await app.request(`/api/v1/repositories/${repository.shortId}/snapshots/snap123`, {
@ -348,9 +348,10 @@ describe("repositories updates", () => {
expect(res.status).toBe(500); expect(res.status).toBe(500);
const body = await res.json(); const body = await res.json();
expect(body.message).toBe( expect(body).toEqual({
"Command failed: An error occurred while executing the command.\nFatal: unexpected HTTP response (403): 403 Forbidden", message: "Command failed: An error occurred while executing the command.",
); details: "Fatal: unexpected HTTP response (403): 403 Forbidden",
});
} finally { } finally {
deleteSnapshotSpy.mockRestore(); deleteSnapshotSpy.mockRestore();
} }

9
app/server/modules/repositories/helpers/doctor.ts
View file

@ -3,14 +3,13 @@ import { type DoctorStep, type DoctorResult, type RepositoryConfig } from "@zero
import { z } from "zod"; import { z } from "zod";
import { getOrganizationId } from "~/server/core/request-context"; import { getOrganizationId } from "~/server/core/request-context";
import { restic } from "~/server/core/restic"; import { restic } from "~/server/core/restic";
import { toMessage } from "~/server/utils/errors"; import { runEffectPromise, toMessage } from "~/server/utils/errors";
import { safeJsonParse } from "@zerobyte/core/utils"; import { safeJsonParse } from "@zerobyte/core/utils";
import { logger } from "@zerobyte/core/node"; import { logger } from "@zerobyte/core/node";
import { db } from "~/server/db/db"; import { db } from "~/server/db/db";
import { repositoriesTable } from "~/server/db/schema"; import { repositoriesTable } from "~/server/db/schema";
import { repoMutex } from "~/server/core/repository-mutex"; import { repoMutex } from "~/server/core/repository-mutex";
import { serverEvents } from "~/server/core/events"; import { serverEvents } from "~/server/core/events";
import { Effect } from "effect";
class AbortError extends Error { class AbortError extends Error {
name = "AbortError"; name = "AbortError";
@ -18,7 +17,7 @@ class AbortError extends Error {
const runUnlockStep = async (config: RepositoryConfig, signal?: AbortSignal) => { const runUnlockStep = async (config: RepositoryConfig, signal?: AbortSignal) => {
const orgId = getOrganizationId(); const orgId = getOrganizationId();
const result = await Effect.runPromise(restic.unlock(config, { signal, organizationId: orgId })).then( const result = await runEffectPromise(restic.unlock(config, { signal, organizationId: orgId })).then(
(result) => ({ success: true, message: result.message, error: null }), (result) => ({ success: true, message: result.message, error: null }),
(error) => ({ success: false, message: null, error: toMessage(error) }), (error) => ({ success: false, message: null, error: toMessage(error) }),
); );
@ -33,7 +32,7 @@ const runUnlockStep = async (config: RepositoryConfig, signal?: AbortSignal) =>
const runCheckStep = async (config: RepositoryConfig, signal: AbortSignal) => { const runCheckStep = async (config: RepositoryConfig, signal: AbortSignal) => {
const orgId = getOrganizationId(); const orgId = getOrganizationId();
const result = await Effect.runPromise( const result = await runEffectPromise(
restic.check(config, { readData: false, signal, organizationId: orgId }), restic.check(config, { readData: false, signal, organizationId: orgId }),
).then( ).then(
(result) => result, (result) => result,
@ -50,7 +49,7 @@ const runCheckStep = async (config: RepositoryConfig, signal: AbortSignal) => {
const runRepairIndexStep = async (config: RepositoryConfig, signal: AbortSignal) => { const runRepairIndexStep = async (config: RepositoryConfig, signal: AbortSignal) => {
const orgId = getOrganizationId(); const orgId = getOrganizationId();
const result = await Effect.runPromise(restic.repairIndex(config, { signal, organizationId: orgId })).then( const result = await runEffectPromise(restic.repairIndex(config, { signal, organizationId: orgId })).then(
(result) => ({ success: true, output: result.output, error: null }), (result) => ({ success: true, output: result.output, error: null }),
(error) => ({ success: false, output: null, error: toMessage(error) }), (error) => ({ success: false, output: null, error: toMessage(error) }),
); );

34
app/server/modules/repositories/repositories.service.ts
View file

@ -21,7 +21,7 @@ import { repoMutex } from "../../core/repository-mutex";
import { db } from "../../db/db"; import { db } from "../../db/db";
import { repositoriesTable, type Repository } from "../../db/schema"; import { repositoriesTable, type Repository } from "../../db/schema";
import { cache, cacheKeys } from "../../utils/cache"; import { cache, cacheKeys } from "../../utils/cache";
import { toMessage } from "../../utils/errors"; import { runEffectPromise, toMessage } from "../../utils/errors";
import { generateShortId } from "../../utils/id"; import { generateShortId } from "../../utils/id";
import { addCommonArgs, buildEnv, buildRepoUrl, cleanupTemporaryKeys } from "@zerobyte/core/restic/server"; import { addCommonArgs, buildEnv, buildRepoUrl, cleanupTemporaryKeys } from "@zerobyte/core/restic/server";
import { restic, resticDeps } from "../../core/restic"; import { restic, resticDeps } from "../../core/restic";
@ -102,13 +102,13 @@ const createRepository = async (name: string, config: RepositoryConfig, compress
let error: string | null = null; let error: string | null = null;
if (config.isExistingRepository) { if (config.isExistingRepository) {
const result = await Effect.runPromise(restic.snapshots(encryptedConfig, { organizationId })) const result = await runEffectPromise(restic.snapshots(encryptedConfig, { organizationId }))
.then(() => ({ error: null })) .then(() => ({ error: null }))
.catch((error) => ({ error })); .catch((error) => ({ error }));
error = result.error; error = result.error;
} else { } else {
const initResult = await Effect.runPromise( const initResult = await runEffectPromise(
restic.init(encryptedConfig, { restic.init(encryptedConfig, {
organizationId, organizationId,
timeoutMs: appConfig.serverIdleTimeout * 1000, timeoutMs: appConfig.serverIdleTimeout * 1000,
@ -147,7 +147,7 @@ const getRepository = async (shortId: ShortId) => {
const runAndStoreRepositoryStats = async (repository: Repository): Promise<ResticStatsDto> => { const runAndStoreRepositoryStats = async (repository: Repository): Promise<ResticStatsDto> => {
const releaseLock = await repoMutex.acquireShared(repository.id, "stats"); const releaseLock = await repoMutex.acquireShared(repository.id, "stats");
try { try {
const stats = await Effect.runPromise( const stats = await runEffectPromise(
restic.stats(repository.config, { organizationId: repository.organizationId }), restic.stats(repository.config, { organizationId: repository.organizationId }),
); );
@ -235,11 +235,11 @@ const listSnapshots = async (shortId: ShortId, backupId?: ShortId) => {
let snapshots = []; let snapshots = [];
if (backupId) { if (backupId) {
snapshots = await Effect.runPromise( snapshots = await runEffectPromise(
restic.snapshots(repository.config, { tags: [backupId], organizationId }), restic.snapshots(repository.config, { tags: [backupId], organizationId }),
); );
} else { } else {
snapshots = await Effect.runPromise(restic.snapshots(repository.config, { organizationId })); snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId }));
} }
cache.set(cacheKey, snapshots); cache.set(cacheKey, snapshots);
@ -286,7 +286,7 @@ const listSnapshotFiles = async (
const releaseLock = await repoMutex.acquireShared(repository.id, `ls:${snapshotId}`); const releaseLock = await repoMutex.acquireShared(repository.id, `ls:${snapshotId}`);
try { try {
const result = await Effect.runPromise( const result = await runEffectPromise(
restic.ls(repository.config, snapshotId, path, { organizationId, offset, limit }), restic.ls(repository.config, snapshotId, path, { organizationId, offset, limit }),
); );
@ -368,7 +368,7 @@ const restoreSnapshot = async (
snapshotId, snapshotId,
}); });
const result = await Effect.runPromise( const result = await runEffectPromise(
restic.restore(repository.config, snapshotId, target, { restic.restore(repository.config, snapshotId, target, {
basePath, basePath,
...options, ...options,
@ -448,7 +448,7 @@ const dumpSnapshot = async (shortId: ShortId, snapshotId: string, path?: string,
} }
} }
dumpStream = await Effect.runPromise(restic.dump(repository.config, preparedDump.snapshotRef, dumpOptions)); dumpStream = await runEffectPromise(restic.dump(repository.config, preparedDump.snapshotRef, dumpOptions));
serverEvents.emit("dump:started", { serverEvents.emit("dump:started", {
organizationId, organizationId,
@ -485,7 +485,7 @@ const getSnapshotDetails = async (shortId: ShortId, snapshotId: string) => {
if (!snapshots) { if (!snapshots) {
const releaseLock = await repoMutex.acquireShared(repository.id, `snapshot_details:${snapshotId}`); const releaseLock = await repoMutex.acquireShared(repository.id, `snapshot_details:${snapshotId}`);
try { try {
snapshots = await Effect.runPromise(restic.snapshots(repository.config, { organizationId })); snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId }));
cache.set(cacheKey, snapshots); cache.set(cacheKey, snapshots);
} finally { } finally {
releaseLock(); releaseLock();
@ -513,7 +513,7 @@ const checkHealth = async (shortId: ShortId) => {
const releaseLock = await repoMutex.acquireExclusive(repository.id, "check"); const releaseLock = await repoMutex.acquireExclusive(repository.id, "check");
try { try {
const { hasErrors, error } = await Effect.runPromise(restic.check(repository.config, { organizationId })); const { hasErrors, error } = await runEffectPromise(restic.check(repository.config, { organizationId }));
await db await db
.update(repositoriesTable) .update(repositoriesTable)
@ -611,7 +611,7 @@ const deleteSnapshot = async (shortId: ShortId, snapshotId: string) => {
const releaseLock = await repoMutex.acquireExclusive(repository.id, `delete:${snapshotId}`); const releaseLock = await repoMutex.acquireExclusive(repository.id, `delete:${snapshotId}`);
try { try {
await Effect.runPromise(restic.deleteSnapshot(repository.config, snapshotId, { organizationId })); await runEffectPromise(restic.deleteSnapshot(repository.config, snapshotId, { organizationId }));
cache.delByPrefix(cacheKeys.repository.all(repository.id)); cache.delByPrefix(cacheKeys.repository.all(repository.id));
void runAndStoreRepositoryStats(repository).catch((error) => { void runAndStoreRepositoryStats(repository).catch((error) => {
logger.error( logger.error(
@ -634,7 +634,7 @@ const deleteSnapshots = async (shortId: ShortId, snapshotIds: string[]) => {
let shouldRefreshStats = false; let shouldRefreshStats = false;
const releaseLock = await repoMutex.acquireExclusive(repository.id, `delete:bulk`); const releaseLock = await repoMutex.acquireExclusive(repository.id, `delete:bulk`);
try { try {
await Effect.runPromise(restic.deleteSnapshots(repository.config, snapshotIds, { organizationId })); await runEffectPromise(restic.deleteSnapshots(repository.config, snapshotIds, { organizationId }));
cache.delByPrefix(cacheKeys.repository.all(repository.id)); cache.delByPrefix(cacheKeys.repository.all(repository.id));
shouldRefreshStats = true; shouldRefreshStats = true;
} finally { } finally {
@ -666,7 +666,7 @@ const tagSnapshots = async (
const releaseLock = await repoMutex.acquireExclusive(repository.id, `tag:bulk`); const releaseLock = await repoMutex.acquireExclusive(repository.id, `tag:bulk`);
try { try {
await Effect.runPromise(restic.tagSnapshots(repository.config, snapshotIds, tags, { organizationId })); await runEffectPromise(restic.tagSnapshots(repository.config, snapshotIds, tags, { organizationId }));
cache.delByPrefix(cacheKeys.repository.all(repository.id)); cache.delByPrefix(cacheKeys.repository.all(repository.id));
} finally { } finally {
releaseLock(); releaseLock();
@ -685,7 +685,7 @@ const refreshSnapshots = async (shortId: ShortId) => {
const releaseLock = await repoMutex.acquireShared(repository.id, "refresh"); const releaseLock = await repoMutex.acquireShared(repository.id, "refresh");
try { try {
const snapshots = await Effect.runPromise(restic.snapshots(repository.config, { organizationId })); const snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId }));
const cacheKey = cacheKeys.repository.snapshots(repository.id); const cacheKey = cacheKeys.repository.snapshots(repository.id);
cache.set(cacheKey, snapshots); cache.set(cacheKey, snapshots);
@ -784,7 +784,7 @@ const unlockRepository = async (shortId: ShortId) => {
const releaseLock = await repoMutex.acquireExclusive(repository.id, "unlock"); const releaseLock = await repoMutex.acquireExclusive(repository.id, "unlock");
try { try {
const result = await Effect.runPromise(restic.unlock(repository.config, { organizationId })); const result = await runEffectPromise(restic.unlock(repository.config, { organizationId }));
return result; return result;
} finally { } finally {
releaseLock(); releaseLock();
@ -847,7 +847,7 @@ const getRetentionCategories = async (repositoryId: ShortId, scheduleId?: ShortI
return new Map<string, RetentionCategory[]>(); return new Map<string, RetentionCategory[]>();
} }
const dryRunResults = await Effect.runPromise( const dryRunResults = await runEffectPromise(
restic.forget(repository.config, schedule.retentionPolicy, { restic.forget(repository.config, schedule.retentionPolicy, {
tag: scheduleId, tag: scheduleId,
organizationId: getOrganizationId(), organizationId: getOrganizationId(),

11
app/server/utils/errors.ts
View file

@ -2,6 +2,7 @@ import { HttpError } from "http-errors-enhanced";
import { sanitizeSensitiveData } from "@zerobyte/core/node"; import { sanitizeSensitiveData } from "@zerobyte/core/node";
import { ResticError } from "@zerobyte/core/restic"; import { ResticError } from "@zerobyte/core/restic";
import { toErrorDetails as getErrorDetails, toMessage as getMessage } from "@zerobyte/core/utils"; import { toErrorDetails as getErrorDetails, toMessage as getMessage } from "@zerobyte/core/utils";
import { Cause, Effect, Exit, Option } from "effect";
const formatAllowedHostsMessage = (message: string) => { const formatAllowedHostsMessage = (message: string) => {
const referencesAllowedHosts = /\ballowed\s+hosts?\b|\ballowedHosts\b/i.test(message); const referencesAllowedHosts = /\ballowed\s+hosts?\b|\ballowedHosts\b/i.test(message);
@ -48,3 +49,13 @@ export const toMessage = (err: unknown): string => {
export const toErrorDetails = (err: unknown): string => { export const toErrorDetails = (err: unknown): string => {
return sanitizeSensitiveData(getErrorDetails(err)); return sanitizeSensitiveData(getErrorDetails(err));
}; };
export const runEffectPromise = <A, E>(effect: Effect.Effect<A, E, never>) => {
return Effect.runPromiseExit(effect).then((exit) => {
if (Exit.isSuccess(exit)) {
return exit.value;
}
throw Option.getOrUndefined(Cause.failureOption(exit.cause)) ?? Cause.squash(exit.cause);
});
};