fix(sailpoint): complete integration validation

Author: waleedlatif1

apps/docs/content/docs/integrations/sailpoint.mdx

@@ -6,6 +6,7 @@ import {
   clearSailPointTokenStateForTests,
   getSailPointAccessToken,
   getSailPointTokenStateForTests,
+  readTotalCount,
   resolveSailPointHosts,
   sailpointFetch,
 } from '@/lib/internal/sailpoint/client'
@@ -65,6 +66,27 @@ describe('SailPoint client', () => {
     await expect(Promise.all([first, second])).resolves.toEqual(['shared', 'shared'])
   })
 
+  it('lets one token waiter abort without cancelling the shared exchange', async () => {
+    let release: ((response: Response) => void) | undefined
+    mockFetch.mockImplementationOnce(
+      () =>
+        new Promise<Response>((resolve) => {
+          release = resolve
+        })
+    )
+    const credentials = { tenant: 'acme', clientId: 'client', clientSecret: 'secret' }
+    const controller = new AbortController()
+    const first = getSailPointAccessToken(credentials, controller.signal)
+    const second = getSailPointAccessToken(credentials)
+
+    controller.abort(new Error('caller stopped'))
+    await expect(first).rejects.toThrow('caller stopped')
+    release?.(tokenResponse('shared'))
+    await expect(second).resolves.toBe('shared')
+    expect(mockFetch).toHaveBeenCalledTimes(1)
+    expect(getSailPointTokenStateForTests().exchangeSize).toBe(0)
+  })
+
   it('expires cached tokens before their provider expiry', async () => {
     vi.useFakeTimers()
     vi.setSystemTime(new Date('2026-01-01T00:00:00.000Z'))
@@ -106,4 +128,46 @@ describe('SailPoint client', () => {
       }))
     ).rejects.toThrow(/maximum|limit|exceeds/i)
   })
+
+  it('aborts during rate-limit backoff without another provider call', async () => {
+    mockFetch
+      .mockResolvedValueOnce(tokenResponse('token'))
+      .mockResolvedValueOnce(new Response(null, { status: 429, headers: { 'retry-after': '30' } }))
+    const controller = new AbortController()
+    const credentials = { tenant: 'acme', clientId: 'client', clientSecret: 'secret' }
+    const pending = sailpointFetch(
+      credentials,
+      (hosts) => ({
+        url: `${hosts.apiBaseUrl}/identities/v1`,
+        init: { method: 'GET' },
+      }),
+      { signal: controller.signal }
+    )
+
+    await vi.waitFor(() => expect(mockFetch).toHaveBeenCalledTimes(2))
+    controller.abort(new Error('stop retrying'))
+    await expect(pending).rejects.toThrow('stop retrying')
+    expect(mockFetch).toHaveBeenCalledTimes(2)
+  })
+
+  it('rejects redirects for token and authenticated provider requests', async () => {
+    mockFetch
+      .mockResolvedValueOnce(tokenResponse('token'))
+      .mockResolvedValueOnce(Response.json({ id: 'identity' }))
+    const credentials = { tenant: 'acme', clientId: 'client', clientSecret: 'secret' }
+
+    await sailpointFetch(credentials, (hosts) => ({
+      url: `${hosts.apiBaseUrl}/identities/v1/id`,
+      init: { method: 'GET' },
+    }))
+
+    expect(mockFetch.mock.calls[0][1]?.redirect).toBe('error')
+    expect(mockFetch.mock.calls[1][1]?.redirect).toBe('error')
+  })
+
+  it('accepts only non-negative integer total counts', () => {
+    expect(readTotalCount(new Headers({ 'x-total-count': '7' }))).toBe(7)
+    expect(readTotalCount(new Headers({ 'x-total-count': '1.5' }))).toBeNull()
+    expect(readTotalCount(new Headers({ 'x-total-count': '-1' }))).toBeNull()
+  })
 })

apps/sim/blocks/blocks/sailpoint.ts

@@ -1,5 +1,5 @@
 import { createHash } from 'node:crypto'
-import { sleep } from '@sim/utils/helpers'
+import { interruptibleSleep } from '@sim/utils/helpers'
 import { isRecordLike } from '@sim/utils/object'
 import { backoffWithJitter, parseRetryAfter } from '@sim/utils/retry'
 import { MAX_JSON_API_RESPONSE_BYTES } from '@/lib/core/security/input-validation.server'
