fix(credentials): pace throttled and Vanta token retries and replay remembered failures without the lock

- A 429 that names no Retry-After is replayed for at least a minute, doubling
  up to the five-minute cap, instead of the 5 s outage curve.
- The outage backoff never undercuts Entra's 5 s first-retry floor through jitter.
- Vanta asks for a token at most once per 15 s while exchanges fail (no
  in-request retry, replay floored), within its documented 5 tokens a minute,
  through a new optional minExchangeIntervalMs on the coordinated resolver.
- A remembered failure and the still-valid fallback are served from committed
  rows before the application lock is taken, so a refresh outage no longer
  serializes every resolution behind the lock.
- retryAfterMs now only carries a provider's Retry-After; a replayed memo
  reports its remainder as the additive replayForMs, which the per-instance
  layer honors exactly.
- One decode/verify path for stored tokens; the stored-failure permanence
  check uses isPermanentTokenFailureCode instead of building an error.

Author: waleedlatif1

apps/sim/lib/credentials/client-credential-accounts/vanta-token.integration.ts

@@ -142,6 +142,15 @@ beforeAll(async () => {
       response.end(JSON.stringify({ error: 'rate_limited' }))
       return
     }
+    if (parsed.client_id.endsWith('-rate-limited')) {
+      response.writeHead(429, { 'Content-Type': 'application/json' })
+      response.end(JSON.stringify({ error: 'rate_limited' }))
+      return
+    }
+    if (parsed.client_id.endsWith('-outage')) {
+      response.writeHead(502).end()
+      return
+    }
     await sleep(15)
     const identity = `${request.headers['x-provider-origin']}:${parsed.client_id}`
     const issued = (applications.get(identity)?.issued ?? 0) + 1
@@ -344,6 +353,33 @@ describe('Vanta application token lifecycle against shared storage', () => {
     expect(stored.expiresAt.getTime() - Date.now()).toBeLessThanOrEqual(90_000)
   })
 
+  it('waits at least a minute after a Vanta 429 that names no Retry-After', async () => {
+    const input = fields('rate-limited')
+    const keys = previousBuildKeys(input, READ)
+    const before = Date.now()
+
+    await expect(resolveVantaClientToken(input)).rejects.toMatchObject({ status: 429 })
+
+    const stored = await readStoredRow(keys.failureKey)
+    expect(stored.expiresAt.getTime() - before).toBeGreaterThanOrEqual(60_000)
+  })
+
+  it('asks a failing Vanta for a token at most once per 15 seconds, within its 5 tokens a minute', async () => {
+    const input = fields('outage')
+    const keys = previousBuildKeys(input, READ)
+    const count = tokenRequests
+    const before = Date.now()
+
+    await expect(resolveVantaClientToken(input)).rejects.toMatchObject({
+      code: 'provider_unavailable',
+      status: 502,
+    })
+
+    expect(tokenRequests).toBe(count + 1)
+    const stored = await readStoredRow(keys.failureKey)
+    expect(stored.expiresAt.getTime() - before).toBeGreaterThanOrEqual(15_000)
+  })
+
   it('fails closed when coordination storage fails', async () => {
     const count = tokenRequests
     vi.spyOn(db, 'transaction').mockRejectedValueOnce(new Error('database unavailable'))

apps/sim/lib/credentials/client-credential-accounts/vanta-token.ts

@@ -15,6 +15,11 @@ import {
 
 const SCOPES = VANTA_PERMISSION_OPTIONS.map((option) => option.value)
 const EXPIRY_BUFFER_MS = 30_000
+/**
+ * Vanta answers 429 when an application mints more than 5 tokens a minute, and every token it
+ * issues revokes the previous one, so a failing exchange is retried at most every 15 s (4 a minute).
+ */
+const MIN_EXCHANGE_INTERVAL_MS = 15_000
 const STEP = 'vanta_token_mint'
 
 export interface VantaClientTokenFields {
@@ -152,6 +157,7 @@ export async function resolveVantaClientToken(
     step: STEP,
     signal: options.signal,
     minRemainingMs: EXPIRY_BUFFER_MS,
+    minExchangeIntervalMs: MIN_EXCHANGE_INTERVAL_MS,
     failureKeyQualifier: scope,
     accessTokenDigestTag: apiDomain,
     pinnedState: {

apps/sim/lib/credentials/coordinated-client-token.integration.ts

@@ -1,6 +1,7 @@
 import { createHash } from 'node:crypto'
 import { db } from '@sim/db'
 import { clientCredentialToken } from '@sim/db/schema'
+import { createDeferred } from '@sim/testing/helpers/deferred'
 import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { isRecordLike } from '@sim/utils/object'
@@ -14,6 +15,7 @@ import {
 } from '@/lib/credentials/coordinated-client-token'
 import { verifyAndBuildServiceAccountSecret } from '@/lib/credentials/service-account-secret'
 import { TokenServiceAccountValidationError } from '@/lib/credentials/token-service-accounts/errors'
+import { acquireAdvisoryXactLock } from '@/lib/db/advisory-locks'
 
 const NAMESPACE = `fixture${generateId().replaceAll('-', '')}`
 const ownedKeyPrefixes = new Set<string>([`${NAMESPACE}:`])
@@ -181,9 +183,11 @@ describe('coordinated client-credentials token against shared storage', () => {
     await expect(
       resolveCoordinatedClientToken(input('memo', exchange, 'revoked'))
     ).rejects.toMatchObject({ code: 'invalid_credentials' })
-    await expect(
-      resolveCoordinatedClientToken(input('memo', exchange, 'revoked'))
-    ).rejects.toMatchObject({ code: 'invalid_credentials' })
+    const replayed = await resolveCoordinatedClientToken(input('memo', exchange, 'revoked')).catch(
+      (error: unknown) => error
+    )
+    expect(replayed).toMatchObject({ code: 'invalid_credentials', replayForMs: expect.any(Number) })
+    expect(replayed).not.toHaveProperty('retryAfterMs', expect.anything())
     expect(exchange).toHaveBeenCalledOnce()
 
     await db
@@ -371,7 +375,7 @@ describe('coordinated token resolution while the provider fails', () => {
       ).rejects.toMatchObject({ code: 'provider_unavailable', status: 429 })
       await expect(
         resolveCoordinatedClientToken(input(application, exchange))
-      ).rejects.toMatchObject({ status: 429, retryAfterMs: expect.any(Number) })
+      ).rejects.toMatchObject({ status: 429, replayForMs: expect.any(Number) })
 
       expect(exchange).toHaveBeenCalledOnce()
       const stored = await failureRowOf(application)
@@ -380,6 +384,87 @@ describe('coordinated token resolution while the provider fails', () => {
     }
   )
 
+  it('replays a throttled exchange that names no Retry-After for at least a minute', async () => {
+    const exchange = failing(429)
+    const before = Date.now()
+
+    await expect(
+      resolveCoordinatedClientToken(input('throttled-bare', exchange))
+    ).rejects.toMatchObject({ code: 'provider_unavailable', status: 429 })
+
+    expect(exchange).toHaveBeenCalledOnce()
+    expect(
+      (await failureRowOf('throttled-bare')).expiresAt.getTime() - before
+    ).toBeGreaterThanOrEqual(60_000)
+  })
+
+  it.each([
+    [
+      'the still-valid token while its refresh is remembered failing',
+      'unlocked-fallback',
+      'outage',
+    ],
+    ['a remembered refusal', 'unlocked-refusal', 'refusal'],
+  ] as const)(
+    'serves %s without waiting for the application lock',
+    async (_case, application, remembered) => {
+      const appKey = applicationKey(NAMESPACE, [application])
+      const fingerprint = sha256Json([application, 'secret-1'])
+      const token = await encryptSecret(
+        JSON.stringify({ accessToken: `${application}:token-1`, secretFingerprint: fingerprint })
+      )
+      const failure = await encryptSecret(
+        JSON.stringify(
+          remembered === 'outage'
+            ? { code: 'provider_unavailable', status: 503, consecutiveFailures: 1 }
+            : { code: 'invalid_credentials', status: 401 }
+        )
+      )
+      await db.insert(clientCredentialToken).values([
+        {
+          id: appKey,
+          encryptedValue: token.encrypted,
+          accessTokenDigest: sha256Json(['coordinated_client_token', `${application}:token-1`]),
+          expiresAt: new Date(Date.now() + 4 * 60_000),
+        },
+        {
+          id: `${appKey}:${sha256Json(fingerprint)}`,
+          encryptedValue: failure.encrypted,
+          expiresAt: new Date(Date.now() + 60_000),
+        },
+      ])
+      const { exchange } = provider(application)
+      const lockHeld = createDeferred<void>()
+      const release = createDeferred<void>()
+      const holder = db.transaction(async (tx) => {
+        await acquireAdvisoryXactLock(tx, `${NAMESPACE}_client_token`, appKey)
+        lockHeld.resolve()
+        await release.promise
+      })
+
+      try {
+        await lockHeld.promise
+        const outcome = await Promise.race([
+          resolveCoordinatedClientToken(input(application, exchange)).then(
+            (value) => ({ value }),
+            (error: unknown) => ({ error })
+          ),
+          sleep(3_000).then(() => 'waited on the lock' as const),
+        ])
+
+        expect(outcome).toEqual(
+          remembered === 'outage'
+            ? { value: expect.objectContaining({ accessToken: `${application}:token-1` }) }
+            : { error: expect.objectContaining({ code: 'invalid_credentials' }) }
+        )
+        expect(exchange).not.toHaveBeenCalled()
+      } finally {
+        release.resolve()
+        await holder
+      }
+    }
+  )
+
   it('backs off an outage from a few seconds, longer while it continues, never past 30 seconds', async () => {
     const exchange = failing(502)
 
@@ -393,6 +478,7 @@ describe('coordinated token resolution while the provider fails', () => {
       await expireFailureRowsOf('outage')
     }
 
+    expect(replays[0]).toBeGreaterThanOrEqual(5_000)
     expect(replays[0]).toBeLessThanOrEqual(6_000 + 2_500)
     expect(replays[1]).toBeGreaterThanOrEqual(8_000)
     expect(Math.max(...replays)).toBeLessThanOrEqual(30_000 + 2_500)