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

refactor(doctor): support abort signal in all operations

Browse source
This commit is contained in:
Nicolas Meienberger 2026-01-16 23:13:19 +01:00
parent d345960afb
commit e423c7327a
5 changed files with 145 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

59
app/server/core/__tests__/repository-mutex.test.ts
View file

@ -35,4 +35,63 @@ describe("RepositoryMutex", () => {
releaseShared2();
});
test("should remove aborted acquisitions from the wait queue", async () => {
const repoId = "abort-test";
const results: string[] = [];
const releaseShared1 = await repoMutex.acquireShared(repoId, "backup-1");
results.push("acquired-shared-1");
const controller = new AbortController();
const exclusivePromise = repoMutex.acquireExclusive(repoId, "unlock", controller.signal).catch((err) => {
results.push("aborted-exclusive");
throw err;
});
const shared2Promise = repoMutex.acquireShared(repoId, "backup-2").then((release) => {
results.push("acquired-shared-2");
return release;
});
await new Promise((resolve) => setTimeout(resolve, 50));
expect(results).toEqual(["acquired-shared-1"]);
controller.abort();
expect(exclusivePromise).rejects.toThrow();
expect(results).toEqual(["acquired-shared-1", "aborted-exclusive"]);
releaseShared1();
// After exclusive is aborted, shared-2 should be next
const releaseShared2 = await shared2Promise;
expect(results).toEqual(["acquired-shared-1", "aborted-exclusive", "acquired-shared-2"]);
releaseShared2();
});
test("should handle multiple aborts correctly", async () => {
const repoId = "multi-abort";
const releaseShared1 = await repoMutex.acquireShared(repoId, "backup-1");
const controller1 = new AbortController();
const controller2 = new AbortController();
const p1 = repoMutex.acquireExclusive(repoId, "ex-1", controller1.signal);
const p2 = repoMutex.acquireExclusive(repoId, "ex-2", controller2.signal);
const p3 = repoMutex.acquireExclusive(repoId, "ex-3");
controller2.abort();
expect(p2).rejects.toThrow();
controller1.abort();
expect(p1).rejects.toThrow();
releaseShared1();
const releaseEx3 = await p3;
expect(releaseEx3).toBeDefined();
releaseEx3();
});
});

68
app/server/core/repository-mutex.ts
View file

