diff --git a/src/agents/session-write-lock.test.ts b/src/agents/session-write-lock.test.ts index 09982b6c446fb..af0a529e1eef2 100644 --- a/src/agents/session-write-lock.test.ts +++ b/src/agents/session-write-lock.test.ts @@ -1,7 +1,11 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import { describe, expect, it, vi } from "vitest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const diagnosticMocks = vi.hoisted(() => ({ + logSessionLockStats: vi.fn(), +})); // Mock getProcessStartTime so PID-recycling detection works on non-Linux // (macOS, CI runners). isPidAlive is left unmocked. @@ -14,6 +18,14 @@ vi.mock("../shared/pid-alive.js", async (importOriginal) => { }; }); +vi.mock("../logging/diagnostic.js", async (importOriginal) => { + const original = await importOriginal(); + return { + ...original, + logSessionLockStats: diagnosticMocks.logSessionLockStats, + }; +}); + import { __testing, acquireSessionWriteLock, @@ -95,6 +107,10 @@ async function expectActiveInProcessLockIsNotReclaimed(params?: { } describe("acquireSessionWriteLock", () => { + beforeEach(() => { + diagnosticMocks.logSessionLockStats.mockClear(); + }); + it("reuses locks across symlinked session paths", async () => { if (process.platform === "win32") { expect(true).toBe(true); @@ -161,6 +177,15 @@ describe("acquireSessionWriteLock", () => { await expect( acquireSessionWriteLock({ sessionFile, timeoutMs: 50, staleMs: 60_000 }), ).rejects.toThrow(/session file locked/); + expect(diagnosticMocks.logSessionLockStats).toHaveBeenCalledWith( + expect.objectContaining({ + action: "timeout", + lockPath: expect.stringContaining(path.basename(lockPath)), + timeoutMs: 50, + ownerStale: true, + ownerStaleReasons: expect.arrayContaining(["missing-pid", "invalid-createdAt"]), + }), + ); await expect(fs.access(lockPath)).resolves.toBeUndefined(); } finally { await fs.rm(root, { recursive: true, force: true }); @@ -174,11 +199,41 @@ describe("acquireSessionWriteLock", () => { await fs.utimes(lockPath, staleDate, staleDate); const lock = await acquireSessionWriteLock({ sessionFile, timeoutMs: 500, staleMs: 10_000 }); + expect(diagnosticMocks.logSessionLockStats).toHaveBeenCalledWith( + expect.objectContaining({ + action: "acquired_after_wait", + lockPath: expect.stringContaining(path.basename(lockPath)), + reclaimedCount: 1, + }), + ); await lock.release(); await expect(fs.access(lockPath)).rejects.toThrow(); }); }); + it("logs long lock holds on final release", async () => { + vi.useFakeTimers(); + try { + await withTempSessionLockFile(async ({ sessionFile, lockPath }) => { + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + const lock = await acquireSessionWriteLock({ sessionFile, timeoutMs: 500 }); + + vi.setSystemTime(new Date("2026-01-01T00:00:31.000Z")); + await lock.release(); + + expect(diagnosticMocks.logSessionLockStats).toHaveBeenCalledWith( + expect.objectContaining({ + action: "released_after_long_hold", + lockPath: expect.stringContaining(path.basename(lockPath)), + holdMs: 31_000, + }), + ); + }); + } finally { + vi.useRealTimers(); + } + }); + it("watchdog releases stale in-process locks", async () => { const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-lock-")); const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); @@ -330,6 +385,15 @@ describe("acquireSessionWriteLock", () => { await writeCurrentProcessLock(lockPath); await expectCurrentPidOwnsLock({ sessionFile, timeoutMs: 500 }); + expect(diagnosticMocks.logSessionLockStats).toHaveBeenCalledWith( + expect.objectContaining({ + action: "acquired_after_wait", + lockPath: expect.stringContaining(path.basename(lockPath)), + ownerStale: true, + ownerStaleReasons: expect.arrayContaining(["orphan-self-pid"]), + reclaimedCount: 1, + }), + ); }); }); diff --git a/src/agents/session-write-lock.ts b/src/agents/session-write-lock.ts index 5f2cfb6fc419d..10d3df88a5bda 100644 --- a/src/agents/session-write-lock.ts +++ b/src/agents/session-write-lock.ts @@ -1,6 +1,7 @@ import fsSync from "node:fs"; import fs from "node:fs/promises"; import path from "node:path"; +import { logSessionLockStats } from "../logging/diagnostic.js"; import { getProcessStartTime, isPidAlive } from "../shared/pid-alive.js"; import { resolveProcessScopedMap } from "../shared/process-scoped-map.js"; @@ -45,6 +46,8 @@ const DEFAULT_STALE_MS = 30 * 60 * 1000; const DEFAULT_MAX_HOLD_MS = 5 * 60 * 1000; const DEFAULT_WATCHDOG_INTERVAL_MS = 60_000; const DEFAULT_TIMEOUT_GRACE_MS = 2 * 60 * 1000; +const DEFAULT_SLOW_LOCK_WAIT_WARN_MS = 1_000; +const DEFAULT_SLOW_LOCK_HOLD_WARN_MS = 30_000; const MAX_LOCK_HOLD_MS = 2_147_000_000; type CleanupState = { @@ -162,6 +165,14 @@ async function releaseHeldLock( try { await held.releasePromise; + const holdMs = Date.now() - held.acquiredAt; + if (holdMs >= DEFAULT_SLOW_LOCK_HOLD_WARN_MS) { + logSessionLockStats({ + action: "released_after_long_hold", + lockPath: held.lockPath, + holdMs, + }); + } return true; } finally { held.releasePromise = undefined; @@ -479,6 +490,8 @@ export async function acquireSessionWriteLock(params: { const startedAt = Date.now(); let attempt = 0; + let reclaimedCount = 0; + let lastInspected: LockInspectionDetails | null = null; while (Date.now() - startedAt < timeoutMs) { attempt += 1; let handle: fs.FileHandle | null = null; @@ -499,6 +512,21 @@ export async function acquireSessionWriteLock(params: { maxHoldMs, }; HELD_LOCKS.set(normalizedSessionFile, createdHeld); + const waitMs = createdHeld.acquiredAt - startedAt; + if (waitMs >= DEFAULT_SLOW_LOCK_WAIT_WARN_MS || attempt > 1 || reclaimedCount > 0) { + logSessionLockStats({ + action: "acquired_after_wait", + lockPath, + waitMs, + attempts: attempt, + timeoutMs, + ownerPid: lastInspected?.pid, + ownerAgeMs: lastInspected?.ageMs, + ownerStale: lastInspected?.stale, + ownerStaleReasons: lastInspected?.staleReasons, + reclaimedCount, + }); + } return { release: async () => { await releaseHeldLock(normalizedSessionFile, createdHeld); @@ -537,8 +565,10 @@ export async function acquireSessionWriteLock(params: { : [...inspected.staleReasons, "orphan-self-pid"], } : inspected; + lastInspected = reclaimDetails; if (await shouldReclaimContendedLockFile(lockPath, reclaimDetails, staleMs, nowMs)) { await fs.rm(lockPath, { force: true }); + reclaimedCount += 1; continue; } @@ -548,7 +578,20 @@ export async function acquireSessionWriteLock(params: { } const payload = await readLockPayload(lockPath); + const inspected = inspectLockPayload(payload, staleMs, Date.now()); const owner = typeof payload?.pid === "number" ? `pid=${payload.pid}` : "unknown"; + logSessionLockStats({ + action: "timeout", + lockPath, + waitMs: Date.now() - startedAt, + attempts: attempt, + timeoutMs, + ownerPid: inspected.pid, + ownerAgeMs: inspected.ageMs, + ownerStale: inspected.stale, + ownerStaleReasons: inspected.staleReasons, + reclaimedCount, + }); throw new Error(`session file locked (timeout ${timeoutMs}ms): ${owner} ${lockPath}`); } diff --git a/src/infra/diagnostic-events.ts b/src/infra/diagnostic-events.ts index 08dafdf26333c..0a4a3e889c1af 100644 --- a/src/infra/diagnostic-events.ts +++ b/src/infra/diagnostic-events.ts @@ -117,6 +117,7 @@ export type DiagnosticLaneDequeueEvent = DiagnosticBaseEvent & { export type DiagnosticLaneSnapshotEvent = DiagnosticBaseEvent & { type: "queue.lane.snapshot"; lane: string; + laneType?: "main" | "session" | "other"; active: number; queued: number; peakActiveLast30s: number; @@ -125,6 +126,21 @@ export type DiagnosticLaneSnapshotEvent = DiagnosticBaseEvent & { oldestQueuedMs: number; }; +export type DiagnosticSessionLockStatsEvent = DiagnosticBaseEvent & { + type: "session.lock.stats"; + action: "acquired_after_wait" | "timeout" | "released_after_long_hold"; + lockPath: string; + waitMs?: number; + holdMs?: number; + attempts?: number; + timeoutMs?: number; + ownerPid?: number | null; + ownerAgeMs?: number | null; + ownerStale?: boolean; + ownerStaleReasons?: string[]; + reclaimedCount?: number; +}; + export type DiagnosticRunAttemptEvent = DiagnosticBaseEvent & { type: "run.attempt"; sessionKey?: string; @@ -170,6 +186,7 @@ export type DiagnosticEventPayload = | DiagnosticLaneEnqueueEvent | DiagnosticLaneDequeueEvent | DiagnosticLaneSnapshotEvent + | DiagnosticSessionLockStatsEvent | DiagnosticRunAttemptEvent | DiagnosticHeartbeatEvent | DiagnosticToolLoopEvent; diff --git a/src/logging/diagnostic.ts b/src/logging/diagnostic.ts index f8942dc9679ed..88323bfe46b24 100644 --- a/src/logging/diagnostic.ts +++ b/src/logging/diagnostic.ts @@ -268,6 +268,7 @@ export function logLaneDequeue(lane: string, waitMs: number, queueSize: number) export function logLaneSnapshot( lane: string, stats: { + laneType?: "main" | "session" | "other"; active: number; queued: number; peakActiveLast30s: number; @@ -276,12 +277,14 @@ export function logLaneSnapshot( oldestQueuedMs: number; }, ) { + const laneType = stats.laneType ? ` laneType=${stats.laneType}` : ""; diag.info( - `lane stats: lane=${lane} active=${stats.active} peakActiveLast30s=${stats.peakActiveLast30s} queued=${stats.queued} maxConcurrent=${stats.maxConcurrent} depth=${stats.depth} oldestQueuedMs=${stats.oldestQueuedMs}`, + `lane stats: lane=${lane}${laneType} active=${stats.active} peakActiveLast30s=${stats.peakActiveLast30s} queued=${stats.queued} maxConcurrent=${stats.maxConcurrent} depth=${stats.depth} oldestQueuedMs=${stats.oldestQueuedMs}`, ); emitDiagnosticEvent({ type: "queue.lane.snapshot", lane, + laneType: stats.laneType, active: stats.active, queued: stats.queued, peakActiveLast30s: stats.peakActiveLast30s, @@ -292,6 +295,71 @@ export function logLaneSnapshot( markActivity(); } +export function logSessionLockStats(params: { + action: "acquired_after_wait" | "timeout" | "released_after_long_hold"; + lockPath: string; + waitMs?: number; + holdMs?: number; + attempts?: number; + timeoutMs?: number; + ownerPid?: number | null; + ownerAgeMs?: number | null; + ownerStale?: boolean; + ownerStaleReasons?: string[]; + reclaimedCount?: number; +}) { + const parts = [`session lock stats: action=${params.action} lockPath=${params.lockPath}`]; + if (params.waitMs !== undefined) { + parts.push(`waitMs=${params.waitMs}`); + } + if (params.holdMs !== undefined) { + parts.push(`holdMs=${params.holdMs}`); + } + if (params.attempts !== undefined) { + parts.push(`attempts=${params.attempts}`); + } + if (params.timeoutMs !== undefined) { + parts.push(`timeoutMs=${params.timeoutMs}`); + } + if (params.ownerPid !== undefined) { + parts.push(`ownerPid=${params.ownerPid ?? "unknown"}`); + } + if (params.ownerAgeMs !== undefined) { + parts.push(`ownerAgeMs=${params.ownerAgeMs ?? "unknown"}`); + } + if (params.ownerStale !== undefined) { + parts.push(`ownerStale=${params.ownerStale ? "true" : "false"}`); + } + if (params.ownerStaleReasons && params.ownerStaleReasons.length > 0) { + parts.push(`ownerStaleReasons=${params.ownerStaleReasons.join(",")}`); + } + if (params.reclaimedCount !== undefined) { + parts.push(`reclaimedCount=${params.reclaimedCount}`); + } + const message = parts.join(" "); + + if (params.action === "timeout") { + diag.error(message); + } else { + diag.warn(message); + } + emitDiagnosticEvent({ + type: "session.lock.stats", + action: params.action, + lockPath: params.lockPath, + waitMs: params.waitMs, + holdMs: params.holdMs, + attempts: params.attempts, + timeoutMs: params.timeoutMs, + ownerPid: params.ownerPid, + ownerAgeMs: params.ownerAgeMs, + ownerStale: params.ownerStale, + ownerStaleReasons: params.ownerStaleReasons, + reclaimedCount: params.reclaimedCount, + }); + markActivity(); +} + export function logRunAttempt(params: SessionRef & { runId: string; attempt: number }) { diag.debug( `run attempt: sessionId=${params.sessionId ?? "unknown"} sessionKey=${ diff --git a/src/process/command-queue.test.ts b/src/process/command-queue.test.ts index 1f245c1f18ef8..23762a5cb1d34 100644 --- a/src/process/command-queue.test.ts +++ b/src/process/command-queue.test.ts @@ -142,6 +142,7 @@ describe("command queue", () => { await Promise.all([firstTask, secondTask, thirdTask]); expect(diagnosticMocks.logLaneSnapshot).toHaveBeenCalledWith("main", { + laneType: "main", active: 2, queued: 1, peakActiveLast30s: 2, @@ -173,6 +174,7 @@ describe("command queue", () => { await vi.advanceTimersByTimeAsync(30_000); expect(diagnosticMocks.logLaneSnapshot).toHaveBeenCalledWith("main", { + laneType: "main", active: 1, queued: 0, peakActiveLast30s: 1, @@ -221,6 +223,7 @@ describe("command queue", () => { await vi.advanceTimersByTimeAsync(30_000); expect(diagnosticMocks.logLaneSnapshot).toHaveBeenCalledWith("main", { + laneType: "main", active: 0, queued: 0, peakActiveLast30s: 3, @@ -234,6 +237,46 @@ describe("command queue", () => { } }); + it("includes session lanes in periodic snapshots", async () => { + vi.useFakeTimers(); + try { + setCommandLaneConcurrency("session:agent:test:sess_123", 1); + + const first = createDeferred(); + const second = createDeferred(); + + const firstTask = enqueueCommandInLane("session:agent:test:sess_123", async () => { + await first.promise; + }); + const secondTask = enqueueCommandInLane("session:agent:test:sess_123", async () => { + await second.promise; + }); + + await vi.waitFor(() => { + expect(getQueueSize("session:agent:test:sess_123")).toBe(2); + }); + + await vi.advanceTimersByTimeAsync(30_000); + + expect(diagnosticMocks.logLaneSnapshot).toHaveBeenCalledWith("session:agent:test:sess_123", { + laneType: "session", + active: 1, + queued: 1, + peakActiveLast30s: 1, + maxConcurrent: 1, + depth: 2, + oldestQueuedMs: expect.any(Number), + }); + + first.resolve(); + second.resolve(); + await Promise.all([firstTask, secondTask]); + } finally { + resetCommandQueueStateForTest(); + vi.useRealTimers(); + } + }); + it("invokes onWait callback when a task waits past the threshold", async () => { let waited: number | null = null; let queuedAhead: number | null = null; diff --git a/src/process/command-queue.ts b/src/process/command-queue.ts index 550f41daba25f..99f8ea45ff705 100644 --- a/src/process/command-queue.ts +++ b/src/process/command-queue.ts @@ -59,36 +59,85 @@ const lanes = new Map(); let nextTaskId = 1; let laneStatsTimer: ReturnType | undefined; +const LANE_STATS_INTERVAL_MS = 30_000; +const MAX_LANE_SNAPSHOT_LOGS = 40; + +function resolveLaneType(lane: string): "main" | "session" | "other" { + if (lane === "main") { + return "main"; + } + if (lane.startsWith("session:")) { + return "session"; + } + return "other"; +} + function getOldestQueuedMs(state: LaneState): number { const oldest = state.queue[0]; return oldest ? Date.now() - oldest.enqueuedAt : 0; } +function getLaneDepth(state: LaneState): number { + return state.queue.length + state.activeTaskIds.size; +} + function startLaneStatsLogger(): void { if (laneStatsTimer) { return; } laneStatsTimer = setInterval(() => { - const state = lanes.get(CommandLane.Main); - if (!state) { - return; + const snapshots = Array.from(lanes.values()) + .map((state) => { + const active = state.activeTaskIds.size; + const queued = state.queue.length; + const peakActiveLast30s = Math.max(state.peakActiveSinceLastSnapshot, active); + return { + state, + laneType: resolveLaneType(state.lane), + active, + queued, + peakActiveLast30s, + maxConcurrent: state.maxConcurrent, + depth: active + queued, + oldestQueuedMs: getOldestQueuedMs(state), + }; + }) + .filter( + (snapshot) => snapshot.active > 0 || snapshot.queued > 0 || snapshot.peakActiveLast30s > 0, + ) + .toSorted((a, b) => { + if (b.queued !== a.queued) { + return b.queued - a.queued; + } + if (b.oldestQueuedMs !== a.oldestQueuedMs) { + return b.oldestQueuedMs - a.oldestQueuedMs; + } + if (b.peakActiveLast30s !== a.peakActiveLast30s) { + return b.peakActiveLast30s - a.peakActiveLast30s; + } + return a.state.lane.localeCompare(b.state.lane); + }) + .slice(0, MAX_LANE_SNAPSHOT_LOGS); + + for (const snapshot of snapshots) { + logLaneSnapshot(snapshot.state.lane, { + laneType: snapshot.laneType, + active: snapshot.active, + queued: snapshot.queued, + peakActiveLast30s: snapshot.peakActiveLast30s, + maxConcurrent: snapshot.maxConcurrent, + depth: snapshot.depth, + oldestQueuedMs: snapshot.oldestQueuedMs, + }); + snapshot.state.peakActiveSinceLastSnapshot = snapshot.active; } - const active = state.activeTaskIds.size; - const queued = state.queue.length; - const peakActiveLast30s = Math.max(state.peakActiveSinceLastSnapshot, active); - if (active === 0 && queued === 0 && peakActiveLast30s === 0) { - return; + + for (const state of lanes.values()) { + if (!snapshots.some((snapshot) => snapshot.state === state)) { + state.peakActiveSinceLastSnapshot = state.activeTaskIds.size; + } } - logLaneSnapshot(state.lane, { - active, - queued, - peakActiveLast30s, - maxConcurrent: state.maxConcurrent, - depth: active + queued, - oldestQueuedMs: getOldestQueuedMs(state), - }); - state.peakActiveSinceLastSnapshot = active; - }, 30_000); + }, LANE_STATS_INTERVAL_MS); laneStatsTimer.unref?.(); } @@ -215,6 +264,7 @@ export function enqueueCommandInLane( if (gatewayDraining) { return Promise.reject(new GatewayDrainingError()); } + startLaneStatsLogger(); const cleaned = lane.trim() || CommandLane.Main; const warnAfterMs = opts?.warnAfterMs ?? 2_000; const state = getLaneState(cleaned); @@ -227,7 +277,7 @@ export function enqueueCommandInLane( warnAfterMs, onWait: opts?.onWait, }); - logLaneEnqueue(cleaned, state.queue.length + state.activeTaskIds.size); + logLaneEnqueue(cleaned, getLaneDepth(state)); drainLane(cleaned); }); }