fix(agents): retry transient stale session locks

steipete · web-flow · commit 210adf1d1189 · 2026-05-31T16:41:37.000+01:00
Follow-up to #88658. Retries transient stale session-lock acquire failures when diagnostics show the old stale report disappeared, was replaced by a fresh valid lock, or was replaced by a fresh payload-less lock still inside the mtime/orphan grace window. Preserves typed `SessionWriteLockStaleError` diagnostics for still-present live OpenClaw-owned stale locks. Proof: 53 focused session-write-lock tests passed locally and in the agents-core CI shard; `pnpm tsgo:test:src`, touched-file oxlint, `git diff --check`, and autoreview passed locally. CI run 26716843811 has unrelated failures in UI deadcode/types and bash-tools tests; session-write-lock tests passed in that run. Refs #87217.
diff --git a/src/agents/session-write-lock.test.ts b/src/agents/session-write-lock.test.ts
@@ -3,6 +3,7 @@ import fsSync from "node:fs";
 import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
+import { fileURLToPath } from "node:url";
 import { MAX_TIMER_TIMEOUT_MS } from "@openclaw/normalization-core/number-coercion";
 import { afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import { SessionWriteLockStaleError } from "./session-write-lock-error.js";
@@ -87,6 +88,19 @@ async function writeCurrentProcessLock(lockPath: string, extra?: Record<string,
   );
 }
 