@ -46,7 +46,11 @@ class RepositoryMutex {
}
}
async acquireShared(repositoryId: string, operation: string): Promise<() => void> {
async acquireShared(repositoryId: string, operation: string, signal?: AbortSignal): Promise<() => void> {
if (signal?.aborted) {
throw signal.reason || new Error("Operation aborted");
}
const state = this.getOrCreateState(repositoryId);
const hasExclusiveInQueue = state.waitQueue.some((item) => item.type === "exclusive");
@ -64,14 +68,42 @@ class RepositoryMutex {
logger.debug(
`[Mutex] Waiting for shared lock on repo ${repositoryId}: ${operation} (exclusive held by: ${state.exclusiveHolder?.operation ?? "none"}, queue: ${state.waitQueue.length})`,
);
const lockId = await new Promise<string>((resolve) => {
state.waitQueue.push({ type: "shared", operation, resolve });
let onAbort: () => void = () => {};
const lockId = await new Promise<string>((resolve, reject) => {
const waiter = { type: "shared" as const, operation, resolve };
state.waitQueue.push(waiter);
if (signal) {
onAbort = () => {
const index = state.waitQueue.indexOf(waiter);
if (index !== -1) {
state.waitQueue.splice(index, 1);
this.cleanupStateIfEmpty(repositoryId);
reject(signal.reason || new Error("Operation aborted"));
}
};
signal.addEventListener("abort", onAbort);
}
});
if (onAbort) {
signal?.removeEventListener("abort", onAbort);
}
if (signal?.aborted) {
this.releaseShared(repositoryId, lockId);
throw signal.reason || new Error("Operation aborted");
}
return () => this.releaseShared(repositoryId, lockId);
}
async acquireExclusive(repositoryId: string, operation: string): Promise<() => void> {
async acquireExclusive(repositoryId: string, operation: string, signal?: AbortSignal): Promise<() => void> {
if (signal?.aborted) {
throw signal.reason || new Error("Operation aborted");
}
const state = this.getOrCreateState(repositoryId);
if (!state.exclusiveHolder && state.sharedHolders.size === 0 && state.waitQueue.length === 0) {
@ -87,10 +119,34 @@ class RepositoryMutex {
logger.debug(
`[Mutex] Waiting for exclusive lock on repo ${repositoryId}: ${operation} (shared: ${state.sharedHolders.size}, exclusive: ${state.exclusiveHolder ? "yes" : "no"}, queue: ${state.waitQueue.length})`,
);
const lockId = await new Promise<string>((resolve) => {
state.waitQueue.push({ type: "exclusive", operation, resolve });
let onAbort: () => void = () => {};
const lockId = await new Promise<string>((resolve, reject) => {
const waiter = { type: "exclusive" as const, operation, resolve };
state.waitQueue.push(waiter);
if (signal) {
onAbort = () => {
const index = state.waitQueue.indexOf(waiter);
if (index !== -1) {
state.waitQueue.splice(index, 1);
this.cleanupStateIfEmpty(repositoryId);
reject(signal.reason || new Error("Operation aborted"));
}
};
signal.addEventListener("abort", onAbort);
}
});
if (onAbort) {
signal?.removeEventListener("abort", onAbort);
}
if (signal?.aborted) {
this.releaseExclusive(repositoryId, lockId);
throw signal.reason || new Error("Operation aborted");
}
logger.debug(`[Mutex] Acquired exclusive lock for repo ${repositoryId}: ${operation} (${lockId})`);
return () => this.releaseExclusive(repositoryId, lockId);
}

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

@ -298,7 +298,7 @@ const executeBackup = async (scheduleId: number, manual = false) => {
backupOptions.include = schedule.includePatterns.map((p) => processPattern(p, volumePath));
}
const releaseBackupLock = await repoMutex.acquireShared(repository.id, `backup:${volume.name}`);
const releaseBackupLock = await repoMutex.acquireShared(repository.id, `backup:${volume.name}`, abortController.signal);
let exitCode: number;
try {
const result = await restic.backup(repository.config, volumePath, {

32
app/server/modules/repositories/doctor.ts
View file

@ -9,8 +9,8 @@ import { type } from "arktype";
import { serverEvents } from "../../core/events";
import { logger } from "../../utils/logger";
const runUnlockStep = async (config: RepositoryConfig): Promise<DoctorStep> => {
const result = await restic.unlock(config).then(
const runUnlockStep = async (config: RepositoryConfig, signal?: AbortSignal) => {
const result = await restic.unlock(config, { signal }).then(
(result) => ({ success: true, message: result.message, error: null }),
(error) => ({ success: false, message: null, error: toMessage(error) }),
);
@ -23,8 +23,8 @@ const runUnlockStep = async (config: RepositoryConfig): Promise<DoctorStep> => {
};
};
const runCheckStep = async (config: RepositoryConfig): Promise<DoctorStep> => {
const result = await restic.check(config, { readData: true }).then(
const runCheckStep = async (config: RepositoryConfig, signal: AbortSignal) => {
const result = await restic.check(config, { readData: true, signal }).then(
(result) => result,
(error) => ({ success: false, output: null, error: toMessage(error), hasErrors: true }),
);
@ -37,8 +37,8 @@ const runCheckStep = async (config: RepositoryConfig): Promise<DoctorStep> => {
};
};
const runRepairIndexStep = async (config: RepositoryConfig) => {
const result = await restic.repairIndex(config).then(
const runRepairIndexStep = async (config: RepositoryConfig, signal: AbortSignal) => {
const result = await restic.repairIndex(config, { signal }).then(
(result) => ({ success: true, output: result.output, error: null }),
(error) => ({ success: false, output: null, error: toMessage(error) }),
);
@ -103,28 +103,28 @@ export const executeDoctor = async (
repositoryId: string,
repositoryConfig: RepositoryConfig,
repositoryName: string,
signal: AbortSignal | undefined,
signal: AbortSignal,
) => {
const steps: DoctorStep[] = [];
try {
const releaseLock = await repoMutex.acquireExclusive(repositoryId, "doctor");
const releaseLock = await repoMutex.acquireExclusive(repositoryId, "doctor", signal);
// Step 1: Unlock repository
const unlockStep = await runUnlockStep(repositoryConfig);
steps.push(unlockStep);
checkAbortSignal(signal);
// Step 2: Check repository with exclusive lock
try {
const checkStep = await runCheckStep(repositoryConfig);
// Step 1: Unlock repository
const unlockStep = await runUnlockStep(repositoryConfig, signal);
steps.push(unlockStep);
checkAbortSignal(signal);
// Step 2: Check repository
const checkStep = await runCheckStep(repositoryConfig, signal);
steps.push(checkStep);
checkAbortSignal(signal);
// Step 3: Repair index if suggested
const checkOutput = parseCheckOutput(checkStep.output);
if (checkOutput?.suggest_repair_index) {
const repairStep = await runRepairIndexStep(repositoryConfig);
const repairStep = await runRepairIndexStep(repositoryConfig, signal);
steps.push(repairStep);
checkAbortSignal(signal);
}

13
app/server/utils/restic.ts
View file

@ -723,14 +723,14 @@ const ls = async (config: RepositoryConfig, snapshotId: string, path?: string) =
return { snapshot, nodes };
};
const unlock = async (config: RepositoryConfig) => {
const unlock = async (config: RepositoryConfig, options?: { signal?: AbortSignal }) => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config);
const args = ["unlock", "--repo", repoUrl, "--remove-all"];
addCommonArgs(args, env, config);
const res = await exec({ command: "restic", args, env });
const res = await exec({ command: "restic", args, env, signal: options?.signal });
await cleanupTemporaryKeys(env);
if (res.exitCode !== 0) {
@ -742,7 +742,7 @@ const unlock = async (config: RepositoryConfig) => {
return { success: true, message: "Repository unlocked successfully" };
};
const check = async (config: RepositoryConfig, options?: { readData?: boolean }) => {
const check = async (config: RepositoryConfig, options?: { readData?: boolean; signal?: AbortSignal }) => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config);
@ -754,7 +754,8 @@ const check = async (config: RepositoryConfig, options?: { readData?: boolean })
addCommonArgs(args, env, config);
const res = await exec({ command: "restic", args, env });
const res = await exec({ command: "restic", args, env, signal: options?.signal });
await cleanupTemporaryKeys(env);
const { stdout, stderr } = res;
@ -780,14 +781,14 @@ const check = async (config: RepositoryConfig, options?: { readData?: boolean })
};
};
const repairIndex = async (config: RepositoryConfig) => {
const repairIndex = async (config: RepositoryConfig, options?: { signal?: AbortSignal }) => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config);
const args = ["repair", "index", "--repo", repoUrl];
addCommonArgs(args, env, config);
const res = await exec({ command: "restic", args, env });
const res = await exec({ command: "restic", args, env, signal: options?.signal });
await cleanupTemporaryKeys(env);
const { stdout, stderr } = res;