fix: cache set race

refactor: controller doesn't shudown abort down to the agent
chore: fix liniting issue
2026-06-05 21:38:27 +02:00 · 2026-06-05 20:34:24 +02:00 · 2026-06-05 20:14:56 +02:00 · 2026-06-05 20:12:13 +02:00 · 2026-06-05 19:29:00 +02:00 · 2026-06-05 18:13:01 +02:00
30 changed files with 1451 additions and 748 deletions
--- a/app/server.ts
+++ b/app/server.ts
@ -1,5 +1,3 @@
-import { logger } from "@zerobyte/core/node";
-import { shutdown } from "./server/modules/lifecycle/shutdown";
 import { runCLI } from "./server/cli";
 import { createStartHandler, defaultStreamHandler, defineHandlerCallback } from "@tanstack/react-start/server";
 import { createServerEntry } from "@tanstack/react-start/server-entry";
@ -18,25 +16,3 @@ const fetch = createStartHandler(customHandler);
 export default createServerEntry({
 	fetch,
 });
-
-process.on("SIGTERM", async () => {
-	logger.info("SIGTERM received, starting graceful shutdown...");
-	try {
-		await shutdown();
-	} catch (err) {
-		logger.error("Error during shutdown", err);
-	} finally {
-		process.exit(0);
-	}
-});
-
-process.on("SIGINT", async () => {
-	logger.info("SIGINT received, starting graceful shutdown...");
-	try {
-		await shutdown();
-	} catch (err) {
-		logger.error("Error during shutdown", err);
-	} finally {
-		process.exit(0);
-	}
-});
--- a/app/server/core/tests/repository-mutex.test.ts
+++ b/app/server/core/tests/repository-mutex.test.ts
@ -2,8 +2,20 @@ import { describe, expect, test, vi } from "vitest";
 import { eq } from "drizzle-orm";
 import { db } from "~/server/db/db";
 import { repositoryLocksTable, repositoryLockWaitersTable } from "~/server/db/schema";
+import {
+	createDeferred,
+	holdExclusiveLock as holdExclusive,
+	holdManyLocks as holdMany,
+	holdSharedLock as holdShared,
+} from "~/test/helpers/repository-mutex";
 import { repoMutex } from "../repository-mutex";

