diff --git a/app/server/core/__tests__/repository-mutex.test.ts b/app/server/core/__tests__/repository-mutex.test.ts index 811ff6fd..b647f238 100644 --- a/app/server/core/__tests__/repository-mutex.test.ts +++ b/app/server/core/__tests__/repository-mutex.test.ts @@ -232,6 +232,76 @@ describe("RepositoryMutex", () => { releaseS3(); }); + test("should acquire many locks in repository order", async () => { + const releaseSharedB = await repoMutex.acquireShared("repo-b", "holder-b"); + const results: string[] = []; + + const manyPromise = repoMutex + .acquireMany([ + { repositoryId: "repo-b", type: "exclusive", operation: "many-b" }, + { repositoryId: "repo-a", type: "shared", operation: "many-a" }, + ]) + .then((release) => { + results.push("many"); + return release; + }); + + await new Promise((resolve) => setTimeout(resolve, 10)); + + const exclusiveAPromise = repoMutex.acquireExclusive("repo-a", "exclusive-a").then((release) => { + results.push("exclusive-a"); + return release; + }); + + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(results).toEqual([]); + + releaseSharedB(); + + const releaseMany = await manyPromise; + expect(results).toEqual(["many"]); + + releaseMany(); + + const releaseExclusiveA = await exclusiveAPromise; + expect(results).toEqual(["many", "exclusive-a"]); + + releaseExclusiveA(); + }); + + test("should release earlier locks when acquireMany is aborted", async () => { + const releaseExclusiveB = await repoMutex.acquireExclusive("repo-b", "holder-b"); + const controller = new AbortController(); + + const manyPromise = repoMutex.acquireMany( + [ + { repositoryId: "repo-b", type: "exclusive", operation: "many-b" }, + { repositoryId: "repo-a", type: "shared", operation: "many-a" }, + ], + controller.signal, + ); + + await new Promise((resolve) => setTimeout(resolve, 10)); + + let exclusiveAAcquired = false; + const exclusiveAPromise = repoMutex.acquireExclusive("repo-a", "exclusive-a").then((release) => { + exclusiveAAcquired = true; + return release; + }); + + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(exclusiveAAcquired).toBe(false); + + controller.abort(new Error("stop")); + await expect(manyPromise).rejects.toThrow("stop"); + + const releaseExclusiveA = await exclusiveAPromise; + expect(exclusiveAAcquired).toBe(true); + + releaseExclusiveA(); + releaseExclusiveB(); + }); + test("should safely handle multiple calls to the release function", async () => { const repoId = "idempotent-release"; diff --git a/app/server/core/repository-mutex.ts b/app/server/core/repository-mutex.ts index 2dee5779..96fb048d 100644 --- a/app/server/core/repository-mutex.ts +++ b/app/server/core/repository-mutex.ts @@ -2,6 +2,12 @@ import { logger } from "@zerobyte/core/node"; export type LockType = "shared" | "exclusive"; +export interface LockRequest { + repositoryId: string; + type: LockType; + operation: string; +} + interface LockHolder { id: string; operation: string; @@ -158,6 +164,54 @@ class RepositoryMutex { return this.createRelease("exclusive", repositoryId, lockId!); } + async acquireMany(requests: LockRequest[], signal?: AbortSignal) { + if (signal?.aborted) { + throw signal.reason || new Error("Operation aborted"); + } + + if (requests.length === 0) { + return () => {}; + } + + const seenRepositoryIds = new Set(); + for (const request of requests) { + if (seenRepositoryIds.has(request.repositoryId)) { + throw new Error(`Duplicate repository lock request: ${request.repositoryId}`); + } + seenRepositoryIds.add(request.repositoryId); + } + + const sortedRequests = [...requests].sort((a, b) => a.repositoryId.localeCompare(b.repositoryId)); + const releases: Array<() => void> = []; + + try { + for (const request of sortedRequests) { + const release = + request.type === "shared" + ? await this.acquireShared(request.repositoryId, request.operation, signal) + : await this.acquireExclusive(request.repositoryId, request.operation, signal); + releases.push(release); + } + } catch (error) { + for (const release of releases.toReversed()) { + release(); + } + throw error; + } + + let released = false; + return () => { + if (released) { + return; + } + + released = true; + for (const release of releases.toReversed()) { + release(); + } + }; + } + private releaseShared(repositoryId: string, lockId: string): void { const state = this.locks.get(repositoryId); if (!state) { diff --git a/app/server/modules/backups/__tests__/backups.execution.test.ts b/app/server/modules/backups/__tests__/backups.execution.test.ts index feb444ee..2aafe12c 100644 --- a/app/server/modules/backups/__tests__/backups.execution.test.ts +++ b/app/server/modules/backups/__tests__/backups.execution.test.ts @@ -525,4 +525,60 @@ describe("mirror operations", () => { // assert expect(resticForgetMock).not.toHaveBeenCalled(); }); + + test("should serialize mirror copies for schedules that share the same mirror repository", async () => { + const { resticCopyMock } = setup(); + const sourceRepository = await createTestRepository(); + const mirrorRepository = await createTestRepository(); + const firstVolume = await createTestVolume(); + const secondVolume = await createTestVolume(); + const firstSchedule = await createTestBackupSchedule({ + volumeId: firstVolume.id, + repositoryId: sourceRepository.id, + }); + const secondSchedule = await createTestBackupSchedule({ + volumeId: secondVolume.id, + repositoryId: sourceRepository.id, + }); + + await createTestBackupScheduleMirror(firstSchedule.id, mirrorRepository.id); + await createTestBackupScheduleMirror(secondSchedule.id, mirrorRepository.id); + + let releaseFirstCopy = () => {}; + let resolveFirstCopyStarted = () => {}; + const firstCopyStarted = new Promise((resolve) => { + resolveFirstCopyStarted = resolve; + }); + + resticCopyMock.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveFirstCopyStarted(); + releaseFirstCopy = () => resolve({ success: true, output: "" }); + }), + ); + resticCopyMock.mockImplementation(() => Promise.resolve({ success: true, output: "" })); + + const firstCopyPromise = backupsExecutionService.copyToMirrors(firstSchedule.id, sourceRepository, null); + await firstCopyStarted; + + const secondCopyPromise = backupsExecutionService.copyToMirrors(secondSchedule.id, sourceRepository, null); + + try { + const secondCopyState = await Promise.race<"resolved" | "timeout">([ + secondCopyPromise.then(() => "resolved"), + new Promise((resolve) => { + setTimeout(() => resolve("timeout"), 50); + }), + ]); + + expect(secondCopyState).toBe("timeout"); + expect(resticCopyMock).toHaveBeenCalledTimes(1); + } finally { + releaseFirstCopy(); + await Promise.all([firstCopyPromise, secondCopyPromise]); + } + + expect(resticCopyMock).toHaveBeenCalledTimes(2); + }); }); diff --git a/app/server/modules/backups/helpers/backup-maintenance.ts b/app/server/modules/backups/helpers/backup-maintenance.ts index 77f6c797..18054eb7 100644 --- a/app/server/modules/backups/helpers/backup-maintenance.ts +++ b/app/server/modules/backups/helpers/backup-maintenance.ts @@ -92,15 +92,16 @@ async function copyToSingleMirror( lastCopyError: null, }); - const releaseSource = await repoMutex.acquireShared(sourceRepository.id, `mirror_source:${scheduleId}`); - const releaseMirror = await repoMutex.acquireShared(mirror.repository.id, `mirror:${scheduleId}`); + const releaseLocks = await repoMutex.acquireMany([ + { repositoryId: sourceRepository.id, type: "shared", operation: `mirror_source:${scheduleId}` }, + { repositoryId: mirror.repository.id, type: "exclusive", operation: `mirror:${scheduleId}` }, + ]); try { await restic.copy(sourceRepository.config, mirror.repository.config, { tag: schedule.shortId, organizationId }); cache.delByPrefix(cacheKeys.repository.all(mirror.repository.id)); } finally { - releaseSource(); - releaseMirror(); + releaseLocks(); } if (retentionPolicy) { diff --git a/tsconfig.json b/tsconfig.json index a4506dd3..8bde7300 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -1,7 +1,7 @@ { "include": ["app/**/*"], "compilerOptions": { - "lib": ["DOM", "DOM.Iterable", "ES2022"], + "lib": ["DOM", "DOM.Iterable", "ES2023"], "types": ["bun", "node", "vite/client"], "target": "ES2022", "module": "ES2022",