fix(oauth): retry refresh intent marker contention

Author: luvs01

src/oauth/index.ts

@@ -1,7 +1,7 @@
 import type { KiroOAuthMetadata, OAuthController, OAuthCredentials } from "./types";
 import { parseCallbackInput } from "./callback-server";
 import type { OcxConfig, OcxProviderConfig, RefreshPolicy } from "../types";
-import { loadConfig, resolveEnvValue, saveConfig } from "../config";
+import { ConfigMutationLockError, loadConfig, resolveEnvValue, saveConfig } from "../config";
 import { maskEmail } from "../lib/privacy";
 import { KiroTokenRefreshError, environmentKiroRoutingMetadata, loginKiro, refreshKiroToken, settleKiroLoginTransaction } from "./kiro";
 import {
@@ -628,27 +628,48 @@ function clearAnthropicRefreshIntentBestEffort(
   }
 }
 
-function clearAnthropicRefreshIntentForKnownFailure(
+const ANTHROPIC_INTENT_MARK_RETRY_DELAYS_MS = [10, 25, 50] as const;
+
+function isConfigMutationLockContention(error: unknown): boolean {
+  if (!(error instanceof ConfigMutationLockError)) return false;
+  const cause = error.cause;
+  const code = cause && typeof cause === "object" && "code" in cause
+    ? String((cause as { code?: unknown }).code)
+    : "";
+  return code === "SQLITE_BUSY" || code === "SQLITE_LOCKED";
+}
+
+async function clearAnthropicRefreshIntentForKnownFailure(
   provider: string,
   accountId: string,
   expected: OAuthRefreshIntent,
   cleanupPending: OAuthRefreshIntentCleanupPending,
   refreshError: unknown,
-): boolean {
+): Promise<boolean> {
   let marked: OAuthRefreshIntent | undefined;
-  try {
-    marked = markOAuthRefreshIntentCleanupPending(
-      provider,
-      accountId,
-      expected,
-      cleanupPending,
-    );
-  } catch (cause) {
-    throw new OAuthRefreshIntentIOError(
-      "mark-cleanup-pending",
-      cause,
-      refreshError,
-    );
+  for (let attempt = 0; attempt <= ANTHROPIC_INTENT_MARK_RETRY_DELAYS_MS.length; attempt += 1) {
+    try {
+      marked = markOAuthRefreshIntentCleanupPending(
+        provider,
+        accountId,
+        expected,
+        cleanupPending,
+      );
+      break;
+    } catch (cause) {
+      const retryDelay = ANTHROPIC_INTENT_MARK_RETRY_DELAYS_MS[attempt];
+      if (!isConfigMutationLockContention(cause) || retryDelay === undefined) {
+        throw new OAuthRefreshIntentIOError(
+          "mark-cleanup-pending",
+          cause,
+          refreshError,
+        );
+      }
+      // The provider has definitively answered, so caller cancellation no longer changes the
+      // settlement obligation. Yield briefly while retaining the per-account refresh lock, then
+      // rerun the existing compare-and-swap marker against current disk state.
+      await Bun.sleep(retryDelay);
+    }
   }
   if (!marked) {
     throw new OAuthRefreshIntentIOError(
@@ -872,7 +893,7 @@ export async function refreshAnthropicAccountWithLock(
         // rotated the token, and replaying it could trip refresh-token-reuse revocation.
         // Those outcomes keep the intent so the guard still refuses a blind replay.
         if ((!refreshMayHaveReachedProvider || definitivelyAnswered(error)) && attemptIntent) {
-          clearAnthropicRefreshIntentForKnownFailure(
+          await clearAnthropicRefreshIntentForKnownFailure(
             provider,
             accountId,
             attemptIntent,

tests/oauth-refresh.test.ts

@@ -641,8 +641,13 @@ describe("oauth refresh hardening", () => {
     const id = getAccountSet("anthropic")!.activeAccountId;
     const credential = getAccountCredential("anthropic", id)!;
     const transient = new AnthropicTokenError("server", 503, undefined);
-    const markerFailure = new Error("intent marker write failed");
+    const markerFailure = new configModule.ConfigMutationLockError(
+      "Could not acquire config mutation transaction",
+      { cause: Object.assign(new Error("intent marker write failed"), { code: "EACCES" }) },
+    );
+    let markerCalls = 0;
     const markerSpy = spyOn(storeModule, "markOAuthRefreshIntentCleanupPending").mockImplementation(() => {
+      markerCalls += 1;
       throw markerFailure;
     });
     try {
@@ -662,6 +667,7 @@ describe("oauth refresh hardening", () => {
         cause: markerFailure,
         refreshError: transient,
       });
+      expect(markerCalls).toBe(1);
       const pending = readOAuthRefreshIntent("anthropic", id);
       expect(pending?.cleanupPending).toBeUndefined();
       expect(getAccountSet("anthropic")!.accounts.find(account => account.id === id)!.needsReauth).toBeUndefined();
@@ -670,6 +676,102 @@ describe("oauth refresh hardening", () => {
     }
   });
 
+  test("transient cleanup-marker lock contention settles a retryable rejection without reauth", async () => {
+    await saveCredential("anthropic", { access: "old", refresh: "rt-old", expires: 1, accountId: "acct" });
+    const id = getAccountSet("anthropic")!.activeAccountId;
+    const credential = getAccountCredential("anthropic", id)!;
+    const transient = new AnthropicTokenError("server", 503, undefined);
+    const realMark = storeModule.markOAuthRefreshIntentCleanupPending;
+    let markerCalls = 0;
+    const markerSpy = spyOn(storeModule, "markOAuthRefreshIntentCleanupPending").mockImplementation((...args) => {
+      markerCalls += 1;
+      if (markerCalls <= 2) {
+        throw new configModule.ConfigMutationLockError(
+          "Config mutation already in progress",
+          { cause: Object.assign(new Error("database is locked"), { code: "SQLITE_BUSY" }) },
+        );
+      }
+      return realMark(...args);
+    });
+    try {
+      let providerCalls = 0;
+      await expect(refreshAnthropicAccountWithLock("anthropic", id, {
+        ...OAUTH_PROVIDERS.anthropic!,
+        refresh: async () => {
+          providerCalls += 1;
+          throw transient;
+        },
+      }, credential)).rejects.toBe(transient);
+
+      expect(providerCalls).toBe(1);
+      expect(markerCalls).toBe(3);
+      expect(readOAuthRefreshIntent("anthropic", id)).toBeUndefined();
+      expect(getAccountSet("anthropic")!.accounts.find(account => account.id === id)!.needsReauth).toBeUndefined();
+
+      await expect(refreshAnthropicAccountWithLock("anthropic", id, {
+        ...OAUTH_PROVIDERS.anthropic!,
+        refresh: async () => ({
+          access: "fresh",
+          refresh: "rt-fresh",
+          expires: Date.now() + 3_600_000,
+        }),
+      }, getAccountCredential("anthropic", id)!)).resolves.toBe("fresh");
+      expect(getAccountCredential("anthropic", id)?.access).toBe("fresh");
+      expect(readOAuthRefreshIntent("anthropic", id)).toBeUndefined();
+    } finally {
+      markerSpy.mockRestore();
+    }
+  });
+
+  test("persistent cleanup-marker lock contention stays bounded and preserves the replay guard", async () => {
+    await saveCredential("anthropic", { access: "old", refresh: "rt-old", expires: 1, accountId: "acct" });
+    const id = getAccountSet("anthropic")!.activeAccountId;
+    const credential = getAccountCredential("anthropic", id)!;
+    const generation = credentialGeneration(credential);
+    const transient = new AnthropicTokenError("server", 503, undefined);
+    let markerCalls = 0;
+    let lastBusy: configModule.ConfigMutationLockError | undefined;
+    const markerSpy = spyOn(storeModule, "markOAuthRefreshIntentCleanupPending").mockImplementation(() => {
+      markerCalls += 1;
+      lastBusy = new configModule.ConfigMutationLockError(
+        "Config mutation already in progress",
+        { cause: Object.assign(new Error("database table is locked"), { code: "SQLITE_LOCKED" }) },
+      );
+      throw lastBusy;
+    });
+    try {
+      let providerCalls = 0;
+      let rejection: unknown;
+      try {
+        await refreshAnthropicAccountWithLock("anthropic", id, {
+          ...OAUTH_PROVIDERS.anthropic!,
+          refresh: async () => {
+            providerCalls += 1;
+            throw transient;
+          },
+        }, credential);
+      } catch (error) {
+        rejection = error;
+      }
+
+      expect(providerCalls).toBe(1);
+      expect(markerCalls).toBe(4);
+      expect(rejection).toBeInstanceOf(OAuthRefreshIntentIOError);
+      expect(rejection).toMatchObject({
+        operation: "mark-cleanup-pending",
+        code: "OAUTH_REFRESH_INTENT_IO",
+        cause: lastBusy,
+        refreshError: transient,
+      });
+      expect(getAccountCredential("anthropic", id)).toEqual(credential);
+      expect(readOAuthRefreshIntent("anthropic", id)).toMatchObject({ generation });
+      expect(readOAuthRefreshIntent("anthropic", id)?.cleanupPending).toBeUndefined();
+      expect(getAccountSet("anthropic")!.accounts.find(account => account.id === id)!.needsReauth).toBeUndefined();
+    } finally {
+      markerSpy.mockRestore();
+    }
+  });
+
   test("a failed definitive-rejection cleanup is retried before the next Anthropic request", async () => {
     await saveCredential("anthropic", { access: "old", refresh: "rt-old", expires: 1, accountId: "acct" });
     const id = getAccountSet("anthropic")!.activeAccountId;