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

fix(repo-mutex): catch errors and remove listener

Browse source
This commit is contained in:
Nicolas Meienberger 2026-01-22 22:06:39 +01:00
parent 51d2ffad17
commit 0e5438897c
3 changed files with 77 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
app/client/modules/repositories/tabs/info.tsx
View file

@ -302,6 +302,7 @@ export const RepositoryInfoTabContent = ({ repository }: Props) => {
<AlertDialogCancel>Cancel</AlertDialogCancel> <AlertDialogCancel>Cancel</AlertDialogCancel>
<AlertDialogAction <AlertDialogAction
onClick={handleConfirmDelete} onClick={handleConfirmDelete}
disabled={deleteRepo.isPending}
className="bg-destructive text-destructive-foreground hover:bg-destructive/90" className="bg-destructive text-destructive-foreground hover:bg-destructive/90"
> >
<Trash2 className="h-4 w-4 mr-2" /> <Trash2 className="h-4 w-4 mr-2" />

28
app/server/core/__tests__/repository-mutex.test.ts
View file

@ -94,4 +94,32 @@ describe("RepositoryMutex", () => {
expect(releaseEx3).toBeDefined(); expect(releaseEx3).toBeDefined();
releaseEx3(); releaseEx3();
}); });
test("should remove signal listener and not release unacquired lock on abort", async () => {
const repoId = "cleanup-test";
const controller = new AbortController();
const signal = controller.signal;
let removed = false;
const originalRemove = signal.removeEventListener.bind(signal);
signal.removeEventListener = (type: string, listener: any, options?: any) => {
if (type === "abort") removed = true;
return originalRemove(type, listener, options);
};
// Hold the lock so the next one has to wait
const release = await repoMutex.acquireExclusive(repoId, "holder");
const abortedAcquisition = repoMutex.acquireShared(repoId, "waiter", signal);
// Trigger abort while it's waiting
controller.abort();
expect(abortedAcquisition).rejects.toThrow();
expect(removed).toBe(true);
expect(repoMutex.isLocked(repoId)).toBe(true);
release();
expect(repoMutex.isLocked(repoId)).toBe(false);
});
}); });

86
app/server/core/repository-mutex.ts
View file

@ -70,31 +70,36 @@ class RepositoryMutex {
); );
let onAbort: () => void = () => {}; let onAbort: () => void = () => {};
const lockId = await new Promise<string>((resolve, reject) => { let lockId: string | undefined;
const waiter = { type: "shared" as const, operation, resolve }; try {
state.waitQueue.push(waiter); lockId = await new Promise<string>((resolve, reject) => {
const waiter = { type: "shared" as const, operation, resolve };
state.waitQueue.push(waiter);
if (signal) { if (signal) {
onAbort = () => { onAbort = () => {
const index = state.waitQueue.indexOf(waiter); const index = state.waitQueue.indexOf(waiter);
if (index !== -1) { if (index !== -1) {
state.waitQueue.splice(index, 1); state.waitQueue.splice(index, 1);
this.cleanupStateIfEmpty(repositoryId); this.cleanupStateIfEmpty(repositoryId);
reject(signal.reason || new Error("Operation aborted")); reject(signal.reason || new Error("Operation aborted"));
} }
}; };
signal.addEventListener("abort", onAbort); signal.addEventListener("abort", onAbort);
} }
}); });
} finally {
signal?.removeEventListener("abort", onAbort); signal?.removeEventListener("abort", onAbort);
}
if (signal?.aborted) { if (signal?.aborted) {
this.releaseShared(repositoryId, lockId); if (lockId) {
this.releaseShared(repositoryId, lockId);
}
throw signal.reason || new Error("Operation aborted"); throw signal.reason || new Error("Operation aborted");
} }
return () => this.releaseShared(repositoryId, lockId); return () => this.releaseShared(repositoryId, lockId!);
} }
async acquireExclusive(repositoryId: string, operation: string, signal?: AbortSignal): Promise<() => void> { async acquireExclusive(repositoryId: string, operation: string, signal?: AbortSignal): Promise<() => void> {
@ -119,32 +124,37 @@ class RepositoryMutex {
); );
let onAbort: () => void = () => {}; let onAbort: () => void = () => {};
const lockId = await new Promise<string>((resolve, reject) => { let lockId: string | undefined;
const waiter = { type: "exclusive" as const, operation, resolve }; try {
state.waitQueue.push(waiter); lockId = await new Promise<string>((resolve, reject) => {
const waiter = { type: "exclusive" as const, operation, resolve };
state.waitQueue.push(waiter);
if (signal) { if (signal) {
onAbort = () => { onAbort = () => {
const index = state.waitQueue.indexOf(waiter); const index = state.waitQueue.indexOf(waiter);
if (index !== -1) { if (index !== -1) {
state.waitQueue.splice(index, 1); state.waitQueue.splice(index, 1);
this.cleanupStateIfEmpty(repositoryId); this.cleanupStateIfEmpty(repositoryId);
reject(signal.reason || new Error("Operation aborted")); reject(signal.reason || new Error("Operation aborted"));
} }
}; };
signal.addEventListener("abort", onAbort); signal.addEventListener("abort", onAbort);
} }
}); });
} finally {
signal?.removeEventListener("abort", onAbort); signal?.removeEventListener("abort", onAbort);
}
if (signal?.aborted) { if (signal?.aborted) {
this.releaseExclusive(repositoryId, lockId); if (lockId) {
this.releaseExclusive(repositoryId, lockId);
}
throw signal.reason || new Error("Operation aborted"); throw signal.reason || new Error("Operation aborted");
} }
logger.debug(`[Mutex] Acquired exclusive lock for repo ${repositoryId}: ${operation} (${lockId})`); logger.debug(`[Mutex] Acquired exclusive lock for repo ${repositoryId}: ${operation} (${lockId})`);
return () => this.releaseExclusive(repositoryId, lockId); return () => this.releaseExclusive(repositoryId, lockId!);
} }
private releaseShared(repositoryId: string, lockId: string): void { private releaseShared(repositoryId: string, lockId: string): void {