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

refactor: get shared lock after semaphore take

Browse source
This commit is contained in:
Nicolas Meienberger 2026-06-05 18:03:30 +02:00
parent 5520d6ca17
commit 991ee5d05c
No known key found for this signature in database
3 changed files with 80 additions and 52 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
app/client/components/file-browsers/__tests__/snapshot-tree-browser.test.tsx
View file

@ -224,29 +224,6 @@ describe("SnapshotTreeBrowser", () => {
});
});
test("prefetches using the query path when display and query roots differ", async () => {
const requests = mockListSnapshotFiles();
renderSnapshotTreeBrowser();
const row = await screen.findByRole("button", { name: "project" });
const initialRequestCount = requests.length;
await userEvent.hover(row);
await waitFor(() => {
expect(requests.length).toBe(initialRequestCount + 1);
});
expect(requests.at(-1)).toEqual({
shortId: "repo-1",
snapshotId: "snap-1",
path: "/mnt/project",
offset: "0",
limit: "500",
});
});
test("shows the query root contents when display and query roots differ", async () => {
mockListSnapshotFiles();

46
app/server/modules/repositories/__tests__/repositories.service.test.ts
View file

@ -218,7 +218,11 @@ describe("repositoriesService.listSnapshotFiles", () => {
let active = 0;
let maxActive = 0;
let releaseAll = false;
let exclusiveAcquired = false;
let releaseExclusive: (() => void) | undefined;
let exclusivePromise: Promise<() => void> | undefined;
const releaseWaiters: Array<() => void> = [];
const exclusiveController = new AbortController();
const releaseWaitingCommands = () => {
const waiters = releaseWaiters.splice(0);
@ -227,6 +231,25 @@ describe("repositoriesService.listSnapshotFiles", () => {
}
};
const resolveWithin = async <T>(promise: Promise<T>, timeoutMs: number) => {
return await new Promise<T>((resolve, reject) => {
const timeout = setTimeout(() => {
reject(new Error(`Expected promise to resolve within ${timeoutMs}ms`));
}, timeoutMs);
promise.then(
(value) => {
clearTimeout(timeout);
resolve(value);
},
(error) => {
clearTimeout(timeout);
reject(error);
},
);
});
};
const lsSpy = vi.spyOn(restic, "ls").mockImplementation((_config, snapshotId, _path, options) =>
Effect.promise(async () => {
active++;
@ -277,8 +300,23 @@ describe("repositoriesService.listSnapshotFiles", () => {
});
expect(maxActive).toBe(2);
exclusivePromise = repoMutex
.acquireExclusive(repository.id, "delete", exclusiveController.signal)
.then((release) => {
exclusiveAcquired = true;
releaseExclusive = release;
return release;
});
releaseWaitingCommands();
releaseExclusive = await resolveWithin(exclusivePromise, 2000);
expect(exclusiveAcquired).toBe(true);
expect(active).toBe(0);
releaseExclusive();
releaseExclusive = undefined;
await waitForExpect(() => {
expect(releaseWaiters).toHaveLength(2);
});
@ -287,9 +325,17 @@ describe("repositoriesService.listSnapshotFiles", () => {
releaseWaitingCommands();
await Promise.all(calls);
} finally {
if (releaseExclusive) {
releaseExclusive();
} else {
exclusiveController.abort();
}
releaseAll = true;
releaseWaitingCommands();
await Promise.allSettled(calls);
if (exclusivePromise) {
await Promise.allSettled([exclusivePromise]);
}
}
expect(lsSpy).toHaveBeenCalledTimes(4);

63
app/server/modules/repositories/repositories.service.ts
View file

@ -103,7 +103,7 @@ const updateActiveRestoreTask = (restoreId: string, eventName: string, update: (
const getLsLimiter = (repositoryId: string) => {
let limiter = lsLimiters.get(repositoryId);
if (!limiter) {
limiter = Effect.unsafeMakeSemaphore(2);
limiter = Effect.runSync(Effect.makeSemaphore(2));
lsLimiters.set(repositoryId, limiter);
}
return limiter;
@ -479,38 +479,43 @@ const listSnapshotFiles = async (
};
}
const releaseLock = await repoMutex.acquireShared(repository.id, `ls:${snapshotId}`);
const limiter = getLsLimiter(repository.id);
await runEffectPromise(limiter.take(1));
try {
const result = await runEffectPromise(
restic
.ls(repository.config, snapshotId, path, { organizationId, offset, limit })
.pipe(getLsLimiter(repository.id).withPermits(1)),
);
const releaseLock = await repoMutex.acquireShared(repository.id, `ls:${snapshotId}`);
try {
const result = await runEffectPromise(
restic.ls(repository.config, snapshotId, path, { organizationId, offset, limit }),
);
if (!result.snapshot) {
throw new NotFoundError("Snapshot not found or empty");
if (!result.snapshot) {
throw new NotFoundError("Snapshot not found or empty");
}
const response = {
snapshot: {
id: result.snapshot.id,
short_id: result.snapshot.short_id,
time: result.snapshot.time,
hostname: result.snapshot.hostname,
paths: result.snapshot.paths,
},
files: result.nodes,
offset: result.pagination.offset,
limit: result.pagination.limit,
total: result.pagination.total,
hasMore: result.pagination.hasMore,
};
cache.set(cacheKey, result);
return response;
} finally {
releaseLock();
}
const response = {
snapshot: {
id: result.snapshot.id,
short_id: result.snapshot.short_id,
time: result.snapshot.time,
hostname: result.snapshot.hostname,
paths: result.snapshot.paths,
},
files: result.nodes,
offset: result.pagination.offset,
limit: result.pagination.limit,
total: result.pagination.total,
hasMore: result.pagination.hasMore,
};
cache.set(cacheKey, result);
return response;
} finally {
releaseLock();
await runEffectPromise(limiter.release(1));
}
};