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

refactor(mutext): run with signal to preserve abort reason

Browse source
This commit is contained in:
Nicolas Meienberger 2026-06-06 13:25:08 +02:00
parent 853a08e2d9
commit 0a36d77cca
No known key found for this signature in database
2 changed files with 117 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -134,6 +134,37 @@ describe("RepositoryMutex", () => {
expect(repoMutex.isLocked(repoId)).toBe(false);
});
test("should reject aborted queued acquisitions only after waiter cleanup finishes", async () => {
const repoId = "abort-waits-for-cleanup";
const releaseHolder = await repoMutex.acquireExclusive(repoId, "holder");
const controller = new AbortController();
const waitingAcquisition = repoMutex.acquireShared(repoId, "waiter", controller.signal);
try {
await new Promise((resolve) => setTimeout(resolve, 50));
const waitersBeforeAbort = await db.query.repositoryLockWaitersTable.findMany({
where: { repositoryId: { eq: repoId } },
});
expect(waitersBeforeAbort.map((waiter) => waiter.operation)).toEqual(["waiter"]);
controller.abort(new Error("stop"));
await expect(waitingAcquisition).rejects.toThrow("stop");
const waitersAfterAbort = await db.query.repositoryLockWaitersTable.findMany({
where: { repositoryId: { eq: repoId } },
});
expect(waitersAfterAbort).toEqual([]);
expect(repoMutex.isLocked(repoId)).toBe(true);
} finally {
releaseHolder();
await db.delete(repositoryLockWaitersTable).where(eq(repositoryLockWaitersTable.repositoryId, repoId));
await db.delete(repositoryLocksTable).where(eq(repositoryLocksTable.repositoryId, repoId));
}
});
test("should allow concurrent shared locks", async () => {
const repoId = "concurrent-shared";
const release1 = await repoMutex.acquireShared(repoId, "op1");

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

@ -34,6 +34,18 @@ const LOCK_HEARTBEAT_MS = 5_000;
const LOCK_POLL_MS = 250;
const LOCK_POLL_CLEANUP_MS = 5_000;
const REPOSITORY_MUTEX_INSTANCE = Symbol.for("zerobyte.repositoryMutex.instance");
function getRepositoryMutex() {
const globalObject = globalThis as typeof globalThis & Record<symbol, RepositoryMutex | undefined>;
const mutex = globalObject[REPOSITORY_MUTEX_INSTANCE];
if (mutex) return mutex;
const newMutex = new RepositoryMutex();
globalObject[REPOSITORY_MUTEX_INSTANCE] = newMutex;
return newMutex;
}
class RepositoryMutex {
private ownerId = `owner_${Bun.randomUUIDv7()}`;
private nextPollCleanupAt = 0;
@ -42,10 +54,8 @@ class RepositoryMutex {
return `lock_${Bun.randomUUIDv7()}`;
}
private throwIfAborted(signal?: AbortSignal) {
if (signal?.aborted) {
throw signal.reason || new Error("Operation aborted");
}
private abortReason(signal: AbortSignal): Error {
return signal.reason || new Error("Operation aborted");
}
private cleanupExpired(tx: RepositoryMutexTransaction, now: number) {
@ -154,9 +164,8 @@ class RepositoryMutex {
}
private createWaiter(request: LockRequest, waiterId: string) {
const now = Date.now();
return Effect.sync(() => {
const now = Date.now();
db.transaction((tx) => {
this.cleanupExpired(tx, now);
tx.insert(repositoryLockWaitersTable)
@ -349,19 +358,23 @@ class RepositoryMutex {
private release(lock: Pick<AcquiredLock, "id">) {
return Effect.gen(this, function* () {
const releasedLock = db.transaction((tx) => {
const row = tx.query.repositoryLocksTable
.findFirst({ where: { AND: [{ id: { eq: lock.id } }, { ownerId: { eq: this.ownerId } }] } })
.sync();
const releasedLock = yield* Effect.sync(() =>
db.transaction((tx) => {
const row = tx.query.repositoryLocksTable
.findFirst({ where: { AND: [{ id: { eq: lock.id } }, { ownerId: { eq: this.ownerId } }] } })
.sync();
if (!row) return null;
if (!row) return null;
tx.delete(repositoryLocksTable)
.where(and(eq(repositoryLocksTable.id, lock.id), eq(repositoryLocksTable.ownerId, this.ownerId)))
.run();
tx.delete(repositoryLocksTable)
.where(
and(eq(repositoryLocksTable.id, lock.id), eq(repositoryLocksTable.ownerId, this.ownerId)),
)
.run();
return row;
});
return row;
}),
);
if (!releasedLock) return;
@ -390,14 +403,14 @@ class RepositoryMutex {
const values = { heartbeatAt: now, expiresAt: now + LOCK_LEASE_MS };
if (target === "lock") {
yield* Effect.sync(() => {
yield* Effect.try(() => {
db.update(repositoryLocksTable)
.set(values)
.where(and(eq(repositoryLocksTable.id, lockId), eq(repositoryLocksTable.ownerId, this.ownerId)))
.run();
});
} else {
yield* Effect.sync(() => {
yield* Effect.try(() => {
db.update(repositoryLockWaitersTable)
.set(values)
.where(
@ -448,9 +461,9 @@ class RepositoryMutex {
return releaseLock;
}
logger.debug(`[Mutex] Waiting for exclusive lock on repo ${repositoryId}: ${operation}`);
yield* logger.effect.debug(`[Mutex] Waiting for exclusive lock on repo ${repositoryId}: ${operation}`);
const queuedReleaseLock = yield* this.waitForQueuedLock(request);
logger.debug(`[Mutex] Acquired exclusive lock for repo ${repositoryId}: ${operation}`);
yield* logger.effect.debug(`[Mutex] Acquired exclusive lock for repo ${repositoryId}: ${operation}`);
return queuedReleaseLock;
});
}
@ -483,20 +496,66 @@ class RepositoryMutex {
});
}
private runWithSignal<A, E>(effect: Effect.Effect<A, E>, signal?: AbortSignal) {
if (!signal) return Effect.runPromise(effect);
if (signal.aborted) {
return Promise.reject(this.abortReason(signal));
}
return new Promise<A>((resolve, reject) => {
const fiber = Effect.runFork(effect);
let settled = false;
let aborting = false;
const complete = (callback: () => void) => {
if (settled) return;
settled = true;
signal.removeEventListener("abort", onAbort);
callback();
};
const onAbort = () => {
aborting = true;
Effect.runPromise(Fiber.interrupt(fiber)).then(
(exit) =>
complete(() => {
if (Exit.isSuccess(exit)) {
resolve(exit.value);
return;
}
reject(this.abortReason(signal));
}),
(error) => complete(() => reject(error)),
);
};
signal.addEventListener("abort", onAbort, { once: true });
Effect.runPromise(Fiber.join(fiber)).then(
(value) => complete(() => resolve(value)),
(error) => {
if (!aborting) {
complete(() => reject(error));
}
},
);
});
}
async acquireShared(repositoryId: string, operation: string, signal?: AbortSignal) {
this.throwIfAborted(signal);
return await Effect.runPromise(this.acquireSharedEffect(repositoryId, operation), { signal });
return await this.runWithSignal(this.acquireSharedEffect(repositoryId, operation), signal);
}
async acquireExclusive(repositoryId: string, operation: string, signal?: AbortSignal) {
this.throwIfAborted(signal);
return await Effect.runPromise(this.acquireExclusiveEffect(repositoryId, operation), { signal });
return await this.runWithSignal(this.acquireExclusiveEffect(repositoryId, operation), signal);
}
async acquireMany(requests: LockRequest[], signal?: AbortSignal) {
this.throwIfAborted(signal);
return await Effect.runPromise(this.acquireManyEffect(requests), { signal });
return await this.runWithSignal(this.acquireManyEffect(requests), signal);
}
}
export const repoMutex = new RepositoryMutex();
export const repoMutex = getRepositoryMutex();