feat(credentials): share token refresh resilience across workers in the coordinated resolver

The coordinated client-credentials resolver (Shopify, Linear, Atlassian,
Vanta) now applies the same policy as the per-instance mint cache, through
the shared row every worker reads:
- a quick outage is retried once under the application lock, within the 15 s
  its waiters allow
- a refusal is remembered 30 s, a throttled answer for its Retry-After (1 s to
  5 min, Vanta's now included), an outage 5 s growing to 30 s while it lasts
- while an outage or throttling is remembered, the stored token for the same
  secret is served until 30 s before it expires; a refusal fails at once

Rolling deploys stay safe: a refusal is stored exactly as before, and outages
add only a consecutiveFailures key that previous decoders ignore (both prior
decoders are pinned in the integration tests). The replay length rides on the
row's existing expiresAt. Expired memos are kept five more minutes so the next
failure can extend the backoff; an older build's sweep only resets it.

Author: waleedlatif1

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

@@ -137,6 +137,11 @@ beforeAll(async () => {
       )
       return
     }
+    if (parsed.client_id.endsWith('-throttled')) {
+      response.writeHead(429, { 'Content-Type': 'application/json', 'Retry-After': '90' })
+      response.end(JSON.stringify({ error: 'rate_limited' }))
+      return
+    }
     await sleep(15)
     const identity = `${request.headers['x-provider-origin']}:${parsed.client_id}`
     const issued = (applications.get(identity)?.issued ?? 0) + 1
@@ -321,6 +326,24 @@ describe('Vanta application token lifecycle against shared storage', () => {
     expect(await isActive(next.accessToken)).toBe(true)
   })
 
+  it('waits as long as a throttled Vanta asks before any worker asks again', async () => {
+    const input = fields('throttled')
+    const keys = previousBuildKeys(input, READ)
+    const count = tokenRequests
+    const before = Date.now()
+
+    await expect(resolveVantaClientToken(input)).rejects.toMatchObject({
+      code: 'provider_unavailable',
+      status: 429,
+    })
+    await expect(resolveVantaClientToken(input)).rejects.toMatchObject({ status: 429 })
+
+    expect(tokenRequests).toBe(count + 1)
+    const stored = await readStoredRow(keys.failureKey)
+    expect(stored.expiresAt.getTime() - before).toBeGreaterThanOrEqual(90_000)
+    expect(stored.expiresAt.getTime() - Date.now()).toBeLessThanOrEqual(90_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

@@ -8,6 +8,7 @@ import {
 import {
   fetchProvider,
   isTransientProviderStatus,
+  providerRetryAfterMs,
   requireClientSecret,
   TokenServiceAccountValidationError,
 } from '@/lib/credentials/token-service-accounts/errors'
@@ -91,7 +92,8 @@ async function exchangeToken(
         ? 'invalid_credentials'
         : 'provider_unavailable',
       response.status,
-      { step: STEP, reason: 'provider rejected token exchange' }
+      { step: STEP, reason: 'provider rejected token exchange' },
+      { retryAfterMs: providerRetryAfterMs(response) }
     )
   }
   const payload = await readResponseJsonWithLimit<unknown>(response, {

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

@@ -3,9 +3,10 @@ import { db } from '@sim/db'
 import { clientCredentialToken } from '@sim/db/schema'
 import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
+import { isRecordLike } from '@sim/utils/object'
 import { eq, like, or } from 'drizzle-orm'
 import { afterAll, afterEach, describe, expect, it, vi } from 'vitest'
-import { decryptSecret } from '@/lib/core/security/encryption'
+import { decryptSecret, encryptSecret } from '@/lib/core/security/encryption'
 import {
   type CoordinatedClientTokenInput,
   invalidateCoordinatedClientToken,
@@ -63,6 +64,68 @@ function rejecting() {
   })
 }
 
+function failing(status: number, retryAfterMs?: number) {
+  return vi.fn(async (): Promise<{ accessToken: string; expiresAt: Date }> => {
+    throw new TokenServiceAccountValidationError(
+      'provider_unavailable',
+      status,
+      { step: 'mint' },
+      {
+        retryAfterMs,
+      }
+    )
+  })
+}
+
+/** An exchange that issues one token valid for `lifetimeMs`, then fails with `then`. */
+function issuingOnce(
+  application: string,
+  lifetimeMs: number,
+  then: CoordinatedClientTokenInput['exchange']
+) {
+  return vi
+    .fn<CoordinatedClientTokenInput['exchange']>()
+    .mockResolvedValueOnce({
+      accessToken: `${application}:token-1`,
+      expiresAt: new Date(Date.now() + lifetimeMs),
+    })
+    .mockImplementation(then)
+}
+
+async function failureRowOf(application: string) {
+  const rows = await db
+    .select()
+    .from(clientCredentialToken)
+    .where(like(clientCredentialToken.id, `${applicationKey(NAMESPACE, [application])}:%`))
+  if (rows.length !== 1) throw new Error(`Expected one failure row, found ${rows.length}`)
+  return readStoredRow(rows[0].id)
+}
+
+async function expireFailureRowsOf(application: string) {
+  await db
+    .update(clientCredentialToken)
+    .set({ expiresAt: new Date(Date.now() - 1000) })
+    .where(like(clientCredentialToken.id, `${applicationKey(NAMESPACE, [application])}:%`))
+}
+
+/** The previous build's failure decoders, which must keep accepting every failure this build stores. */
+const previousBuildDecodesFailure = {
+  coordinated: (value: unknown) =>
+    isRecordLike(value) &&
+    typeof value.code === 'string' &&
+    [
+      'invalid_credentials',
+      'grant_not_permitted',
+      'site_not_found',
+      'provider_unavailable',
+    ].includes(value.code) &&
+    typeof value.status === 'number',
+  vantaBeforeCoordination: (value: unknown) =>
+    isRecordLike(value) &&
+    (value.code === 'invalid_credentials' || value.code === 'provider_unavailable') &&
+    typeof value.status === 'number',
+}
+
 afterEach(() => {
   vi.unstubAllGlobals()
 })
@@ -204,27 +267,207 @@ describe('coordinated client-credentials token against shared storage', () => {
     ).resolves.toMatchObject({ accessToken: 'fresh-rejection:token-2' })
   })
 
-  it('sweeps expired failure memos', async () => {
+  it('sweeps failure memos once they can no longer extend a backoff', async () => {
+    for (const application of ['swept', 'retained']) {
+      await expect(
+        resolveCoordinatedClientToken(input(application, rejecting(), 'revoked'))
+      ).rejects.toMatchObject({ code: 'invalid_credentials' })
+    }
+    const failurePrefix = (application: string) => `${applicationKey(NAMESPACE, [application])}:`
+    const expire = (application: string, agoMs: number) =>
+      db
+        .update(clientCredentialToken)
+        .set({ expiresAt: new Date(Date.now() - agoMs) })
+        .where(like(clientCredentialToken.id, `${failurePrefix(application)}%`))
+    await expire('swept', 6 * 60_000)
+    await expire('retained', 60_000)
+
+    await resolveCoordinatedClientToken(input('sweeper', provider('sweeper').exchange))
+
+    const remaining = async (application: string) =>
+      db
+        .select({ id: clientCredentialToken.id })
+        .from(clientCredentialToken)
+        .where(like(clientCredentialToken.id, `${failurePrefix(application)}%`))
+    expect(await remaining('swept')).toEqual([])
+    expect(await remaining('retained')).toHaveLength(1)
+  })
+})
+
+describe('coordinated token resolution while the provider fails', () => {
+  it('serves the stored token while it is still valid when its early refresh is throttled', async () => {
+    const exchange = issuingOnce('throttled-refresh', 4 * 60_000, failing(429))
+    await resolveCoordinatedClientToken(input('throttled-refresh', exchange))
+
+    for (let call = 0; call < 3; call++) {
+      await expect(
+        resolveCoordinatedClientToken(input('throttled-refresh', exchange))
+      ).resolves.toMatchObject({ accessToken: 'throttled-refresh:token-1' })
+    }
+    expect(exchange).toHaveBeenCalledTimes(2)
+  })
+
+  it('reports the failed refresh once the stored token has run out', async () => {
+    const exchange = issuingOnce('expired-refresh', 20_000, failing(429))
+    await resolveCoordinatedClientToken(input('expired-refresh', exchange))
+
     await expect(
-      resolveCoordinatedClientToken(input('swept', rejecting(), 'revoked'))
+      resolveCoordinatedClientToken(input('expired-refresh', exchange))
+    ).rejects.toMatchObject({ code: 'provider_unavailable', status: 429 })
+  })
+
+  it('stops serving the stored token once the provider refuses the credentials', async () => {
+    const exchange = issuingOnce('refused-refresh', 4 * 60_000, rejecting())
+    await resolveCoordinatedClientToken(input('refused-refresh', exchange))
+
+    await expect(
+      resolveCoordinatedClientToken(input('refused-refresh', exchange))
     ).rejects.toMatchObject({ code: 'invalid_credentials' })
-    const failurePrefix = `${applicationKey(NAMESPACE, ['swept'])}:`
-    await db
-      .update(clientCredentialToken)
-      .set({ expiresAt: new Date(Date.now() - 1000) })
-      .where(like(clientCredentialToken.id, `${failurePrefix}%`))
+    await expect(
+      resolveCoordinatedClientToken(input('refused-refresh', exchange))
+    ).rejects.toMatchObject({ code: 'invalid_credentials' })
+    expect(exchange).toHaveBeenCalledTimes(2)
+  })
 
-    await resolveCoordinatedClientToken(input('sweeper', provider('sweeper').exchange))
+  it('never serves a still-valid token to a different secret whose refresh fails', async () => {
+    const exchange = issuingOnce('rotated-refresh', 4 * 60_000, failing(429))
+    await resolveCoordinatedClientToken(input('rotated-refresh', exchange))
 
-    const rows = await db
-      .select({ id: clientCredentialToken.id })
-      .from(clientCredentialToken)
-      .where(like(clientCredentialToken.id, `${failurePrefix}%`))
-    expect(rows).toEqual([])
+    await expect(
+      resolveCoordinatedClientToken(input('rotated-refresh', exchange, 'secret-2'))
+    ).rejects.toMatchObject({ code: 'provider_unavailable' })
+  })
+
+  it('retries a quick outage once under the application lock, for every waiting resolution', async () => {
+    const { exchange: issue } = provider('retried')
+    const exchange = vi
+      .fn<CoordinatedClientTokenInput['exchange']>()
+      .mockRejectedValueOnce(
+        new TokenServiceAccountValidationError('provider_unavailable', 503, { step: 'mint' })
+      )
+      .mockImplementation(issue)
+
+    const results = await Promise.all(
+      Array.from({ length: 5 }, () => resolveCoordinatedClientToken(input('retried', exchange)))
+    )
+
+    expect(new Set(results.map((result) => result.accessToken))).toEqual(
+      new Set(['retried:token-1'])
+    )
+    expect(exchange).toHaveBeenCalledTimes(2)
+  })
+
+  it.each([
+    ['as long as the provider asks', 'throttled-90s', 90_000, 90_000],
+    ['for at most five minutes', 'throttled-1h', 3_600_000, 300_000],
+  ])(
+    "shares a throttled exchange's Retry-After with every worker %s",
+    async (_case, application, retryAfterMs, replayMs) => {
+      const exchange = failing(429, retryAfterMs)
+      const before = Date.now()
+
+      await expect(
+        resolveCoordinatedClientToken(input(application, exchange))
+      ).rejects.toMatchObject({ code: 'provider_unavailable', status: 429 })
+      await expect(
+        resolveCoordinatedClientToken(input(application, exchange))
+      ).rejects.toMatchObject({ status: 429, retryAfterMs: expect.any(Number) })
+
+      expect(exchange).toHaveBeenCalledOnce()
+      const stored = await failureRowOf(application)
+      expect(stored.expiresAt.getTime() - before).toBeGreaterThanOrEqual(replayMs)
+      expect(stored.expiresAt.getTime() - Date.now()).toBeLessThanOrEqual(replayMs)
+    }
+  )
+
+  it('backs off an outage from a few seconds, longer while it continues, never past 30 seconds', async () => {
+    const exchange = failing(502)
+
+    const replays: number[] = []
+    for (let attempt = 0; attempt < 5; attempt++) {
+      const before = Date.now()
+      await expect(resolveCoordinatedClientToken(input('outage', exchange))).rejects.toMatchObject({
+        code: 'provider_unavailable',
+      })
+      replays.push((await failureRowOf('outage')).expiresAt.getTime() - before)
+      await expireFailureRowsOf('outage')
+    }
+
+    expect(replays[0]).toBeLessThanOrEqual(6_000 + 2_500)
+    expect(replays[1]).toBeGreaterThanOrEqual(8_000)
+    expect(Math.max(...replays)).toBeLessThanOrEqual(30_000 + 2_500)
+  })
+
+  it('starts the backoff over after the provider recovers', async () => {
+    const { exchange: issue } = provider('recovered')
+    const exchange = vi
+      .fn<CoordinatedClientTokenInput['exchange']>()
+      .mockImplementation(failing(502))
+    for (let attempt = 0; attempt < 3; attempt++) {
+      await expect(resolveCoordinatedClientToken(input('recovered', exchange))).rejects.toThrow()
+      await expireFailureRowsOf('recovered')
+    }
+    exchange.mockImplementationOnce(issue)
+    const { accessToken } = await resolveCoordinatedClientToken(input('recovered', exchange))
+    await invalidateCoordinatedClientToken(accessToken)
+    exchange.mockImplementation(failing(502))
+
+    const before = Date.now()
+    await expect(resolveCoordinatedClientToken(input('recovered', exchange))).rejects.toThrow()
+
+    expect((await failureRowOf('recovered')).expiresAt.getTime() - before).toBeLessThanOrEqual(
+      6_000 + 2_500
+    )
   })
 })
 
 describe('default token rows shared with the previous build during a rolling deploy', () => {
+  it('serves a still-valid token when the previous build remembered an outage', async () => {
+    const application = 'legacy-outage'
+    const appKey = applicationKey(NAMESPACE, [application])
+    const fingerprint = sha256Json([application, 'secret-1'])
+    const token = await encryptSecret(
+      JSON.stringify({ accessToken: 'legacy-token', secretFingerprint: fingerprint })
+    )
+    const outage = await encryptSecret(
+      JSON.stringify({ code: 'provider_unavailable', status: 503 })
+    )
+    await db.insert(clientCredentialToken).values([
+      {
+        id: appKey,
+        encryptedValue: token.encrypted,
+        accessTokenDigest: sha256Json(['coordinated_client_token', 'legacy-token']),
+        expiresAt: new Date(Date.now() + 4 * 60_000),
+      },
+      {
+        id: `${appKey}:${sha256Json(fingerprint)}`,
+        encryptedValue: outage.encrypted,
+        expiresAt: new Date(Date.now() + 20_000),
+      },
+    ])
+    const { exchange } = provider(application)
+
+    await expect(
+      resolveCoordinatedClientToken(input(application, exchange))
+    ).resolves.toMatchObject({ accessToken: 'legacy-token' })
+    expect(exchange).not.toHaveBeenCalled()
+  })
+
+  it('stores every outage and throttled answer as a failure the previous build reads', async () => {
+    for (const [application, exchange] of [
+      ['legacy-readable-outage', failing(502)],
+      ['legacy-readable-throttle', failing(429, 60_000)],
+    ] as const) {
+      await expect(resolveCoordinatedClientToken(input(application, exchange))).rejects.toThrow()
+
+      const stored = await failureRowOf(application)
+      const payload: unknown = JSON.parse(stored.decrypted)
+      expect(previousBuildDecodesFailure.coordinated(payload)).toBe(true)
+      expect(previousBuildDecodesFailure.vantaBeforeCoordination(payload)).toBe(true)
+      expect(stored.accessTokenDigest).toBeNull()
+    }
+  })
+
   it('stores tokens and refusals under the ids, digest, and payloads the previous build used', async () => {
     const application = ['legacy.myshopify.com', 'legacy-client']
     const appKey = applicationKey(NAMESPACE, application)