fix(connectors): validate retry response lifecycles

Author: waleedlatif1

apps/sim/lib/atlassian/discovery.test.ts

@@ -217,4 +217,18 @@ describe('resolveAtlassianCloudId', () => {
       'Invalid Jira accessible-resources response'
     )
   })
+
+  it.each([
+    [{ url: SITE }],
+    [{ id: CLOUD_ID }],
+    [{ id: '', url: SITE }],
+    [{ id: CLOUD_ID, url: '' }],
+    [null],
+  ])('rejects malformed resource entries in an otherwise valid array', async (resources) => {
+    fetchMock.mockResolvedValue(createMockResponse({ json: resources }))
+
+    await expect(resolveAtlassianCloudId(options())).rejects.toThrow(
+      'Invalid Jira accessible-resources response'
+    )
+  })
 })

apps/sim/lib/atlassian/discovery.ts

@@ -102,6 +102,17 @@ interface AccessibleResource {
   url: string
 }
 
+function isAccessibleResource(value: unknown): value is AccessibleResource {
+  if (typeof value !== 'object' || value === null) return false
+  const resource = value as Record<string, unknown>
+  return (
+    typeof resource.id === 'string' &&
+    resource.id.trim().length > 0 &&
+    typeof resource.url === 'string' &&
+    resource.url.trim().length > 0
+  )
+}
+
 interface ResolveAtlassianCloudIdOptions {
   domain: string
   accessToken: string
@@ -203,7 +214,7 @@ export function selectAtlassianCloudId(
   domain: string,
   product: string
 ): string {
-  if (!Array.isArray(resources)) {
+  if (!Array.isArray(resources) || !resources.every(isAccessibleResource)) {
     throw new Error(`Invalid ${product} accessible-resources response`)
   }
 
@@ -215,16 +226,14 @@ export function selectAtlassianCloudId(
   }
 
   const siteUrl = normalizeAtlassianSiteUrl(domain)
-  const match = (resources as AccessibleResource[]).find(
-    (r) => normalizeAtlassianSiteUrl(r.url) === siteUrl
-  )
+  const match = resources.find((r) => normalizeAtlassianSiteUrl(r.url) === siteUrl)
   if (match) return match.id
 
-  if (resources.length === 1) return (resources as AccessibleResource[])[0].id
+  if (resources.length === 1) return resources[0].id
 
   throw new Error(
     `Could not match ${product} domain "${domain}" to any accessible resource. ` +
-      `Available sites: ${(resources as AccessibleResource[]).map((r) => r.url).join(', ')}`
+      `Available sites: ${resources.map((r) => r.url).join(', ')}`
   )
 }
 

apps/sim/lib/knowledge/documents/utils.test.ts

@@ -547,6 +547,26 @@ describe('fetchWithRetry rate-limit handling', () => {
     expect(fetchMock).toHaveBeenCalledTimes(1)
   })
 
+  it('cancels an omitted rate-limit response body before throwing', async () => {
+    let cancelled = false
+    const body = new ReadableStream<Uint8Array>({
+      cancel() {
+        cancelled = true
+      },
+    })
+    globalThis.fetch = vi.fn().mockResolvedValue(
+      new Response(body, {
+        status: 429,
+        headers: { 'retry-after': '900' },
+      })
+    )
+
+    await expect(
+      fetchWithRetry('https://api.github.com/repos', {}, { ...FAST_RETRY, maxRetries: 0 })
+    ).rejects.toThrow('HTTP 429 - upstream rate limit exceeded')
+    expect(cancelled).toBe(true)
+  })
+
   it('waits until an admitted x-rate-limit-reset instant before retrying', async () => {
     vi.useFakeTimers()
     const now = 1_700_000_000_000

apps/sim/lib/knowledge/documents/utils.ts

@@ -286,6 +286,16 @@ interface RetryableHttpResponse {
   text?: () => Promise<string>
 }
 
+/** Releases a response stream when its provider-controlled body is intentionally omitted. */
+async function cancelHttpResponseBody(response: RetryableHttpResponse): Promise<void> {
+  if (!response.body) return
+  try {
+    await response.body.cancel()
+  } catch {
+    return
+  }
+}
+
 /**
  * Builds the bounded error shared by direct and SSRF-safe connector fetches.
  * Rate-limit responses are named from trusted status/header evidence while all
@@ -296,6 +306,9 @@ export async function createRetryableHttpError(
 ): Promise<HTTPError> {
   const rateLimited =
     response.status === 429 || (response.status === 403 && hasRateLimitEvidence(response.headers))
+  if (rateLimited) {
+    await cancelHttpResponseBody(response)
+  }
   const diagnostic = rateLimited
     ? 'upstream rate limit exceeded'
     : await readBoundedHttpErrorBody(response)