refactor(copy-mirror): implement acquireMany to avoid deadlock cycles
This commit is contained in:
parent
955f77bb0b
commit
397afa28b0
4 changed files with 130 additions and 5 deletions
|
|
@ -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";
|
||||
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -92,15 +92,16 @@ async function copyToSingleMirror(
|
|||
lastCopyError: null,
|
||||
});
|
||||
|
||||
const releaseSource = await repoMutex.acquireShared(sourceRepository.id, `mirror_source:${scheduleId}`);
|
||||
const releaseMirror = await repoMutex.acquireExclusive(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) {
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
Loading…
Reference in a new issue