fix: serialize mirror copy operations per destination repository (#747)

* fix: serialize mirror copy operations per destination repository

* refactor(copy-mirror): implement acquireMany to avoid deadlock cycles

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";
 

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<string>();
+		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) {

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<void>((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);
+	});
 });

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) {

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",