feat(cms): improve catalog, editorial recovery and operations

src/lib/services/furni-job-worker.ts

@@ -1,4 +1,5 @@
 import { randomUUID } from "node:crypto";
+import type { ImportJob } from "@/lib/furni/import-job";
 import { redis } from "@/lib/redis";
 import { logServerError } from "@/lib/server-log";
 import { logAudit } from "./audit";
@@ -13,6 +14,7 @@ import { repairFurniture } from "./furniture-repair";
 import { rcon } from "./rcon";
 
 const LOCK = "furniture-import-worker:v1";
+class ImportLeaseLostError extends Error {}
 let running: Promise<void> | undefined;
 export function drainFurnitureImports(): Promise<void> {
 	if (running) return running;
@@ -30,6 +32,28 @@ async function drain() {
 	const token = randomUUID();
 	if ((await redis.set(LOCK, token, "EX", 600, "NX")) !== "OK") return;
 	let lease = true;
+	async function requireLease() {
+		if (!lease) throw new ImportLeaseLostError("Import worker lost its lease");
+		try {
+			const owned = await redis?.eval(
+				"if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('expire',KEYS[1],600) else return 0 end",
+				1,
+				LOCK,
+				token,
+			);
+			if (owned !== 1 || !lease) {
+				lease = false;
+				throw new ImportLeaseLostError("Import worker lost its lease");
+			}
+		} catch (error) {
+			lease = false;
+			if (error instanceof ImportLeaseLostError) throw error;
+			throw new ImportLeaseLostError(
+				"Import worker could not verify its lease",
+				{ cause: error },
+			);
+		}
+	}
 	const timer = setInterval(() => {
 		void redis
 			?.eval(
@@ -47,29 +71,48 @@ async function drain() {
 	}, 30000);
 	try {
 		const store = new ImportJobStore();
+		// Redis ownership checks and file writes are separate operations. These checks
+		// prevent known stale writes; they do not provide atomic filesystem fencing.
+		async function saveOwned(job: ImportJob) {
+			await requireLease();
+			await store.save(job);
+		}
 		for (const job of await store.list()) {
-			if (!lease) break;
+			await requireLease();
 			if (job.state === "running") {
 				job.state = "interrupted";
 				for (const item of job.items)
-					if (item.state === "running" || item.state === "pending") {
-						item.state = "failed";
+					if (item.state === "running") {
+						item.state = "interrupted";
 						item.error =
-							"Server restarted during import. Review local data and retry to complete missing parts.";
+							"Server restarted during import. Its outcome is uncertain. Inspect local data before starting a new repair.";
 					}
-				await store.save(job);
+				await saveOwned(job);
 				continue;
 			}
 			if (job.state !== "queued") continue;
+			if (await store.isCancellationRequested(job.id)) {
+				job.state = "cancelled";
+				for (const item of job.items)
+					if (item.state === "pending") item.state = "cancelled";
+				await saveOwned(job);
+				continue;
+			}
 			job.state = "running";
-			await store.save(job);
+			await saveOwned(job);
 			await withCatalogExport(async () => {
 				await ensureDirectories();
 				const source = job.sourceId ? await getSource(job.sourceId) : null;
 				for (const item of job.items) {
-					if (!lease) throw Error("Import worker lost its lease");
+					await requireLease();
+					if (item.state !== "pending") continue;
+					if (await store.isCancellationRequested(job.id)) {
+						for (const remaining of job.items)
+							if (remaining.state === "pending") remaining.state = "cancelled";
+						break;
+					}
 					item.state = "running";
-					await store.save(job);
+					await saveOwned(job);
 					try {
 						if (job.sourceId && !source)
 							throw Error("Import source no longer exists");
@@ -80,6 +123,7 @@ async function drain() {
 									job.userId,
 								)
 							: undefined;
+						await requireLease();
 						const result = await (job.mode === "repair"
 							? repairFurniture
 							: importSingleFurni)({
@@ -90,6 +134,7 @@ async function drain() {
 							nitroBaseUrl: source?.nitroBaseUrl,
 							iconBaseUrl: source?.iconBaseUrl,
 						});
+						await requireLease();
 						item.warnings = result.warnings;
 						item.itemId = result.itemId;
 						if (!result.ok) throw Error(result.error || "Import failed");
@@ -114,16 +159,21 @@ async function drain() {
 								);
 							}
 					} catch (error) {
+						if (error instanceof ImportLeaseLostError) throw error;
+						await requireLease();
 						item.state = "failed";
 						item.error =
 							error instanceof Error ? error.message : "Import failed";
 					}
-					await store.save(job);
+					await saveOwned(job);
 				}
+				await requireLease();
 				try {
 					await rcon.updateCatalog();
+					await requireLease();
 					await rcon.updateItems();
-				} catch {
+				} catch (error) {
+					if (error instanceof ImportLeaseLostError) throw error;
 					for (const item of job.items)
 						if (item.state === "done") {
 							item.warnings ??= [];
@@ -131,17 +181,25 @@ async function drain() {
 						}
 				}
 			});
-			job.state = "completed";
-			await store.save(job);
+			await requireLease();
+			job.state = job.items.some((item) => item.state === "cancelled")
+				? "cancelled"
+				: "completed";
+			await saveOwned(job);
 		}
+		await requireLease();
 		await runCatalogExport();
 	} finally {
 		clearInterval(timer);
-		await redis.eval(
-			"if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end",
-			1,
-			LOCK,
-			token,
-		);
+		await redis
+			.eval(
+				"if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end",
+				1,
+				LOCK,
+				token,
+			)
+			.catch((error) => {
+				logServerError("furni.worker_release_failed", error);
+			});
 	}
 }