+const loadRepositoryMutexModule = async () => {
+	const moduleUrl = new URL("../repository-mutex.ts", import.meta.url);
+	moduleUrl.searchParams.set("test", crypto.randomUUID());
+	return import(moduleUrl.href) as Promise<typeof import("../repository-mutex")>;
+};
+
 const acquireWithin = <T>(promise: Promise<T>, ms = 500) =>
 	Promise.race([
 		promise,
@ -17,15 +29,15 @@ describe("RepositoryMutex", () => {
 		const repoId = "test-repo";
 		const results: string[] = [];

-		const releaseShared1 = await repoMutex.acquireShared(repoId, "backup-1");
+		const releaseShared1 = await holdShared(repoId, "backup-1");
 		results.push("acquired-shared-1");

-		const exclusivePromise = repoMutex.acquireExclusive(repoId, "unlock").then((release) => {
+		const exclusivePromise = holdExclusive(repoId, "unlock").then((release) => {
 			results.push("acquired-exclusive");
 			return release;
 		});

-		const shared2Promise = repoMutex.acquireShared(repoId, "backup-2").then((release) => {
+		const shared2Promise = holdShared(repoId, "backup-2").then((release) => {
 			results.push("acquired-shared-2");
 			return release;
 		});
@ -34,33 +46,33 @@ describe("RepositoryMutex", () => {

 		expect(results).toEqual(["acquired-shared-1"]);

-		releaseShared1();
+		await releaseShared1();

 		const releaseExclusive = await exclusivePromise;
 		expect(results).toEqual(["acquired-shared-1", "acquired-exclusive"]);

-		releaseExclusive();
+		await releaseExclusive();

 		const releaseShared2 = await shared2Promise;
 		expect(results).toEqual(["acquired-shared-1", "acquired-exclusive", "acquired-shared-2"]);

-		releaseShared2();
+		await releaseShared2();
 	});

 	test("should remove aborted acquisitions from the wait queue", async () => {
 		const repoId = "abort-test";
 		const results: string[] = [];

-		const releaseShared1 = await repoMutex.acquireShared(repoId, "backup-1");
+		const releaseShared1 = await holdShared(repoId, "backup-1");
 		results.push("acquired-shared-1");

 		const controller = new AbortController();
-		const exclusivePromise = repoMutex.acquireExclusive(repoId, "unlock", controller.signal).catch((err) => {
+		const exclusivePromise = holdExclusive(repoId, "unlock", controller.signal).catch((err) => {
 			results.push("aborted-exclusive");
 			throw err;
 		});

-		const shared2Promise = repoMutex.acquireShared(repoId, "backup-2").then((release) => {
+		const shared2Promise = holdShared(repoId, "backup-2").then((release) => {
 			results.push("acquired-shared-2");
 			return release;
 		});
@ -73,25 +85,25 @@ describe("RepositoryMutex", () => {
 		await expect(exclusivePromise).rejects.toThrow();
 		expect(results).toEqual(["acquired-shared-1", "aborted-exclusive"]);

-		releaseShared1();
+		await releaseShared1();

 		// After exclusive is aborted, shared-2 should be next
 		const releaseShared2 = await shared2Promise;
 		expect(results).toEqual(["acquired-shared-1", "aborted-exclusive", "acquired-shared-2"]);

-		releaseShared2();
+		await releaseShared2();
 	});

 	test("should handle multiple aborts correctly", async () => {
 		const repoId = "multi-abort";
-		const releaseShared1 = await repoMutex.acquireShared(repoId, "backup-1");
+		const releaseShared1 = await holdShared(repoId, "backup-1");

 		const controller1 = new AbortController();
 		const controller2 = new AbortController();

-		const p1 = repoMutex.acquireExclusive(repoId, "ex-1", controller1.signal);
-		const p2 = repoMutex.acquireExclusive(repoId, "ex-2", controller2.signal);
-		const p3 = repoMutex.acquireExclusive(repoId, "ex-3");
+		const p1 = holdExclusive(repoId, "ex-1", controller1.signal);
+		const p2 = holdExclusive(repoId, "ex-2", controller2.signal);
+		const p3 = holdExclusive(repoId, "ex-3");

 		controller2.abort();
 		await expect(p2).rejects.toThrow();
@ -99,11 +111,11 @@ describe("RepositoryMutex", () => {
 		controller1.abort();
 		await expect(p1).rejects.toThrow();

-		releaseShared1();
+		await releaseShared1();

 		const releaseEx3 = await p3;
 		expect(releaseEx3).toBeDefined();
-		releaseEx3();
+		await releaseEx3();
 	});

 	test("should remove signal listener and not release unacquired lock on abort", async () => {
@ -119,9 +131,9 @@ describe("RepositoryMutex", () => {
 		};

 		// Hold the lock so the next one has to wait
-		const release = await repoMutex.acquireExclusive(repoId, "holder");
+		const release = await holdExclusive(repoId, "holder");

-		const abortedAcquisition = repoMutex.acquireShared(repoId, "waiter", signal);
+		const abortedAcquisition = holdShared(repoId, "waiter", signal);

 		// Trigger abort while it's waiting
 		controller.abort();
@ -130,52 +142,21 @@ describe("RepositoryMutex", () => {
 		expect(removed).toBe(true);
 		expect(repoMutex.isLocked(repoId)).toBe(true);

-		release();
+		await release();
 		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");
-		const release2 = await repoMutex.acquireShared(repoId, "op2");
-		const release3 = await repoMutex.acquireShared(repoId, "op3");
+		const release1 = await holdShared(repoId, "op1");
+		const release2 = await holdShared(repoId, "op2");
+		const release3 = await holdShared(repoId, "op3");

 		expect(repoMutex.isLocked(repoId)).toBe(true);

-		release1();
-		release2();
-		release3();
+		await release1();
+		await release2();
+		await release3();

 		expect(repoMutex.isLocked(repoId)).toBe(false);
 	});
@ -184,10 +165,10 @@ describe("RepositoryMutex", () => {
 		const repoId = "shared-blocks-exclusive";
 		let exclusiveAcquired = false;

-		const releaseShared1 = await repoMutex.acquireShared(repoId, "s1");
-		const releaseShared2 = await repoMutex.acquireShared(repoId, "s2");
+		const releaseShared1 = await holdShared(repoId, "s1");
+		const releaseShared2 = await holdShared(repoId, "s2");

-		const exclusivePromise = repoMutex.acquireExclusive(repoId, "e1").then((release) => {
+		const exclusivePromise = holdExclusive(repoId, "e1").then((release) => {
 			exclusiveAcquired = true;
 			return release;
 		});
@ -195,29 +176,29 @@ describe("RepositoryMutex", () => {
 		await new Promise((resolve) => setTimeout(resolve, 10));
 		expect(exclusiveAcquired).toBe(false);

-		releaseShared1();
+		await releaseShared1();
 		await new Promise((resolve) => setTimeout(resolve, 10));
 		expect(exclusiveAcquired).toBe(false); // still waiting for s2

-		releaseShared2();
+		await releaseShared2();
 		const releaseExclusive = await exclusivePromise;
 		expect(exclusiveAcquired).toBe(true);

-		releaseExclusive();
+		await releaseExclusive();
 	});

 	test("should block all locks while exclusive lock is held", async () => {
 		const repoId = "exclusive-blocks-all";
 		const results: string[] = [];

-		const releaseExclusive = await repoMutex.acquireExclusive(repoId, "e1");
+		const releaseExclusive = await holdExclusive(repoId, "e1");
 		results.push("e1-acquired");

-		const s1Promise = repoMutex.acquireShared(repoId, "s1").then((release) => {
+		const s1Promise = holdShared(repoId, "s1").then((release) => {
 			results.push("s1-acquired");
 			return release;
 		});
-		const e2Promise = repoMutex.acquireExclusive(repoId, "e2").then((release) => {
+		const e2Promise = holdExclusive(repoId, "e2").then((release) => {
 			results.push("e2-acquired");
 			return release;
 		});
@ -225,34 +206,34 @@ describe("RepositoryMutex", () => {
 		await new Promise((resolve) => setTimeout(resolve, 10));
 		expect(results).toEqual(["e1-acquired"]);

-		releaseExclusive();
+		await releaseExclusive();

 		const releaseS1 = await s1Promise;
 		expect(results).toEqual(["e1-acquired", "s1-acquired"]);

-		releaseS1();
+		await releaseS1();

 		const releaseE2 = await e2Promise;
 		expect(results).toEqual(["e1-acquired", "s1-acquired", "e2-acquired"]);

-		releaseE2();
+		await releaseE2();
 	});

 	test("should grant all waiting shared locks at once when exclusive lock is released", async () => {
 		const repoId = "batch-shared";
 		const results: string[] = [];

-		const releaseExclusive = await repoMutex.acquireExclusive(repoId, "e1");
+		const releaseExclusive = await holdExclusive(repoId, "e1");

-		const s1Promise = repoMutex.acquireShared(repoId, "s1").then((release) => {
+		const s1Promise = holdShared(repoId, "s1").then((release) => {
 			results.push("s1");
 			return release;
 		});
-		const s2Promise = repoMutex.acquireShared(repoId, "s2").then((release) => {
+		const s2Promise = holdShared(repoId, "s2").then((release) => {
 			results.push("s2");
 			return release;
 		});
-		const s3Promise = repoMutex.acquireShared(repoId, "s3").then((release) => {
+		const s3Promise = holdShared(repoId, "s3").then((release) => {
 			results.push("s3");
 			return release;
 		});
@ -260,7 +241,7 @@ describe("RepositoryMutex", () => {
 		await new Promise((resolve) => setTimeout(resolve, 10));
 		expect(results).toEqual([]);

-		releaseExclusive();
+		await releaseExclusive();

 		const [releaseS1, releaseS2, releaseS3] = await Promise.all([s1Promise, s2Promise, s3Promise]);

@ -269,29 +250,27 @@ describe("RepositoryMutex", () => {
 		expect(results).toContain("s2");
 		expect(results).toContain("s3");

-		releaseS1();
-		releaseS2();
-		releaseS3();
+		await releaseS1();
+		await releaseS2();
+		await releaseS3();
 	});

 	test("should wait to acquire all many-lock requests before locking any repository", async () => {
-		const releaseSharedB = await repoMutex.acquireShared("repo-b", "holder-b");
+		const releaseSharedB = await holdShared("repo-b", "holder-b");
 		let manyAcquired = false;
 		let exclusiveAAcquired = false;

-		const manyPromise = repoMutex
-			.acquireMany([
+		const manyPromise = holdMany([
 			{ repositoryId: "repo-b", type: "exclusive", operation: "many-b" },
 			{ repositoryId: "repo-a", type: "shared", operation: "many-a" },
-			])
-			.then((release) => {
+		]).then((release) => {
 			manyAcquired = true;
 			return release;
 		});

 		await new Promise((resolve) => setTimeout(resolve, 10));

-		const exclusiveAPromise = repoMutex.acquireExclusive("repo-a", "exclusive-a").then((release) => {
+		const exclusiveAPromise = holdExclusive("repo-a", "exclusive-a").then((release) => {
 			exclusiveAAcquired = true;
 			return release;
 		});
@ -300,23 +279,23 @@ describe("RepositoryMutex", () => {
 		expect(exclusiveAAcquired).toBe(true);
 		expect(manyAcquired).toBe(false);

-		releaseSharedB();
+		await releaseSharedB();
 		await new Promise((resolve) => setTimeout(resolve, 10));
 		expect(manyAcquired).toBe(false);

 		const releaseExclusiveA = await exclusiveAPromise;
-		releaseExclusiveA();
+		await releaseExclusiveA();

 		const releaseMany = await manyPromise;
 		expect(manyAcquired).toBe(true);
-		releaseMany();
+		await releaseMany();
 	});

-	test("should abort acquireMany without leaving partial locks behind", async () => {
-		const releaseExclusiveB = await repoMutex.acquireExclusive("repo-b", "holder-b");
+	test("should abort runMany without leaving partial locks behind", async () => {
+		const releaseExclusiveB = await holdExclusive("repo-b", "holder-b");
 		const controller = new AbortController();

-		const manyPromise = repoMutex.acquireMany(
+		const manyPromise = holdMany(
 			[
 				{ repositoryId: "repo-b", type: "exclusive", operation: "many-b" },
 				{ repositoryId: "repo-a", type: "shared", operation: "many-a" },
@ -327,7 +306,7 @@ describe("RepositoryMutex", () => {
 		await new Promise((resolve) => setTimeout(resolve, 10));

 		let exclusiveAAcquired = false;
-		const exclusiveAPromise = repoMutex.acquireExclusive("repo-a", "exclusive-a").then((release) => {
+		const exclusiveAPromise = holdExclusive("repo-a", "exclusive-a").then((release) => {
 			exclusiveAAcquired = true;
 			return release;
 		});
@ -341,17 +320,17 @@ describe("RepositoryMutex", () => {
 		const releaseExclusiveA = await exclusiveAPromise;
 		expect(exclusiveAAcquired).toBe(true);

-		releaseExclusiveA();
-		releaseExclusiveB();
+		await releaseExclusiveA();
+		await releaseExclusiveB();
 	});

-	test("should abort acquireMany if the signal aborts after waiting resolves", async () => {
+	test("should abort runMany if the signal aborts after waiting resolves", async () => {
 		const repoA = "many-abort-after-wake-a";
 		const repoB = "many-abort-after-wake-b";
-		const releaseExclusiveB = await repoMutex.acquireExclusive(repoB, "holder-b");
+		const releaseExclusiveB = await holdExclusive(repoB, "holder-b");
 		const controller = new AbortController();

-		const manyPromise = repoMutex.acquireMany(
+		const manyPromise = holdMany(
 			[
 				{ repositoryId: repoB, type: "exclusive", operation: "many-b" },
 				{ repositoryId: repoA, type: "shared", operation: "many-a" },
@ -361,7 +340,7 @@ describe("RepositoryMutex", () => {

 		await new Promise((resolve) => setTimeout(resolve, 10));

-		releaseExclusiveB();
+		await releaseExclusiveB();
 		controller.abort(new Error("stop after wake"));

 		await expect(manyPromise).rejects.toThrow("stop after wake");
@ -372,17 +351,17 @@ describe("RepositoryMutex", () => {
 	test("should not leave a promoted shared lock behind if that waiter aborts before observing acquisition", async () => {
 		vi.useFakeTimers();
 		const repoId = "shared-abort-after-promotion";
-		const releaseExclusive = await repoMutex.acquireExclusive(repoId, "holder");
-		const firstSharedPromise = repoMutex.acquireShared(repoId, "shared-1");
+		const releaseExclusive = await holdExclusive(repoId, "holder");
+		const firstSharedPromise = holdShared(repoId, "shared-1");

 		try {
 			await vi.advanceTimersByTimeAsync(100);

 			const controller = new AbortController();
-			const secondSharedPromise = repoMutex.acquireShared(repoId, "shared-2", controller.signal);
+			const secondSharedPromise = holdShared(repoId, "shared-2", controller.signal);

 			await vi.advanceTimersByTimeAsync(140);
-			releaseExclusive();
+			await releaseExclusive();

 			// The first shared waiter wakes first and promotes both shared waiters.
 			await vi.advanceTimersByTimeAsync(10);
@ -391,7 +370,7 @@ describe("RepositoryMutex", () => {
 			controller.abort(new Error("abort after promotion"));
 			await expect(secondSharedPromise).rejects.toThrow("abort after promotion");

-			releaseShared1();
+			await releaseShared1();

 			const remainingLocks = await db.query.repositoryLocksTable.findMany({
 				where: { repositoryId: { eq: repoId } },
@ -409,14 +388,14 @@ describe("RepositoryMutex", () => {
 	test("should safely handle multiple calls to the release function", async () => {
 		const repoId = "idempotent-release";

-		const releaseShared = await repoMutex.acquireShared(repoId, "s1");
-		releaseShared();
-		releaseShared(); // Should not throw or cause issues
-		releaseShared();
+		const releaseShared = await holdShared(repoId, "s1");
+		await releaseShared();
+		await releaseShared(); // Should not throw or cause issues
+		await releaseShared();

-		const releaseExclusive = await repoMutex.acquireExclusive(repoId, "e1");
-		releaseExclusive();
-		releaseExclusive(); // Should not throw
+		const releaseExclusive = await holdExclusive(repoId, "e1");
+		await releaseExclusive();
+		await releaseExclusive(); // Should not throw

 		expect(repoMutex.isLocked(repoId)).toBe(false);
 	});
@ -426,8 +405,8 @@ describe("RepositoryMutex", () => {
 		const controller = new AbortController();
 		controller.abort(new Error("pre-aborted"));

-		await expect(repoMutex.acquireShared(repoId, "s1", controller.signal)).rejects.toThrow("pre-aborted");
-		await expect(repoMutex.acquireExclusive(repoId, "e1", controller.signal)).rejects.toThrow("pre-aborted");
+		await expect(holdShared(repoId, "s1", controller.signal)).rejects.toThrow("pre-aborted");
+		await expect(holdExclusive(repoId, "e1", controller.signal)).rejects.toThrow("pre-aborted");

 		expect(repoMutex.isLocked(repoId)).toBe(false);
 	});
@ -437,22 +416,22 @@ describe("RepositoryMutex", () => {

 		expect(repoMutex.isLocked(repoId)).toBe(false);

-		const releaseShared1 = await repoMutex.acquireShared(repoId, "s1");
+		const releaseShared1 = await holdShared(repoId, "s1");
 		expect(repoMutex.isLocked(repoId)).toBe(true);

-		const releaseShared2 = await repoMutex.acquireShared(repoId, "s2");
+		const releaseShared2 = await holdShared(repoId, "s2");
 		expect(repoMutex.isLocked(repoId)).toBe(true);

-		releaseShared1();
+		await releaseShared1();
 		expect(repoMutex.isLocked(repoId)).toBe(true); // still locked by s2

-		releaseShared2();
+		await releaseShared2();
 		expect(repoMutex.isLocked(repoId)).toBe(false); // all shared released

-		const releaseExclusive = await repoMutex.acquireExclusive(repoId, "e1");
+		const releaseExclusive = await holdExclusive(repoId, "e1");
 		expect(repoMutex.isLocked(repoId)).toBe(true);

-		releaseExclusive();
+		await releaseExclusive();
 		expect(repoMutex.isLocked(repoId)).toBe(false);
 	});

@ -472,7 +451,7 @@ describe("RepositoryMutex", () => {
 			heartbeatAt: now - 10_000,
 		});

-		const releaseShared = await acquireWithin(repoMutex.acquireShared(repoId, "backup"));
+		const releaseShared = await acquireWithin(holdShared(repoId, "backup"));

 		try {
 			const expiredLock = await db.query.repositoryLocksTable.findFirst({
@ -482,7 +461,7 @@ describe("RepositoryMutex", () => {
 			expect(expiredLock).toBeUndefined();
 			expect(repoMutex.isLocked(repoId)).toBe(true);
 		} finally {
-			releaseShared();
+			await releaseShared();
 			await db.delete(repositoryLocksTable).where(eq(repositoryLocksTable.repositoryId, repoId));
 		}
 	});
@ -503,7 +482,7 @@ describe("RepositoryMutex", () => {
 			heartbeatAt: now - 10_000,
 		});

-		const releaseShared = await acquireWithin(repoMutex.acquireShared(repoId, "backup"));
+		const releaseShared = await acquireWithin(holdShared(repoId, "backup"));

 		try {
 			const expiredWaiter = await db.query.repositoryLockWaitersTable.findFirst({
@ -513,7 +492,7 @@ describe("RepositoryMutex", () => {
 			expect(expiredWaiter).toBeUndefined();
 			expect(repoMutex.isLocked(repoId)).toBe(true);
 		} finally {
-			releaseShared();
+			await releaseShared();
 			await db.delete(repositoryLockWaitersTable).where(eq(repositoryLockWaitersTable.repositoryId, repoId));
 			await db.delete(repositoryLocksTable).where(eq(repositoryLocksTable.repositoryId, repoId));
 		}
@ -522,7 +501,7 @@ describe("RepositoryMutex", () => {
 	test("should release only the caller lock row", async () => {
 		const repoId = "release-own-row";
 		const foreignLockId = "foreign-release-own-row";
-		const releaseShared = await repoMutex.acquireShared(repoId, "owned-shared");
+		const releaseShared = await holdShared(repoId, "owned-shared");
 		const now = Date.now();

 		try {
@ -537,7 +516,7 @@ describe("RepositoryMutex", () => {
 				heartbeatAt: now,
 			});

-			releaseShared();
+			await releaseShared();

 			const remainingLocks = await db.query.repositoryLocksTable.findMany({
 				where: { repositoryId: { eq: repoId } },
@ -547,8 +526,129 @@ describe("RepositoryMutex", () => {
 			expect(remainingLocks.map((lock) => lock.operation)).toEqual(["foreign-shared"]);
 			expect(repoMutex.isLocked(repoId)).toBe(true);
 		} finally {
-			releaseShared();
+			await releaseShared();
 			await db.delete(repositoryLocksTable).where(eq(repositoryLocksTable.repositoryId, repoId));
 		}
 	});
+
+	test("shutdown should abort an active managed operation and release its lock", async () => {
+		const repoId = "shutdown-active";
+		const started = createDeferred<AbortSignal>();
+		const operation = repoMutex.runShared(repoId, "active", async ({ signal }) => {
+			started.resolve(signal);
+			await new Promise<void>((resolve) => signal.addEventListener("abort", () => resolve(), { once: true }));
+		});
+
+		const signal = await started.promise;
+		expect(repoMutex.isLocked(repoId)).toBe(true);
+
+		await repoMutex.shutdown({ timeoutMs: 100 });
+		await operation;
+
+		expect(signal.aborted).toBe(true);
+		expect(repoMutex.isLocked(repoId)).toBe(false);
+	});
+
+	test("shutdown should abort queued managed operations before they start", async () => {
+		const repoId = "shutdown-queued";
+		const holderStarted = createDeferred<AbortSignal>();
+		const holder = repoMutex.runExclusive(repoId, "holder", async ({ signal }) => {
+			holderStarted.resolve(signal);
+			await new Promise<void>((resolve) => signal.addEventListener("abort", () => resolve(), { once: true }));
+		});
+		await holderStarted.promise;
+
+		let queuedStarted = false;
+		const queued = repoMutex.runShared(repoId, "queued", async () => {
+			queuedStarted = true;
+		});
+		const queuedResult = queued.then(
+			() => null,
+			(error: unknown) => error,
+		);
+
+		await new Promise((resolve) => setTimeout(resolve, 20));
+		await repoMutex.shutdown({ timeoutMs: 100 });
+		await holder;
+
+		const queuedError = await queuedResult;
+		expect(queuedError).toBeInstanceOf(Error);
+		expect(String((queuedError as Error).message)).toContain("Repository mutex is shutting down");
+		expect(queuedStarted).toBe(false);
+
+		const remainingWaiters = await db.query.repositoryLockWaitersTable.findMany({
+			where: { repositoryId: { eq: repoId } },
+		});
+		expect(remainingWaiters).toEqual([]);
+		expect(repoMutex.isLocked(repoId)).toBe(false);
+	});
+
+	test("should release the lease without invoking the operation when aborted after acquisition", async () => {
+		const repoId = "abort-after-acquire";
+		const abortController = new AbortController();
+		let operationStarted = false;
+		const abortReason = new Error("caller aborted");
+
+		const operation = repoMutex.runShared(
+			repoId,
+			"abort-before-callback",
+			async () => {
+				operationStarted = true;
+			},
+			abortController.signal,
+		);
+		abortController.abort(abortReason);
+
+		await expect(operation).rejects.toThrow("caller aborted");
+		expect(operationStarted).toBe(false);
+		expect(repoMutex.isLocked(repoId)).toBe(false);
+	});
+
+	test("shutdown should release owned lock rows after the timeout when an operation ignores abort", async () => {
+		const repoId = "shutdown-timeout-release";
+		const started = createDeferred<AbortSignal>();
+		const finish = createDeferred();
+		const operation = repoMutex.runExclusive(repoId, "stuck", async ({ signal }) => {
+			started.resolve(signal);
+			await finish.promise;
+		});
+
+		const signal = await started.promise;
+		expect(repoMutex.isLocked(repoId)).toBe(true);
+
+		await repoMutex.shutdown({ timeoutMs: 10 });
+
+		expect(signal.aborted).toBe(true);
+		expect(repoMutex.isLocked(repoId)).toBe(false);
+
+		finish.resolve();
+		await operation;
+		expect(repoMutex.isLocked(repoId)).toBe(false);
+	});
+
+	test("should reuse the same global repository mutex when the module is evaluated more than once", async () => {
+		const { repoMutex: reloadedMutex } = await loadRepositoryMutexModule();
+		const repoId = "shutdown-other-instance";
+		const started = createDeferred<AbortSignal>();
+		const operation = reloadedMutex.runShared(repoId, "active", async ({ signal }) => {
+			started.resolve(signal);
+			await new Promise<void>((resolve) => signal.addEventListener("abort", () => resolve(), { once: true }));
+		});
+
+		const signal = await started.promise;
+		expect(reloadedMutex).toBe(repoMutex);
+		expect(reloadedMutex.isLocked(repoId)).toBe(true);
+
+		await repoMutex.shutdown({ timeoutMs: 100 });
+		await operation;
+
+		expect(signal.aborted).toBe(true);
+		expect(reloadedMutex.isLocked(repoId)).toBe(false);
+	});
+
+	test("shutdown should be idempotent", async () => {
+		await Promise.all([repoMutex.shutdown({ timeoutMs: 10 }), repoMutex.shutdown({ timeoutMs: 10 })]);
+
+		await expect(repoMutex.runShared("post-shutdown", "op", () => "ok")).resolves.toBe("ok");
+	});
 });
--- a/app/server/core/repository-mutex.ts
+++ b/app/server/core/repository-mutex.ts
@ -7,7 +7,6 @@ import {
 	type RepositoryLock,
 	type RepositoryLockWaiter,
 } from "../db/schema";
-import { Effect, Exit, Fiber, Schedule, Scope } from "effect";

 type LockType = "shared" | "exclusive";

@ -25,6 +24,24 @@ interface AcquiredLock {
 	acquiredAt: number;
 }

+interface RepositoryLease {
+	signal: AbortSignal;
+	release: () => void;
+}
+
+interface ActiveRepositoryOperation {
+	abortController: AbortController;
+	cleanup: () => void;
+	completion: Promise<unknown>;
+	release: (() => void) | null;
+}
+
+type RepositoryOperationContext = {
+	signal: AbortSignal;
+};
+
+type RepositoryOperation<T> = (context: RepositoryOperationContext) => T | Promise<T>;
+
 type RepositoryMutexTransaction = Parameters<Parameters<typeof db.transaction>[0]>[0];
 type HeartbeatTarget = "lock" | "waiter";
 type QueueAttempt = { status: "acquired"; lock: AcquiredLock } | { status: "waiting" } | { status: "missing" };
@ -33,8 +50,9 @@ const LOCK_LEASE_MS = 30_000;
 const LOCK_HEARTBEAT_MS = 5_000;
 const LOCK_POLL_MS = 250;
 const LOCK_POLL_CLEANUP_MS = 5_000;
-
+const SHUTDOWN_WAIT_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];
@ -46,18 +64,183 @@ function getRepositoryMutex() {
 	return newMutex;
 }

-class RepositoryMutex {
+export class RepositoryMutex {
 	private ownerId = `owner_${Bun.randomUUIDv7()}`;
+	private heartbeatTimers = new Map<string, ReturnType<typeof setInterval>>();
+	private activeOperations = new Map<string, ActiveRepositoryOperation>();
 	private nextPollCleanupAt = 0;
+	private shuttingDown = false;
+	private shutdownPromise: Promise<void> | null = null;

 	private generateLockId(): string {
 		return `lock_${Bun.randomUUIDv7()}`;
 	}

-	private abortReason(signal: AbortSignal): Error {
+	private abortReason(signal: AbortSignal) {
 		return signal.reason || new Error("Operation aborted");
 	}

+	private throwIfAborted(signal?: AbortSignal) {
+		if (signal?.aborted) {
+			throw this.abortReason(signal);
+		}
+	}
+
+	private releaseIfAborted(releaseLock: () => void, signal?: AbortSignal) {
+		if (!signal?.aborted) return;
+		releaseLock();
+		throw this.abortReason(signal);
+	}
+
+	private throwIfShuttingDown() {
+		if (this.shuttingDown) {
+			throw new Error("Repository mutex is shutting down");
+		}
+	}
+
+	private createOperationController(signal?: AbortSignal) {
+		const abortController = new AbortController();
+		const abortOperation = () => {
+			if (!abortController.signal.aborted) {
+				abortController.abort(signal ? this.abortReason(signal) : new Error("Operation aborted"));
+			}
+		};
+
+		if (signal?.aborted) {
+			abortOperation();
+		} else {
+			signal?.addEventListener("abort", abortOperation, { once: true });
+		}
+
+		return {
+			abortController,
+			cleanup: () => signal?.removeEventListener("abort", abortOperation),
+		};
+	}
+
+	private createLease(releaseLock: () => void, signal?: AbortSignal): RepositoryLease {
+		const abortController = new AbortController();
+		let released = false;
+		const abortLease = () => {
+			if (!abortController.signal.aborted) {
+				abortController.abort(signal ? this.abortReason(signal) : new Error("Operation aborted"));
+			}
+		};
+
+		if (signal?.aborted) {
+			abortLease();
+		} else {
+			signal?.addEventListener("abort", abortLease, { once: true });
+		}
+
+		return {
+			signal: abortController.signal,
+			release: () => {
+				if (released) return;
+
+				released = true;
+				signal?.removeEventListener("abort", abortLease);
+				releaseLock();
+			},
+		};
+	}
+
+	private async runWithLease<T>(lease: RepositoryLease, operation: RepositoryOperation<T>) {
+		try {
+			this.throwIfAborted(lease.signal);
+			return await operation({ signal: lease.signal });
+		} finally {
+			lease.release();
+		}
+	}
+
+	private runManagedOperation<T>(
+		openLease: (signal: AbortSignal) => Promise<RepositoryLease>,
+		operation: RepositoryOperation<T>,
+		signal?: AbortSignal,
+	) {
+		this.throwIfShuttingDown();
+
+		const operationId = `operation_${Bun.randomUUIDv7()}`;
+		const { abortController, cleanup } = this.createOperationController(signal);
+		const activeOperation: ActiveRepositoryOperation = {
+			abortController,
+			cleanup,
+			completion: Promise.resolve(),
+			release: null,
+		};
+
+		const completion = (async () => {
+			const lease = await openLease(abortController.signal);
+			activeOperation.release = lease.release;
+			return await this.runWithLease(lease, operation);
+		})();
+
+		activeOperation.completion = completion.finally(() => {
+			cleanup();
+			this.activeOperations.delete(operationId);
+		});
+		this.activeOperations.set(operationId, activeOperation);
+
+		return activeOperation.completion as Promise<T>;
+	}
+
+	private waitForShutdownSettled(operations: ActiveRepositoryOperation[], timeoutMs: number) {
+		if (operations.length === 0) {
+			return Promise.resolve();
+		}
+
+		return new Promise<void>((resolve) => {
+			const timeout = setTimeout(resolve, timeoutMs);
+			void Promise.allSettled(operations.map((operation) => operation.completion)).then(() => {
+				clearTimeout(timeout);
+				resolve();
+			});
+		});
+	}
+
+	private releaseOwnedRows() {
+		db.transaction((tx) => {
+			tx.delete(repositoryLockWaitersTable).where(eq(repositoryLockWaitersTable.ownerId, this.ownerId)).run();
+			tx.delete(repositoryLocksTable).where(eq(repositoryLocksTable.ownerId, this.ownerId)).run();
+		});
+	}
+
+	private stopAllHeartbeats() {
+		for (const timer of this.heartbeatTimers.values()) {
+			clearInterval(timer);
+		}
+		this.heartbeatTimers.clear();
+	}
+
+	private waitForPoll(signal?: AbortSignal) {
+		this.throwIfAborted(signal);
+
+		return new Promise<void>((resolve, reject) => {
+			let settled = false;
+			const timeout = setTimeout(() => settle(resolve), LOCK_POLL_MS);
+
+			const onAbort = () => {
+				settle(() => reject(this.abortReason(signal!)));
+			};
+
+			const cleanup = () => {
+				clearTimeout(timeout);
+				signal?.removeEventListener("abort", onAbort);
+			};
+
+			const settle = (callback: () => void) => {
+				if (settled) return;
+
+				settled = true;
+				cleanup();
+				callback();
+			};
+
+			signal?.addEventListener("abort", onAbort, { once: true });
+		});
+	}
+
 	private cleanupExpired(tx: RepositoryMutexTransaction, now: number) {
 		tx.delete(repositoryLocksTable).where(lte(repositoryLocksTable.expiresAt, now)).run();
 		tx.delete(repositoryLockWaitersTable).where(lte(repositoryLockWaitersTable.expiresAt, now)).run();
@ -153,19 +336,20 @@ class RepositoryMutex {
 		});
 	}

-	private tryAcquireImmediately(request: LockRequest) {
-		return Effect.gen(this, function* () {
+	private tryAcquireImmediately(request: LockRequest, signal?: AbortSignal) {
 		const locks = this.tryAcquireManyRows([request]);
 		if (!locks || locks.length === 0) return null;

 		const [lock] = locks;
-			return yield* this.createRelease(lock);
-		});
+		const releaseLock = this.createRelease(lock);
+		this.releaseIfAborted(releaseLock, signal);
+
+		return releaseLock;
 	}

 	private createWaiter(request: LockRequest, waiterId: string) {
-		return Effect.sync(() => {
 		const now = Date.now();
+
 		db.transaction((tx) => {
 			this.cleanupExpired(tx, now);
 			tx.insert(repositoryLockWaitersTable)
@ -181,21 +365,14 @@ class RepositoryMutex {
 				})
 				.run();
 		});
-		});
 	}

 	private deleteWaiter(waiterId: string) {
-		return Effect.sync(() =>
-			db
-				.delete(repositoryLockWaitersTable)
+		db.delete(repositoryLockWaitersTable)
 			.where(
-					and(
-						eq(repositoryLockWaitersTable.id, waiterId),
-						eq(repositoryLockWaitersTable.ownerId, this.ownerId),
-					),
+				and(eq(repositoryLockWaitersTable.id, waiterId), eq(repositoryLockWaitersTable.ownerId, this.ownerId)),
 			)
-				.run(),
-		);
+			.run();
 	}

 	private deleteWaiterRow(tx: RepositoryMutexTransaction, waiterId: string): void {
@ -271,51 +448,156 @@ class RepositoryMutex {
 		});
 	}

-	private waitForQueuedLock(request: LockRequest) {
-		const waiterId = this.generateLockId();
+	private async waitForQueuedLock(request: LockRequest, signal?: AbortSignal) {
+		this.throwIfAborted(signal);

-		const attempt = Effect.sync(() => this.tryPromoteWaiter(waiterId)).pipe(
-			Effect.flatMap((attempt) => {
+		const waiterId = this.generateLockId();
+		this.createWaiter(request, waiterId);
+		this.startHeartbeat("waiter", waiterId);
+
+		try {
+			while (true) {
+				this.throwIfAborted(signal);
+
+				const attempt = this.tryPromoteWaiter(waiterId);
 				if (attempt.status === "acquired") {
-					return Effect.succeed(attempt.lock);
+					this.stopHeartbeat(waiterId);
+					const releaseLock = this.createRelease(attempt.lock);
+					this.releaseIfAborted(releaseLock, signal);
+
+					return releaseLock;
 				}

 				if (attempt.status === "missing") {
-					return Effect.gen(this, function* () {
-						yield* this.createWaiter(request, waiterId);
-						yield* this.startHeartbeat("waiter", waiterId);
-
-						return yield* Effect.fail("retry");
-					});
+					this.createWaiter(request, waiterId);
+					this.startHeartbeat("waiter", waiterId);
 				}

-				return Effect.fail("retry");
-			}),
-		);
-
-		const cleanupAbandonedWaiter = Effect.gen(this, function* () {
-			yield* this.deleteWaiter(waiterId);
-			yield* this.release({ id: waiterId });
-		});
-
-		return Effect.scoped(
-			Effect.gen(this, function* () {
-				const lock = yield* attempt.pipe(
-					Effect.retry(Schedule.spaced(LOCK_POLL_MS)),
-					Effect.onExit((exit) => {
-						if (Exit.isSuccess(exit)) {
-							return Effect.void;
+				await this.waitForPoll(signal);
+			}
+		} catch (error) {
+			this.stopHeartbeat(waiterId);
+			this.deleteWaiter(waiterId);
+			this.release({ id: waiterId });
+			throw error;
+		}
 	}

-						return cleanupAbandonedWaiter;
-					}),
-				);
+	private async openSingleLease(request: LockRequest, signal?: AbortSignal) {
+		this.throwIfAborted(signal);

-				return yield* this.createRelease(lock);
-			}),
+		const releaseLock = this.tryAcquireImmediately(request, signal);
+		if (releaseLock) {
+			return this.createLease(releaseLock, signal);
+		}
+
+		logger.debug(`[Mutex] Waiting for ${request.type} lock on repo ${request.repositoryId}: ${request.operation}`);
+		return this.createLease(await this.waitForQueuedLock(request, signal), signal);
+	}
+
+	private async openManyLease(requests: LockRequest[], signal?: AbortSignal) {
+		this.throwIfAborted(signal);
+
+		if (requests.length === 0) {
+			return this.createLease(() => {}, signal);
+		}
+
+		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));
+		while (true) {
+			const locks = this.tryAcquireManyRows(sortedRequests);
+			if (locks) {
+				const releaseLocks = this.createReleaseMany(locks);
+				this.releaseIfAborted(releaseLocks, signal);
+
+				return this.createLease(releaseLocks, signal);
+			}
+
+			await this.waitForPoll(signal);
+		}
+	}
+
+	async runShared<T>(
+		repositoryId: string,
+		operation: string,
+		callback: RepositoryOperation<T>,
+		signal?: AbortSignal,
+	) {
+		return this.runManagedOperation(
+			(operationSignal) => this.openSingleLease({ repositoryId, type: "shared", operation }, operationSignal),
+			callback,
+			signal,
 		);
 	}

+	async runExclusive<T>(
+		repositoryId: string,
+		operation: string,
+		callback: RepositoryOperation<T>,
+		signal?: AbortSignal,
+	) {
+		return this.runManagedOperation(
+			async (operationSignal) => {
+				const lease = await this.openSingleLease(
+					{ repositoryId, type: "exclusive", operation },
+					operationSignal,
+				);
+				logger.debug(`[Mutex] Acquired exclusive lock for repo ${repositoryId}: ${operation}`);
+				return lease;
+			},
+			callback,
+			signal,
+		);
+	}
+
+	async runMany<T>(requests: LockRequest[], callback: RepositoryOperation<T>, signal?: AbortSignal) {
+		return this.runManagedOperation(
+			(operationSignal) => this.openManyLease(requests, operationSignal),
+			callback,
+			signal,
+		);
+	}
+
+	async shutdown(options: { timeoutMs?: number } = {}) {
+		if (this.shutdownPromise) {
+			return await this.shutdownPromise;
+		}
+
+		this.shuttingDown = true;
+		this.shutdownPromise = (async () => {
+			const activeOperations = [...this.activeOperations.values()];
+			const reason = new Error("Repository mutex is shutting down");
+
+			for (const operation of activeOperations) {
+				if (!operation.abortController.signal.aborted) {
+					operation.abortController.abort(reason);
+				}
+			}
+
+			await this.waitForShutdownSettled(activeOperations, options.timeoutMs ?? SHUTDOWN_WAIT_MS);
+
+			for (const operation of activeOperations) {
+				operation.release?.();
+				operation.cleanup();
+			}
+
+			this.releaseOwnedRows();
+			this.stopAllHeartbeats();
+		})().finally(() => {
+			this.shuttingDown = false;
+			this.shutdownPromise = null;
+		});
+
+		return await this.shutdownPromise;
+	}
+
 	isLocked(repositoryId: string) {
 		const now = Date.now();

@ -326,8 +608,7 @@ class RepositoryMutex {
 	}

 	private createReleaseMany(locks: AcquiredLock[]) {
-		return Effect.gen(this, function* () {
-			const releases = yield* Effect.all(locks.map((lock) => this.createRelease(lock)));
+		const releases = locks.map((lock) => this.createRelease(lock));
 		let released = false;

 		return () => {
@ -338,28 +619,23 @@ class RepositoryMutex {
 				release();
 			}
 		};
-		});
 	}

 	private createRelease(lock: AcquiredLock) {
-		return Effect.gen(this, function* () {
-			const heartbeatFiber = yield* this.startHeartbeat("lock", lock.id);
+		this.startHeartbeat("lock", lock.id);
 		let released = false;

 		return () => {
 			if (released) return;

 			released = true;
-				Effect.runFork(Fiber.interrupt(heartbeatFiber));
-				Effect.runSync(this.release(lock));
+			this.stopHeartbeat(lock.id);
+			this.release(lock);
 		};
-		});
 	}

 	private release(lock: Pick<AcquiredLock, "id">) {
-		return Effect.gen(this, function* () {
-			const releasedLock = yield* Effect.sync(() =>
-				db.transaction((tx) => {
+		const releasedLock = db.transaction((tx) => {
 			const row = tx.query.repositoryLocksTable
 				.findFirst({ where: { AND: [{ id: { eq: lock.id } }, { ownerId: { eq: this.ownerId } }] } })
 				.sync();
@ -367,50 +643,34 @@ class RepositoryMutex {
 			if (!row) return null;

 			tx.delete(repositoryLocksTable)
-						.where(
-							and(eq(repositoryLocksTable.id, lock.id), eq(repositoryLocksTable.ownerId, this.ownerId)),
-						)
+				.where(and(eq(repositoryLocksTable.id, lock.id), eq(repositoryLocksTable.ownerId, this.ownerId)))
 				.run();

 			return row;
-				}),
-			);
+		});

 		if (!releasedLock) return;

 		const duration = Date.now() - releasedLock.acquiredAt;
-
-			yield* logger.effect.debug(
+		logger.debug(
 			`[Mutex] Released ${releasedLock.type} lock for repo ${releasedLock.repositoryId}: ${releasedLock.operation} (held for ${duration}ms)`,
 		);
-		});
 	}

-	private startHeartbeat(
-		target: "waiter",
-		lockId: string,
-	): Effect.Effect<Fiber.RuntimeFiber<void, never>, never, Scope.Scope>;
-	private startHeartbeat(
-		target: "lock",
-		lockId: string,
-	): Effect.Effect<Fiber.RuntimeFiber<void, never>, never, never>;
-	private startHeartbeat(
-		target: HeartbeatTarget,
-		lockId: string,
-	): Effect.Effect<Fiber.RuntimeFiber<unknown, never>, never, Scope.Scope> {
-		const heartbeat = Effect.gen(this, function* () {
+	private startHeartbeat(target: HeartbeatTarget, lockId: string) {
+		this.stopHeartbeat(lockId);
+
+		const heartbeat = () => {
 			const now = Date.now();
 			const values = { heartbeatAt: now, expiresAt: now + LOCK_LEASE_MS };

+			try {
 				if (target === "lock") {
-				yield* Effect.try(() => {
 					db.update(repositoryLocksTable)
 						.set(values)
 						.where(and(eq(repositoryLocksTable.id, lockId), eq(repositoryLocksTable.ownerId, this.ownerId)))
 						.run();
-				});
 				} else {
-				yield* Effect.try(() => {
 					db.update(repositoryLockWaitersTable)
 						.set(values)
 						.where(
@ -420,141 +680,28 @@ class RepositoryMutex {
 							),
 						)
 						.run();
-				});
 				}
-		}).pipe(
-			Effect.catchAll((error) =>
-				logger.effect.warn(`[Mutex] Failed to heartbeat ${target} ${lockId}: ${String(error)}`),
-			),
-		);
-
-		const repeat = heartbeat.pipe(Effect.repeat(Schedule.spaced(LOCK_HEARTBEAT_MS)));
-
-		if (target === "waiter") {
-			// For waiters, we can stop heartbeating when the releaser is dropped, so we use a scoped fiber
-			return repeat.pipe(Effect.forkScoped);
+			} catch (error) {
+				logger.warn(`[Mutex] Failed to heartbeat ${target} ${lockId}: ${String(error)}`);
 			}
-
-		// For locks, the heartbeat must outlive the acquire scope.
-		// It is interrupted manually by the returned release function.
-		// TODO: max lifetime for lock heartbeats to prevent leaks if the releaser is never called?
-		return repeat.pipe(Effect.forkDaemon);
-	}
-
-	private acquireSharedEffect(repositoryId: string, operation: string) {
-		return Effect.gen(this, function* () {
-			const request: LockRequest = { repositoryId, type: "shared", operation };
-			const releaseLock = yield* this.tryAcquireImmediately(request);
-			if (releaseLock) return releaseLock;
-
-			yield* logger.effect.debug(`[Mutex] Waiting for shared lock on repo ${repositoryId}: ${operation}`);
-			return yield* this.waitForQueuedLock(request);
-		});
-	}
-
-	private acquireExclusiveEffect(repositoryId: string, operation: string) {
-		return Effect.gen(this, function* () {
-			const request: LockRequest = { repositoryId, type: "exclusive", operation };
-			const releaseLock = yield* this.tryAcquireImmediately(request);
-			if (releaseLock) {
-				yield* logger.effect.debug(`[Mutex] Acquired exclusive lock for repo ${repositoryId}: ${operation}`);
-				return releaseLock;
-			}
-
-			yield* logger.effect.debug(`[Mutex] Waiting for exclusive lock on repo ${repositoryId}: ${operation}`);
-			const queuedReleaseLock = yield* this.waitForQueuedLock(request);
-			yield* logger.effect.debug(`[Mutex] Acquired exclusive lock for repo ${repositoryId}: ${operation}`);
-			return queuedReleaseLock;
-		});
-	}
-
-	private acquireManyEffect(requests: LockRequest[]) {
-		return Effect.gen(this, function* () {
-			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 locks = yield* Effect.sync(() => this.tryAcquireManyRows(sortedRequests)).pipe(
-				Effect.flatMap((locks) => {
-					if (locks) return Effect.succeed(locks);
-					return Effect.fail("retry");
-				}),
-				Effect.retry(Schedule.spaced(LOCK_POLL_MS)),
-			);
-
-			return yield* this.createReleaseMany(locks);
-		});
-	}
-
-	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);
+		const timer = setInterval(heartbeat, LOCK_HEARTBEAT_MS);
+		if (timer && "unref" in timer) {
+			timer.unref();
+		}
+
+		this.heartbeatTimers.set(lockId, timer);
+	}
+
+	private stopHeartbeat(lockId: string) {
+		const timer = this.heartbeatTimers.get(lockId);
+		if (!timer) {
 			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) {
-		return await this.runWithSignal(this.acquireSharedEffect(repositoryId, operation), signal);
-	}
-
-	async acquireExclusive(repositoryId: string, operation: string, signal?: AbortSignal) {
-		return await this.runWithSignal(this.acquireExclusiveEffect(repositoryId, operation), signal);
-	}
-
-	async acquireMany(requests: LockRequest[], signal?: AbortSignal) {
-		return await this.runWithSignal(this.acquireManyEffect(requests), signal);
+		clearInterval(timer);
+		this.heartbeatTimers.delete(lockId);
 	}
 }

--- a/app/server/modules/agents/tests/agents-manager.events.test.ts
+++ b/app/server/modules/agents/tests/agents-manager.events.test.ts
@ -270,34 +270,6 @@ test("runBackup rejects before sending when the abort signal is already aborted"
 	await stopAgentController();
 });

-test("runBackup requests cancellation when the abort signal fires while sending", async () => {
-	resetAgentRuntime();
-	const abortController = new AbortController();
-	controllerMock.sendBackup.mockImplementation(() =>
-		Effect.sync(() => {
-			abortController.abort();
-			return true;
-		}),
-	);
-	controllerMock.cancelBackup.mockImplementation(() => Effect.succeed(false));
-	const { agentManager, startAgentController, stopAgentController } = await import("../agents-manager");
-
-	await startAgentController();
-	const result = await agentManager.runBackup("local", {
-		scheduleId: 42,
-		payload: backupPayload,
-		signal: abortController.signal,
-		onProgress: vi.fn(),
-	});
-
-	expect(result).toEqual({ status: "cancelled" });
-	expect(controllerMock.cancelBackup).toHaveBeenCalledWith("local", {
-		jobId: "job-1",
-		scheduleId: "schedule-1",
-	});
-	await stopAgentController();
-});
-
 test("restore events are delivered to the running restore callbacks", async () => {
 	resetAgentRuntime();
 	controllerMock.sendRestore.mockImplementation(() => Effect.succeed(true));
--- a/app/server/modules/agents/agents-manager.ts
+++ b/app/server/modules/agents/agents-manager.ts
@ -431,7 +431,6 @@ export const agentManager = {
 			});
 			getActiveBackupScheduleIdsByJobId().set(request.payload.jobId, request.scheduleId);
 		});
-
 		try {
 			if (!(await Effect.runPromise(runtime.sendBackup(agentId, request.payload)))) {
 				clearActiveBackupRun(request.scheduleId);
@ -441,11 +440,7 @@ export const agentManager = {
 				} satisfies BackupExecutionResult;
 			}

-			if (request.signal.aborted) {
-				await requestBackupCancellation(agentId, request.scheduleId);
-			}
-
-			return completion;
+			return await completion;
 		} catch (error) {
 			clearActiveBackupRun(request.scheduleId);
 			throw error;
@ -494,7 +489,6 @@ export const agentManager = {
 				cancellationRequested: false,
 			});
 		});
-
 		try {
 			if (!(await Effect.runPromise(runtime.sendRestore(agentId, request.payload)))) {
 				clearActiveRestoreRun(request.payload.restoreId);
@ -504,11 +498,10 @@ export const agentManager = {
 				};
 			}

-			if (request.signal.aborted) {
-				await requestRestoreCancellation(agentId, request.payload.restoreId);
-			}
-
-			return { status: "started", result: completion };
+			return {
+				status: "started",
+				result: completion,
+			};
 		} catch (error) {
 			clearActiveRestoreRun(request.payload.restoreId);
 			throw error;
--- a/app/server/modules/backups/tests/backups.service.execution.test.ts
+++ b/app/server/modules/backups/tests/backups.service.execution.test.ts
@ -16,7 +16,6 @@ import { NotFoundError } from "http-errors-enhanced";
 import { fromAny } from "@total-typescript/shoehorn";
 import { scheduleQueries } from "../backups.queries";
 import { repositoriesService } from "~/server/modules/repositories/repositories.service";
-import { repoMutex } from "~/server/core/repository-mutex";
 import { notificationsService } from "~/server/modules/notifications/notifications.service";
 import { agentManager } from "~/server/modules/agents/agents-manager";
 import { createAgentBackupMocks } from "~/test/helpers/agent-mock";
@ -26,6 +25,8 @@ import { db } from "~/server/db/db";
 import { config } from "~/server/core/config";
 import { Effect } from "effect";
 import { taskStore } from "~/server/modules/tasks/tasks.store";
+import { holdExclusiveLock } from "~/test/helpers/repository-mutex";
+import { repoMutex } from "~/server/core/repository-mutex";

 const setup = () => {
 	const resticBackupMock = vi.fn((_: SafeSpawnParams) =>
@ -842,7 +843,7 @@ describe("stop backup", () => {
 			repositoryId: repository.id,
 		});

-		const releaseLock = await repoMutex.acquireExclusive(repository.id, "test");
+		const releaseLock = await holdExclusiveLock(repository.id, "test");
 		const executePromise = backupsService.executeBackup(schedule.id);

 		try {
@ -857,7 +858,7 @@ describe("stop backup", () => {

 			await backupsService.stopBackup(schedule.id);
 		} finally {
-			releaseLock();
+			await releaseLock();
 		}

 		await executePromise;
@ -871,6 +872,42 @@ describe("stop backup", () => {
 		expect(resticBackupMock).not.toHaveBeenCalled();
 	});

+	test("should cancel a queued backup when repository mutex shutdown aborts it", async () => {
+		const { resticBackupMock } = setup();
+		const volume = await createTestVolume();
+		const repository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: repository.id,
+		});
+
+		const releaseLock = await holdExclusiveLock(repository.id, "test");
+		const executePromise = backupsService.executeBackup(schedule.id);
+
+		try {
+			await waitForExpect(async () => {
+				const task = await getBackupTaskForSchedule(schedule.id);
+				expect(task?.status).toBe("queued");
+			});
+
+			expect(resticBackupMock).not.toHaveBeenCalled();
+
+			await repoMutex.shutdown({ timeoutMs: 10 });
+		} finally {
+			await releaseLock();
+		}
+
+		await executePromise;
+
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		const task = await getBackupTaskForSchedule(schedule.id);
+		expect(updatedSchedule.lastBackupStatus).toBe("warning");
+		expect(updatedSchedule.lastBackupError).toBe("Repository mutex is shutting down");
+		expect(task?.status).toBe("cancelled");
+		expect(task?.error).toBe("Repository mutex is shutting down");
+		expect(resticBackupMock).not.toHaveBeenCalled();
+	});
+
 	test("should clear failureRetryCount when a scheduled retry is cancelled", async () => {
 		const { resticBackupMock } = setup();
 		const volume = await createTestVolume();
@ -1219,6 +1256,32 @@ describe("mirror operations", () => {
 		expect(updatedMirror?.lastCopyAt).not.toBeNull();
 	});

+	test("should clear mirror in-progress status when shutdown aborts copy", async () => {
+		const { resticCopyMock } = setup();
+		const volume = await createTestVolume();
+		const sourceRepository = await createTestRepository();
+		const mirrorRepository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: sourceRepository.id,
+		});
+
+		const mirror = await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id);
+		resticCopyMock.mockImplementationOnce(() =>
+			Effect.sync(() => {
+				throw new Error("Repository mutex is shutting down");
+			}),
+		);
+
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
+
+		const mirrors = await backupsService.getMirrors(schedule.id);
+		const updatedMirror = mirrors.find((m) => m.id === mirror.id);
+		expect(updatedMirror?.lastCopyStatus).toBeNull();
+		expect(updatedMirror?.lastCopyError).toBeNull();
+		expect(resticCopyMock).toHaveBeenCalledTimes(1);
+	});
+
 	test("should run forget on mirror after successful copy when retention policy exists", async () => {
 		// arrange
 		const { resticCopyMock, resticForgetMock } = setup();
--- a/app/server/modules/backups/backup-executor.ts
+++ b/app/server/modules/backups/backup-executor.ts
@ -17,6 +17,7 @@ const IGNORE_INODE_FLAG = "--ignore-inode";
 type BackupExecutionRequest = {
 	jobId: string;
 	scheduleId: number;
+	trackedAbortController: AbortController;
 	schedule: BackupSchedule;
 	volume: Volume;
 	repository: Repository;
@ -109,6 +110,19 @@ const executeBackupWithoutAgent = async (
 	);
 };

+const getTrackedExecution = (scheduleId: number, abortController: AbortController) => {
+	const trackedExecution = activeControllersByScheduleId.get(scheduleId);
+	if (!trackedExecution) {
+		throw new Error(`Backup execution for schedule ${scheduleId} was not tracked`);
+	}
+
+	if (trackedExecution.abortController !== abortController) {
+		throw new Error(`Backup execution for schedule ${scheduleId} was superseded`);
+	}
+
+	return trackedExecution;
+};
+
 export const backupExecutor = {
 	track: (scheduleId: number) => {
 		const abortController = new AbortController();
@ -121,10 +135,7 @@ export const backupExecutor = {
 		}
 	},
 	execute: async (request: BackupExecutionRequest) => {
-		const trackedExecution = activeControllersByScheduleId.get(request.scheduleId);
-		if (!trackedExecution || trackedExecution.abortController.signal !== request.signal) {
-			throw new Error(`Backup execution for schedule ${request.scheduleId} was not tracked`);
-		}
+		getTrackedExecution(request.scheduleId, request.trackedAbortController);

 		if (request.signal.aborted) {
 			throw request.signal.reason || new Error("Operation aborted");
@ -137,6 +148,7 @@ export const backupExecutor = {
 		}

 		const executionAgentId = getBackupExecutionAgentId(request.volume, request.repository);
+		const trackedExecution = getTrackedExecution(request.scheduleId, request.trackedAbortController);
 		trackedExecution.agentId = executionAgentId;

 		const executionResult = await agentManager.runBackup(executionAgentId, {
--- a/app/server/modules/backups/backups.queries.ts
+++ b/app/server/modules/backups/backups.queries.ts
@ -7,7 +7,7 @@ type MirrorStatusType = "in_progress" | "success" | "error";

 type MirrorStatusUpdate = {
 	lastCopyAt?: number | null;
-	lastCopyStatus?: MirrorStatusType;
+	lastCopyStatus?: MirrorStatusType | null;
 	lastCopyError?: string | null;
 };

@ -48,7 +48,9 @@ export const scheduleQueries = {
 		return db
 			.update(backupSchedulesTable)
 			.set({ ...status, updatedAt: Date.now() })
-			.where(and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId)));
+			.where(
+				and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId)),
+			);
 	},
 };

--- a/app/server/modules/backups/backups.service.ts
+++ b/app/server/modules/backups/backups.service.ts
@ -451,23 +451,21 @@ const executeBackup = async (scheduleId: number, manual = false) => {
 	let domainHandlerCompleted = false;

 	try {
-		const releaseLock = await repoMutex.acquireShared(
+		await repoMutex.runShared(
 			ctx.repository.id,
 			`backup:${ctx.volume.name}`,
-			abortController.signal,
-		);
-
-		try {
+			async ({ signal }) => {
 				taskStore.markRunning(task.id);

 				const executionResult = await backupExecutor.execute({
 					jobId: task.id,
 					scheduleId,
+					trackedAbortController: abortController,
 					schedule: ctx.schedule,
 					volume: ctx.volume,
 					repository: ctx.repository,
 					organizationId: ctx.organizationId,
-				signal: abortController.signal,
+					signal,
 					onProgress: (progress) => {
 						updateBackupProgress(ctx, progress);
 						try {
@ -512,12 +510,21 @@ const executeBackup = async (scheduleId: number, manual = false) => {
 						taskStore.cancel(task.id, executionResult.message ?? "Backup was stopped by the user");
 						return;
 				}
-		} finally {
-			releaseLock();
-		}
+			},
+			abortController.signal,
+		);
 	} catch (error) {
 		if (abortController.signal.aborted) {
-			taskStore.cancel(task.id, "Backup was stopped by the user");
+			const message = "Backup was stopped by the user";
+			await handleBackupCancellation(scheduleId, ctx.organizationId, message);
+			taskStore.cancel(task.id, message);
+			return;
+		}
+
+		if (toMessage(error) === "Repository mutex is shutting down") {
+			const message = toMessage(error);
+			await handleBackupCancellation(scheduleId, ctx.organizationId, message);
+			taskStore.cancel(task.id, message);
 			return;
 		}

--- a/app/server/modules/backups/helpers/backup-maintenance.ts
+++ b/app/server/modules/backups/helpers/backup-maintenance.ts
@ -9,6 +9,11 @@ import { runEffectPromise, toMessage } from "../../../utils/errors";
 import { getOrganizationId } from "~/server/core/request-context";
 import { mirrorQueries, repositoryQueries, scheduleQueries } from "../backups.queries";

+const isMirrorCopyCancellation = (error: unknown) => {
+	const message = toMessage(error);
+	return message === "Repository mutex is shutting down" || message === "Operation aborted";
+};
+
 export async function runForget(scheduleId: number, repositoryId?: string, organizationIdOverride?: string) {
 	const organizationId = organizationIdOverride ?? getOrganizationId();
 	const schedule = await scheduleQueries.findById(scheduleId, organizationId);
@ -20,6 +25,7 @@ export async function runForget(scheduleId: number, repositoryId?: string, organ
 	if (!schedule.retentionPolicy) {
 		throw new BadRequestError("No retention policy configured for this schedule");
 	}
+	const retentionPolicy = schedule.retentionPolicy;

 	const repository = await repositoryQueries.findById(repositoryId ?? schedule.repositoryId, organizationId);

@ -28,16 +34,12 @@ export async function runForget(scheduleId: number, repositoryId?: string, organ
 	}

 	logger.info(`running retention policy (forget) for schedule ${scheduleId}`);
-	const releaseLock = await repoMutex.acquireExclusive(repository.id, `forget:${scheduleId}`);
-
-	try {
+	await repoMutex.runExclusive(repository.id, `forget:${scheduleId}`, async ({ signal }) => {
 		await runEffectPromise(
-			restic.forget(repository.config, schedule.retentionPolicy, { tag: schedule.shortId, organizationId }),
+			restic.forget(repository.config, retentionPolicy, { tag: schedule.shortId, organizationId, signal }),
 		);
 		cache.delByPrefix(cacheKeys.repository.all(repository.id));
-	} finally {
-		releaseLock();
-	}
+	});

 	logger.info(`Retention policy applied successfully for schedule ${scheduleId}`);
 }
@ -114,23 +116,23 @@ export async function syncSnapshotsToMirror(
 			lastCopyError: null,
 		});

-		const releaseLocks = await repoMutex.acquireMany([
+		await repoMutex.runMany(
+			[
 				{ repositoryId: sourceRepository.id, type: "shared", operation: `mirror_sync_source:${scheduleId}` },
 				{ repositoryId: mirrorRepository.id, type: "exclusive", operation: `mirror_sync:${scheduleId}` },
-		]);
-
-		try {
+			],
+			async ({ signal }) => {
 				await runEffectPromise(
 					restic.copy(sourceRepository.config, mirrorRepository.config, {
 						tag: schedule.shortId,
 						organizationId,
 						snapshotIds,
+						signal,
 					}),
 				);
 				cache.delByPrefix(cacheKeys.repository.all(mirrorRepository.id));
-		} finally {
-			releaseLocks();
-		}
+			},
+		);

 		if (schedule.retentionPolicy) {
 			void runForget(scheduleId, mirrorRepository.id, organizationId).catch((error) => {
@ -157,6 +159,17 @@ export async function syncSnapshotsToMirror(
 		});
 	} catch (error) {
 		const errorMessage = toMessage(error);
+		if (isMirrorCopyCancellation(error)) {
+			logger.info(
+				`[Background] Mirror sync to repository ${mirrorRepository.name} was cancelled: ${errorMessage}`,
+			);
+			await mirrorQueries.updateStatus(scheduleId, mirrorRepositoryId, {
+				lastCopyStatus: null,
+				lastCopyError: null,
+			});
+			return;
+		}
+
 		logger.error(
 			`[Background] Failed to sync all snapshots to mirror repository ${mirrorRepository.name}: ${errorMessage}`,
 		);
@ -204,22 +217,22 @@ async function copyToSingleMirror(
 			lastCopyError: null,
 		});

-		const releaseLocks = await repoMutex.acquireMany([
+		await repoMutex.runMany(
+			[
 				{ repositoryId: sourceRepository.id, type: "shared", operation: `mirror_source:${scheduleId}` },
 				{ repositoryId: mirror.repository.id, type: "exclusive", operation: `mirror:${scheduleId}` },
-		]);
-
-		try {
+			],
+			async ({ signal }) => {
 				await runEffectPromise(
 					restic.copy(sourceRepository.config, mirror.repository.config, {
 						tag: schedule.shortId,
 						organizationId,
+						signal,
 					}),
 				);
 				cache.delByPrefix(cacheKeys.repository.all(mirror.repository.id));
-		} finally {
-			releaseLocks();
-		}
+			},
+		);

 		if (retentionPolicy) {
 			void runForget(scheduleId, mirror.repository.id, organizationId).catch((error) => {
@ -246,6 +259,17 @@ async function copyToSingleMirror(
 		});
 	} catch (error) {
 		const errorMessage = toMessage(error);
+		if (isMirrorCopyCancellation(error)) {
+			logger.info(
+				`[Background] Mirror copy to repository ${mirror.repository.name} was cancelled: ${errorMessage}`,
+			);
+			await mirrorQueries.updateStatus(scheduleId, mirror.repositoryId, {
+				lastCopyStatus: null,
+				lastCopyError: null,
+			});
+			return;
+		}
+
 		logger.error(`[Background] Failed to copy to mirror repository ${mirror.repository.name}: ${errorMessage}`);

 		await mirrorQueries.updateStatus(scheduleId, mirror.repositoryId, {
--- a/app/server/modules/lifecycle/tests/shutdown.test.ts
+++ b/app/server/modules/lifecycle/tests/shutdown.test.ts
@ -1,5 +1,6 @@
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
 import { Scheduler } from "../../../core/scheduler";
+import { repoMutex } from "../../../core/repository-mutex";
 import * as bootstrapModule from "../bootstrap";
 import { agentManager } from "../../agents/agents-manager";
 import { createTestVolume } from "~/test/helpers/volume";
@ -32,6 +33,9 @@ describe("shutdown", () => {
 		const stopApplicationRuntime = vi.fn(async () => {
 			events.push("agents.stop");
 		});
+		const shutdownRepoMutex = vi.fn(async () => {
+			events.push("repo-mutex.shutdown");
+		});
 		const runVolumeCommand = vi.spyOn(agentManager, "runVolumeCommand");

 		const volume = await createTestVolume({
@ -45,13 +49,14 @@ describe("shutdown", () => {
 		});

 		vi.spyOn(Scheduler, "stop").mockImplementation(stopScheduler);
+		vi.spyOn(repoMutex, "shutdown").mockImplementation(shutdownRepoMutex);
 		vi.spyOn(bootstrapModule, "stopApplicationRuntime").mockImplementation(stopApplicationRuntime);

 		const { shutdown } = await loadShutdownModule();

 		await shutdown();

-		expect(events).toEqual(["scheduler.stop", "agents.stop"]);
+		expect(events).toEqual(["scheduler.stop", "repo-mutex.shutdown", "agents.stop"]);
 		expect(runVolumeCommand).not.toHaveBeenCalled();
 		const updated = await db.query.volumesTable.findFirst({ where: { id: volume.id } });
 		expect(updated).toBeDefined();
@ -64,6 +69,9 @@ describe("shutdown", () => {
 		vi.spyOn(Scheduler, "stop").mockImplementation(async () => {
 			events.push("scheduler.stop");
 		});
+		vi.spyOn(repoMutex, "shutdown").mockImplementation(async () => {
+			events.push("repo-mutex.shutdown");
+		});
 		vi.spyOn(bootstrapModule, "stopApplicationRuntime").mockImplementation(async () => {
 			events.push("agents.stop");
 		});
@ -82,7 +90,7 @@ describe("shutdown", () => {
 		await shutdown();

 		const updated = await db.query.volumesTable.findFirst({ where: { id: volume.id } });
-		expect(events).toEqual(["scheduler.stop", "agents.stop"]);
+		expect(events).toEqual(["scheduler.stop", "repo-mutex.shutdown", "agents.stop"]);
 		expect(runVolumeCommand).not.toHaveBeenCalled();
 		expect(updated).toBeDefined();
 		expect(updated!.status).toBe("mounted");
--- a/app/server/modules/lifecycle/shutdown.ts
+++ b/app/server/modules/lifecycle/shutdown.ts
@ -1,4 +1,5 @@
 import { Scheduler } from "../../core/scheduler";
+import { repoMutex } from "../../core/repository-mutex";
 import { db } from "../../db/db";
 import { logger } from "@zerobyte/core/node";
 import { stopApplicationRuntime } from "./bootstrap";
@ -8,9 +9,7 @@ import { toMessage } from "../../utils/errors";
 import { config } from "../../core/config";
 import { LOCAL_AGENT_ID } from "../agents/constants";

-export const shutdown = async () => {
-	await Scheduler.stop();
-
+const unmountVolumes = async () => {
 	if (!config.flags.enableLocalAgent) {
 		const volumes = await db.query.volumesTable.findMany({
 			where: { AND: [{ status: "mounted" }, { agentId: LOCAL_AGENT_ID }] },
@ -24,6 +23,8 @@ export const shutdown = async () => {
 			logger.info(`Volume ${volume.name} unmount status: ${status}${error ? `, error: ${error}` : ""}`);
 		}
 	}
-
-	await stopApplicationRuntime();
+};
+
+export const shutdown = async () => {
+	await Promise.allSettled([Scheduler.stop(), repoMutex.shutdown(), unmountVolumes(), stopApplicationRuntime()]);
 };
--- a/app/server/modules/repositories/tests/repositories.service.test.ts
+++ b/app/server/modules/repositories/tests/repositories.service.test.ts
@ -22,6 +22,7 @@ import { ResticError } from "@zerobyte/core/restic/server";
 import { repoMutex } from "~/server/core/repository-mutex";
 import { taskStore } from "~/server/modules/tasks/tasks.store";
 import { repositoriesService } from "../repositories.service";
+import { holdExclusiveLock, holdSharedLock } from "~/test/helpers/repository-mutex";

 const createTestRepository = async (organizationId: string, overrides: Partial<RepositoryInsert> = {}) => {
 	const id = randomUUID();
@ -220,9 +221,17 @@ describe("repositoriesService.listSnapshotFiles", () => {
 		let releaseAll = false;
 		let exclusiveAcquired = false;
 		let releaseExclusive: (() => void) | undefined;
-		let exclusivePromise: Promise<() => void> | undefined;
+		let exclusivePromise: Promise<void> | undefined;
 		const releaseWaiters: Array<() => void> = [];
 		const exclusiveController = new AbortController();
+		let resolveExclusiveStarted: () => void = () => {};
+		let resolveExclusiveReleased: () => void = () => {};
+		const exclusiveStarted = new Promise<void>((resolve) => {
+			resolveExclusiveStarted = resolve;
+		});
+		const exclusiveReleased = new Promise<void>((resolve) => {
+			resolveExclusiveReleased = resolve;
+		});

 		const releaseWaitingCommands = () => {
 			const waiters = releaseWaiters.splice(0);
@ -300,22 +309,27 @@ describe("repositoriesService.listSnapshotFiles", () => {
 			});
 			expect(maxActive).toBe(2);

-			exclusivePromise = repoMutex
-				.acquireExclusive(repository.id, "delete", exclusiveController.signal)
-				.then((release) => {
+			exclusivePromise = repoMutex.runExclusive(
+				repository.id,
+				"delete",
+				async () => {
 					exclusiveAcquired = true;
-					releaseExclusive = release;
-					return release;
-				});
+					releaseExclusive = resolveExclusiveReleased;
+					resolveExclusiveStarted();
+					await exclusiveReleased;
+				},
+				exclusiveController.signal,
+			);

 			releaseWaitingCommands();

-			releaseExclusive = await resolveWithin(exclusivePromise, 2000);
+			await resolveWithin(exclusiveStarted, 2000);
 			expect(exclusiveAcquired).toBe(true);
 			expect(active).toBe(0);

-			releaseExclusive();
+			releaseExclusive?.();
 			releaseExclusive = undefined;
+			await resolveWithin(exclusivePromise, 2000);

 			await waitForExpect(() => {
 				expect(releaseWaiters).toHaveLength(2);
@ -343,6 +357,78 @@ describe("repositoriesService.listSnapshotFiles", () => {
 	});
 });

+describe("repositoriesService.checkHealth", () => {
+	afterEach(() => {
+		vi.restoreAllMocks();
+	});
+
+	test("does not persist an aborted health check as a repository error", async () => {
+		const repository = await createTestRepository(session.organizationId, {
+			status: "healthy",
+			lastError: "previous error",
+		});
+		let shutdownPromise: Promise<void> | null = null;
+		vi.spyOn(restic, "check").mockImplementation(() => {
+			shutdownPromise = repoMutex.shutdown({ timeoutMs: 10 });
+			return Effect.succeed({ success: false, hasErrors: true, output: "", error: "Operation aborted" });
+		});
+
+		await expect(
+			withContext({ organizationId: session.organizationId, userId: session.user.id }, () =>
+				repositoriesService.checkHealth(repository.shortId),
+			),
+		).rejects.toThrow("Repository mutex is shutting down");
+		await shutdownPromise;
+
+		const persistedRepository = await db.query.repositoriesTable.findFirst({
+			where: { id: repository.id },
+		});
+		expect(persistedRepository?.status).toBe("healthy");
+		expect(persistedRepository?.lastError).toBe("previous error");
+	});
+});
+
+describe("repositoriesService.startDoctor", () => {
+	afterEach(() => {
+		vi.restoreAllMocks();
+	});
+
+	test("marks a queued doctor as cancelled when repository mutex shutdown aborts it", async () => {
+		const repository = await createTestRepository(session.organizationId, {
+			status: "healthy",
+			lastError: "previous error",
+		});
+		const releaseLock = await holdSharedLock(repository.id, "backup");
+
+		try {
+			await withContext({ organizationId: session.organizationId, userId: session.user.id }, () =>
+				repositoriesService.startDoctor(repository.shortId),
+			);
+
+			await waitForExpect(async () => {
+				const runningRepository = await db.query.repositoriesTable.findFirst({ where: { id: repository.id } });
+				expect(runningRepository?.status).toBe("doctor");
+			});
+			await waitForExpect(async () => {
+				const waiters = await db.query.repositoryLockWaitersTable.findMany({
+					where: { repositoryId: repository.id },
+				});
+				expect(waiters).toHaveLength(1);
+			});
+
+			await repoMutex.shutdown({ timeoutMs: 10 });
+		} finally {
+			await releaseLock();
+		}
+
+		await waitForExpect(async () => {
+			const cancelledRepository = await db.query.repositoriesTable.findFirst({ where: { id: repository.id } });
+			expect(cancelledRepository?.status).toBe("cancelled");
+			expect(cancelledRepository?.lastError).toBe("Repository mutex is shutting down");
+		});
+	});
+});
+
 describe("repositoriesService.dumpSnapshot", () => {
 	afterEach(() => {
 		vi.restoreAllMocks();
@ -373,10 +459,11 @@ describe("repositoriesService.dumpSnapshot", () => {
 		snapshotPaths?: string[];
 	}) => {
 		const organizationId = session.organizationId;
+		const repositoryId = randomUUID();
 		const shortId = generateShortId();

 		await db.insert(repositoriesTable).values({
-			id: randomUUID(),
+			id: repositoryId,
 			shortId,
 			name: `Repository-${randomUUID()}`,
 			type: "local",
@ -421,10 +508,27 @@ describe("repositoriesService.dumpSnapshot", () => {
 		return {
 			organizationId,
 			userId: session.user.id,
+			repositoryId,
 			shortId,
 			basePath,
 		};
 	};
+	const rejectWithin = async <T>(promise: Promise<T>, timeoutMs: number) => {
+		let timeout: ReturnType<typeof setTimeout> | undefined;
+		try {
+			return await Promise.race([
+				promise,
+				new Promise<T>((_resolve, reject) => {
+					timeout = setTimeout(
+						() => reject(new Error(`Expected promise to reject within ${timeoutMs}ms`)),
+						timeoutMs,
+					);
+				}),
+			]);
+		} finally {
+			clearTimeout(timeout);
+		}
+	};

 	test("returns a tar download rooted at the selected directory within the snapshot", async () => {
 		const { organizationId, userId, shortId, basePath } = await setupDumpSnapshotScenario({
@ -547,6 +651,29 @@ describe("repositoriesService.dumpSnapshot", () => {
 			}),
 		);
 	});
+
+	test("rejects when shutdown cancels the dump before it acquires the mutex", async () => {
+		const { organizationId, userId, repositoryId, shortId } = await setupDumpSnapshotScenario({
+			snapshotId: "snapshot-shutdown",
+			basePath: "/var/lib/zerobyte/volumes/vol123/_data",
+		});
+		const holderStarted = new Promise<AbortSignal>((resolve) => {
+			void repoMutex.runExclusive(repositoryId, "holder", async ({ signal }) => {
+				resolve(signal);
+				await new Promise<void>((done) => signal.addEventListener("abort", () => done(), { once: true }));
+			});
+		});
+
+		await holderStarted;
+		const dump = withContext({ organizationId, userId }, () =>
+			repositoriesService.dumpSnapshot(shortId, "snapshot-shutdown"),
+		);
+
+		await new Promise((resolve) => setTimeout(resolve, 20));
+		await repoMutex.shutdown({ timeoutMs: 100 });
+
+		await expect(rejectWithin(dump, 1000)).rejects.toThrow("Repository mutex is shutting down");
+	});
 });

 describe("repositoriesService.restoreSnapshot", () => {
@ -696,7 +823,7 @@ describe("repositoriesService.restoreSnapshot", () => {
 		});
 		restoreMock.mockResolvedValueOnce({ status: "started", result: restoreResult });

-		const releaseExclusive = await repoMutex.acquireExclusive(repositoryId, "check");
+		const releaseExclusive = await holdExclusiveLock(repositoryId, "check");
 		let restoreId = "";
 		try {
 			const restoreStart = withContext({ organizationId, userId }, () =>
@ -720,7 +847,7 @@ describe("repositoriesService.restoreSnapshot", () => {
 			expect(task?.id).toBe(restoreId);
 			expect(task?.status).toBe("queued");
 		} finally {
-			releaseExclusive();
+			await releaseExclusive();
 			await fs.rm(targetPath, { recursive: true, force: true });
 		}

@ -749,6 +876,34 @@ describe("repositoriesService.restoreSnapshot", () => {
 		});
 	});

+	test("cancels a mutex-aborted restore before it starts", async () => {
+		const { organizationId, userId, repositoryShortId, restoreMock } = await setupRestoreSnapshotScenario();
+		const targetPath = await fs.mkdtemp(nodePath.join(process.cwd(), "restore-target-"));
+		await withContext({ organizationId, userId }, () =>
+			repositoriesService.getSnapshotDetails(repositoryShortId, "snapshot-restore"),
+		);
+		vi.spyOn(repoMutex, "runShared").mockRejectedValueOnce(new Error("Repository mutex is shutting down"));
+
+		try {
+			const result = await withContext({ organizationId, userId }, () =>
+				repositoriesService.restoreSnapshot(repositoryShortId, "snapshot-restore", {
+					targetPath,
+				}),
+			);
+
+			expect(restoreMock).not.toHaveBeenCalled();
+			await waitForExpect(async () => {
+				const task = await db.query.tasksTable.findFirst({ where: { id: result.restoreId } });
+				expect(task?.status).toBe("cancelled");
+				expect(task?.error).toBe("Restore was cancelled");
+			});
+		} finally {
+			await fs.rm(targetPath, { recursive: true, force: true });
+		}
+
+		expect(restoreMock).not.toHaveBeenCalled();
+	});
+
 	test("routes restore to the requested target agent", async () => {
 		const organizationId = session.organizationId;
 		const agentId = `agent-${randomUUID()}`;
--- a/app/server/modules/repositories/helpers/doctor.ts
+++ b/app/server/modules/repositories/helpers/doctor.ts
@ -130,29 +130,30 @@ export const executeDoctor = async (
 	const organizationId = getOrganizationId();

 	try {
-		const releaseLock = await repoMutex.acquireExclusive(repositoryId, "doctor", signal);
-
-		try {
+		await repoMutex.runExclusive(
+			repositoryId,
+			"doctor",
+			async ({ signal: operationSignal }) => {
 				// Step 1: Unlock repository
-			const unlockStep = await runUnlockStep(repositoryConfig, signal);
+				const unlockStep = await runUnlockStep(repositoryConfig, operationSignal);
 				steps.push(unlockStep);
-			checkAbortSignal(signal);
+				checkAbortSignal(operationSignal);

 				// Step 2: Check repository
-			const checkStep = await runCheckStep(repositoryConfig, signal);
+				const checkStep = await runCheckStep(repositoryConfig, operationSignal);
 				steps.push(checkStep);
-			checkAbortSignal(signal);
+				checkAbortSignal(operationSignal);

 				// Step 3: Repair index if suggested
 				const checkOutput = parseCheckOutput(checkStep.output);
 				if (checkOutput?.suggest_repair_index) {
-				const repairStep = await runRepairIndexStep(repositoryConfig, signal);
+					const repairStep = await runRepairIndexStep(repositoryConfig, operationSignal);
 					steps.push(repairStep);
-				checkAbortSignal(signal);
-			}
-		} finally {
-			releaseLock();
+					checkAbortSignal(operationSignal);
 				}
+			},
+			signal,
+		);

 		const finalStatus = determineRepositoryStatus(steps);
 		await saveDoctorResults(repositoryId, steps, finalStatus);
@ -166,7 +167,13 @@ export const executeDoctor = async (
 			completedAt: Date.now(),
 		});
 	} catch (error) {
-		if (error instanceof AbortError) {
+		const errorMessage = toMessage(error);
+		const isCancellation =
+			error instanceof AbortError ||
+			errorMessage === "Repository mutex is shutting down" ||
+			errorMessage === "Operation aborted";
+
+		if (isCancellation) {
 			const doctorResult: DoctorResult = {
 				success: false,
 				steps,
@ -178,7 +185,7 @@ export const executeDoctor = async (
 				.set({
 					status: "cancelled",
 					lastChecked: Date.now(),
-					lastError: toMessage(error),
+					lastError: errorMessage,
 					doctorResult,
 				})
 				.where(eq(repositoriesTable.id, repositoryId));
@ -187,7 +194,7 @@ export const executeDoctor = async (
 				organizationId,
 				repositoryId: repositoryShortId,
 				repositoryName,
-				error: toMessage(error),
+				error: errorMessage,
 			});
 		} else {
 			await db
--- a/app/server/modules/repositories/repositories.service.ts
+++ b/app/server/modules/repositories/repositories.service.ts
@ -334,10 +334,9 @@ const getRepository = async (shortId: ShortId) => {
 };

 const runAndStoreRepositoryStats = async (repository: Repository): Promise<ResticStatsDto> => {
-	const releaseLock = await repoMutex.acquireShared(repository.id, "stats");
-	try {
+	return repoMutex.runShared(repository.id, "stats", async ({ signal }) => {
 		const stats = await runEffectPromise(
-			restic.stats(repository.config, { organizationId: repository.organizationId }),
+			restic.stats(repository.config, { organizationId: repository.organizationId, signal }),
 		);

 		await db
@ -351,9 +350,7 @@ const runAndStoreRepositoryStats = async (repository: Repository): Promise<Resti
 			);

 		return stats;
-	} finally {
-		releaseLock();
-	}
+	});
 };

 const refreshRepositoryStats = async (shortId: ShortId) => {
@ -419,24 +416,21 @@ const listSnapshots = async (shortId: ShortId, backupId?: ShortId) => {
 		return cached;
 	}

-	const releaseLock = await repoMutex.acquireShared(repository.id, "snapshots");
-	try {
+	return repoMutex.runShared(repository.id, "snapshots", async ({ signal }) => {
 		let snapshots = [];

 		if (backupId) {
 			snapshots = await runEffectPromise(
-				restic.snapshots(repository.config, { tags: [backupId], organizationId }),
+				restic.snapshots(repository.config, { tags: [backupId], organizationId, signal }),
 			);
 		} else {
-			snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId }));
+			snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId, signal }));
 		}

 		cache.set(cacheKey, snapshots);

 		return snapshots;
-	} finally {
-		releaseLock();
-	}
+	});
 };

 const listSnapshotFiles = async (
@ -483,10 +477,9 @@ const listSnapshotFiles = async (
 	await runEffectPromise(limiter.take(1));

 	try {
-		const releaseLock = await repoMutex.acquireShared(repository.id, `ls:${snapshotId}`);
-		try {
+		return await repoMutex.runShared(repository.id, `ls:${snapshotId}`, async ({ signal }) => {
 			const result = await runEffectPromise(
-				restic.ls(repository.config, snapshotId, path, { organizationId, offset, limit }),
+				restic.ls(repository.config, snapshotId, path, { organizationId, offset, limit, signal }),
 			);

 			if (!result.snapshot) {
@ -511,9 +504,7 @@ const listSnapshotFiles = async (
 			cache.set(cacheKey, result);

 			return response;
-		} finally {
-			releaseLock();
-		}
+		});
 	} finally {
 		await runEffectPromise(limiter.release(1));
 	}
@ -616,9 +607,18 @@ const dumpSnapshot = async (shortId: ShortId, snapshotId: string, path?: string,
 		throw new NotFoundError("Repository not found");
 	}

-	const releaseLock = await repoMutex.acquireShared(repository.id, `dump:${snapshotId}`);
-	let dumpStream: ResticDumpStream | null = null;
+	type DumpSnapshotResult = ResticDumpStream & { filename: string; contentType: string };
+	let resolveStarted: (result: DumpSnapshotResult) => void = () => {};
+	let rejectStarted: (error: unknown) => void = () => {};
+	const started = new Promise<DumpSnapshotResult>((resolve, reject) => {
+		resolveStarted = resolve;
+		rejectStarted = reject;
+	});
+	let lockCompletion: Promise<void>;

+	// The response returns once the stream opens, but the lock must stay held until the stream finishes.
+	lockCompletion = repoMutex.runShared(repository.id, `dump:${snapshotId}`, async ({ signal }) => {
+		let dumpStream: ResticDumpStream | null = null;
 		try {
 			const snapshot = await getSnapshotDetails(repository.shortId, snapshotId);
 			const preparedDump = prepareSnapshotDump({
@ -629,6 +629,7 @@ const dumpSnapshot = async (shortId: ShortId, snapshotId: string, path?: string,
 			const dumpOptions: Parameters<typeof restic.dump>[2] = {
 				organizationId,
 				path: preparedDump.path,
+				signal,
 			};

 			let filename = preparedDump.filename;
@ -659,17 +660,18 @@ const dumpSnapshot = async (shortId: ShortId, snapshotId: string, path?: string,
 				filename,
 			});

-		const completion = dumpStream.completion.finally(releaseLock);
-		void completion.catch(() => {});
+			resolveStarted({ ...dumpStream, completion: lockCompletion, filename, contentType });

-		return { ...dumpStream, completion, filename, contentType };
+			await dumpStream.completion;
 		} catch (error) {
-		if (dumpStream) {
-			dumpStream.abort();
-		}
-		releaseLock();
+			rejectStarted(error);
+			dumpStream?.abort();
 			throw error;
 		}
+	});
+	void lockCompletion.catch(rejectStarted);
+
+	return await started;
 };

 const getSnapshotDetails = async (shortId: ShortId, snapshotId: string) => {
@ -684,13 +686,11 @@ const getSnapshotDetails = async (shortId: ShortId, snapshotId: string) => {
 	let snapshots = cache.get<Effect.Effect.Success<ReturnType<typeof restic.snapshots>>>(cacheKey);

 	if (!snapshots) {
-		const releaseLock = await repoMutex.acquireShared(repository.id, `snapshot_details:${snapshotId}`);
-		try {
-			snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId }));
+		snapshots = await repoMutex.runShared(repository.id, `snapshot_details:${snapshotId}`, async ({ signal }) => {
+			const snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId, signal }));
 			cache.set(cacheKey, snapshots);
-		} finally {
-			releaseLock();
-		}
+			return snapshots;
+		});
 	}

 	const snapshot = snapshots.find((snap) => snap.id === snapshotId || snap.short_id === snapshotId);
@ -712,9 +712,13 @@ const checkHealth = async (shortId: ShortId) => {
 		throw new NotFoundError("Repository not found");
 	}

-	const releaseLock = await repoMutex.acquireExclusive(repository.id, "check");
-	try {
-		const { hasErrors, error } = await runEffectPromise(restic.check(repository.config, { organizationId }));
+	return repoMutex.runExclusive(repository.id, "check", async ({ signal }) => {
+		const { hasErrors, error } = await runEffectPromise(
+			restic.check(repository.config, { organizationId, signal }),
+		);
+		if (signal.aborted) {
+			throw signal.reason ?? new Error("Operation aborted");
+		}

 		await db
 			.update(repositoriesTable)
@ -731,9 +735,7 @@ const checkHealth = async (shortId: ShortId) => {
 			);

 		return { lastError: error };
-	} finally {
-		releaseLock();
-	}
+	});
 };

 const startDoctor = async (shortId: ShortId) => {
@ -810,18 +812,15 @@ const deleteSnapshot = async (shortId: ShortId, snapshotId: string) => {
 		throw new NotFoundError("Repository not found");
 	}

-	const releaseLock = await repoMutex.acquireExclusive(repository.id, `delete:${snapshotId}`);
-	try {
-		await runEffectPromise(restic.deleteSnapshot(repository.config, snapshotId, { organizationId }));
+	await repoMutex.runExclusive(repository.id, `delete:${snapshotId}`, async ({ signal }) => {
+		await runEffectPromise(restic.deleteSnapshot(repository.config, snapshotId, { organizationId, signal }));
 		cache.delByPrefix(cacheKeys.repository.all(repository.id));
 		void runAndStoreRepositoryStats(repository).catch((error) => {
 			logger.error(
 				`Failed to refresh repository stats after snapshot deletion for ${repository.shortId}: ${toMessage(error)}`,
 			);
 		});
-	} finally {
-		releaseLock();
-	}
+	});
 };

 const deleteSnapshots = async (shortId: ShortId, snapshotIds: string[]) => {
@ -833,14 +832,11 @@ const deleteSnapshots = async (shortId: ShortId, snapshotIds: string[]) => {
 	}

 	let shouldRefreshStats = false;
-	const releaseLock = await repoMutex.acquireExclusive(repository.id, `delete:bulk`);
-	try {
-		await runEffectPromise(restic.deleteSnapshots(repository.config, snapshotIds, { organizationId }));
+	await repoMutex.runExclusive(repository.id, `delete:bulk`, async ({ signal }) => {
+		await runEffectPromise(restic.deleteSnapshots(repository.config, snapshotIds, { organizationId, signal }));
 		cache.delByPrefix(cacheKeys.repository.all(repository.id));
 		shouldRefreshStats = true;
-	} finally {
-		releaseLock();
-	}
+	});

 	if (!shouldRefreshStats) {
 		return;
@ -865,13 +861,10 @@ const tagSnapshots = async (
 		throw new NotFoundError("Repository not found");
 	}

-	const releaseLock = await repoMutex.acquireExclusive(repository.id, `tag:bulk`);
-	try {
-		await runEffectPromise(restic.tagSnapshots(repository.config, snapshotIds, tags, { organizationId }));
+	await repoMutex.runExclusive(repository.id, `tag:bulk`, async ({ signal }) => {
+		await runEffectPromise(restic.tagSnapshots(repository.config, snapshotIds, tags, { organizationId, signal }));
 		cache.delByPrefix(cacheKeys.repository.all(repository.id));
-	} finally {
-		releaseLock();
-	}
+	});
 };

 const refreshSnapshots = async (shortId: ShortId) => {
@ -884,9 +877,8 @@ const refreshSnapshots = async (shortId: ShortId) => {

 	cache.delByPrefix(cacheKeys.repository.all(repository.id));

-	const releaseLock = await repoMutex.acquireShared(repository.id, "refresh");
-	try {
-		const snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId }));
+	return repoMutex.runShared(repository.id, "refresh", async ({ signal }) => {
+		const snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId, signal }));
 		const cacheKey = cacheKeys.repository.snapshots(repository.id);
 		cache.set(cacheKey, snapshots);

@ -894,9 +886,7 @@ const refreshSnapshots = async (shortId: ShortId) => {
 			message: "Snapshot cache cleared and refreshed",
 			count: snapshots.length,
 		};
-	} finally {
-		releaseLock();
-	}
+	});
 };

 const updateRepository = async (shortId: ShortId, updates: UpdateRepositoryBody) => {
@ -982,13 +972,9 @@ const unlockRepository = async (shortId: ShortId) => {
 		throw new NotFoundError("Repository not found");
 	}

-	const releaseLock = await repoMutex.acquireExclusive(repository.id, "unlock");
-	try {
-		const result = await runEffectPromise(restic.unlock(repository.config, { organizationId }));
-		return result;
-	} finally {
-		releaseLock();
-	}
+	return repoMutex.runExclusive(repository.id, "unlock", async ({ signal }) => {
+		return await runEffectPromise(restic.unlock(repository.config, { organizationId, signal }));
+	});
 };

 const execResticCommand = async (
--- a/app/server/modules/repositories/restore-executor.ts
+++ b/app/server/modules/repositories/restore-executor.ts
@ -29,6 +29,11 @@ type RestoreExecutionHandle = {

 const shouldRunInController = (agentId: string) => agentId === LOCAL_AGENT_ID && !appConfig.flags.enableLocalAgent;

+const isAbortLikeError = (error: unknown) => {
+	const message = toMessage(error);
+	return message === "Repository mutex is shutting down" || message === "Operation aborted";
+};
+
 const createRestoreRunPayload = async (request: RestoreExecutionRequest): Promise<RestoreRunPayload> => {
 	const encryptedResticPassword = await resticDeps.getOrganizationResticPassword(request.organizationId);
 	const resticPassword = await resticDeps.resolveSecret(encryptedResticPassword);
@ -70,7 +75,7 @@ const executeControllerRestore = async (

 		return { status: "completed", result };
 	} catch (error) {
-		if (signal.aborted) {
+		if (signal.aborted || isAbortLikeError(error)) {
 			return { status: "cancelled", message: "Restore was cancelled" };
 		}

@ -101,7 +106,7 @@ const executeAgentRestore = async (

 		return await started.result;
 	} catch (error) {
-		if (signal.aborted) {
+		if (signal.aborted || isAbortLikeError(error)) {
 			return { status: "cancelled", message: "Restore was cancelled" };
 		}

@ -113,24 +118,25 @@ const executeRestoreWithRepositoryLock = async (
 	request: RestoreExecutionRequest,
 	signal: AbortSignal,
 ): Promise<RestoreExecutionResult> => {
-	let releaseLock: (() => void) | null = null;
-
 	try {
-		releaseLock = await repoMutex.acquireShared(request.repositoryId, `restore:${request.restoreId}`, signal);
-
+		return await repoMutex.runShared(
+			request.repositoryId,
+			`restore:${request.restoreId}`,
+			async ({ signal: operationSignal }) => {
 				if (shouldRunInController(request.executionAgentId)) {
-			return await executeControllerRestore(request, signal);
+					return await executeControllerRestore(request, operationSignal);
 				}

-		return await executeAgentRestore(request, signal);
+				return await executeAgentRestore(request, operationSignal);
+			},
+			signal,
+		);
 	} catch (error) {
-		if (signal.aborted) {
+		if (signal.aborted || isAbortLikeError(error)) {
 			return { status: "cancelled", message: "Restore was cancelled" };
 		}

 		return { status: "failed", error: toMessage(error) };
-	} finally {
-		releaseLock?.();
 	}
 };

--- a/app/server/plugins/bootstrap.ts
+++ b/app/server/plugins/bootstrap.ts
@ -1,10 +1,36 @@
 import { definePlugin } from "nitro";
-import { bootstrapApplication, stopApplicationRuntime } from "../modules/lifecycle/bootstrap";
+import { bootstrapApplication } from "../modules/lifecycle/bootstrap";
+import { shutdown } from "../modules/lifecycle/shutdown";
 import { logger } from "@zerobyte/core/node";
 import { toMessage } from "../utils/errors";

+let shutdownPromise: Promise<void> | undefined;
+
+const runGracefulShutdown = async (reason: string) => {
+	if (!shutdownPromise) {
+		logger.info(`${reason}, starting graceful shutdown...`);
+		shutdownPromise = shutdown().catch((err) => {
+			logger.error("Error during shutdown", err);
+		});
+	}
+
+	await shutdownPromise;
+};
+
+const runSignalShutdown = (reason: string) => {
+	void runGracefulShutdown(reason).finally(() => {
+		process.exit(0);
+	});
+};
+
 export default definePlugin(async (nitroApp) => {
-	nitroApp.hooks.hook("close", stopApplicationRuntime);
+	process.on("SIGTERM", () => {
+		runSignalShutdown("SIGTERM received");
+	});
+	process.on("SIGINT", () => {
+		runSignalShutdown("SIGINT received");
+	});
+	nitroApp.hooks.hook("close", () => runGracefulShutdown("Server closing"));

 	await bootstrapApplication().catch((err) => {
 		logger.error(`Bootstrap failed: ${toMessage(err)}`);
--- a/app/test/helpers/repository-mutex.ts
+++ b/app/test/helpers/repository-mutex.ts
@ -0,0 +1,43 @@
+import { repoMutex } from "~/server/core/repository-mutex";
+
+export const createDeferred = <T = void>() => {
+	let resolve: (value: T | PromiseLike<T>) => void = () => {};
+	let reject: (reason?: unknown) => void = () => {};
+	const promise = new Promise<T>((promiseResolve, promiseReject) => {
+		resolve = promiseResolve;
+		reject = promiseReject;
+	});
+
+	return { promise, resolve, reject };
+};
+
+const holdWithRun = async (run: (callback: () => Promise<void>) => Promise<unknown>) => {
+	const started = createDeferred();
+	const finished = createDeferred();
+	const operation = run(async () => {
+		started.resolve();
+		await finished.promise;
+	});
+	operation.catch((error) => started.reject(error));
+
+	await started.promise;
+
+	let released = false;
+	return async () => {
+		if (!released) {
+			released = true;
+			finished.resolve();
+		}
+
+		await operation;
+	};
+};
+
+export const holdSharedLock = (repositoryId: string, operation: string, signal?: AbortSignal) =>
+	holdWithRun((callback) => repoMutex.runShared(repositoryId, operation, callback, signal));
+
+export const holdExclusiveLock = (repositoryId: string, operation: string, signal?: AbortSignal) =>
+	holdWithRun((callback) => repoMutex.runExclusive(repositoryId, operation, callback, signal));
+
+export const holdManyLocks = (requests: Parameters<typeof repoMutex.runMany>[0], signal?: AbortSignal) =>
+	holdWithRun((callback) => repoMutex.runMany(requests, callback, signal));
--- a/packages/core/src/restic/commands/tests/copy.test.ts
+++ b/packages/core/src/restic/commands/tests/copy.test.ts
@ -102,4 +102,24 @@ describe("copy command", () => {
 		expect(separatorIndex).toBeGreaterThan(-1);
 		expect(getArgs().slice(separatorIndex + 1)).toEqual(["latest"]);
 	});
+
+	test("does not treat a cleanup-time abort as a failed successful copy", async () => {
+		const controller = new AbortController();
+		setup();
+		vi.spyOn(cleanupModule, "cleanupTemporaryKeys").mockImplementation(async () => {
+			controller.abort(new Error("aborted during cleanup"));
+		});
+
+		const result = await Effect.runPromise(
+			copy(
+				sourceConfig,
+				destConfig,
+				{ organizationId: "org-1", tag: "daily", signal: controller.signal },
+				mockDeps,
+			),
+		);
+
+		expect(result).toEqual({ success: true, output: "copied" });
+		expect(cleanupModule.cleanupTemporaryKeys).toHaveBeenCalledTimes(2);
+	});
 });
--- a/packages/core/src/restic/commands/tests/delete-snapshots.test.ts
+++ b/packages/core/src/restic/commands/tests/delete-snapshots.test.ts
@ -68,4 +68,19 @@ describe("deleteSnapshots command", () => {
 		expect(nodeModule.safeExec).not.toHaveBeenCalled();
 		expect(cleanupModule.cleanupTemporaryKeys).not.toHaveBeenCalled();
 	});
+
+	test("does not treat a cleanup-time abort as a failed successful deletion", async () => {
+		const controller = new AbortController();
+		setup();
+		vi.spyOn(cleanupModule, "cleanupTemporaryKeys").mockImplementation(async () => {
+			controller.abort(new Error("aborted during cleanup"));
+		});
+
+		const result = await Effect.runPromise(
+			deleteSnapshots(config, ["snapshot-1"], { organizationId: "org-1", signal: controller.signal }, mockDeps),
+		);
+
+		expect(result).toEqual({ success: true });
+		expect(cleanupModule.cleanupTemporaryKeys).toHaveBeenCalledTimes(1);
+	});
 });
--- a/packages/core/src/restic/commands/tests/forget.test.ts
+++ b/packages/core/src/restic/commands/tests/forget.test.ts
@ -0,0 +1,55 @@
+import { afterEach, describe, expect, test, vi } from "vitest";
+import * as cleanupModule from "../../helpers/cleanup-temporary-keys";
+import * as nodeModule from "../../../node";
+import { forget } from "../forget";
+import type { ResticDeps } from "../../types";
+import { Effect } from "effect";
+
+const mockDeps: ResticDeps = {
+	resolveSecret: async (s) => s,
+	getOrganizationResticPassword: async () => "org-restic-password",
+	resticCacheDir: "/tmp/restic-cache",
+	resticPassFile: "/tmp/restic.pass",
+	defaultExcludes: ["/tmp/restic.pass", "/var/lib/zerobyte/repositories"],
+	rcloneConfigFile: "/root/.config/rclone/rclone.conf",
+};
+
+const config = {
+	backend: "local" as const,
+	path: "/tmp/restic-repo",
+	isExistingRepository: true,
+	customPassword: "custom-password",
+};
+
+const setup = () => {
+	vi.spyOn(cleanupModule, "cleanupTemporaryKeys").mockImplementation(() => Promise.resolve());
+	vi.spyOn(nodeModule, "safeExec").mockImplementation(async () => {
+		return { exitCode: 0, stdout: "", stderr: "", timedOut: false };
+	});
+};
+
+afterEach(() => {
+	vi.restoreAllMocks();
+});
+
+describe("forget command", () => {
+	test("does not treat a cleanup-time abort as a failed successful forget", async () => {
+		const controller = new AbortController();
+		setup();
+		vi.spyOn(cleanupModule, "cleanupTemporaryKeys").mockImplementation(async () => {
+			controller.abort(new Error("aborted during cleanup"));
+		});
+
+		const result = await Effect.runPromise(
+			forget(
+				config,
+				{ keepLast: 1 },
+				{ organizationId: "org-1", tag: "daily", signal: controller.signal },
+				mockDeps,
+			),
+		);
+
+		expect(result).toEqual({ success: true, data: null });
+		expect(cleanupModule.cleanupTemporaryKeys).toHaveBeenCalledTimes(1);
+	});
+});
--- a/packages/core/src/restic/commands/tests/tag-snapshots.test.ts
+++ b/packages/core/src/restic/commands/tests/tag-snapshots.test.ts
@ -52,4 +52,25 @@ describe("tagSnapshots command", () => {
 		expect(separatorIndex).toBeGreaterThan(-1);
 		expect(getArgs().slice(separatorIndex + 1)).toEqual(snapshotIds);
 	});
+
+	test("does not treat a cleanup-time abort as a failed successful tag", async () => {
+		const controller = new AbortController();
+		setup();
+		vi.spyOn(cleanupModule, "cleanupTemporaryKeys").mockImplementation(async () => {
+			controller.abort(new Error("aborted during cleanup"));
+		});
+
+		const result = await Effect.runPromise(
+			tagSnapshots(
+				config,
+				["snapshot-1"],
+				{ add: ["keep"] },
+				{ organizationId: "org-1", signal: controller.signal },
+				mockDeps,
+			),
+		);
+
+		expect(result).toEqual({ success: true });
+		expect(cleanupModule.cleanupTemporaryKeys).toHaveBeenCalledTimes(1);
+	});
 });
--- a/packages/core/src/restic/commands/copy.ts
+++ b/packages/core/src/restic/commands/copy.ts
@ -18,7 +18,7 @@ class ResticCopyCommandError extends Data.TaggedError("ResticCopyCommandError")<
 export const copy = (
 	sourceConfig: RepositoryConfig,
 	destConfig: RepositoryConfig,
-	options: { organizationId: string; tag?: string; snapshotIds?: string[] },
+	options: { organizationId: string; tag?: string; snapshotIds?: string[]; signal?: AbortSignal },
 	deps: ResticDeps,
 ) => {
 	return Effect.tryPromise({
@ -63,14 +63,22 @@ export const copy = (
 			logger.info(`Copying snapshots from ${sourceRepoUrl} to ${destRepoUrl}...`);
 			logger.debug(`Executing: restic ${args.join(" ")}`);

-			const res = await safeExec({ command: "restic", args, env });
-
+			let res: Awaited<ReturnType<typeof safeExec>>;
+			try {
+				res = await safeExec({ command: "restic", args, env, signal: options.signal });
+			} finally {
 				await cleanupTemporaryKeys(sourceEnv, deps);
 				await cleanupTemporaryKeys(destEnv, deps);
+			}

 			const { stdout, stderr } = res;

 			if (res.exitCode !== 0) {
+				if (options.signal?.aborted) {
+					logger.warn("Restic copy was aborted by signal.");
+					throw new Error("Operation aborted");
+				}
+
 				logger.error(`Restic copy failed: ${stderr}`);
 				throw createResticError(res.exitCode, stderr);
 			}
--- a/packages/core/src/restic/commands/delete-snapshots.ts
+++ b/packages/core/src/restic/commands/delete-snapshots.ts
@ -17,7 +17,7 @@ class ResticDeleteSnapshotCommandError extends Data.TaggedError("ResticDeleteSna
 export const deleteSnapshots = (
 	config: RepositoryConfig,
 	snapshotIds: string[],
-	options: { organizationId: string },
+	options: { organizationId: string; signal?: AbortSignal },
 	deps: ResticDeps,
 ) => {
 	return Effect.tryPromise({
@ -33,10 +33,19 @@ export const deleteSnapshots = (
 			addCommonArgs(args, env, config);
 			args.push("--", ...snapshotIds);

-			const res = await safeExec({ command: "restic", args, env });
+			let res: Awaited<ReturnType<typeof safeExec>>;
+			try {
+				res = await safeExec({ command: "restic", args, env, signal: options.signal });
+			} finally {
 				await cleanupTemporaryKeys(env, deps);
+			}

 			if (res.exitCode !== 0) {
+				if (options.signal?.aborted) {
+					logger.warn("Restic snapshot deletion was aborted by signal.");
+					throw new Error("Operation aborted");
+				}
+
 				logger.error(`Restic snapshot deletion failed: ${res.stderr}`);
 				throw createResticError(res.exitCode, res.stderr);
 			}
@ -59,7 +68,7 @@ export const deleteSnapshots = (
 export const deleteSnapshot = (
 	config: RepositoryConfig,
 	snapshotId: string,
-	options: { organizationId: string },
+	options: { organizationId: string; signal?: AbortSignal },
 	deps: ResticDeps,
 ) => {
 	return deleteSnapshots(config, [snapshotId], options, deps);
--- a/packages/core/src/restic/commands/dump.ts
+++ b/packages/core/src/restic/commands/dump.ts
@ -32,6 +32,7 @@ export const dump = (
 		organizationId: string;
 		path?: string;
 		archive?: false;
+		signal?: AbortSignal;
 	},
 	deps: ResticDeps,
 ) => {
@ -64,6 +65,12 @@ export const dump = (

 			let stream: Readable | null = null;
 			let abortController: AbortController | null = new AbortController();
+			const abortDump = () => abortController?.abort(options.signal?.reason);
+			if (options.signal?.aborted) {
+				abortDump();
+			} else {
+				options.signal?.addEventListener("abort", abortDump, { once: true });
+			}

 			const maxStderrChars = 64 * 1024;
 			let stderrTail = "";
@ -97,6 +104,7 @@ export const dump = (
 					throw createResticError(result.exitCode, stderr);
 				})
 				.finally(async () => {
+					options.signal?.removeEventListener("abort", abortDump);
 					abortController = null;
 					await cleanup();
 				});
--- a/packages/core/src/restic/commands/forget.ts
+++ b/packages/core/src/restic/commands/forget.ts
@ -18,7 +18,7 @@ class ResticForgetCommandError extends Data.TaggedError("ResticForgetCommandErro
 export const forget = (
 	config: RepositoryConfig,
 	options: RetentionPolicy,
-	extra: { tag: string; organizationId: string; dryRun?: boolean },
+	extra: { tag: string; organizationId: string; dryRun?: boolean; signal?: AbortSignal },
 	deps: ResticDeps,
 ) => {
 	return Effect.tryPromise({
@ -60,10 +60,19 @@ export const forget = (

 			addCommonArgs(args, env, config);

-			const res = await safeExec({ command: "restic", args, env });
+			let res: Awaited<ReturnType<typeof safeExec>>;
+			try {
+				res = await safeExec({ command: "restic", args, env, signal: extra.signal });
+			} finally {
 				await cleanupTemporaryKeys(env, deps);
+			}

 			if (res.exitCode !== 0) {
+				if (extra.signal?.aborted) {
+					logger.warn("Restic forget was aborted by signal.");
+					throw new Error("Operation aborted");
+				}
+
 				logger.error(`Restic forget failed: ${res.stderr}`);
 				throw createResticError(res.exitCode, res.stderr);
 			}
--- a/packages/core/src/restic/commands/ls.ts
+++ b/packages/core/src/restic/commands/ls.ts
@ -60,7 +60,7 @@ export const ls = (
 	config: RepositoryConfig,
 	snapshotId: string,
 	path: string | undefined,
-	options: { organizationId: string; offset?: number; limit?: number },
+	options: { organizationId: string; offset?: number; limit?: number; signal?: AbortSignal },
 	deps: ResticDeps,
 ) => {
 	return Effect.tryPromise({
@ -92,6 +92,7 @@ export const ls = (
 				command: "restic",
 				args,
 				env,
+				signal: options.signal,
 				onStdout: (line) => {
 					const trimmedLine = line.trim();
 					if (!trimmedLine) {
@ -132,6 +133,11 @@ export const ls = (

 			await cleanupTemporaryKeys(env, deps);

+			if (options.signal?.aborted) {
+				logger.warn("Restic ls was aborted by signal.");
+				throw new Error("Operation aborted");
+			}
+
 			if (res.exitCode !== 0) {
 				logger.error(`Restic ls failed: ${res.error}`);
 				throw createResticError(res.exitCode, res.stderr || res.error);
--- a/packages/core/src/restic/commands/snapshots.ts
+++ b/packages/core/src/restic/commands/snapshots.ts
@ -33,7 +33,7 @@ const snapshotInfoSchema = z.object({

 export const snapshots = (
 	config: RepositoryConfig,
-	options: { tags?: string[]; organizationId: string },
+	options: { tags?: string[]; organizationId: string; signal?: AbortSignal },
 	deps: ResticDeps,
 ) => {
 	return Effect.tryPromise({
@ -58,12 +58,18 @@ export const snapshots = (
 				command: "restic",
 				args,
 				env,
+				signal: options.signal,
 				onStdout: (line) => {
 					stdoutLines.push(line);
 				},
 			});
 			await cleanupTemporaryKeys(env, deps);

+			if (options.signal?.aborted) {
+				logger.warn("Restic snapshots was aborted by signal.");
+				throw new Error("Operation aborted");
+			}
+
 			if (res.exitCode !== 0) {
 				const errorMessage = res.stderr || res.error;
 				logger.error(`Restic snapshots retrieval failed: ${errorMessage}`);
--- a/packages/core/src/restic/commands/stats.ts
+++ b/packages/core/src/restic/commands/stats.ts
@ -16,7 +16,11 @@ class ResticStatsCommandError extends Data.TaggedError("ResticStatsCommandError"
 	message: string;
 }> {}

-export const stats = (config: RepositoryConfig, options: { organizationId: string }, deps: ResticDeps) => {
+export const stats = (
+	config: RepositoryConfig,
+	options: { organizationId: string; signal?: AbortSignal },
+	deps: ResticDeps,
+) => {
 	return Effect.tryPromise({
 		try: async () => {
 			const repoUrl = buildRepoUrl(config);
@ -25,9 +29,14 @@ export const stats = (config: RepositoryConfig, options: { organizationId: strin
 			const args = ["--repo", repoUrl, "stats", "--mode", "raw-data"];
 			addCommonArgs(args, env, config);

-			const res = await safeExec({ command: "restic", args, env });
+			const res = await safeExec({ command: "restic", args, env, signal: options.signal });
 			await cleanupTemporaryKeys(env, deps);

+			if (options.signal?.aborted) {
+				logger.warn("Restic stats was aborted by signal.");
+				throw new Error("Operation aborted");
+			}
+
 			if (res.exitCode !== 0) {
 				logger.error(`Restic stats retrieval failed: ${res.stderr}`);
 				throw createResticError(res.exitCode, res.stderr);
--- a/packages/core/src/restic/commands/tag-snapshots.ts
+++ b/packages/core/src/restic/commands/tag-snapshots.ts
@ -18,7 +18,7 @@ export const tagSnapshots = (
 	config: RepositoryConfig,
 	snapshotIds: string[],
 	tags: { add?: string[]; remove?: string[]; set?: string[] },
-	options: { organizationId: string },
+	options: { organizationId: string; signal?: AbortSignal },
 	deps: ResticDeps,
 ) => {
 	return Effect.tryPromise({
@ -53,10 +53,19 @@ export const tagSnapshots = (
 			addCommonArgs(args, env, config);
 			args.push("--", ...snapshotIds);

-			const res = await safeExec({ command: "restic", args, env });
+			let res: Awaited<ReturnType<typeof safeExec>>;
+			try {
+				res = await safeExec({ command: "restic", args, env, signal: options.signal });
+			} finally {
 				await cleanupTemporaryKeys(env, deps);
+			}

 			if (res.exitCode !== 0) {
+				if (options.signal?.aborted) {
+					logger.warn("Restic snapshot tagging was aborted by signal.");
+					throw new Error("Operation aborted");
+				}
+
 				logger.error(`Restic snapshot tagging failed: ${res.stderr}`);
 				throw createResticError(res.exitCode, res.stderr);
 			}
Author	SHA1	Message	Date
Nicolas Meienberger	9553f8b436	fix: cache set race	2026-06-05 21:38:27 +02:00
Nicolas Meienberger	729f6822d5	refactor: controller doesn't shudown abort down to the agent	2026-06-05 20:34:24 +02:00
Nicolas Meienberger	c27e717e0f	chore: fix liniting issue	2026-06-05 20:14:56 +02:00
Nicolas Meienberger	6f0bac785b	fix: handle shutdown aborts without persisting failures	2026-06-05 20:12:13 +02:00
Nicolas Meienberger	e674a8f6bf	refactor: pr feedbacks	2026-06-05 19:29:00 +02:00
Nicolas Meienberger	31112f4f9f	refactor: gracefully abort restic tasks on shutdown	2026-06-05 18:13:01 +02:00