fix(selectors): close residual cancellation gaps

apps/sim/lib/selectors/application/execute-selector.test.ts

@@ -390,6 +390,34 @@ describe('executeSelector', () => {
     ])
   })
 
+  it('binds the request signal to credentials and does not present a late provider result', async () => {
+    const controller = new AbortController()
+    let markProviderStarted!: () => void
+    let finishProvider!: (result: { kind: 'list'; items: never[] }) => void
+    const providerStarted = new Promise<void>((resolve) => {
+      markProviderStarted = resolve
+    })
+    mocks.executeAttachment.mockImplementationOnce(
+      (args: ExecuteServerSelectorArgs) =>
+        new Promise((resolve) => {
+          expect(args.credential?.signal).toBe(controller.signal)
+          markProviderStarted()
+          finishProvider = resolve
+        })
+    )
+
+    const pending = execute({ signal: controller.signal })
+    await providerStarted
+
+    const abortReason = new DOMException('Selector request canceled', 'AbortError')
+    controller.abort(abortReason)
+    finishProvider({ kind: 'list', items: [] })
+
+    await expect(pending).rejects.toBe(abortReason)
+    expect(mocks.sanitize).not.toHaveBeenCalled()
+    expect(mocks.logger.info).not.toHaveBeenCalledWith('Executed selector', expect.anything())
+  })
+
   it('records legacy service-account use once with its trusted provider id', async () => {
     mocks.authorizeCredential.mockResolvedValueOnce({
       suppliedId: 'credential-1',

apps/sim/lib/selectors/application/execute-selector.ts

@@ -151,15 +151,18 @@ async function executeAuthorizedSelector(args: {
     }
 
     const credential = attachment.credential
-      ? await authorizeSelectorCredential({
-          principal: args.principal,
-          context: resolvedContext,
-          scope: args.input.scope,
-          workspaceId: args.context.workspaceId,
-          policy: attachment.credential,
-          protectedValues,
-          references: resolved.references,
-        })
+      ? {
+          ...(await authorizeSelectorCredential({
+            principal: args.principal,
+            context: resolvedContext,
+            scope: args.input.scope,
+            workspaceId: args.context.workspaceId,
+            policy: attachment.credential,
+            protectedValues,
+            references: resolved.references,
+          })),
+          signal: args.input.signal,
+        }
       : undefined
 
     /**
@@ -220,6 +223,7 @@ async function executeAuthorizedSelector(args: {
         ? undefined
         : await attachment.destination.prepare(selectorArgs)
     const providerResult = await attachment.execute(selectorArgs, preparedDestination)
+    args.input.signal?.throwIfAborted()
     if (providerResult.diagnostics?.truncated) {
       logger.warn('Selector provider result reached a configured cap', {
         selectorKey: args.input.selectorKey,

apps/sim/lib/selectors/server/credentials.test.ts

@@ -10,22 +10,26 @@ const mocks = vi.hoisted(() => ({
   authorizeCredentialUse: vi.fn(),
   credentialProviderMatchesService: vi.fn(),
   getServiceConfig: vi.fn(),
+  resolveCredentialTokenBundle: vi.fn(),
 }))
 
 vi.mock('@/lib/auth/credential-access', () => ({
   authorizeCredentialUseForAuth: mocks.authorizeCredentialUse,
 }))
 
 vi.mock('@/lib/oauth/credential-service', () => ({
-  resolveCredentialTokenBundle: vi.fn(),
+  resolveCredentialTokenBundle: mocks.resolveCredentialTokenBundle,
 }))
 
 vi.mock('@/lib/oauth/utils', () => ({
   credentialProviderMatchesService: mocks.credentialProviderMatchesService,
   getServiceConfigByServiceId: mocks.getServiceConfig,
 }))
 
-import { authorizeSelectorCredential } from '@/lib/selectors/server/credentials'
+import {
+  authorizeSelectorCredential,
+  resolveSelectorOAuthAccessToken,
+} from '@/lib/selectors/server/credentials'
 import { SelectorConnectionUnavailableError } from '@/lib/selectors/server/errors'
 import { createSelectorProtectedValues } from '@/lib/selectors/server/protected-values'
 
@@ -137,3 +141,119 @@ describe('authorizeSelectorCredential', () => {
     })
   })
 })
+
+describe('resolveSelectorOAuthAccessToken', () => {
+  beforeEach(() => vi.clearAllMocks())
+
+  it('rejects only the canceled waiter while shared credential work serves another caller', async () => {
+    let resolveShared!: (value: { accessToken: string }) => void
+    const sharedResolution = new Promise<{ accessToken: string }>((resolve) => {
+      resolveShared = resolve
+    })
+    mocks.resolveCredentialTokenBundle.mockReturnValue(sharedResolution)
+    const canceledController = new AbortController()
+    const liveController = new AbortController()
+    const canceledProtectedValues = createSelectorProtectedValues()
+    const liveProtectedValues = createSelectorProtectedValues()
+    const canceledRecordUse = vi.fn()
+    const liveRecordUse = vi.fn()
+    const access = {
+      ok: true as const,
+      credentialOwnerUserId: 'owner-1',
+      resolvedCredentialId: 'credential-1',
+    }
+
+    const canceledWaiter = resolveSelectorOAuthAccessToken({
+      credential: {
+        suppliedId: 'credential-1',
+        access,
+        signal: canceledController.signal,
+      },
+      serviceId: 'gmail',
+      protectedValues: canceledProtectedValues,
+      recordCredentialUse: canceledRecordUse,
+    })
+    const liveWaiter = resolveSelectorOAuthAccessToken({
+      credential: {
+        suppliedId: 'credential-1',
+        access,
+        signal: liveController.signal,
+      },
+      serviceId: 'gmail',
+      protectedValues: liveProtectedValues,
+      recordCredentialUse: liveRecordUse,
+    })
+
+    const abortReason = new DOMException('Selector request canceled', 'AbortError')
+    canceledController.abort(abortReason)
+    await expect(canceledWaiter).rejects.toBe(abortReason)
+
+    resolveShared({ accessToken: 'shared-access-token' })
+    await expect(liveWaiter).resolves.toBe('shared-access-token')
+
+    expect(canceledRecordUse).not.toHaveBeenCalled()
+    expect(canceledProtectedValues.contains('shared-access-token')).toBe(false)
+    expect(liveRecordUse).toHaveBeenCalledOnce()
+    expect(liveProtectedValues.contains('shared-access-token')).toBe(true)
+    expect(mocks.resolveCredentialTokenBundle).toHaveBeenCalledTimes(2)
+    for (const call of mocks.resolveCredentialTokenBundle.mock.calls) {
+      expect(call[5]).toEqual({ privacyMode: 'selector' })
+      expect(call).not.toContain(canceledController.signal)
+      expect(call).not.toContain(liveController.signal)
+    }
+  })
+
+  it('does not start credential resolution for an already canceled selector', async () => {
+    const controller = new AbortController()
+    const abortReason = new DOMException('Selector request canceled', 'AbortError')
+    controller.abort(abortReason)
+
+    await expect(
+      resolveSelectorOAuthAccessToken({
+        credential: {
+          suppliedId: 'credential-1',
+          access: {
+            ok: true,
+            credentialOwnerUserId: 'owner-1',
+            resolvedCredentialId: 'credential-1',
+          },
+          signal: controller.signal,
+        },
+        serviceId: 'gmail',
+        protectedValues: createSelectorProtectedValues(),
+      })
+    ).rejects.toBe(abortReason)
+
+    expect(mocks.resolveCredentialTokenBundle).not.toHaveBeenCalled()
+  })
+
+  it('rechecks cancellation before consuming a fulfilled credential result', async () => {
+    mocks.resolveCredentialTokenBundle.mockResolvedValue({
+      accessToken: 'fulfilled-access-token',
+    })
+    const controller = new AbortController()
+    const protectedValues = createSelectorProtectedValues()
+    const recordCredentialUse = vi.fn()
+    const abortReason = new DOMException('Selector request canceled', 'AbortError')
+
+    const pending = resolveSelectorOAuthAccessToken({
+      credential: {
+        suppliedId: 'credential-1',
+        access: {
+          ok: true,
+          credentialOwnerUserId: 'owner-1',
+          resolvedCredentialId: 'credential-1',
+        },
+        signal: controller.signal,
+      },
+      serviceId: 'gmail',
+      protectedValues,
+      recordCredentialUse,
+    })
+    queueMicrotask(() => controller.abort(abortReason))
+
+    await expect(pending).rejects.toBe(abortReason)
+    expect(protectedValues.contains('fulfilled-access-token')).toBe(false)
+    expect(recordCredentialUse).not.toHaveBeenCalled()
+  })
+})

apps/sim/lib/selectors/server/credentials.ts

@@ -18,6 +18,41 @@ import type {
 } from '@/lib/selectors/server/types'
 import type { SelectorContext, SelectorScope } from '@/lib/selectors/types'
 
+function selectorAbortReason(signal: AbortSignal): unknown {
+  return signal.reason ?? new DOMException('The operation was aborted.', 'AbortError')
+}
+
+/**
+ * Makes one selector's wait abortable without attaching its signal to shared
+ * refresh or mint work that may still be serving other callers.
+ */
+export function waitForSelectorCredentialResolution<T>(
+  resolution: Promise<T>,
+  signal?: AbortSignal
+): Promise<T> {
+  if (!signal) return resolution
+  if (signal.aborted) return Promise.reject(selectorAbortReason(signal))
+
+  return new Promise<T>((resolve, reject) => {
+    const onAbort = () => {
+      signal.removeEventListener('abort', onAbort)
+      reject(selectorAbortReason(signal))
+    }
+    signal.addEventListener('abort', onAbort, { once: true })
+    resolution.then(
+      (value) => {
+        signal.removeEventListener('abort', onAbort)
+        resolve(value)
+      },
+      (error) => {
+        signal.removeEventListener('abort', onAbort)
+        reject(error)
+      }
+    )
+    if (signal.aborted) onAbort()
+  })
+}
+
 async function resolveCredentialProviderId(input: {
   credentialId: string
   credentialOwnerUserId: string
@@ -116,21 +151,26 @@ export async function resolveSelectorOAuthAccessToken(input: {
   protectedValues: SelectorProtectedValues
   recordCredentialUse?: (providerId: string) => void
 }): Promise<string> {
+  input.credential.signal?.throwIfAborted()
   if (input.credential.fixedToken) return input.credential.fixedToken
 
   const access = input.credential.access
   if (!access?.credentialOwnerUserId || !access.resolvedCredentialId) {
     throw new SelectorConnectionUnavailableError()
   }
 
-  const result = await resolveCredentialTokenBundle(
-    input.credential.suppliedId,
-    access.credentialOwnerUserId,
-    'selector-execution',
-    input.scopes ? [...input.scopes] : undefined,
-    input.impersonateEmail,
-    { privacyMode: 'selector' }
+  const result = await waitForSelectorCredentialResolution(
+    resolveCredentialTokenBundle(
+      input.credential.suppliedId,
+      access.credentialOwnerUserId,
+      'selector-execution',
+      input.scopes ? [...input.scopes] : undefined,
+      input.impersonateEmail,
+      { privacyMode: 'selector' }
+    ),
+    input.credential.signal
   )
+  input.credential.signal?.throwIfAborted()
   const token = result?.accessToken
 
   if (!token) throw new SelectorConnectionUnavailableError()

apps/sim/lib/selectors/server/internal.test.ts

@@ -5,11 +5,16 @@ import { environmentUtilsMockFns, resetEnvironmentUtilsMock } from '@sim/testing
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
 const mockListWorkflows = vi.hoisted(() => vi.fn())
+const mockFetchOpenRouterEmbeddingModelCatalog = vi.hoisted(() => vi.fn())
 
 vi.mock('@/lib/workflows/application/list-workflows', () => ({
   listWorkflows: { execute: mockListWorkflows },
 }))
 
+vi.mock('@/lib/embeddings/openrouter-model-catalog.server', () => ({
+  fetchOpenRouterEmbeddingModelCatalog: mockFetchOpenRouterEmbeddingModelCatalog,
+}))
+
 import { SelectorOptionsUnavailableError } from '@/lib/selectors/server/errors'
 import { internalSelectorAttachments } from '@/lib/selectors/server/internal'
 import { createSelectorProtectedValues } from '@/lib/selectors/server/protected-values'
@@ -112,3 +117,36 @@ describe('sim.workflows selector', () => {
     expect(mockListWorkflows).toHaveBeenCalledTimes(40)
   })
 })
+
+describe('providers.openrouterEmbeddingModels selector', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetEnvironmentUtilsMock()
+  })
+
+  it('passes the selector signal to the OpenRouter catalog fetch', async () => {
+    const controller = new AbortController()
+    mockFetchOpenRouterEmbeddingModelCatalog.mockResolvedValue([
+      { id: 'openai/text-embedding-3-small', maxInputTokens: 8_191 },
+    ])
+
+    await expect(
+      internalSelectorAttachments['providers.openrouterEmbeddingModels'].execute({
+        ...workflowArgs(),
+        selectorKey: 'providers.openrouterEmbeddingModels',
+        signal: controller.signal,
+      })
+    ).resolves.toEqual({
+      kind: 'list',
+      items: [
+        {
+          id: 'openai/text-embedding-3-small',
+          label: 'openai/text-embedding-3-small',
+        },
+      ],
+    })
+
+    expect(mockFetchOpenRouterEmbeddingModelCatalog).toHaveBeenCalledOnce()
+    expect(mockFetchOpenRouterEmbeddingModelCatalog).toHaveBeenCalledWith(controller.signal)
+  })
+})

apps/sim/lib/selectors/server/internal.ts

@@ -298,10 +298,10 @@ export const internalSelectorAttachments = {
   },
   'providers.openrouterEmbeddingModels': {
     destination: 'fixed',
-    async execute() {
+    async execute(args: ExecuteServerSelectorArgs) {
       if (isProviderBlacklisted('openrouter')) return listSelectorResult([])
       const models = filterBlacklistedModels(
-        (await fetchOpenRouterEmbeddingModelCatalog()).map((model) => model.id)
+        (await fetchOpenRouterEmbeddingModelCatalog(args.signal)).map((model) => model.id)
       )
       return listSelectorResult([...new Set(models)].map((model) => ({ id: model, label: model })))
     },