+function readFilePathToString(filePath: Parameters<typeof fs.readFile>[0]): string | undefined {
+  if (typeof filePath === "string") {
+    return filePath;
+  }
+  if (Buffer.isBuffer(filePath)) {
+    return filePath.toString("utf8");
+  }
+  if (filePath instanceof URL) {
+    return fileURLToPath(filePath);
+  }
+  return undefined;
+}
+
 async function withSymlinkedSessionPaths(
   run: (params: {
     sessionReal: string;
@@ -453,6 +467,157 @@ describe("acquireSessionWriteLock", () => {
     });
   });
 
+  it("retries when a stale lock report disappears before diagnostics", async () => {
+    await withTempSessionLockFile(async ({ sessionFile, lockPath }) => {
+      const owner = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)", "openclaw"], {
+        stdio: "ignore",
+      });
+      if (!owner.pid) {
+        throw new Error("missing lock owner pid");
+      }
+      await fs.writeFile(
+        lockPath,
+        JSON.stringify({
+          pid: owner.pid,
+          createdAt: new Date(Date.now() - 120_000).toISOString(),
+        }),
+        "utf8",
+      );
+
+      const originalReadFile = fs.readFile.bind(fs);
+      let lockReads = 0;
+      const readFileSpy = vi.spyOn(fs, "readFile").mockImplementation((async (
+        filePath,
+        options,
+      ) => {
+        const lockFilePath = readFilePathToString(filePath);
+        if (lockFilePath && path.basename(lockFilePath) === path.basename(lockPath)) {
+          lockReads += 1;
+          if (lockReads === 3) {
+            await fs.rm(lockFilePath, { force: true });
+            await fs.rm(lockPath, { force: true });
+            throw Object.assign(new Error("lock disappeared"), { code: "ENOENT" });
+          }
+        }
+        return await originalReadFile(filePath, options as never);
+      }) as typeof fs.readFile);
+
+      try {
+        const lock = await acquireSessionWriteLock({ sessionFile, timeoutMs: 500, staleMs: 10 });
+        await lock.release();
+        expect(lockReads).toBeGreaterThanOrEqual(3);
+        await expectPathMissing(lockPath);
+      } finally {
+        readFileSpy.mockRestore();
+        owner.kill("SIGTERM");
+      }
+    });
+  });
+
+  it("retries when a stale lock report is replaced before diagnostics", async () => {
+    await withTempSessionLockFile(async ({ sessionFile, lockPath }) => {
+      const owner = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)", "openclaw"], {
+        stdio: "ignore",
+      });
+      if (!owner.pid) {
+        throw new Error("missing lock owner pid");
+      }
+      await fs.writeFile(
+        lockPath,
+        JSON.stringify({
+          pid: owner.pid,
+          createdAt: new Date(Date.now() - 120_000).toISOString(),
+        }),
+        "utf8",
+      );
+
+      const originalReadFile = fs.readFile.bind(fs);
+      let lockReads = 0;
+      const readFileSpy = vi.spyOn(fs, "readFile").mockImplementation((async (
+        filePath,
+        options,
+      ) => {
+        const lockFilePath = readFilePathToString(filePath);
+        if (lockFilePath && path.basename(lockFilePath) === path.basename(lockPath)) {
+          lockReads += 1;
+          if (lockReads === 3) {
+            await fs.rm(lockFilePath, { force: true });
+            await fs.rm(lockPath, { force: true });
+            await fs.writeFile(
+              lockFilePath,
+              JSON.stringify({ pid: owner.pid, createdAt: new Date().toISOString() }),
+              "utf8",
+            );
+            setTimeout(() => {
+              void fs.rm(lockFilePath, { force: true });
+            }, 10);
+            throw Object.assign(new Error("lock disappeared"), { code: "ENOENT" });
+          }
+        }
+        return await originalReadFile(filePath, options as never);
+      }) as typeof fs.readFile);
+
+      try {
+        const lock = await acquireSessionWriteLock({ sessionFile, timeoutMs: 500, staleMs: 10 });
+        await lock.release();
+        expect(lockReads).toBeGreaterThanOrEqual(3);
+        await expectPathMissing(lockPath);
+      } finally {
+        readFileSpy.mockRestore();
+        owner.kill("SIGTERM");
+      }
+    });
+  });
+
+  it("retries when a stale lock report is replaced by a fresh payload-less lock", async () => {
+    await withTempSessionLockFile(async ({ sessionFile, lockPath }) => {
+      const owner = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)", "openclaw"], {
+        stdio: "ignore",
+      });
+      if (!owner.pid) {
+        throw new Error("missing lock owner pid");
+      }
+      await fs.writeFile(
+        lockPath,
+        JSON.stringify({
+          pid: owner.pid,
+          createdAt: new Date(Date.now() - 120_000).toISOString(),
+        }),
+        "utf8",
+      );
+
+      const originalReadFile = fs.readFile.bind(fs);
+      let lockReads = 0;
+      const readFileSpy = vi.spyOn(fs, "readFile").mockImplementation((async (
+        filePath,
+        options,
+      ) => {
+        const lockFilePath = readFilePathToString(filePath);
+        if (lockFilePath && path.basename(lockFilePath) === path.basename(lockPath)) {
+          lockReads += 1;
+          if (lockReads === 3) {
+            await fs.rm(lockFilePath, { force: true });
+            await fs.writeFile(lockFilePath, "", "utf8");
+            setTimeout(() => {
+              void fs.rm(lockFilePath, { force: true });
+            }, 10);
+          }
+        }
+        return await originalReadFile(filePath, options as never);
+      }) as typeof fs.readFile);
+
+      try {
+        const lock = await acquireSessionWriteLock({ sessionFile, timeoutMs: 800, staleMs: 10 });
+        await lock.release();
+        expect(lockReads).toBeGreaterThanOrEqual(3);
+        await expectPathMissing(lockPath);
+      } finally {
+        readFileSpy.mockRestore();
+        owner.kill("SIGTERM");
+      }
+    });
+  });
+
   it("watchdog releases stale in-process locks", async () => {
     const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-lock-"));
     const stderrSpy = vi.spyOn(process.stderr, "write").mockImplementation(() => true);
@@ -600,6 +765,14 @@ describe("acquireSessionWriteLock", () => {
     });
   });
 
