fix(oauth): take over a lapsed refresh lease, share overlapping service-account key reads, and label unsearched accounts

apps/sim/lib/concurrency/leader-lock.test.ts

@@ -119,6 +119,32 @@ describe('withLeaderLock', () => {
     }
   })
 
+  describe('when the follower wait ends with nothing', () => {
+    const waitedOut = {
+      key: 'k',
+      pollIntervalMs: 5,
+      maxWaitMs: 20,
+      onFollower: async () => null,
+    }
+
+    it('leads once a leader that vanished leaves its lease to lapse', async () => {
+      redisConfigMockFns.mockAcquireLock.mockResolvedValueOnce(false)
+      const onLeader = vi.fn(async () => 'took-over')
+      await expect(withLeaderLock<string>({ ...waitedOut, onLeader })).resolves.toBe('took-over')
+      expect(onLeader).toHaveBeenCalledOnce()
+      expect(redisConfigMockFns.mockAcquireLock).toHaveBeenCalledTimes(2)
+      expect(redisConfigMockFns.mockReleaseLock).toHaveBeenCalledOnce()
+    })
+
+    it('returns the follower result when a live holder still has the lease', async () => {
+      redisConfigMockFns.mockAcquireLock.mockResolvedValue(false)
+      const onLeader = vi.fn(async () => 'should-not-run')
+      await expect(withLeaderLock<string>({ ...waitedOut, onLeader })).resolves.toBeNull()
+      expect(onLeader).not.toHaveBeenCalled()
+      expect(redisConfigMockFns.mockAcquireLock).toHaveBeenCalledTimes(2)
+    })
+  })
+
   it('only one of N concurrent callers acquires the lock', async () => {
     // Track which calls won the lock: first one returns true, rest return false.
     let acquired = false

apps/sim/lib/concurrency/leader-lock.ts

@@ -19,6 +19,13 @@ export interface LeaderLockOptions<T> {
   onFollower: () => Promise<T | null>
 }
 
+/**
+ * Runs `onLeader` in the one caller that takes the lock for `key`; every other caller polls
+ * `onFollower` until it reports the leader's result or `maxWaitMs` passes. A follower whose wait
+ * ends with nothing tries the lock once more: a leader whose process died without releasing it
+ * leaves the lease to lapse, and that follower then leads instead of failing. Losing that try
+ * means a live holder has the lease, and the follower returns `null`.
+ */
 export async function withLeaderLock<T>(opts: LeaderLockOptions<T>): Promise<T | null> {
   const {
     key,
@@ -42,7 +49,7 @@ export async function withLeaderLock<T>(opts: LeaderLockOptions<T>): Promise<T |
     return onLeader()
   }
 
-  if (acquired) {
+  const leadHoldingLock = async () => {
     try {
       return await onLeader()
     } finally {
@@ -57,6 +64,8 @@ export async function withLeaderLock<T>(opts: LeaderLockOptions<T>): Promise<T |
     }
   }
 
+  if (acquired) return leadHoldingLock()
+
   const deadline = Date.now() + maxWaitMs
   while (Date.now() < deadline) {
     await sleep(pollIntervalMs)
@@ -68,6 +77,20 @@ export async function withLeaderLock<T>(opts: LeaderLockOptions<T>): Promise<T |
   const lastChance = await onFollower()
   if (lastChance !== null) return lastChance
 
+  let tookOver = false
+  try {
+    tookOver = await acquireLock(key, ownerToken, ttlSec)
+  } catch (error) {
+    logger.warn('Lock takeover failed after the follower wait', {
+      key,
+      error: toError(error).message,
+    })
+  }
+  if (tookOver) {
+    logger.warn('Follower took over a lease its leader left to lapse', { key, maxWaitMs })
+    return leadHoldingLock()
+  }
+
   logger.warn('Follower timed out waiting for leader', { key, maxWaitMs })
   return null
 }

apps/sim/lib/core/utils/concurrency.test.ts

@@ -0,0 +1,50 @@
+import { flushMacrotask } from '@sim/testing/helpers/async'
+import { createDeferred } from '@sim/testing/helpers/deferred'
+import { describe, expect, it } from 'vitest'
+import { createConcurrencyLimit } from '@/lib/core/utils/concurrency'
+
+describe('createConcurrencyLimit', () => {
+  it('runs at most the limit at once and starts waiting tasks in call order', async () => {
+    const limit = createConcurrencyLimit(2)
+    const gates = Array.from({ length: 5 }, () => createDeferred<void>())
+    const started: number[] = []
+    let active = 0
+    let peak = 0
+    const runs = gates.map((gate, index) =>
+      limit(async () => {
+        started.push(index)
+        peak = Math.max(peak, ++active)
+        await gate.promise
+        active--
+        return index
+      })
+    )
+    await flushMacrotask()
+    expect(started).toEqual([0, 1])
+    for (const index of [1, 0, 4, 3, 2]) {
+      gates[index].resolve()
+      await flushMacrotask()
+    }
+    await expect(Promise.all(runs)).resolves.toEqual([0, 1, 2, 3, 4])
+    expect(started).toEqual([0, 1, 2, 3, 4])
+    expect(peak).toBe(2)
+  })
+
+  it('releases the slot of a task that throws synchronously or rejects', async () => {
+    const limit = createConcurrencyLimit(1)
+    const error = new Error('task failed')
+    const throwsSynchronously = (): Promise<never> => {
+      throw error
+    }
+    const thrown = limit(throwsSynchronously)
+    const rejected = limit(async () => Promise.reject(error))
+    let next: string | undefined
+    void limit(async () => 'next').then((value) => {
+      next = value
+    })
+    await expect(thrown).rejects.toBe(error)
+    await expect(rejected).rejects.toBe(error)
+    await flushMacrotask()
+    expect(next).toBe('next')
+  })
+})

apps/sim/lib/knowledge/search/diagnostics.ts

@@ -148,6 +148,8 @@ const FAILURE_REASONS = [
   'mcp_payload_too_large',
   'mcp_payload_unreadable',
   'mcp_account_required',
+  /** The account never got to search: no slot freed with enough of the search deadline left. */
+  'not_started',
 ] as const
 export type SearchFailureReason = (typeof FAILURE_REASONS)[number]