From e423c7327a915e1a0ce57d09b94798c5cc669435 Mon Sep 17 00:00:00 2001 From: Nicolas Meienberger Date: Fri, 16 Jan 2026 23:13:19 +0100 Subject: [PATCH] refactor(doctor): support abort signal in all operations --- .../core/__tests__/repository-mutex.test.ts | 59 ++++++++++++++++ app/server/core/repository-mutex.ts | 68 +++++++++++++++++-- app/server/modules/backups/backups.service.ts | 2 +- app/server/modules/repositories/doctor.ts | 32 ++++----- app/server/utils/restic.ts | 13 ++-- 5 files changed, 145 insertions(+), 29 deletions(-) diff --git a/app/server/core/__tests__/repository-mutex.test.ts b/app/server/core/__tests__/repository-mutex.test.ts index 623881e0..ade8307f 100644 --- a/app/server/core/__tests__/repository-mutex.test.ts +++ b/app/server/core/__tests__/repository-mutex.test.ts @@ -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(); + }); }); diff --git a/app/server/core/repository-mutex.ts b/app/server/core/repository-mutex.ts index a1bb5ad4..959c8778 100644 --- a/app/server/core/repository-mutex.ts +++ b/app/server/core/repository-mutex.ts @@ -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((resolve) => { - state.waitQueue.push({ type: "shared", operation, resolve }); + + let onAbort: () => void = () => {}; + const lockId = await new Promise((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((resolve) => { - state.waitQueue.push({ type: "exclusive", operation, resolve }); + + let onAbort: () => void = () => {}; + const lockId = await new Promise((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); } diff --git a/app/server/modules/backups/backups.service.ts b/app/server/modules/backups/backups.service.ts index d18e507b..8b7ca06d 100644 --- a/app/server/modules/backups/backups.service.ts +++ b/app/server/modules/backups/backups.service.ts @@ -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, { diff --git a/app/server/modules/repositories/doctor.ts b/app/server/modules/repositories/doctor.ts index 2600d95f..13818d98 100644 --- a/app/server/modules/repositories/doctor.ts +++ b/app/server/modules/repositories/doctor.ts @@ -9,8 +9,8 @@ import { type } from "arktype"; import { serverEvents } from "../../core/events"; import { logger } from "../../utils/logger"; -const runUnlockStep = async (config: RepositoryConfig): Promise => { - 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 => { }; }; -const runCheckStep = async (config: RepositoryConfig): Promise => { - 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 => { }; }; -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); } diff --git a/app/server/utils/restic.ts b/app/server/utils/restic.ts index dd5f7b64..89dc951e 100644 --- a/app/server/utils/restic.ts +++ b/app/server/utils/restic.ts @@ -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;