+  it("preserves one acquire timeout budget across retries", () => {
+    expect(testing.resolveRemainingAcquireTimeoutMs(500, 1_000, 1_125)).toBe(375);
+    expect(testing.resolveRemainingAcquireTimeoutMs(500, 1_000, 1_500)).toBe(0);
+    expect(testing.resolveRemainingAcquireTimeoutMs(Number.POSITIVE_INFINITY, 1_000, 9_000)).toBe(
+      Number.POSITIVE_INFINITY,
+    );
+  });
+
   it("uses resolved stale policy when cleaning stale lock files", async () => {
     const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-lock-policy-"));
     const sessionsDir = path.join(root, "sessions");
diff --git a/src/agents/session-write-lock.ts b/src/agents/session-write-lock.ts
@@ -389,29 +389,45 @@ function unregisterCleanupHandlers(): void {
   cleanupState.registered = false;
 }
 
+function parseLockPayload(raw: string): LockFilePayload | null {
+  const parsed = JSON.parse(raw) as Record<string, unknown>;
+  const payload: LockFilePayload = {};
+  if (isValidLockNumber(parsed.pid) && parsed.pid > 0) {
+    payload.pid = parsed.pid;
+  }
+  if (typeof parsed.createdAt === "string") {
+    payload.createdAt = parsed.createdAt;
+  }
+  if (isValidLockNumber(parsed.starttime)) {
+    payload.starttime = parsed.starttime;
+  }
+  if (isValidLockNumber(parsed.maxHoldMs) && parsed.maxHoldMs > 0) {
+    payload.maxHoldMs = parsed.maxHoldMs;
+  }
+  return payload;
+}
+
 async function readLockPayload(lockPath: string): Promise<LockFilePayload | null> {
   try {
     const raw = await fs.readFile(lockPath, "utf8");
-    const parsed = JSON.parse(raw) as Record<string, unknown>;
-    const payload: LockFilePayload = {};
-    if (isValidLockNumber(parsed.pid) && parsed.pid > 0) {
-      payload.pid = parsed.pid;
-    }
-    if (typeof parsed.createdAt === "string") {
-      payload.createdAt = parsed.createdAt;
-    }
-    if (isValidLockNumber(parsed.starttime)) {
-      payload.starttime = parsed.starttime;
-    }
-    if (isValidLockNumber(parsed.maxHoldMs) && parsed.maxHoldMs > 0) {
-      payload.maxHoldMs = parsed.maxHoldMs;
-    }
-    return payload;
+    return parseLockPayload(raw);
   } catch {
     return null;
   }
 }
 
+async function readLockPayloadForDiagnostics(
+  lockPath: string,
+): Promise<{ payload: LockFilePayload | null; missing: boolean }> {
+  try {
+    const raw = await fs.readFile(lockPath, "utf8");
+    return { payload: parseLockPayload(raw), missing: false };
+  } catch (error) {
+    const code = (error as { code?: string } | null)?.code;
+    return { payload: null, missing: code === "ENOENT" };
+  }
+}
+
 async function resolveNormalizedSessionFile(sessionFile: string): Promise<string> {
   const resolvedSessionFile = path.resolve(sessionFile);
   const sessionDir = path.dirname(resolvedSessionFile);
@@ -626,6 +642,40 @@ function resolveOrphanLockPayloadGraceMs(timeoutMs: number): number {
   return ORPHAN_LOCK_PAYLOAD_GRACE_MS;
 }
 
+function resolveRemainingAcquireTimeoutMs(
+  timeoutMs: number,
+  startedAtMs: number,
+  nowMs: number,
+): number {
+  if (timeoutMs === Number.POSITIVE_INFINITY) {
+    return Number.POSITIVE_INFINITY;
+  }
+  const elapsedMs = Math.max(0, nowMs - startedAtMs);
+  return Math.max(0, timeoutMs - elapsedMs);
+}
+
+async function shouldRetryStaleAcquireFailure(params: {
+  lockPath: string;
+  lockMissingAtDiagnostics: boolean;
+  inspected: LockInspectionDetails;
+  heldByThisProcess: boolean;
+  staleMs: number;
+  nowMs: number;
+  orphanPayloadGraceMs: number;
+}): Promise<boolean> {
+  if (params.lockMissingAtDiagnostics) {
+    return true;
+  }
+  return !(await shouldReportContendedLockStale({
+    lockPath: params.lockPath,
+    details: params.inspected,
+    heldByThisProcess: params.heldByThisProcess,
+    staleMs: params.staleMs,
+    nowMs: params.nowMs,
+    orphanPayloadGraceMs: params.orphanPayloadGraceMs,
+  }));
+}
+
 async function shouldRemoveLockDuringCleanup(
   lockPath: string,
   details: LockInspectionDetails,
@@ -845,12 +895,30 @@ export async function acquireSessionWriteLock(params: {
   const normalizedSessionFile = await resolveNormalizedSessionFile(sessionFile);
   const lockPath = `${normalizedSessionFile}.lock`;
   await fs.mkdir(sessionDir, { recursive: true });
+  const startedAtMs = Date.now();
 
   while (true) {
+    const remainingTimeoutMs = resolveRemainingAcquireTimeoutMs(timeoutMs, startedAtMs, Date.now());
+    if (remainingTimeoutMs <= 0) {
+      const payload = await readLockPayload(lockPath);
+      const nowMs = Date.now();
+      const heldByThisProcess = sessionLockHeldByThisProcess(normalizedSessionFile);
+      const inspected = inspectLockPayloadForSession({
+        payload,
+        staleMs,
+        nowMs,
+        heldByThisProcess,
+        reclaimLockWithoutStarttime: true,
+        readOwnerProcessArgs: readProcessArgsSync,
+        respectMaxHold: !heldByThisProcess,
+      });
+      const owner = describeLockOwnerForError({ payload, inspected });
+      throw new SessionWriteLockTimeoutError({ timeoutMs, owner, lockPath });
+    }
     try {
       const lock = await SESSION_LOCKS.acquire(sessionFile, {
         staleMs,
-        timeoutMs,
+        timeoutMs: remainingTimeoutMs,
         retry: { minTimeout: 50, maxTimeout: 1000, factor: 1 },
         staleRecovery: "remove-if-unchanged",
         allowReentrant,
@@ -914,7 +982,8 @@ export async function acquireSessionWriteLock(params: {
         throw err;
       }
       const errorLockPath = (err as { lockPath?: string }).lockPath ?? lockPath;
-      const payload = await readLockPayload(errorLockPath);
+      const { payload, missing: lockMissingAtDiagnostics } =
+        await readLockPayloadForDiagnostics(errorLockPath);
       const nowMs = Date.now();
       const heldByThisProcess = sessionLockHeldByThisProcess(normalizedSessionFile);
       const inspected = inspectLockPayloadForSession({
@@ -928,6 +997,20 @@ export async function acquireSessionWriteLock(params: {
       });
       const owner = describeLockOwnerForError({ payload, inspected });
       if (isFileLockError(err, "file_lock_stale")) {
+        if (
+          resolveRemainingAcquireTimeoutMs(timeoutMs, startedAtMs, Date.now()) > 0 &&
+          (await shouldRetryStaleAcquireFailure({
+            lockPath: errorLockPath,
+            lockMissingAtDiagnostics,
+            inspected,
+            heldByThisProcess,
+            staleMs,
+            nowMs,
+            orphanPayloadGraceMs,
+          }))
+        ) {
+          continue;
+        }
         throw new SessionWriteLockStaleError({
           owner,
           lockPath: errorLockPath,
@@ -945,6 +1028,7 @@ export const testing = {
   inspectLockPayloadForTest: inspectLockPayload,
   releaseAllLocksSync,
   runLockWatchdogCheck,
+  resolveRemainingAcquireTimeoutMs,
   setProcessStartTimeResolverForTest(resolver: ((pid: number) => number | null) | null): void {
     resolveProcessStartTimeForLock = resolver ?? getProcessStartTime;
   },
