Merge pull request #1674 from kaitranntt/kai/fix/1670-concurrent-account-locks

fix(cliproxy): retry contended state locks

docs/reports/hardening-inventory.json

@@ -1,10 +1,10 @@
 {
   "scope": "src/**/*.{ts,tsx,js,jsx,mjs,cjs}",
   "syncFs": {
-    "totalOccurrences": 2378,
-    "filesAffected": 254,
-    "hotpathOccurrences": 1102,
-    "hotpathFilesAffected": 150,
+    "totalOccurrences": 2379,
+    "filesAffected": 255,
+    "hotpathOccurrences": 1103,
+    "hotpathFilesAffected": 151,
     "topHotpathFiles": [
       {
         "file": "src/utils/browser/mcp-installer.ts",
@@ -573,8 +573,8 @@
     },
     "loggerCoverage": {
       "filesWithCreateLogger": 65,
-      "totalSourceFiles": 752,
-      "coverageRatio": 0.0864,
+      "totalSourceFiles": 753,
+      "coverageRatio": 0.0863,
       "subdomainsWithZeroCreateLogger": [
         "api",
         "bin",
@@ -605,7 +605,7 @@
         },
         {
           "subdomain": "utils",
-          "count": 88,
+          "count": 89,
           "withLogger": 1
         },
         {

docs/reports/hardening-inventory.md

@@ -6,10 +6,10 @@ Scope: `src/**/*.{ts,tsx,js,jsx,mjs,cjs}`
 
 | Metric | Value |
 |---|---:|
-| Sync fs occurrences (all) | 2378 |
-| Sync fs files affected (all) | 254 |
-| Sync fs occurrences (runtime hotpaths) | 1102 |
-| Sync fs files affected (runtime hotpaths) | 150 |
+| Sync fs occurrences (all) | 2379 |
+| Sync fs files affected (all) | 255 |
+| Sync fs occurrences (runtime hotpaths) | 1103 |
+| Sync fs files affected (runtime hotpaths) | 151 |
 | Legacy shim markers | 456 |
 | Legacy shim files affected | 171 |
 
@@ -60,7 +60,7 @@ Scope: `src/**/*.{ts,tsx,js,jsx,mjs,cjs}`
 | typed-error adoption (P4 locked subdomains) | 91.7% (22/24), target 40% |
 | hotpath console.error/warn occurrences | 266 (571 total, 305 CLI-UX exempt) |
 | hotpath console.error/warn files | 82 |
-| files with createLogger | 65/752 |
+| files with createLogger | 65/753 |
 | subdomains with zero createLogger | 15 (api, bin, channels, cliproxy, cliproxy/accounts, cliproxy/ai-providers, cliproxy/binary, cliproxy/config, cliproxy/management, cliproxy/sync, cliproxy/types, config, dispatcher, shared, types) |
 | files > 400 LOC | 91 |
 | files > 600 LOC | 42 |

src/cliproxy/accounts/registry.ts

@@ -5,11 +5,11 @@
 
 import * as fs from 'fs';
 import * as path from 'path';
-import * as lockfile from 'proper-lockfile';
 import { CLIProxyProvider } from '../types';
 import { PROVIDER_CAPABILITIES } from '../provider-capabilities';
 import { PROVIDER_TYPE_VALUES } from '../auth/auth-types';
 import { getAuthDir, getCliproxyDir } from '../config/config-generator';
+import { withSyncLockRetry } from '../../utils/sync-lock-retry';
 import { AccountsRegistry, AccountInfo, PROVIDERS_WITHOUT_EMAIL, DrainOrderConfig } from './types';
 import {
   getAccountsRegistryPath,
@@ -338,24 +338,24 @@ function recoverAccountsRegistryFromCorruption(registryPath: string): AccountsRe
 
 function withAccountsRegistryLock<T>(callback: () => T): T {
   const lockTarget = getCliproxyDir();
-  let release: (() => void) | undefined;
 
   if (!fs.existsSync(lockTarget)) {
     fs.mkdirSync(lockTarget, { recursive: true, mode: 0o700 });
   }
 
-  try {
-    release = lockfile.lockSync(lockTarget, { stale: 10000 }) as () => void;
-    return callback();
-  } finally {
-    if (release) {
-      try {
-        release();
-      } catch {
-        // Best-effort release
-      }
-    }
-  }
+  // proper-lockfile rejects built-in retries in synchronous mode.
+  // Keep this API sync while waiting for short-lived concurrent launches.
+  // Preserve the existing stale-lock window.
+  // Retry only contention failures.
+  // Wait in short intervals so released locks are picked up promptly.
+  // Bound the wait so a genuinely stuck process still produces an error.
+  // The helper adds recovery guidance when that timeout is reached.
+  return withSyncLockRetry(lockTarget, callback, {
+    description: 'CLIProxy account registry lock',
+    staleMs: 10000,
+    retryDelayMs: 200,
+    retryTimeoutMs: 10000,
+  });
 }
 
 function readAccountsRegistryFromDisk(): AccountsRegistry {

src/cliproxy/session-tracker.ts

@@ -17,10 +17,10 @@
 import * as fs from 'fs';
 import * as path from 'path';
 import * as crypto from 'crypto';
-import * as lockfile from 'proper-lockfile';
 import { getCliproxyDir } from './config/config-generator';
 import { getPortProcess, isCLIProxyProcess } from '../utils/port-utils';
 import { CLIPROXY_DEFAULT_PORT } from './config/config-generator';
+import { withSyncLockRetry } from '../utils/sync-lock-retry';
 
 /** Session lock file structure */
 interface SessionLock {
@@ -61,22 +61,9 @@ function ensureCliproxyDir(): string {
 
 function withSessionTrackerLock<T>(fn: () => T): T {
   const dir = ensureCliproxyDir();
-  let release: (() => void) | undefined;
-
-  try {
-    release = lockfile.lockSync(dir, {
-      stale: 10000,
-    }) as () => void;
-    return fn();
-  } finally {
-    if (release) {
-      try {
-        release();
-      } catch {
-        // Best-effort release
-      }
-    }
-  }
+  return withSyncLockRetry(dir, fn, {
+    description: 'CLIProxy session tracker lock',
+  });
 }
 
 /** Get path to session lock file (default port) - kept for future use */

src/utils/sync-lock-retry.ts

@@ -0,0 +1,115 @@
+import * as fs from 'fs';
+import { performance } from 'perf_hooks';
+import * as lockfile from 'proper-lockfile';
+
+const DEFAULT_STALE_MS = 10000;
+const DEFAULT_RETRY_DELAY_MS = 200;
+const DEFAULT_RETRY_TIMEOUT_MS = 10000;
+
+interface SyncLockRetryOptions {
+  description: string;
+  staleMs?: number;
+  retryDelayMs?: number;
+  retryTimeoutMs?: number;
+}
+
+export function withSyncLockRetry<T>(
+  lockTarget: string,
+  callback: () => T,
+  options: SyncLockRetryOptions
+): T {
+  const staleMs = options.staleMs ?? DEFAULT_STALE_MS;
+  const retryDelayMs = options.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS;
+  const retryTimeoutMs = options.retryTimeoutMs ?? DEFAULT_RETRY_TIMEOUT_MS;
+  const startedAt = performance.now();
+  let release: (() => void) | undefined;
+  let contendedLockPath: string | undefined;
+  let lastContentionError: unknown;
+
+  for (;;) {
+    if (contendedLockPath && fs.existsSync(contendedLockPath)) {
+      sleepBeforeRetry(
+        startedAt,
+        retryDelayMs,
+        retryTimeoutMs,
+        options.description,
+        lastContentionError
+      );
+      continue;
+    }
+
+    if (contendedLockPath && performance.now() - startedAt >= retryTimeoutMs) {
+      throw buildLockTimeoutError(options.description, retryTimeoutMs, lastContentionError);
+    }
+
+    try {
+      release = lockfile.lockSync(lockTarget, { stale: staleMs }) as () => void;
+      break;
+    } catch (error) {
+      if (!isLockContentionError(error)) {
+        throw error;
+      }
+      contendedLockPath ??= `${fs.realpathSync(lockTarget)}.lock`;
+      lastContentionError = error;
+      sleepBeforeRetry(startedAt, retryDelayMs, retryTimeoutMs, options.description, error);
+    }
+  }
+
+  try {
+    return callback();
+  } finally {
+    try {
+      release();
+    } catch {
+      // Best-effort release.
+    }
+  }
+}
+
+function isLockContentionError(error: unknown): boolean {
+  const code = (error as NodeJS.ErrnoException | undefined)?.code;
+  return code === 'ELOCKED' || code === 'ENOTACQUIRED';
+}
+
+function sleepSync(ms: number): void {
+  if (ms <= 0) {
+    return;
+  }
+
+  try {
+    Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms);
+  } catch {
+    const deadline = performance.now() + ms;
+    while (performance.now() < deadline) {
+      // Fall back for runtimes without Atomics.wait.
+    }
+  }
+}
+
+function sleepBeforeRetry(
+  startedAt: number,
+  retryDelayMs: number,
+  retryTimeoutMs: number,
+  description: string,
+  cause: unknown
+): void {
+  const elapsedMs = performance.now() - startedAt;
+  if (elapsedMs >= retryTimeoutMs) {
+    throw buildLockTimeoutError(description, retryTimeoutMs, cause);
+  }
+  sleepSync(Math.min(retryDelayMs, retryTimeoutMs - elapsedMs));
+}
+
+function buildLockTimeoutError(
+  description: string,
+  timeoutMs: number,
+  cause: unknown
+): NodeJS.ErrnoException {
+  const causeCode = (cause as NodeJS.ErrnoException | undefined)?.code;
+  const error = new Error(
+    `Failed to acquire ${description} after ${timeoutMs}ms; another CCS process may still be updating CLIProxy state. Wait for it to finish, then retry.`
+  ) as NodeJS.ErrnoException & { cause?: unknown };
+  error.code = causeCode === 'ENOTACQUIRED' ? 'ENOTACQUIRED' : 'ELOCKED';
+  error.cause = cause;
+  return error;
+}

tests/unit/cliproxy/concurrent-state-locks.test.ts

@@ -0,0 +1,259 @@
+import { afterEach, beforeEach, describe, expect, it } from 'bun:test';
+import { spawn, type ChildProcess } from 'child_process';
+import * as fs from 'fs';
+import * as os from 'os';
+import * as path from 'path';
+import { pathToFileURL } from 'url';
+import { loadAccountsRegistry } from '../../../src/cliproxy/accounts/registry';
+import { registerSession } from '../../../src/cliproxy/session-tracker';
+import { withSyncLockRetry } from '../../../src/utils/sync-lock-retry';
+
+const ORIGINAL_CCS_HOME = process.env.CCS_HOME;
+const LOCK_HOLD_MS = 400;
+
+let tempDir: string;
+let cliproxyDir: string;
+
+beforeEach(() => {
+  tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ccs-concurrent-state-locks-'));
+  process.env.CCS_HOME = tempDir;
+  cliproxyDir = path.join(tempDir, '.ccs', 'cliproxy');
+  fs.mkdirSync(cliproxyDir, { recursive: true, mode: 0o700 });
+});
+
+afterEach(() => {
+  if (ORIGINAL_CCS_HOME === undefined) {
+    delete process.env.CCS_HOME;
+  } else {
+    process.env.CCS_HOME = ORIGINAL_CCS_HOME;
+  }
+  fs.rmSync(tempDir, { recursive: true, force: true });
+});
+
+describe('CLIProxy state locks', () => {
+  it('waits for a contended account registry lock before loading', async () => {
+    const holder = await holdLock(cliproxyDir);
+
+    try {
+      expect(loadAccountsRegistry()).toEqual({ version: 1, providers: {} });
+    } finally {
+      await stopLockHolder(holder);
+    }
+  });
+
+  it('waits for a contended session tracker lock before registering', async () => {
+    const holder = await holdLock(cliproxyDir);
+
+    try {
+      const sessionId = registerSession(8317, process.pid);
+      const sessionState = JSON.parse(
+        fs.readFileSync(path.join(cliproxyDir, 'sessions.json'), 'utf8')
+      ) as { sessions: string[] };
+
+      expect(sessionState.sessions).toContain(sessionId);
+    } finally {
+      await stopLockHolder(holder);
+    }
+  });
+
+  it('preserves stale lock recovery for account registry access', () => {
+    const staleLockPath = `${cliproxyDir}.lock`;
+    fs.mkdirSync(staleLockPath);
+    const staleTime = new Date(Date.now() - 15000);
+    fs.utimesSync(staleLockPath, staleTime, staleTime);
+
+    expect(loadAccountsRegistry()).toEqual({ version: 1, providers: {} });
+    expect(fs.existsSync(staleLockPath)).toBe(false);
+  });
+
+  it('gives actionable guidance when contention outlasts the bound', async () => {
+    const holder = await holdLock(cliproxyDir);
+
+    try {
+      expect(() =>
+        withSyncLockRetry(cliproxyDir, () => undefined, {
+          description: 'test CLIProxy state lock',
+          retryTimeoutMs: 0,
+        })
+      ).toThrow(
+        'Failed to acquire test CLIProxy state lock after 0ms; another CCS process may still be updating CLIProxy state. Wait for it to finish, then retry.'
+      );
+    } finally {
+      await stopLockHolder(holder);
+    }
+  });
+
+  it('does not steal a fresh lock that ages past stale while its holder remains active', async () => {
+    const holder = await holdRawLock(cliproxyDir, 2800);
+    let callbackRan = false;
+
+    try {
+      expect(() =>
+        withSyncLockRetry(
+          cliproxyDir,
+          () => {
+            callbackRan = true;
+          },
+          {
+            description: 'test active CLIProxy state lock',
+            staleMs: 2000,
+            retryDelayMs: 50,
+            retryTimeoutMs: 2300,
+          }
+        )
+      ).toThrow('Failed to acquire test active CLIProxy state lock after 2300ms');
+      expect(callbackRan).toBe(false);
+      expect(holder.exitCode).toBeNull();
+    } finally {
+      await stopLockHolder(holder);
+    }
+  });
+
+  it('preserves all concurrent session registrations', async () => {
+    const workerCount = 6;
+    const gatePath = path.join(tempDir, 'session-writer-gate');
+    const workerScript = path.join(tempDir, 'session-writer.ts');
+    const sessionTrackerUrl = pathToFileURL(
+      path.join(process.cwd(), 'src/cliproxy/session-tracker.ts')
+    ).href;
+    fs.writeFileSync(
+      workerScript,
+      `
+import * as fs from 'fs';
+import { registerSession } from ${JSON.stringify(sessionTrackerUrl)};
+
+const gatePath = process.argv[2];
+const readyPath = process.argv[3];
+const proxyPid = Number(process.argv[4]);
+fs.writeFileSync(readyPath, String(process.pid));
+while (!fs.existsSync(gatePath)) {
+  await new Promise((resolve) => setTimeout(resolve, 5));
+}
+registerSession(8317, proxyPid);
+`,
+      'utf8'
+    );
+
+    const workers = Array.from({ length: workerCount }, (_, index) => {
+      const readyPath = path.join(tempDir, `session-writer-ready-${index}`);
+      return {
+        readyPath,
+        child: spawn(process.execPath, [workerScript, gatePath, readyPath, String(process.pid)], {
+          cwd: process.cwd(),
+          env: { ...process.env, CCS_HOME: tempDir },
+          stdio: ['ignore', 'ignore', 'pipe'],
+        }),
+      };
+    });
+
+    try {
+      await Promise.all(workers.map(({ readyPath, child }) => waitForFile(readyPath, child)));
+      fs.writeFileSync(gatePath, 'go');
+      await Promise.all(workers.map(({ child }) => waitForSuccessfulExit(child)));
+
+      const sessionState = JSON.parse(
+        fs.readFileSync(path.join(cliproxyDir, 'sessions.json'), 'utf8')
+      ) as { sessions: string[] };
+      expect(sessionState.sessions).toHaveLength(workerCount);
+      expect(new Set(sessionState.sessions).size).toBe(workerCount);
+    } finally {
+      await Promise.all(workers.map(({ child }) => stopLockHolder(child)));
+    }
+  });
+});
+
+async function holdLock(lockTarget: string): Promise<ChildProcess> {
+  const readyPath = path.join(tempDir, `holder-ready-${Date.now()}-${Math.random()}`);
+  const holderScript = path.join(tempDir, `hold-lock-${Date.now()}-${Math.random()}.cjs`);
+  fs.writeFileSync(
+    holderScript,
+    `
+const fs = require('fs');
+const lockfile = require(process.argv[5]);
+const release = lockfile.lockSync(process.argv[2], { stale: 10000 });
+fs.writeFileSync(process.argv[3], String(process.pid));
+setTimeout(() => {
+  release();
+  process.exit(0);
+}, Number(process.argv[4]));
+setTimeout(() => process.exit(2), 5000);
+process.on('SIGTERM', () => {
+  try { release(); } finally { process.exit(0); }
+});
+`,
+    'utf8'
+  );
+
+  const child = spawn(
+    process.execPath,
+    [holderScript, lockTarget, readyPath, String(LOCK_HOLD_MS), require.resolve('proper-lockfile')],
+    {
+      cwd: process.cwd(),
+      stdio: ['ignore', 'ignore', 'pipe'],
+    }
+  );
+  await waitForFile(readyPath, child);
+  return child;
+}
+
+async function holdRawLock(lockTarget: string, holdMs: number): Promise<ChildProcess> {
+  const readyPath = path.join(tempDir, `raw-holder-ready-${Date.now()}-${Math.random()}`);
+  const holderScript = path.join(tempDir, `hold-raw-lock-${Date.now()}-${Math.random()}.cjs`);
+  fs.writeFileSync(
+    holderScript,
+    `
+const fs = require('fs');
+const lockPath = process.argv[2] + '.lock';
+fs.mkdirSync(lockPath);
+fs.writeFileSync(process.argv[3], String(process.pid));
+setTimeout(() => {
+  try { fs.rmdirSync(lockPath); } catch {}
+  process.exit(0);
+}, Number(process.argv[4]));
+setTimeout(() => process.exit(2), 5000);
+process.on('SIGTERM', () => {
+  try { fs.rmdirSync(lockPath); } catch {}
+  process.exit(0);
+});
+`,
+    'utf8'
+  );
+
+  const child = spawn(process.execPath, [holderScript, lockTarget, readyPath, String(holdMs)], {
+    cwd: process.cwd(),
+    stdio: ['ignore', 'ignore', 'pipe'],
+  });
+  await waitForFile(readyPath, child);
+  return child;
+}
+
+async function waitForFile(filePath: string, child: ChildProcess, timeoutMs = 2000): Promise<void> {
+  const startedAt = Date.now();
+  while (!fs.existsSync(filePath)) {
+    if (child.exitCode !== null || child.signalCode !== null) {
+      throw new Error(`Lock holder exited before acquiring lock (code ${child.exitCode})`);
+    }
+    if (Date.now() - startedAt >= timeoutMs) {
+      throw new Error(`Timed out waiting for lock holder at ${filePath}`);
+    }
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+}
+
+async function stopLockHolder(child: ChildProcess): Promise<void> {
+  if (child.exitCode === null && child.signalCode === null) {
+    child.kill();
+  }
+  if (child.exitCode === null && child.signalCode === null) {
+    await new Promise<void>((resolve) => child.once('exit', () => resolve()));
+  }
+}
+
+async function waitForSuccessfulExit(child: ChildProcess): Promise<void> {
+  if (child.exitCode === null && child.signalCode === null) {
+    await new Promise<void>((resolve) => child.once('exit', () => resolve()));
+  }
+  if (child.exitCode !== 0) {
+    throw new Error(`Concurrent session writer exited with code ${child.exitCode}`);
+  }
+}