Function withLeaderLock

apps/sim/lib/concurrency/leader-lock.ts:22–73

(opts: LeaderLockOptions<T>)

}

export async function withLeaderLock<T>(opts: LeaderLockOptions<T>): Promise<T | null> {
  const {
    key,
    ttlSec = DEFAULT_TTL_SEC,
    pollIntervalMs = DEFAULT_POLL_INTERVAL_MS,
    maxWaitMs = DEFAULT_MAX_WAIT_MS,
    onLeader,
    onFollower,
  } = opts

  const ownerToken = generateShortId()

  let acquired = false
  try {
    acquired = await acquireLock(key, ownerToken, ttlSec)
  } catch (error) {
    logger.warn('Lock acquisition failed; running leader path uncoordinated', {
      key,
      error: toError(error).message,
    })
    return onLeader()
  }

  if (acquired) {
    try {
      return await onLeader()
    } finally {
      try {
        await releaseLock(key, ownerToken)
      } catch (error) {
        logger.warn('Lock release failed (will expire via TTL)', {
          key,
          error: toError(error).message,
        })
      }
    }
  }

  const deadline = Date.now() + maxWaitMs
  while (Date.now() < deadline) {
    await sleep(pollIntervalMs)
    const value = await onFollower()
    if (value !== null) return value
  }

  // The leader may have persisted between our final poll and now; one last check.
  const lastChance = await onFollower()
  if (lastChance !== null) return lastChance

  logger.warn('Follower timed out waiting for leader', { key, maxWaitMs })
  return null
}