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

refactor(restic): all commands return effects (#924)

Browse source 
* refactor(restic): all commands are effects

* fix(restic): preserve effect failure errors

* chore: pr feedbacks
This commit is contained in:
Nico 2026-05-30 10:10:54 +02:00 committed by GitHub
parent ddf7cab503
commit d4436b0cdc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
36 changed files with 1567 additions and 888 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
app/server/modules/backups/__tests__/backups.mirror-sync.test.ts
View file

@ -10,24 +10,27 @@ import * as context from "~/server/core/request-context";
import * as resticModule from "~/server/core/restic";
import * as spawnModule from "@zerobyte/core/node";
import type { ShortId } from "~/server/utils/branded";
import { Effect } from "effect";
const setup = () => {
vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID);
vi.spyOn(spawnModule, "safeSpawn").mockImplementation(() => Promise.resolve({ exitCode: 0, summary: "", error: "" }));
vi.spyOn(spawnModule, "safeSpawn").mockImplementation(() =>
Promise.resolve({ exitCode: 0, summary: "", error: "" }),
);
return {
mockSnapshots: (sourceSnapshots: unknown[], mirrorSnapshots: unknown[]) => {
let callCount = 0;
vi.spyOn(resticModule.restic, "snapshots").mockImplementation(() => {
callCount++;
if (callCount === 1) return Promise.resolve(sourceSnapshots as never);
return Promise.resolve(mirrorSnapshots as never);
if (callCount === 1) return Effect.succeed(sourceSnapshots as never);
return Effect.succeed(mirrorSnapshots as never);
});
},
mockCopy: () => {
const copyMock = vi
.spyOn(resticModule.restic, "copy")
.mockImplementation(() => Promise.resolve({ success: true, output: "" }));
.mockImplementation(() => Effect.succeed({ success: true, output: "" }));
return copyMock;
},
};
@ -190,12 +193,14 @@ describe("syncMirror", () => {
const copyMock = mockCopy();
let releaseCopy: (() => void) | undefined;
const copyStarted = new Promise<void>((resolve) => {
copyMock.mockImplementation(
copyMock.mockImplementation(() =>
Effect.promise(
() =>
new Promise((copyResolve) => {
releaseCopy = () => copyResolve({ success: true, output: "" });
resolve();
}),
),
);
});
@ -208,9 +213,7 @@ describe("syncMirror", () => {
});
await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id);
await expect(
backupsService.syncMirror(schedule.shortId, mirrorRepository.shortId as ShortId, ["snap1"]),
).resolves.toEqual({ success: true });
const firstSync = backupsService.syncMirror(schedule.shortId, mirrorRepository.shortId as ShortId, ["snap1"]);
await copyStarted;
@ -225,6 +228,8 @@ describe("syncMirror", () => {
releaseCopy?.();
await expect(firstSync).resolves.toEqual({ success: true });
await waitForExpect(async () => {
const mirrors = await backupsService.getMirrors(schedule.shortId);
expect(mirrors[0]?.lastCopyStatus).toBe("success");

25
app/server/modules/backups/__tests__/backups.service.execution.test.ts
View file

@ -24,13 +24,14 @@ import { getScheduleByIdOrShortId } from "../helpers/backup-schedule-lookups";
import { volumeService } from "~/server/modules/volumes/volume.service";
import { db } from "~/server/db/db";
import { config } from "~/server/core/config";
import { Effect } from "effect";
const setup = () => {
const resticBackupMock = vi.fn((_: SafeSpawnParams) =>
Promise.resolve({ exitCode: 0, summary: generateBackupOutput(), error: "" }),
);
const resticForgetMock = vi.fn(() => Promise.resolve({ success: true, data: null }));
const resticCopyMock = vi.fn(() => Promise.resolve({ success: true, output: "" }));
const resticForgetMock = vi.fn(() => Effect.succeed({ success: true, data: null }));
const resticCopyMock = vi.fn(() => Effect.succeed({ success: true, output: "" }));
const { runBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock);
const refreshStatsMock = vi.fn(() =>
Promise.resolve({
@ -925,12 +926,14 @@ describe("mirror operations", () => {
const originalMirror = await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id);
resticCopyMock.mockImplementationOnce(async () => {
resticCopyMock.mockImplementationOnce(() =>
Effect.promise(async () => {
await backupsService.updateMirrors(schedule.id, {
mirrors: [{ repositoryId: mirrorRepository.id, enabled: true }],
});
return { success: true, output: "" };
});
}),
);
// act
await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
@ -957,7 +960,11 @@ describe("mirror operations", () => {
const mirror = await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id);
resticCopyMock.mockImplementationOnce(() => Promise.reject(new Error("Copy failed")));
resticCopyMock.mockImplementationOnce(() =>
Effect.sync(() => {
throw new Error("Copy failed");
}),
);
// act
await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
@ -985,7 +992,7 @@ describe("mirror operations", () => {
await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id);
resticCopyMock.mockClear();
resticCopyMock.mockImplementation(() => Promise.resolve({ success: true, output: "" }));
resticCopyMock.mockImplementation(() => Effect.succeed({ success: true, output: "" }));
// act
await backupsService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
@ -1053,14 +1060,16 @@ describe("mirror operations", () => {
resolveFirstCopyStarted = resolve;
});
resticCopyMock.mockImplementationOnce(
resticCopyMock.mockImplementationOnce(() =>
Effect.promise(
() =>
new Promise((resolve) => {
resolveFirstCopyStarted();
releaseFirstCopy = () => resolve({ success: true, output: "" });
}),
),
);
resticCopyMock.mockImplementation(() => Promise.resolve({ success: true, output: "" }));
resticCopyMock.mockImplementation(() => Effect.succeed({ success: true, output: "" }));
const firstCopyPromise = backupsService.copyToMirrors(firstSchedule.id, sourceRepository, null);
await firstCopyStarted;

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

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

@ -28,7 +28,8 @@ import { getScheduleByIdOrShortId } from "./helpers/backup-schedule-lookups";
import { copyToMirrors, runForget, syncSnapshotsToMirror } from "./helpers/backup-maintenance";
import { restic } from "../../core/restic";
import { mirrorQueries } from "./backups.queries";
import { toMessage } from "../../utils/errors";
import { runEffectPromise, toMessage } from "../../utils/errors";
import { Effect } from "effect";
const listSchedules = async () => {
const organizationId = getOrganizationId();
const schedules = await db.query.backupSchedulesTable.findMany({
@ -73,7 +74,10 @@ const createSchedule = async (data: CreateBackupScheduleBody) => {
const repository = await db.query.repositoriesTable.findFirst({
where: {
AND: [{ OR: [{ id: data.repositoryId }, { shortId: { eq: asShortId(data.repositoryId) } }] }, { organizationId }],
AND: [
{ OR: [{ id: data.repositoryId }, { shortId: { eq: asShortId(data.repositoryId) } }] },
{ organizationId },
],
},
});
@ -149,7 +153,10 @@ const updateSchedule = async (scheduleIdOrShortId: number | string, data: Update
const repository = await db.query.repositoriesTable.findFirst({
where: {
AND: [{ OR: [{ id: data.repositoryId }, { shortId: { eq: asShortId(data.repositoryId) } }] }, { organizationId }],
AND: [
{ OR: [{ id: data.repositoryId }, { shortId: { eq: asShortId(data.repositoryId) } }] },
{ organizationId },
],
},
});
@ -159,7 +166,11 @@ const updateSchedule = async (scheduleIdOrShortId: number | string, data: Update
const cronExpression = data.cronExpression ?? schedule.cronExpression;
const nextBackupAt =
data.cronExpression === "" ? null : data.cronExpression ? calculateNextRun(cronExpression) : schedule.nextBackupAt;
data.cronExpression === ""
? null
: data.cronExpression
? calculateNextRun(cronExpression)
: schedule.nextBackupAt;
const [updated] = await db
.update(backupSchedulesTable)
@ -351,7 +362,12 @@ const reorderSchedules = async (scheduleShortIds: ShortId[]) => {
for (const [index, scheduleId] of scheduleIds.entries()) {
tx.update(backupSchedulesTable)
.set({ sortOrder: index, updatedAt: now })
.where(and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId)))
.where(
and(
eq(backupSchedulesTable.id, scheduleId),
eq(backupSchedulesTable.organizationId, organizationId),
),
)
.run();
}
});
@ -498,10 +514,15 @@ const getMirrorSyncStatus = async (scheduleIdOrShortId: number | string, mirrorS
throw new NotFoundError("Mirror not found for this schedule");
}
const [sourceSnapshots, mirrorSnapshots] = await Promise.all([
const [sourceSnapshots, mirrorSnapshots] = await runEffectPromise(
Effect.all(
[
restic.snapshots(schedule.repository.config, { tags: [schedule.shortId], organizationId }),
restic.snapshots(mirrorRepo.config, { tags: [schedule.shortId], organizationId }),
]);
],
{ concurrency: "unbounded" },
),
);
const mirrorSnapshotTimes = new Set(mirrorSnapshots.map((s) => s.time));

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

@ -5,7 +5,7 @@ import { restic } from "../../../core/restic";
import { repoMutex } from "../../../core/repository-mutex";
import { serverEvents } from "../../../core/events";
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 { mirrorQueries, repositoryQueries, scheduleQueries } from "../backups.queries";
@ -31,7 +31,9 @@ export async function runForget(scheduleId: number, repositoryId?: string, organ
const releaseLock = await repoMutex.acquireExclusive(repository.id, `forget:${scheduleId}`);
try {
await restic.forget(repository.config, schedule.retentionPolicy, { tag: schedule.shortId, organizationId });
await runEffectPromise(
restic.forget(repository.config, schedule.retentionPolicy, { tag: schedule.shortId, organizationId }),
);
cache.delByPrefix(cacheKeys.repository.all(repository.id));
} finally {
releaseLock();
@ -118,11 +120,13 @@ export async function syncSnapshotsToMirror(
]);
try {
await restic.copy(sourceRepository.config, mirrorRepository.config, {
await runEffectPromise(
restic.copy(sourceRepository.config, mirrorRepository.config, {
tag: schedule.shortId,
organizationId,
snapshotIds,
});
}),
);
cache.delByPrefix(cacheKeys.repository.all(mirrorRepository.id));
} finally {
releaseLocks();
@ -206,7 +210,12 @@ async function copyToSingleMirror(
]);
try {
await restic.copy(sourceRepository.config, mirror.repository.config, { tag: schedule.shortId, organizationId });
await runEffectPromise(
restic.copy(sourceRepository.config, mirror.repository.config, {
tag: schedule.shortId,
organizationId,
}),
);
cache.delByPrefix(cacheKeys.repository.all(mirror.repository.id));
} finally {
releaseLocks();

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

@ -8,6 +8,7 @@ import { restic } from "~/server/core/restic";
import { createTestSession } from "~/test/helpers/auth";
import { generateShortId } from "~/server/utils/id";
import { provisionedResourcesSchema, readProvisionedResourcesFile, syncProvisionedResources } from "./provisioning";
import { Effect } from "effect";
describe("provisioning", () => {
let session: Awaited<ReturnType<typeof createTestSession>>;
@ -17,7 +18,7 @@ describe("provisioning", () => {
});
beforeEach(async () => {
vi.spyOn(restic, "init").mockResolvedValue({ success: true, error: null });
vi.spyOn(restic, "init").mockReturnValue(Effect.succeed({ success: true, error: null }));
await db.delete(backupSchedulesTable);
await db.delete(volumesTable);
@ -361,7 +362,7 @@ describe("provisioning", () => {
const { organizationId } = session;
const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-provisioning-"));
const provisioningPath = path.join(tempDir, "provisioning.json");
const initMock = vi.fn(() => Promise.resolve({ success: true, error: null }));
const initMock = vi.fn(() => Effect.succeed({ success: true, error: null }));
vi.spyOn(restic, "init").mockImplementation(initMock);

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

@ -18,7 +18,7 @@ import { mapRepositoryConfigSecrets } from "~/server/modules/repositories/reposi
import { mapVolumeConfigSecrets } from "~/server/modules/volumes/volume-config-secrets";
import { BACKEND_TYPES, volumeConfigSchema, type BackendConfig } from "@zerobyte/contracts/volumes";
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";
const envSecretPrefix = "env://";
@ -171,9 +171,12 @@ const syncProvisionedRepositories = async (repositories: ProvisionedRepository[]
});
if (!repository.config.isExistingRepository) {
const result = await restic
.init(encryptedConfig, repository.organizationId, { timeoutMs: appConfig.serverIdleTimeout * 1000 })
.catch((error) => ({ success: false, error }));
const result = await runEffectPromise(
restic.init(encryptedConfig, {
organizationId: repository.organizationId,
timeoutMs: appConfig.serverIdleTimeout * 1000,
}),
).catch((error) => ({ success: false, error }));
await db
.update(repositoriesTable)
@ -186,7 +189,9 @@ const syncProvisionedRepositories = async (repositories: ProvisionedRepository[]
.where(eq(repositoriesTable.id, id));
if (result.error) {
logger.error(`Provisioned repository ${repository.name} failed to initialize: ${toMessage(result.error)}`);
logger.error(
`Provisioned repository ${repository.name} failed to initialize: ${toMessage(result.error)}`,
);
}
}
continue;

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

@ -8,6 +8,7 @@ import { generateShortId } from "~/server/utils/id";
import { createTestSession, getAuthHeaders } from "~/test/helpers/auth";
import type { RepositoryConfig } from "@zerobyte/core/restic";
import { restic } from "~/server/core/restic";
import { Effect } from "effect";
const app = createApp();
@ -18,7 +19,7 @@ beforeAll(async () => {
});
beforeEach(() => {
vi.spyOn(restic, "init").mockResolvedValue({ success: true, error: null });
vi.spyOn(restic, "init").mockReturnValue(Effect.succeed({ success: true, error: null }));
});
afterEach(() => {
@ -327,15 +328,17 @@ describe("repositories updates", () => {
});
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 { restic } = await import("~/server/core/restic");
const { ResticError } = await import("@zerobyte/core/restic");
const deleteSnapshotSpy = vi.spyOn(restic, "deleteSnapshot").mockImplementation(async () => {
throw new ResticError(1, "Fatal: unexpected HTTP response (403): 403 Forbidden");
});
const deleteSnapshotSpy = vi
.spyOn(restic, "deleteSnapshot")
.mockImplementation(() =>
Effect.fail(new ResticError(1, "Fatal: unexpected HTTP response (403): 403 Forbidden")),
);
try {
const res = await app.request(`/api/v1/repositories/${repository.shortId}/snapshots/snap123`, {
@ -345,8 +348,10 @@ describe("repositories updates", () => {
expect(res.status).toBe(500);
const body = await res.json();
expect(body.message).toBe("Command failed: An error occurred while executing the command.");
expect(body.details).toBe("Fatal: unexpected HTTP response (403): 403 Forbidden");
expect(body).toEqual({
message: "Command failed: An error occurred while executing the command.",
details: "Fatal: unexpected HTTP response (403): 403 Forbidden",
});
} finally {
deleteSnapshotSpy.mockRestore();
}
@ -359,7 +364,8 @@ describe("repositories updates", () => {
const stream = new PassThrough();
const expectedContent = "downloaded snapshot contents";
const snapshotsSpy = vi.spyOn(restic, "snapshots").mockResolvedValue([
const snapshotsSpy = vi.spyOn(restic, "snapshots").mockReturnValue(
Effect.succeed([
{
id: "test-snapshot",
short_id: "test-snapshot",
@ -367,19 +373,25 @@ describe("repositories updates", () => {
paths: ["/mnt/project"],
hostname: "host",
},
]);
const dumpSpy = vi.spyOn(restic, "dump").mockResolvedValue({
stream,
]),
);
const dumpSpy = vi.spyOn(restic, "dump").mockReturnValue(
Effect.succeed({
stream: stream as never,
completion: Promise.resolve(),
abort: vi.fn(),
});
}),
);
try {
const controller = new AbortController();
const response = await app.request(`/api/v1/repositories/${repository.shortId}/snapshots/test-snapshot/dump`, {
const response = await app.request(
`/api/v1/repositories/${repository.shortId}/snapshots/test-snapshot/dump`,
{
headers: session.headers,
signal: controller.signal,
});
},
);
queueMicrotask(() => {
controller.abort();
@ -397,7 +409,8 @@ describe("repositories updates", () => {
const repository = await createRepositoryRecord(session.organizationId);
const stream = new PassThrough();
const snapshotsSpy = vi.spyOn(restic, "snapshots").mockResolvedValue([
const snapshotsSpy = vi.spyOn(restic, "snapshots").mockReturnValue(
Effect.succeed([
{
id: "test-snapshot",
short_id: "test-snapshot",
@ -405,12 +418,15 @@ describe("repositories updates", () => {
paths: ["/mnt/project"],
hostname: "host",
},
]);
const dumpSpy = vi.spyOn(restic, "dump").mockResolvedValue({
stream,
]),
);
const dumpSpy = vi.spyOn(restic, "dump").mockReturnValue(
Effect.succeed({
stream: stream as never,
completion: Promise.resolve(),
abort: vi.fn(),
});
}),
);
try {
stream.end("downloaded snapshot contents");

39
app/server/modules/repositories/__tests__/repositories.service.test.ts
View file

@ -17,6 +17,7 @@ import { createTestBackupSchedule } from "~/test/helpers/backup";
import { cache, cacheKeys } from "~/server/utils/cache";
import { ResticError } from "@zerobyte/core/restic/server";
import { repositoriesService } from "../repositories.service";
import { Effect } from "effect";
const createTestRepository = async (organizationId: string) => {
const id = randomUUID();
@ -44,7 +45,7 @@ beforeAll(async () => {
});
describe("repositoriesService.createRepository", () => {
const initMock = vi.fn(() => Promise.resolve({ success: true, error: null }));
const initMock = vi.fn(() => Effect.succeed({ success: true, error: null }));
beforeEach(() => {
initMock.mockClear();
@ -116,7 +117,7 @@ describe("repositoriesService.createRepository", () => {
const explicitPath = `${REPOSITORY_BASE}/custom-${randomUUID()}`;
const config: RepositoryConfig = { backend: "local", path: explicitPath, isExistingRepository: true };
vi.spyOn(restic, "snapshots").mockImplementation(() => Promise.resolve([]));
vi.spyOn(restic, "snapshots").mockImplementation(() => Effect.succeed([]));
// act
const result = await withContext({ organizationId: session.organizationId, userId: session.user.id }, () =>
@ -174,7 +175,7 @@ describe("repositoriesService repository stats", () => {
snapshots_count: 3,
};
const statsSpy = vi.spyOn(restic, "stats").mockResolvedValue(expectedStats);
const statsSpy = vi.spyOn(restic, "stats").mockReturnValue(Effect.succeed(expectedStats));
const refreshed = await withContext({ organizationId: session.organizationId, userId: session.user.id }, () =>
repositoriesService.refreshRepositoryStats(repository.shortId),
@ -202,7 +203,7 @@ describe("repositoriesService.dumpSnapshot", () => {
});
const createDumpResult = (payload: string) => ({
stream: Readable.from([payload]),
stream: Readable.from([payload]) as never,
completion: Promise.resolve(),
abort: () => {},
});
@ -242,7 +243,8 @@ describe("repositoriesService.dumpSnapshot", () => {
organizationId,
});
vi.spyOn(restic, "snapshots").mockResolvedValue([
vi.spyOn(restic, "snapshots").mockReturnValue(
Effect.succeed([
{
id: snapshotId,
short_id: snapshotId,
@ -250,14 +252,15 @@ describe("repositoriesService.dumpSnapshot", () => {
paths: snapshotPaths ?? [basePath],
hostname: "host",
},
]);
]),
);
const dumpMock = vi.fn((_config: unknown, snapshotRef: string, options: Parameters<typeof restic.dump>[2]) => {
if (!options.path) {
throw new Error("Expected dump path in test");
}
return Promise.resolve(
return Effect.succeed(
createDumpResult(
JSON.stringify({
snapshotRef,
@ -409,7 +412,8 @@ describe("repositoriesService.restoreSnapshot", () => {
const organizationId = session.organizationId;
const repository = await createTestRepository(organizationId);
vi.spyOn(restic, "snapshots").mockResolvedValue([
vi.spyOn(restic, "snapshots").mockReturnValue(
Effect.succeed([
{
id: "snapshot-restore",
short_id: "snapshot-restore",
@ -417,10 +421,11 @@ describe("repositoriesService.restoreSnapshot", () => {
paths,
hostname: "host",
},
]);
]),
);
const restoreMock = vi.fn(() =>
Promise.resolve({
Effect.succeed({
message_type: "summary" as const,
seconds_elapsed: 1,
percent_done: 100,
@ -569,8 +574,8 @@ describe("repositoriesService.getRetentionCategories", () => {
});
const forgetSpy = vi.spyOn(restic, "forget");
forgetSpy.mockResolvedValueOnce(buildForgetResponse(oldSnapshotId));
forgetSpy.mockResolvedValueOnce(buildForgetResponse(newSnapshotId));
forgetSpy.mockReturnValueOnce(Effect.succeed(buildForgetResponse(oldSnapshotId)));
forgetSpy.mockReturnValueOnce(Effect.succeed(buildForgetResponse(newSnapshotId)));
const firstCategories = await withContext({ organizationId, userId: session.user.id }, () =>
repositoriesService.getRetentionCategories(repository.shortId, schedule.shortId),
@ -606,8 +611,8 @@ describe("repositoriesService.deleteSnapshot", () => {
snapshots_count: 1,
};
vi.spyOn(restic, "deleteSnapshot").mockResolvedValue({ success: true });
const statsSpy = vi.spyOn(restic, "stats").mockResolvedValue(expectedStats);
vi.spyOn(restic, "deleteSnapshot").mockReturnValue(Effect.succeed({ success: true }));
const statsSpy = vi.spyOn(restic, "stats").mockReturnValue(Effect.succeed(expectedStats));
await withContext({ organizationId: session.organizationId, userId: session.user.id }, () =>
repositoriesService.deleteSnapshot(repository.shortId, "snap-1"),
@ -625,9 +630,9 @@ describe("repositoriesService.deleteSnapshot", () => {
test("should throw original error when restic deleteSnapshot fails", async () => {
const repository = await createTestRepository(session.organizationId);
vi.spyOn(restic, "deleteSnapshot").mockImplementation(async () => {
throw new ResticError(1, "Fatal: unexpected HTTP response (403): 403 Forbidden");
});
vi.spyOn(restic, "deleteSnapshot").mockImplementation(() =>
Effect.fail(new ResticError(1, "Fatal: unexpected HTTP response (403): 403 Forbidden")),
);
await expect(
withContext({ organizationId: session.organizationId, userId: session.user.id }, () =>

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

@ -3,7 +3,7 @@ import { type DoctorStep, type DoctorResult, type RepositoryConfig } from "@zero
import { z } from "zod";
import { getOrganizationId } from "~/server/core/request-context";
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 { logger } from "@zerobyte/core/node";
import { db } from "~/server/db/db";
@ -17,7 +17,7 @@ class AbortError extends Error {
const runUnlockStep = async (config: RepositoryConfig, signal?: AbortSignal) => {
const orgId = getOrganizationId();
const result = await 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 }),
(error) => ({ success: false, message: null, error: toMessage(error) }),
);
@ -32,7 +32,9 @@ const runUnlockStep = async (config: RepositoryConfig, signal?: AbortSignal) =>
const runCheckStep = async (config: RepositoryConfig, signal: AbortSignal) => {
const orgId = getOrganizationId();
const result = await restic.check(config, { readData: false, signal, organizationId: orgId }).then(
const result = await runEffectPromise(
restic.check(config, { readData: false, signal, organizationId: orgId }),
).then(
(result) => result,
(error) => ({ success: false, output: null, error: toMessage(error), hasErrors: true }),
);
@ -47,7 +49,7 @@ const runCheckStep = async (config: RepositoryConfig, signal: AbortSignal) => {
const runRepairIndexStep = async (config: RepositoryConfig, signal: AbortSignal) => {
const orgId = getOrganizationId();
const result = await 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 }),
(error) => ({ success: false, output: null, error: toMessage(error) }),
);

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

@ -6,6 +6,7 @@ import {
type CompressionMode,
type OverwriteMode,
type RepositoryConfig,
type ResticDumpStream,
type ResticStatsDto,
repositoryConfigSchema,
} from "@zerobyte/core/restic";
@ -20,7 +21,7 @@ import { repoMutex } from "../../core/repository-mutex";
import { db } from "../../db/db";
import { repositoriesTable, type Repository } from "../../db/schema";
import { cache, cacheKeys } from "../../utils/cache";
import { toMessage } from "../../utils/errors";
import { runEffectPromise, toMessage } from "../../utils/errors";
import { generateShortId } from "../../utils/id";
import { addCommonArgs, buildEnv, buildRepoUrl, cleanupTemporaryKeys } from "@zerobyte/core/restic/server";
import { restic, resticDeps } from "../../core/restic";
@ -32,6 +33,7 @@ import { executeDoctor } from "./helpers/doctor";
import type { ShortId } from "~/server/utils/branded";
import { decryptRepositoryConfig, encryptRepositoryConfig } from "./repository-config-secrets";
import { getScheduleByIdOrShortId } from "../backups/helpers/backup-schedule-lookups";
import { Effect } from "effect";
const runningDoctors = new Map<string, AbortController>();
@ -100,16 +102,18 @@ const createRepository = async (name: string, config: RepositoryConfig, compress
let error: string | null = null;
if (config.isExistingRepository) {
const result = await restic
.snapshots(encryptedConfig, { organizationId })
const result = await runEffectPromise(restic.snapshots(encryptedConfig, { organizationId }))
.then(() => ({ error: null }))
.catch((error) => ({ error }));
error = result.error;
} else {
const initResult = await restic.init(encryptedConfig, organizationId, {
const initResult = await runEffectPromise(
restic.init(encryptedConfig, {
organizationId,
timeoutMs: appConfig.serverIdleTimeout * 1000,
});
}),
);
error = initResult.error;
}
@ -143,7 +147,9 @@ const getRepository = async (shortId: ShortId) => {
const runAndStoreRepositoryStats = async (repository: Repository): Promise<ResticStatsDto> => {
const releaseLock = await repoMutex.acquireShared(repository.id, "stats");
try {
const stats = await restic.stats(repository.config, { organizationId: repository.organizationId });
const stats = await runEffectPromise(
restic.stats(repository.config, { organizationId: repository.organizationId }),
);
await db
.update(repositoriesTable)
@ -219,7 +225,7 @@ const listSnapshots = async (shortId: ShortId, backupId?: ShortId) => {
}
const cacheKey = cacheKeys.repository.snapshots(repository.id, backupId);
const cached = cache.get<Awaited<ReturnType<typeof restic.snapshots>>>(cacheKey);
const cached = cache.get<Effect.Effect.Success<ReturnType<typeof restic.snapshots>>>(cacheKey);
if (cached) {
return cached;
}
@ -229,9 +235,11 @@ const listSnapshots = async (shortId: ShortId, backupId?: ShortId) => {
let snapshots = [];
if (backupId) {
snapshots = await restic.snapshots(repository.config, { tags: [backupId], organizationId });
snapshots = await runEffectPromise(
restic.snapshots(repository.config, { tags: [backupId], organizationId }),
);
} else {
snapshots = await restic.snapshots(repository.config, { organizationId });
snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId }));
}
cache.set(cacheKey, snapshots);
@ -278,7 +286,9 @@ const listSnapshotFiles = async (
const releaseLock = await repoMutex.acquireShared(repository.id, `ls:${snapshotId}`);
try {
const result = await restic.ls(repository.config, snapshotId, organizationId, path, { offset, limit });
const result = await runEffectPromise(
restic.ls(repository.config, snapshotId, path, { organizationId, offset, limit }),
);
if (!result.snapshot) {
throw new NotFoundError("Snapshot not found or empty");
@ -358,7 +368,8 @@ const restoreSnapshot = async (
snapshotId,
});
const result = await restic.restore(repository.config, snapshotId, target, {
const result = await runEffectPromise(
restic.restore(repository.config, snapshotId, target, {
basePath,
...options,
organizationId,
@ -370,7 +381,8 @@ const restoreSnapshot = async (
...progress,
});
},
});
}),
);
serverEvents.emit("restore:completed", {
organizationId,
@ -408,7 +420,7 @@ const dumpSnapshot = async (shortId: ShortId, snapshotId: string, path?: string,
}
const releaseLock = await repoMutex.acquireShared(repository.id, `dump:${snapshotId}`);
let dumpStream: Awaited<ReturnType<typeof restic.dump>> | undefined = undefined;
let dumpStream: ResticDumpStream | null = null;
try {
const snapshot = await getSnapshotDetails(repository.shortId, snapshotId);
@ -436,7 +448,7 @@ const dumpSnapshot = async (shortId: ShortId, snapshotId: string, path?: string,
}
}
dumpStream = await restic.dump(repository.config, preparedDump.snapshotRef, dumpOptions);
dumpStream = await runEffectPromise(restic.dump(repository.config, preparedDump.snapshotRef, dumpOptions));
serverEvents.emit("dump:started", {
organizationId,
@ -468,12 +480,12 @@ const getSnapshotDetails = async (shortId: ShortId, snapshotId: string) => {
}
const cacheKey = cacheKeys.repository.snapshots(repository.id);
let snapshots = cache.get<Awaited<ReturnType<typeof restic.snapshots>>>(cacheKey);
let snapshots = cache.get<Effect.Effect.Success<ReturnType<typeof restic.snapshots>>>(cacheKey);
if (!snapshots) {
const releaseLock = await repoMutex.acquireShared(repository.id, `snapshot_details:${snapshotId}`);
try {
snapshots = await restic.snapshots(repository.config, { organizationId });
snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId }));
cache.set(cacheKey, snapshots);
} finally {
releaseLock();
@ -501,7 +513,7 @@ const checkHealth = async (shortId: ShortId) => {
const releaseLock = await repoMutex.acquireExclusive(repository.id, "check");
try {
const { hasErrors, error } = await restic.check(repository.config, { organizationId });
const { hasErrors, error } = await runEffectPromise(restic.check(repository.config, { organizationId }));
await db
.update(repositoriesTable)
@ -599,7 +611,7 @@ const deleteSnapshot = async (shortId: ShortId, snapshotId: string) => {
const releaseLock = await repoMutex.acquireExclusive(repository.id, `delete:${snapshotId}`);
try {
await restic.deleteSnapshot(repository.config, snapshotId, organizationId);
await runEffectPromise(restic.deleteSnapshot(repository.config, snapshotId, { organizationId }));
cache.delByPrefix(cacheKeys.repository.all(repository.id));
void runAndStoreRepositoryStats(repository).catch((error) => {
logger.error(
@ -622,7 +634,7 @@ const deleteSnapshots = async (shortId: ShortId, snapshotIds: string[]) => {
let shouldRefreshStats = false;
const releaseLock = await repoMutex.acquireExclusive(repository.id, `delete:bulk`);
try {
await restic.deleteSnapshots(repository.config, snapshotIds, organizationId);
await runEffectPromise(restic.deleteSnapshots(repository.config, snapshotIds, { organizationId }));
cache.delByPrefix(cacheKeys.repository.all(repository.id));
shouldRefreshStats = true;
} finally {
@ -654,7 +666,7 @@ const tagSnapshots = async (
const releaseLock = await repoMutex.acquireExclusive(repository.id, `tag:bulk`);
try {
await restic.tagSnapshots(repository.config, snapshotIds, tags, organizationId);
await runEffectPromise(restic.tagSnapshots(repository.config, snapshotIds, tags, { organizationId }));
cache.delByPrefix(cacheKeys.repository.all(repository.id));
} finally {
releaseLock();
@ -673,7 +685,7 @@ const refreshSnapshots = async (shortId: ShortId) => {
const releaseLock = await repoMutex.acquireShared(repository.id, "refresh");
try {
const snapshots = await restic.snapshots(repository.config, { organizationId });
const snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId }));
const cacheKey = cacheKeys.repository.snapshots(repository.id);
cache.set(cacheKey, snapshots);
@ -772,7 +784,7 @@ const unlockRepository = async (shortId: ShortId) => {
const releaseLock = await repoMutex.acquireExclusive(repository.id, "unlock");
try {
const result = await restic.unlock(repository.config, { organizationId });
const result = await runEffectPromise(restic.unlock(repository.config, { organizationId }));
return result;
} finally {
releaseLock();
@ -835,11 +847,13 @@ const getRetentionCategories = async (repositoryId: ShortId, scheduleId?: ShortI
return new Map<string, RetentionCategory[]>();
}
const dryRunResults = await restic.forget(repository.config, schedule.retentionPolicy, {
const dryRunResults = await runEffectPromise(
restic.forget(repository.config, schedule.retentionPolicy, {
tag: scheduleId,
organizationId: getOrganizationId(),
dryRun: true,
});
}),
);
if (!dryRunResults.data) {
return new Map<string, RetentionCategory[]>();

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 { ResticError } from "@zerobyte/core/restic";
import { toErrorDetails as getErrorDetails, toMessage as getMessage } from "@zerobyte/core/utils";
import { Cause, Effect, Exit, Option } from "effect";
const formatAllowedHostsMessage = (message: string) => {
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 => {
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);
});
};

16
app/test/backend-integration/runner.ts
View file

@ -138,17 +138,19 @@ async function verifySnapshot(
fixtureRootPath: string,
scenario: IntegrationScenario,
) {
const snapshots = await resticClient.snapshots(repositoryConfig, {
const snapshots = await Effect.runPromise(
resticClient.snapshots(repositoryConfig, {
organizationId,
tags: [uniqueBackupTag],
});
}),
);
const snapshot = snapshots.find((candidate) => candidate.id === snapshotId || candidate.short_id === snapshotId);
if (!snapshot) {
throw new Error(`Unable to find snapshot ${snapshotId} by integration tag ${uniqueBackupTag}`);
}
const res = await resticClient.ls(repositoryConfig, snapshotId, organizationId, undefined, {});
const res = await Effect.runPromise(resticClient.ls(repositoryConfig, snapshotId, undefined, { organizationId }));
await verifySnapshotEntries(fixtureRootPath, res.nodes, scenario.expectedEntries);
}
@ -163,12 +165,14 @@ async function restoreSnapshot(
restoreOptions: IntegrationScenario["restore"],
) {
await fs.mkdir(restoreTarget, { recursive: true });
await resticClient.restore(repositoryConfig, snapshotId, restoreTarget, {
await Effect.runPromise(
resticClient.restore(repositoryConfig, snapshotId, restoreTarget, {
organizationId,
basePath: fixtureRootPath,
excludeXattr: restoreOptions?.excludeXattr,
overwrite: restoreOptions?.overwrite,
});
}),
);
}
async function cleanupScenario(backend: VolumeBackend, restoreTarget: string, scenarioWorkspace: string) {
@ -226,7 +230,7 @@ async function runScenario(scenario: IntegrationScenario, runId: string): Promis
await runStage(stages, "init-repository", async () => {
if (scenario.repository.isExistingRepository) return;
const initResult = await resticClient.init(scenario.repository, organizationId, undefined);
const initResult = await Effect.runPromise(resticClient.init(scenario.repository, { organizationId }));
if (!initResult.success) {
throw new Error(initResult.error ?? "restic init failed");
}

20
packages/core/src/restic/commands/__tests__/copy.test.ts
View file

@ -3,6 +3,7 @@ import * as cleanupModule from "../../helpers/cleanup-temporary-keys";
import * as nodeModule from "../../../node";
import { copy } from "../copy";
import type { ResticDeps } from "../../types";
import { Effect } from "effect";
const mockDeps: ResticDeps = {
resolveSecret: async (s) => s,
@ -49,7 +50,14 @@ describe("copy command", () => {
test("treats flag-like snapshot IDs as positional args", async () => {
const { getArgs } = setup();
await copy(sourceConfig, destConfig, { organizationId: "org-1", snapshotIds: ["--help"], tag: "daily" }, mockDeps);
await Effect.runPromise(
copy(
sourceConfig,
destConfig,
{ organizationId: "org-1", snapshotIds: ["--help"], tag: "daily" },
mockDeps,
),
);
const separatorIndex = getArgs().indexOf("--");
expect(separatorIndex).toBeGreaterThan(-1);
@ -59,7 +67,7 @@ describe("copy command", () => {
test("defaults to 'latest' when no snapshotIds are provided", async () => {
const { getArgs } = setup();
await copy(sourceConfig, destConfig, { organizationId: "org-1", tag: "daily" }, mockDeps);
await Effect.runPromise(copy(sourceConfig, destConfig, { organizationId: "org-1", tag: "daily" }, mockDeps));
const separatorIndex = getArgs().indexOf("--");
expect(separatorIndex).toBeGreaterThan(-1);
@ -69,11 +77,13 @@ describe("copy command", () => {
test("passes multiple snapshot IDs after separator", async () => {
const { getArgs } = setup();
await copy(
await Effect.runPromise(
copy(
sourceConfig,
destConfig,
{ organizationId: "org-1", snapshotIds: ["abc123", "def456", "ghi789"], tag: "daily" },
mockDeps,
),
);
const separatorIndex = getArgs().indexOf("--");
@ -84,7 +94,9 @@ describe("copy command", () => {
test("defaults to 'latest' when snapshotIds is empty array", async () => {
const { getArgs } = setup();
await copy(sourceConfig, destConfig, { organizationId: "org-1", snapshotIds: [], tag: "daily" }, mockDeps);
await Effect.runPromise(
copy(sourceConfig, destConfig, { organizationId: "org-1", snapshotIds: [], tag: "daily" }, mockDeps),
);
const separatorIndex = getArgs().indexOf("--");
expect(separatorIndex).toBeGreaterThan(-1);

21
packages/core/src/restic/commands/__tests__/delete-snapshots.test.ts
View file

@ -3,6 +3,7 @@ import * as cleanupModule from "../../helpers/cleanup-temporary-keys";
import * as nodeModule from "../../../node";
import { deleteSnapshots } from "../delete-snapshots";
import type { ResticDeps } from "../../types";
import { Effect } from "effect";
const mockDeps: ResticDeps = {
resolveSecret: async (s) => s,
@ -43,10 +44,28 @@ describe("deleteSnapshots command", () => {
const { getArgs } = setup();
const snapshotIds = ["--help", "--password-command=sh -c 'id'"];
await deleteSnapshots(config, snapshotIds, "org-1", mockDeps);
await Effect.runPromise(deleteSnapshots(config, snapshotIds, { organizationId: "org-1" }, mockDeps));
const separatorIndex = getArgs().indexOf("--");
expect(separatorIndex).toBeGreaterThan(-1);
expect(getArgs().slice(separatorIndex + 1)).toEqual(snapshotIds);
});
test("rejects empty snapshot IDs before building restic env", async () => {
setup();
const deps = {
...mockDeps,
resolveSecret: vi.fn(mockDeps.resolveSecret),
};
await expect(
Effect.runPromise(deleteSnapshots(config, [], { organizationId: "org-1" }, deps)),
).rejects.toMatchObject({
message: "No snapshot IDs provided for deletion.",
});
expect(deps.resolveSecret).not.toHaveBeenCalled();
expect(nodeModule.safeExec).not.toHaveBeenCalled();
expect(cleanupModule.cleanupTemporaryKeys).not.toHaveBeenCalled();
});
});

5
packages/core/src/restic/commands/__tests__/dump.test.ts
View file

@ -4,6 +4,7 @@ import * as cleanupModule from "../../helpers/cleanup-temporary-keys";
import * as nodeModule from "../../../node";
import { dump } from "../dump";
import type { ResticDeps } from "../../types";
import { Effect } from "effect";
const mockDeps: ResticDeps = {
resolveSecret: async (s) => s,
@ -45,7 +46,9 @@ describe("dump command", () => {
test("treats snapshot reference as a positional arg", async () => {
const { getArgs } = setup();
const result = await dump(config, "--help", { organizationId: "org-1", path: "folder/file.txt" }, mockDeps);
const result = await Effect.runPromise(
dump(config, "--help", { organizationId: "org-1", path: "folder/file.txt" }, mockDeps),
);
await result.completion;
const separatorIndex = getArgs().indexOf("--");

3
packages/core/src/restic/commands/__tests__/ls.test.ts
View file

@ -4,6 +4,7 @@ import * as spawnModule from "../../../node/spawn";
import { ls } from "../ls";
import type { ResticDeps } from "../../types";
import type { SafeSpawnParams } from "../../../node/spawn";
import { Effect } from "effect";
const mockDeps: ResticDeps = {
resolveSecret: async (s) => s,
@ -57,7 +58,7 @@ describe("ls command", () => {
const snapshotId = "--password-command=sh -c 'id'";
const path = "--help";
await ls(config, snapshotId, "org-1", path, undefined, mockDeps);
await Effect.runPromise(ls(config, snapshotId, path, { organizationId: "org-1" }, mockDeps));
const separatorIndex = getArgs().indexOf("--");
expect(separatorIndex).toBeGreaterThan(-1);

37
packages/core/src/restic/commands/__tests__/restore.test.ts
View file

@ -5,6 +5,7 @@ import { ResticError } from "../../error";
import { restore } from "../restore";
import type { ResticDeps } from "../../types";
import type { SafeSpawnParams, SpawnResult } from "../../../node/spawn";
import { Effect } from "effect";
const mockDeps: ResticDeps = {
resolveSecret: async (s) => s,
@ -93,7 +94,8 @@ describe("restore command", () => {
test("uses the common ancestor as restore root and strips includes for non-root targets", async () => {
const { getRestoreArg, getOptionValues } = setup();
await restore(
await Effect.runPromise(
restore(
config,
"snapshot-456",
"/tmp/restore-target",
@ -105,6 +107,7 @@ describe("restore command", () => {
],
},
mockDeps,
),
);
expect(getRestoreArg()).toBe("snapshot-456:/var/lib/zerobyte/volumes/vol123/_data");
@ -114,7 +117,8 @@ describe("restore command", () => {
test("restores a selected file from its parent directory for non-root targets", async () => {
const { getRestoreArg, getOptionValues } = setup();
await restore(
await Effect.runPromise(
restore(
config,
"snapshot-single-file",
"/tmp/restore-target",
@ -124,6 +128,7 @@ describe("restore command", () => {
selectedItemKind: "file",
},
mockDeps,
),
);
expect(getRestoreArg()).toBe("snapshot-single-file:/var/lib/zerobyte/volumes/vol123/_data/archive");
@ -133,7 +138,8 @@ describe("restore command", () => {
test("treats flag-like snapshot IDs as positional restore args", async () => {
const { getArgs, getRestoreArg } = setup();
await restore(
await Effect.runPromise(
restore(
config,
"--help",
"/tmp/restore-target",
@ -142,6 +148,7 @@ describe("restore command", () => {
basePath: "/var/lib/zerobyte/volumes/vol123/_data",
},
mockDeps,
),
);
const separatorIndex = getArgs().indexOf("--");
@ -154,12 +161,14 @@ describe("restore command", () => {
test("returns a parsed restore summary on success", async () => {
setup();
const result = await restore(
const result = await Effect.runPromise(
restore(
config,
"snapshot-123",
"/tmp/restore-target",
{ organizationId: "org-1", basePath: "/var/lib/zerobyte/volumes/vol123/_data" },
mockDeps,
),
);
expect(result).toMatchObject({
@ -172,7 +181,8 @@ describe("restore command", () => {
test("throws ResticError when the command fails", async () => {
setup({ spawnResult: { exitCode: 1, summary: "", error: "restore failed" } });
await expect(
const error = await Effect.runPromise(
Effect.flip(
restore(
config,
"snapshot-123",
@ -180,18 +190,23 @@ describe("restore command", () => {
{ organizationId: "org-1", basePath: "/var/lib/zerobyte/volumes/vol123/_data" },
mockDeps,
),
).rejects.toBeInstanceOf(ResticError);
),
);
expect(error).toBeInstanceOf(ResticError);
});
test("falls back to an empty summary when restic output cannot be parsed", async () => {
setup({ spawnResult: { summary: "not-json" } });
const result = await restore(
const result = await Effect.runPromise(
restore(
config,
"snapshot-123",
"/tmp/restore-target",
{ organizationId: "org-1", basePath: "/var/lib/zerobyte/volumes/vol123/_data" },
mockDeps,
),
);
expect(result).toEqual({
@ -209,7 +224,8 @@ describe("restore command", () => {
const progressUpdates: unknown[] = [];
setup({ onSpawnCall: (params) => params.onStdout?.(validProgressLine) });
await restore(
await Effect.runPromise(
restore(
config,
"snapshot-123",
"/tmp/restore-target",
@ -219,6 +235,7 @@ describe("restore command", () => {
onProgress: (progress) => progressUpdates.push(progress),
},
mockDeps,
),
);
expect(progressUpdates).toHaveLength(1);
@ -238,7 +255,8 @@ describe("restore command", () => {
},
});
await restore(
await Effect.runPromise(
restore(
config,
"snapshot-123",
"/tmp/restore-target",
@ -248,6 +266,7 @@ describe("restore command", () => {
onProgress: (progress) => progressUpdates.push(progress),
},
mockDeps,
),
);
expect(progressUpdates).toHaveLength(0);

3
packages/core/src/restic/commands/__tests__/snapshots.test.ts
View file

@ -4,6 +4,7 @@ import * as nodeModule from "../../../node";
import { snapshots } from "../snapshots";
import type { SafeSpawnParams } from "../../../node";
import type { ResticDeps } from "../../types";
import { Effect } from "effect";
const mockDeps: ResticDeps = {
resolveSecret: async (s) => s,
@ -57,7 +58,7 @@ describe("snapshots command", () => {
return Promise.resolve({ exitCode: 0, summary: "", error: "" });
});
const result = await snapshots(config, { organizationId: "org-1" }, mockDeps);
const result = await Effect.runPromise(snapshots(config, { organizationId: "org-1" }, mockDeps));
expect(result).toEqual(snapshotsOutput);
});

5
packages/core/src/restic/commands/__tests__/tag-snapshots.test.ts
View file

@ -3,6 +3,7 @@ import * as cleanupModule from "../../helpers/cleanup-temporary-keys";
import * as nodeModule from "../../../node";
import { tagSnapshots } from "../tag-snapshots";
import type { ResticDeps } from "../../types";
import { Effect } from "effect";
const mockDeps: ResticDeps = {
resolveSecret: async (s) => s,
@ -43,7 +44,9 @@ describe("tagSnapshots command", () => {
const { getArgs } = setup();
const snapshotIds = ["--help", "--password-command=sh -c 'id'"];
await tagSnapshots(config, snapshotIds, { add: ["keep"] }, "org-1", mockDeps);
await Effect.runPromise(
tagSnapshots(config, snapshotIds, { add: ["keep"] }, { organizationId: "org-1" }, mockDeps),
);
const separatorIndex = getArgs().indexOf("--");
expect(separatorIndex).toBeGreaterThan(-1);

24
packages/core/src/restic/commands/check.ts
View file

@ -1,3 +1,4 @@
import { Data, Effect } from "effect";
import { logger, safeExec } from "../../node";
import { addCommonArgs } from "../helpers/add-common-args";
import { buildEnv } from "../helpers/build-env";
@ -5,8 +6,15 @@ import { buildRepoUrl } from "../helpers/build-repo-url";
import { cleanupTemporaryKeys } from "../helpers/cleanup-temporary-keys";
import type { RepositoryConfig } from "../schemas";
import type { ResticDeps } from "../types";
import { ResticError } from "../error";
import { toMessage } from "../../utils";
export const check = async (
class ResticCheckCommandError extends Data.TaggedError("ResticCheckCommandError")<{
cause: unknown;
message: string;
}> {}
export const check = (
config: RepositoryConfig,
options: {
readData?: boolean;
@ -15,6 +23,8 @@ export const check = async (
},
deps: ResticDeps,
) => {
return Effect.tryPromise({
try: async () => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, options.organizationId, deps);
@ -65,4 +75,16 @@ export const check = async (
output: stdout,
error: hasErrors ? "Repository contains errors" : null,
};
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticCheckCommandError({
cause: error,
message: toMessage(error),
});
},
});
};

23
packages/core/src/restic/commands/copy.ts
View file

@ -7,13 +7,22 @@ import type { RepositoryConfig } from "../schemas";
import { ResticError } from "../error";
import { logger, safeExec } from "../../node";
import type { ResticDeps } from "../types";
import { Data, Effect } from "effect";
import { toMessage } from "../../utils";
export const copy = async (
class ResticCopyCommandError extends Data.TaggedError("ResticCopyCommandError")<{
cause: unknown;
message: string;
}> {}
export const copy = (
sourceConfig: RepositoryConfig,
destConfig: RepositoryConfig,
options: { organizationId: string; tag?: string; snapshotIds?: string[] },
deps: ResticDeps,
) => {
return Effect.tryPromise({
try: async () => {
const sourceRepoUrl = buildRepoUrl(sourceConfig);
const destRepoUrl = buildRepoUrl(destConfig);
@ -71,4 +80,16 @@ export const copy = async (
success: true,
output: stdout,
};
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticCopyCommandError({
cause: error,
message: toMessage(error),
});
},
});
};

37
packages/core/src/restic/commands/delete-snapshots.ts
View file

@ -1,3 +1,4 @@
import { Data, Effect } from "effect";
import { logger, safeExec } from "../../node";
import { ResticError } from "../error";
import { addCommonArgs } from "../helpers/add-common-args";
@ -6,20 +7,28 @@ import { buildRepoUrl } from "../helpers/build-repo-url";
import { cleanupTemporaryKeys } from "../helpers/cleanup-temporary-keys";
import type { RepositoryConfig } from "../schemas";
import type { ResticDeps } from "../types";
import { toMessage } from "../../utils";
export const deleteSnapshots = async (
class ResticDeleteSnapshotCommandError extends Data.TaggedError("ResticDeleteSnapshotCommandError")<{
cause: unknown;
message: string;
}> {}
export const deleteSnapshots = (
config: RepositoryConfig,
snapshotIds: string[],
organizationId: string,
options: { organizationId: string },
deps: ResticDeps,
) => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, organizationId, deps);
return Effect.tryPromise({
try: async () => {
if (snapshotIds.length === 0) {
throw new Error("No snapshot IDs provided for deletion.");
}
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, options.organizationId, deps);
const args: string[] = ["--repo", repoUrl, "forget", "--prune"];
addCommonArgs(args, env, config);
args.push("--", ...snapshotIds);
@ -33,13 +42,25 @@ export const deleteSnapshots = async (
}
return { success: true };
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticDeleteSnapshotCommandError({
cause: error,
message: toMessage(error),
});
},
});
};
export const deleteSnapshot = async (
export const deleteSnapshot = (
config: RepositoryConfig,
snapshotId: string,
organizationId: string,
options: { organizationId: string },
deps: ResticDeps,
) => {
return deleteSnapshots(config, [snapshotId], organizationId, deps);
return deleteSnapshots(config, [snapshotId], options, deps);
};

27
packages/core/src/restic/commands/dump.ts
View file

@ -1,6 +1,6 @@
import { normalizeAbsolutePath } from "../../utils/path";
import type { Readable } from "node:stream";
import type { ResticDeps, ResticDumpStream } from "../types";
import type { ResticDeps } from "../types";
import { addCommonArgs } from "../helpers/add-common-args";
import { buildEnv } from "../helpers/build-env";
import { buildRepoUrl } from "../helpers/build-repo-url";
@ -8,6 +8,13 @@ import { cleanupTemporaryKeys } from "../helpers/cleanup-temporary-keys";
import type { RepositoryConfig } from "../schemas";
import { logger, safeSpawn } from "../../node";
import { ResticError } from "../error";
import { Data, Effect } from "effect";
import { toMessage } from "../../utils";
class ResticDumpCommandError extends Data.TaggedError("ResticDumpCommandError")<{
cause: unknown;
message: string;
}> {}
const normalizeDumpPath = (pathToDump?: string): string => {
const trimmedPath = pathToDump?.trim();
@ -18,7 +25,7 @@ const normalizeDumpPath = (pathToDump?: string): string => {
return normalizeAbsolutePath(trimmedPath);
};
export const dump = async (
export const dump = (
config: RepositoryConfig,
snapshotRef: string,
options: {
@ -27,7 +34,9 @@ export const dump = async (
archive?: false;
},
deps: ResticDeps,
): Promise<ResticDumpStream> => {
) => {
return Effect.tryPromise({
try: async () => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, options.organizationId, deps);
const pathToDump = normalizeDumpPath(options.path);
@ -109,4 +118,16 @@ export const dump = async (
}
},
};
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticDumpCommandError({
cause: error,
message: toMessage(error),
});
},
});
};

23
packages/core/src/restic/commands/forget.ts
View file

@ -7,13 +7,22 @@ import { cleanupTemporaryKeys } from "../helpers/cleanup-temporary-keys";
import type { RepositoryConfig } from "../schemas";
import { logger, safeExec } from "../../node";
import { ResticError } from "../error";
import { toMessage } from "../../utils";
import { Data, Effect } from "effect";
export const forget = async (
class ResticForgetCommandError extends Data.TaggedError("ResticForgetCommandError")<{
cause: unknown;
message: string;
}> {}
export const forget = (
config: RepositoryConfig,
options: RetentionPolicy,
extra: { tag: string; organizationId: string; dryRun?: boolean },
deps: ResticDeps,
) => {
return Effect.tryPromise({
try: async () => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, extra.organizationId, deps);
@ -63,4 +72,16 @@ export const forget = async (
const result = extra.dryRun ? safeJsonParse<ResticForgetResponse>(lines.at(-1) ?? "[]") : null;
return { success: true, data: result };
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticForgetCommandError({
cause: error,
message: toMessage(error),
});
},
});
};

47
packages/core/src/restic/commands/init.ts
View file

@ -6,22 +6,31 @@ import { keyAdd } from "./key-add";
import type { RepositoryConfig } from "../schemas";
import { logger, safeExec } from "../../node";
import type { ResticDeps } from "../types";
import { Data, Effect } from "effect";
import { ResticError } from "../error";
import { toMessage } from "../../utils";
class ResticInitCommandError extends Data.TaggedError("ResticInitCommandError")<{
cause: unknown;
message: string;
}> {}
const addDefaultKey = async (
config: RepositoryConfig,
organizationId: string,
options: { timeoutMs?: number },
options: { organizationId: string; timeoutMs?: number },
deps: ResticDeps,
) => {
if (deps?.hostname) {
const keyResult = await keyAdd(
const keyResult = await Effect.runPromise(
keyAdd(
config,
organizationId,
{
organizationId: options.organizationId,
host: deps.hostname,
timeoutMs: options?.timeoutMs,
},
deps,
),
);
if (!keyResult.success) {
@ -30,17 +39,18 @@ const addDefaultKey = async (
}
};
export const init = async (
export const init = (
config: RepositoryConfig,
organizationId: string,
options: { timeoutMs?: number } | undefined,
options: { organizationId: string; timeoutMs?: number },
deps: ResticDeps,
) => {
return Effect.tryPromise({
try: async () => {
const repoUrl = buildRepoUrl(config);
logger.info(`Initializing restic repository at ${repoUrl}...`);
const env = await buildEnv(config, organizationId, deps);
const env = await buildEnv(config, options.organizationId, deps);
const args = ["init", "--repo", repoUrl];
addCommonArgs(args, env, config);
@ -55,14 +65,19 @@ export const init = async (
logger.info(`Restic repository initialized: ${repoUrl}`);
void addDefaultKey(
config,
organizationId,
{
timeoutMs: options?.timeoutMs,
},
deps,
);
void addDefaultKey(config, { organizationId: options.organizationId, timeoutMs: options?.timeoutMs }, deps);
return { success: true, error: null };
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticInitCommandError({
cause: error,
message: toMessage(error),
});
},
});
};

29
packages/core/src/restic/commands/key-add.ts
View file

@ -1,3 +1,4 @@
import { Data, Effect } from "effect";
import { logger, safeExec } from "../../node";
import { addCommonArgs } from "../helpers/add-common-args";
import { buildEnv } from "../helpers/build-env";
@ -5,18 +6,26 @@ import { buildRepoUrl } from "../helpers/build-repo-url";
import { cleanupTemporaryKeys } from "../helpers/cleanup-temporary-keys";
import type { RepositoryConfig } from "../schemas";
import type { ResticDeps } from "../types";
import { toMessage } from "../../utils";
import { ResticError } from "../error";
export const keyAdd = async (
class ResticKeyAddCommandError extends Data.TaggedError("ResticKeyAddCommandError")<{
cause: unknown;
message: string;
}> {}
export const keyAdd = (
config: RepositoryConfig,
organizationId: string,
options: { host: string; timeoutMs?: number },
options: { organizationId: string; host: string; timeoutMs?: number },
deps: ResticDeps,
) => {
return Effect.tryPromise({
try: async () => {
const repoUrl = buildRepoUrl(config);
logger.info(`Adding restic key with host "${options.host}" for repository at ${repoUrl}...`);
const env = await buildEnv(config, organizationId, deps);
const env = await buildEnv(config, options.organizationId, deps);
const args = [
"key",
@ -41,4 +50,16 @@ export const keyAdd = async (
logger.info(`Restic key added with host "${options.host}" for repository: ${repoUrl}`);
return { success: true, error: null };
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticKeyAddCommandError({
cause: error,
message: toMessage(error),
});
},
});
};

32
packages/core/src/restic/commands/ls.ts
View file

@ -7,6 +7,13 @@ import type { RepositoryConfig } from "../schemas";
import { logger, safeSpawn } from "../../node";
import { ResticError } from "../error";
import type { ResticDeps } from "../types";
import { Data, Effect } from "effect";
import { toMessage } from "../../utils";
class ResticLsCommandError extends Data.TaggedError("ResticLsCommandError")<{
cause: unknown;
message: string;
}> {}
const lsNodeSchema = z.object({
name: z.string(),
@ -49,16 +56,17 @@ type ResticLsResult = {
};
};
export const ls = async (
export const ls = (
config: RepositoryConfig,
snapshotId: string,
organizationId: string,
path: string | undefined,
options: { offset?: number; limit?: number } | undefined,
options: { organizationId: string; offset?: number; limit?: number },
deps: ResticDeps,
): Promise<ResticLsResult> => {
) => {
return Effect.tryPromise({
try: async () => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, organizationId, deps);
const env = await buildEnv(config, options.organizationId, deps);
const args: string[] = ["--repo", repoUrl, "ls", "--long"];
@ -142,5 +150,17 @@ export const ls = async (
total: totalNodes,
hasMore,
},
};
} as ResticLsResult;
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticLsCommandError({
cause: error,
message: toMessage(error),
});
},
});
};

23
packages/core/src/restic/commands/repair-index.ts
View file

@ -1,3 +1,4 @@
import { Data, Effect } from "effect";
import { logger, safeExec } from "../../node";
import { ResticError } from "../error";
import { addCommonArgs } from "../helpers/add-common-args";
@ -6,12 +7,20 @@ import { buildRepoUrl } from "../helpers/build-repo-url";
import { cleanupTemporaryKeys } from "../helpers/cleanup-temporary-keys";
import type { RepositoryConfig } from "../schemas";
import type { ResticDeps } from "../types";
import { toMessage } from "../../utils";
export const repairIndex = async (
class ResticRepairIndexCommandError extends Data.TaggedError("ResticRepairIndexCommandError")<{
cause: unknown;
message: string;
}> {}
export const repairIndex = (
config: RepositoryConfig,
options: { signal?: AbortSignal; organizationId: string },
deps: ResticDeps,
) => {
return Effect.tryPromise({
try: async () => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, options.organizationId, deps);
@ -44,4 +53,16 @@ export const repairIndex = async (
output: stdout,
message: "Index repaired successfully",
};
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticRepairIndexCommandError({
cause: error,
message: toMessage(error),
});
},
});
};

31
packages/core/src/restic/commands/restore.ts
View file

@ -11,6 +11,13 @@ import { logger, safeSpawn } from "../../node";
import { ResticError } from "../error";
import { resticRestoreOutputSchema, type ResticRestoreOutputDto } from "../restic-dto";
import type { ResticDeps } from "../types";
import { Data, Effect } from "effect";
import { toMessage } from "../../utils";
class ResticRestoreCommandError extends Data.TaggedError("ResticRestoreCommandError")<{
cause: unknown;
message: string;
}> {}
const restoreProgressSchema = z.object({
message_type: z.enum(["status", "summary"]),
@ -24,7 +31,7 @@ const restoreProgressSchema = z.object({
export type RestoreProgress = z.infer<typeof restoreProgressSchema>;
export const restore = async (
export const restore = (
config: RepositoryConfig,
snapshotId: string,
target: string,
@ -42,6 +49,8 @@ export const restore = async (
},
deps: ResticDeps,
) => {
return Effect.tryPromise({
try: async () => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, options.organizationId, deps);
@ -69,8 +78,12 @@ export const restore = async (
args.push("--include", pattern);
}
} else {
const strippedIncludes = options.include.map((pattern) => path.posix.relative(commonAncestor, pattern));
const includesCoverRestoreRoot = strippedIncludes.some((pattern) => pattern === "" || pattern === ".");
const strippedIncludes = options.include.map((pattern) =>
path.posix.relative(commonAncestor, pattern),
);
const includesCoverRestoreRoot = strippedIncludes.some(
(pattern) => pattern === "" || pattern === ".",
);
if (!includesCoverRestoreRoot) {
for (const pattern of strippedIncludes) {
@ -168,4 +181,16 @@ export const restore = async (
);
return result.data;
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticRestoreCommandError({
cause: error,
message: toMessage(error),
});
},
});
};

25
packages/core/src/restic/commands/snapshots.ts
View file

@ -7,6 +7,14 @@ import { resticSnapshotSummarySchema } from "../restic-dto";
import type { RepositoryConfig } from "../schemas";
import { logger, safeSpawn } from "../../node";
import type { ResticDeps } from "../types";
import { Data, Effect } from "effect";
import { toMessage } from "../../utils";
import { ResticError } from "../error";
class ResticSnapshotsCommandError extends Data.TaggedError("ResticSnapshotsCommandError")<{
cause: unknown;
message: string;
}> {}
const snapshotInfoSchema = z.object({
gid: z.number().optional(),
@ -23,11 +31,13 @@ const snapshotInfoSchema = z.object({
summary: resticSnapshotSummarySchema.optional(),
});
export const snapshots = async (
export const snapshots = (
config: RepositoryConfig,
options: { tags?: string[]; organizationId: string },
deps: ResticDeps,
) => {
return Effect.tryPromise({
try: async () => {
const { tags, organizationId } = options;
const repoUrl = buildRepoUrl(config);
@ -68,4 +78,17 @@ export const snapshots = async (
}
return result.data;
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticSnapshotsCommandError({
cause: error,
message: toMessage(error),
});
},
});
};

23
packages/core/src/restic/commands/stats.ts
View file

@ -8,8 +8,17 @@ import { logger, safeExec } from "../../node";
import { ResticError } from "../error";
import { resticStatsSchema } from "../restic-dto";
import type { ResticDeps } from "../types";
import { Data, Effect } from "effect";
import { toMessage } from "../../utils";
export const stats = async (config: RepositoryConfig, options: { organizationId: string }, deps: ResticDeps) => {
class ResticStatsCommandError extends Data.TaggedError("ResticStatsCommandError")<{
cause: unknown;
message: string;
}> {}
export const stats = (config: RepositoryConfig, options: { organizationId: string }, deps: ResticDeps) => {
return Effect.tryPromise({
try: async () => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, options.organizationId, deps);
@ -33,4 +42,16 @@ export const stats = async (config: RepositoryConfig, options: { organizationId:
}
return result.data;
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticStatsCommandError({
cause: error,
message: toMessage(error),
});
},
});
};

28
packages/core/src/restic/commands/tag-snapshots.ts
View file

@ -1,3 +1,4 @@
import { Data, Effect } from "effect";
import { logger, safeExec } from "../../node";
import { ResticError } from "../error";
import { addCommonArgs } from "../helpers/add-common-args";
@ -6,20 +7,28 @@ import { buildRepoUrl } from "../helpers/build-repo-url";
import { cleanupTemporaryKeys } from "../helpers/cleanup-temporary-keys";
import type { RepositoryConfig } from "../schemas";
import type { ResticDeps } from "../types";
import { toMessage } from "../../utils";
export const tagSnapshots = async (
class ResticTagSnapshotsCommandError extends Data.TaggedError("ResticTagSnapshotsCommandError")<{
cause: unknown;
message: string;
}> {}
export const tagSnapshots = (
config: RepositoryConfig,
snapshotIds: string[],
tags: { add?: string[]; remove?: string[]; set?: string[] },
organizationId: string,
options: { organizationId: string },
deps: ResticDeps,
) => {
return Effect.tryPromise({
try: async () => {
if (snapshotIds.length === 0) {
throw new Error("No snapshot IDs provided for tagging.");
}
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, organizationId, deps);
const env = await buildEnv(config, options.organizationId, deps);
const args: string[] = ["--repo", repoUrl, "tag"];
@ -53,4 +62,17 @@ export const tagSnapshots = async (
}
return { success: true };
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticTagSnapshotsCommandError({
cause: error,
message: toMessage(error),
});
},
});
};

23
packages/core/src/restic/commands/unlock.ts
View file

@ -1,3 +1,4 @@
import { Data, Effect } from "effect";
import { logger, safeExec } from "../../node";
import { ResticError } from "../error";
import { addCommonArgs } from "../helpers/add-common-args";
@ -6,12 +7,20 @@ import { buildRepoUrl } from "../helpers/build-repo-url";
import { cleanupTemporaryKeys } from "../helpers/cleanup-temporary-keys";
import type { RepositoryConfig } from "../schemas";
import type { ResticDeps } from "../types";
import { toMessage } from "../../utils";
export const unlock = async (
class ResticUnlockCommandError extends Data.TaggedError("ResticUnlockCommandError")<{
cause: unknown;
message: string;
}> {}
export const unlock = (
config: RepositoryConfig,
options: { signal?: AbortSignal; organizationId: string },
deps: ResticDeps,
) => {
return Effect.tryPromise({
try: async () => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, options.organizationId, deps);
@ -38,4 +47,16 @@ export const unlock = async (
logger.info(`Restic unlock succeeded for repository: ${repoUrl}`);
return { success: true, message: "Repository unlocked successfully" };
},
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticUnlockCommandError({
cause: error,
message: toMessage(error),
});
},
});
};

169
packages/core/src/restic/lock-diagnostics.ts Normal file
View file

@ -0,0 +1,169 @@
import { logger, safeExec } from "../node";
import { toMessage } from "../utils";
import { ResticError } from "./error";
import { addCommonArgs } from "./helpers/add-common-args";
import { buildEnv } from "./helpers/build-env";
import { buildRepoUrl } from "./helpers/build-repo-url";
import { cleanupTemporaryKeys } from "./helpers/cleanup-temporary-keys";
import type { RepositoryConfig } from "./schemas";
import type { ResticDeps } from "./types";
type ResticLockDiagnosticContext = {
error: unknown;
operation: string;
repositoryConfig: RepositoryConfig;
organizationId: string;
resticDeps: ResticDeps;
relatedRepositoryConfigs?: RepositoryConfig[];
};
const LOCK_ERROR_PATTERNS = [
/unable to create lock in backend/i,
/repository is already locked/i,
/failed to lock repository/i,
/"code"\s*:\s*11/i,
/\bexit_error\b.*\b11\b/i,
];
export const isResticLockFailure = (error: unknown) => {
if (error instanceof ResticError && error.code === 11) {
return true;
}
const message = toMessage(error);
return LOCK_ERROR_PATTERNS.some((pattern) => pattern.test(message));
};
const parseLockIds = (stdout: string) => {
const ids = new Set<string>();
for (const line of stdout.split("\n")) {
const trimmed = line.trim();
if (!trimmed) continue;
const jsonId = trimmed.match(/"id"\s*:\s*"([^"]+)"/)?.[1];
if (jsonId) {
ids.add(jsonId);
continue;
}
const hexId = trimmed.match(/[a-f0-9]{64}/i)?.[0];
if (hexId) {
ids.add(hexId);
}
}
return [...ids].slice(0, 20);
};
const inspectResticLocks = async (config: RepositoryConfig, organizationId: string, deps: ResticDeps) => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, organizationId, deps);
const baseArgs = ["--repo", repoUrl];
addCommonArgs(baseArgs, env, config);
try {
const listResult = await safeExec({
command: "restic",
args: [...baseArgs, "list", "locks"],
env,
timeout: 15_000,
});
const lockIds = listResult.exitCode === 0 ? parseLockIds(listResult.stdout) : [];
const lockDetails = [];
for (const lockId of lockIds) {
const catResult = await safeExec({
command: "restic",
args: [...baseArgs, "cat", "lock", lockId],
env,
timeout: 15_000,
});
lockDetails.push({
lockId,
exitCode: catResult.exitCode,
stdout: catResult.stdout,
stderr: catResult.stderr,
timedOut: catResult.timedOut,
});
}
return {
repoUrl,
list: {
exitCode: listResult.exitCode,
stdout: listResult.stdout,
stderr: listResult.stderr,
timedOut: listResult.timedOut,
},
lockIds,
lockDetails,
};
} finally {
await cleanupTemporaryKeys(env, deps);
}
};
export const logResticLockFailureDiagnostics = async ({
error,
operation,
repositoryConfig,
organizationId,
resticDeps,
relatedRepositoryConfigs = [],
}: ResticLockDiagnosticContext) => {
if (!isResticLockFailure(error)) {
return false;
}
try {
const configsByRepoUrl = new Map(
[repositoryConfig, ...relatedRepositoryConfigs].map((config) => [buildRepoUrl(config), config]),
);
const configsToInspect = [...configsByRepoUrl.entries()];
logger.error("[ResticLockFailure] Restic repository lock failure detected", {
operation,
error: toMessage(error),
process: {
pid: process.pid,
hostname: process.env.HOSTNAME,
nodeEnv: process.env.NODE_ENV,
},
repository: {
repoUrl: buildRepoUrl(repositoryConfig),
},
relatedRepositories: relatedRepositoryConfigs.map((config) => ({
repoUrl: buildRepoUrl(config),
})),
});
for (const [repoUrl, config] of configsToInspect) {
try {
const resticLocks = await inspectResticLocks(config, organizationId, resticDeps);
logger.error("[ResticLockFailure] Restic backend lock dump", {
operation,
repoUrl,
resticLocks,
});
} catch (diagnosticError) {
logger.error("[ResticLockFailure] Failed to inspect restic backend locks", {
operation,
repoUrl,
error: toMessage(diagnosticError),
});
}
}
return true;
} catch (diagnosticError) {
logger.error("[ResticLockFailure] Failed to collect lock diagnostics", {
operation,
repoUrl: buildRepoUrl(repositoryConfig),
error: toMessage(diagnosticError),
});
return true;
}
};

96
packages/core/src/restic/server.ts
View file

@ -1,3 +1,4 @@
import { Effect } from "effect";
import { backup } from "./commands/backup";
import { check } from "./commands/check";
import { copy } from "./commands/copy";
@ -13,6 +14,8 @@ import { snapshots } from "./commands/snapshots";
import { stats } from "./commands/stats";
import { tagSnapshots } from "./commands/tag-snapshots";
import { unlock } from "./commands/unlock";
import { logResticLockFailureDiagnostics } from "./lock-diagnostics";
import type { RepositoryConfig } from "./schemas";
import type { ResticDeps } from "./types";
export { addCommonArgs } from "./helpers/add-common-args";
@ -20,30 +23,83 @@ export { buildEnv } from "./helpers/build-env";
export { buildRepoUrl } from "./helpers/build-repo-url";
export { cleanupTemporaryKeys } from "./helpers/cleanup-temporary-keys";
export { validateCustomResticParams } from "./helpers/validate-custom-params";
export { isResticLockFailure, logResticLockFailureDiagnostics } from "./lock-diagnostics";
export { ResticError } from "./error";
function withDeps<Args extends unknown[], Result>(
command: (...args: [...Args, ResticDeps]) => Result,
type LockDiagnosticCommandContext = {
repositoryConfig: RepositoryConfig;
organizationId: string;
relatedRepositoryConfigs?: RepositoryConfig[];
};
type ResticCommandOptions = { organizationId: string };
type ResticCommandResult = { error?: unknown };
const getCommandContext = (operation: string, args: unknown[]): LockDiagnosticCommandContext => {
const firstRepositoryConfig = args[0] as RepositoryConfig;
const options = args.at(-1) as ResticCommandOptions;
if (operation === "restic.copy") {
return {
repositoryConfig: args[1] as RepositoryConfig,
organizationId: options.organizationId,
relatedRepositoryConfigs: [firstRepositoryConfig],
};
}
return {
repositoryConfig: firstRepositoryConfig,
organizationId: options.organizationId,
};
};
const logLockFailure = async (
error: unknown,
operation: string,
context: LockDiagnosticCommandContext,
deps: ResticDeps,
): (...args: Args) => Result {
return (...args: Args) => command(...args, deps);
) =>
logResticLockFailureDiagnostics({
error,
operation,
repositoryConfig: context.repositoryConfig,
organizationId: context.organizationId,
resticDeps: deps,
relatedRepositoryConfigs: context.relatedRepositoryConfigs,
});
function withDeps<Args extends unknown[], Success, Failure, Requirements>(
operation: string,
command: (...args: [...Args, ResticDeps]) => Effect.Effect<Success, Failure, Requirements>,
deps: ResticDeps,
): (...args: Args) => Effect.Effect<Success, Failure, Requirements> {
return (...args: Args) => {
const context = getCommandContext(operation, args);
return command(...args, deps).pipe(
Effect.tapError((error) => Effect.promise(() => logLockFailure(error, operation, context, deps))),
Effect.tap((result) => {
const { error } = result as ResticCommandResult;
return error ? Effect.promise(() => logLockFailure(error, operation, context, deps)) : Effect.void;
}),
);
};
}
export const createRestic = (deps: ResticDeps) => ({
init: withDeps(init, deps),
keyAdd: withDeps(keyAdd, deps),
backup: withDeps(backup, deps),
restore: withDeps(restore, deps),
dump: withDeps(dump, deps),
snapshots: withDeps(snapshots, deps),
stats: withDeps(stats, deps),
forget: withDeps(forget, deps),
deleteSnapshot: withDeps(deleteSnapshot, deps),
deleteSnapshots: withDeps(deleteSnapshots, deps),
tagSnapshots: withDeps(tagSnapshots, deps),
unlock: withDeps(unlock, deps),
ls: withDeps(ls, deps),
check: withDeps(check, deps),
repairIndex: withDeps(repairIndex, deps),
copy: withDeps(copy, deps),
init: withDeps("restic.init", init, deps),
keyAdd: withDeps("restic.keyAdd", keyAdd, deps),
backup: withDeps("restic.backup", backup, deps),
restore: withDeps("restic.restore", restore, deps),
dump: withDeps("restic.dump", dump, deps),
snapshots: withDeps("restic.snapshots", snapshots, deps),
stats: withDeps("restic.stats", stats, deps),
forget: withDeps("restic.forget", forget, deps),
deleteSnapshot: withDeps("restic.deleteSnapshot", deleteSnapshot, deps),
deleteSnapshots: withDeps("restic.deleteSnapshots", deleteSnapshots, deps),
tagSnapshots: withDeps("restic.tagSnapshots", tagSnapshots, deps),
unlock: withDeps("restic.unlock", unlock, deps),
ls: withDeps("restic.ls", ls, deps),
check: withDeps("restic.check", check, deps),
repairIndex: withDeps("restic.repairIndex", repairIndex, deps),
copy: withDeps("restic.copy", copy, deps),
});