File size: 4,465 Bytes
fc93158 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | import fs from "node:fs/promises";
import path from "node:path";
import { isPidAlive } from "../shared/pid-alive.js";
import { resolveProcessScopedMap } from "../shared/process-scoped-map.js";
export type FileLockOptions = {
retries: {
retries: number;
factor: number;
minTimeout: number;
maxTimeout: number;
randomize?: boolean;
};
stale: number;
};
type LockFilePayload = {
pid: number;
createdAt: string;
};
type HeldLock = {
count: number;
handle: fs.FileHandle;
lockPath: string;
};
const HELD_LOCKS_KEY = Symbol.for("openclaw.fileLockHeldLocks");
const HELD_LOCKS = resolveProcessScopedMap<HeldLock>(HELD_LOCKS_KEY);
function computeDelayMs(retries: FileLockOptions["retries"], attempt: number): number {
const base = Math.min(
retries.maxTimeout,
Math.max(retries.minTimeout, retries.minTimeout * retries.factor ** attempt),
);
const jitter = retries.randomize ? 1 + Math.random() : 1;
return Math.min(retries.maxTimeout, Math.round(base * jitter));
}
async function readLockPayload(lockPath: string): Promise<LockFilePayload | null> {
try {
const raw = await fs.readFile(lockPath, "utf8");
const parsed = JSON.parse(raw) as Partial<LockFilePayload>;
if (typeof parsed.pid !== "number" || typeof parsed.createdAt !== "string") {
return null;
}
return { pid: parsed.pid, createdAt: parsed.createdAt };
} catch {
return null;
}
}
async function resolveNormalizedFilePath(filePath: string): Promise<string> {
const resolved = path.resolve(filePath);
const dir = path.dirname(resolved);
await fs.mkdir(dir, { recursive: true });
try {
const realDir = await fs.realpath(dir);
return path.join(realDir, path.basename(resolved));
} catch {
return resolved;
}
}
async function isStaleLock(lockPath: string, staleMs: number): Promise<boolean> {
const payload = await readLockPayload(lockPath);
if (payload?.pid && !isPidAlive(payload.pid)) {
return true;
}
if (payload?.createdAt) {
const createdAt = Date.parse(payload.createdAt);
if (!Number.isFinite(createdAt) || Date.now() - createdAt > staleMs) {
return true;
}
}
try {
const stat = await fs.stat(lockPath);
return Date.now() - stat.mtimeMs > staleMs;
} catch {
return true;
}
}
export type FileLockHandle = {
lockPath: string;
release: () => Promise<void>;
};
async function releaseHeldLock(normalizedFile: string): Promise<void> {
const current = HELD_LOCKS.get(normalizedFile);
if (!current) {
return;
}
current.count -= 1;
if (current.count > 0) {
return;
}
HELD_LOCKS.delete(normalizedFile);
await current.handle.close().catch(() => undefined);
await fs.rm(current.lockPath, { force: true }).catch(() => undefined);
}
export async function acquireFileLock(
filePath: string,
options: FileLockOptions,
): Promise<FileLockHandle> {
const normalizedFile = await resolveNormalizedFilePath(filePath);
const lockPath = `${normalizedFile}.lock`;
const held = HELD_LOCKS.get(normalizedFile);
if (held) {
held.count += 1;
return {
lockPath,
release: () => releaseHeldLock(normalizedFile),
};
}
const attempts = Math.max(1, options.retries.retries + 1);
for (let attempt = 0; attempt < attempts; attempt += 1) {
try {
const handle = await fs.open(lockPath, "wx");
await handle.writeFile(
JSON.stringify({ pid: process.pid, createdAt: new Date().toISOString() }, null, 2),
"utf8",
);
HELD_LOCKS.set(normalizedFile, { count: 1, handle, lockPath });
return {
lockPath,
release: () => releaseHeldLock(normalizedFile),
};
} catch (err) {
const code = (err as { code?: string }).code;
if (code !== "EEXIST") {
throw err;
}
if (await isStaleLock(lockPath, options.stale)) {
await fs.rm(lockPath, { force: true }).catch(() => undefined);
continue;
}
if (attempt >= attempts - 1) {
break;
}
await new Promise((resolve) => setTimeout(resolve, computeDelayMs(options.retries, attempt)));
}
}
throw new Error(`file lock timeout for ${normalizedFile}`);
}
export async function withFileLock<T>(
filePath: string,
options: FileLockOptions,
fn: () => Promise<T>,
): Promise<T> {
const lock = await acquireFileLock(filePath, options);
try {
return await fn();
} finally {
await lock.release();
}
}
|