@@ -37,10 +37,23 @@ const MAX_FETCH_RETRIES = 4
 const MAX_TOKEN_CACHE_ENTRIES = 100
 const MAX_TOKEN_EXCHANGES = 100
 const MAX_TOKEN_RESPONSE_BYTES = 1024 * 1024
+const TOKEN_EXCHANGE_TIMEOUT_MS = 30_000
 const TOKEN_EXPIRY_BUFFER_MS = 60_000
 const tokenCache = new Map<string, CachedToken>()
 const tokenExchanges = new Map<string, Promise<string>>()
 
+async function waitForPromiseWithSignal<T>(promise: Promise<T>, signal?: AbortSignal): Promise<T> {
+  if (!signal) return promise
+  signal.throwIfAborted()
+
+  return new Promise<T>((resolve, reject) => {
+    const onAbort = () =>
+      reject(signal.reason ?? new DOMException('The operation was aborted', 'AbortError'))
+    signal.addEventListener('abort', onAbort, { once: true })
+    promise.then(resolve, reject).finally(() => signal.removeEventListener('abort', onAbort))
+  })
+}
+
 export function resolveSailPointHosts(tenant: string): SailPointHosts {
   let host = tenant.trim().replace(/^https?:\/\//i, '')
   host = host
@@ -155,14 +168,16 @@ async function exchangeAccessToken(
         client_secret: credentials.clientSecret,
       }).toString(),
       cache: 'no-store',
+      redirect: 'error',
       signal,
     })
 
     if (response.status === 429 && attempt < MAX_FETCH_RETRIES) {
       const retryAfterMs = parseRetryAfter(response.headers.get('retry-after'))
       await consumeOrCancelBody(response, DEFAULT_MAX_ERROR_BODY_BYTES)
       attempt += 1
-      await sleep(backoffWithJitter(attempt, retryAfterMs))
+      await interruptibleSleep(backoffWithJitter(attempt, retryAfterMs), signal)
+      signal?.throwIfAborted()
       continue
     }
 
@@ -210,16 +225,19 @@ export async function getSailPointAccessToken(
   if (cached) tokenCache.delete(key)
 
   const existing = tokenExchanges.get(key)
-  if (existing) return existing
+  if (existing) return waitForPromiseWithSignal(existing, signal)
   if (tokenExchanges.size >= MAX_TOKEN_EXCHANGES) {
     throw new Error('Too many concurrent SailPoint token exchanges')
   }
 
-  const exchange = exchangeAccessToken(credentials, signal).finally(() => {
+  const exchange = exchangeAccessToken(
+    credentials,
+    AbortSignal.timeout(TOKEN_EXCHANGE_TIMEOUT_MS)
+  ).finally(() => {
     tokenExchanges.delete(key)
   })
   tokenExchanges.set(key, exchange)
-  return exchange
+  return waitForPromiseWithSignal(exchange, signal)
 }
 
 export async function sailpointFetch(
@@ -244,6 +262,7 @@ export async function sailpointFetch(
       ...init,
       cache: 'no-store',
       headers,
+      redirect: 'error',
       signal: options.signal,
     })
 
@@ -257,7 +276,8 @@ export async function sailpointFetch(
       const retryAfterMs = parseRetryAfter(response.headers.get('retry-after'))
       await consumeOrCancelBody(response, DEFAULT_MAX_ERROR_BODY_BYTES)
       attempt += 1
-      await sleep(backoffWithJitter(attempt, retryAfterMs))
+      await interruptibleSleep(backoffWithJitter(attempt, retryAfterMs), options.signal)
+      options.signal?.throwIfAborted()
       continue
     }
 
@@ -279,7 +299,7 @@ export function readTotalCount(headers: Headers): number | null {
   const raw = headers.get('x-total-count')
   if (!raw) return null
   const parsed = Number(raw)
-  return Number.isFinite(parsed) && parsed >= 0 ? parsed : null
+  return Number.isInteger(parsed) && parsed >= 0 ? parsed : null
 }
 
 /** Clears process-local authentication state for deterministic tests. */