fix(file-search): isolate and queue concurrent searches

apps/sim/lib/workspace-files/application/search-workspace-file-content.test.ts

@@ -0,0 +1,75 @@
+/** @vitest-environment node */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({
+  load: vi.fn(),
+  permission: vi.fn(),
+  search: vi.fn(),
+  folders: vi.fn(),
+}))
+vi.mock('@sim/platform-authz/workspace', () => ({
+  permissionSatisfies: (actual: string | null) => actual !== null,
+  resolveEffectiveWorkspacePermission: mocks.permission,
+}))
+vi.mock('@/lib/uploads/contexts/workspace', () => ({ loadActiveWorkspaceContext: mocks.load }))
+vi.mock('@/lib/workspace-files/search/repository', () => ({
+  searchWorkspaceFileIndex: mocks.search,
+}))
+vi.mock('@/lib/workspace-files/resolve-folder-scope', () => ({
+  resolveWorkspaceFolderScope: mocks.folders,
+}))
+
+import { searchWorkspaceFileContent } from '@/lib/workspace-files/application/search-workspace-file-content'
+
+const principal = { kind: 'session', userId: 'user-1', sessionId: 'session-1' } as const
+const input = {
+  workspaceId: 'workspace-1',
+  query: 'needle',
+  mode: 'exact',
+  maxResults: 10,
+} as const
+
+describe('searchWorkspaceFileContent cancellation', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mocks.load.mockResolvedValue({
+      workspaceId: 'workspace-1',
+      workspaceOrganizationId: null,
+      allowPersonalApiKeys: true,
+      billedAccountUserId: 'user-1',
+    })
+    mocks.permission.mockResolvedValue('read')
+    mocks.search.mockResolvedValue({ results: [] })
+  })
+
+  it.each(['request', 'input'] as const)(
+    'propagates the %s signal through the authorized application operation',
+    async (source) => {
+      const controller = new AbortController()
+      await searchWorkspaceFileContent.execute({
+        principal,
+        input: { ...input, ...(source === 'input' ? { signal: controller.signal } : {}) },
+        request: {
+          headers: new Headers(),
+          ...(source === 'request' ? { signal: controller.signal } : {}),
+        },
+      })
+      expect(mocks.search).toHaveBeenCalledWith(
+        expect.objectContaining({ signal: controller.signal })
+      )
+    }
+  )
+
+  it('does not resolve folders or enqueue database work for a cancelled HTTP request', async () => {
+    const signal = AbortSignal.abort(new Error('cancelled'))
+    await expect(
+      searchWorkspaceFileContent.execute({
+        principal,
+        input: { ...input, folderPaths: ['/notes'] },
+        request: { headers: new Headers(), signal },
+      })
+    ).rejects.toBe(signal.reason)
+    expect(mocks.folders).not.toHaveBeenCalled()
+    expect(mocks.search).not.toHaveBeenCalled()
+  })
+})

apps/sim/lib/workspace-files/application/search-workspace-file-content.ts

@@ -3,15 +3,13 @@ import { loadActiveWorkspaceContext } from '@/lib/uploads/contexts/workspace'
 import { defineAuthorizedWorkspaceFileUseCase } from '@/lib/workspace-files/application/authorized-workspace-file-use-case'
 import { fileOperations } from '@/lib/workspace-files/application/operations'
 import { resolveWorkspaceFolderScope } from '@/lib/workspace-files/resolve-folder-scope'
+import { WorkspaceFileSearchUnavailableError } from '@/lib/workspace-files/search/errors'
 import {
   compileFileSearchPattern,
   type FileSearchMode,
   FileSearchPatternError,
 } from '@/lib/workspace-files/search/pattern'
-import {
-  searchWorkspaceFileIndex,
-  WorkspaceFileSearchUnavailableError,
-} from '@/lib/workspace-files/search/repository'
+import { searchWorkspaceFileIndex } from '@/lib/workspace-files/search/repository'
 
 export interface SearchWorkspaceFileContentInput {
   workspaceId: string
@@ -37,14 +35,16 @@ export const searchWorkspaceFileContent = defineAuthorizedWorkspaceFileUseCase({
   operation: fileOperations.searchContent,
   resolveContext: ({ input }: { input: SearchWorkspaceFileContentInput }) =>
     resolveSearchWorkspaceFileContext(input),
-  execute: async ({ principal, input, context }) => {
-    /*
+  execute: async ({ principal, input, context, request }) => {
+    const signal = input.signal ?? request?.signal
+    signal?.throwIfAborted()
+    /**
      * Resolved here rather than at the surface so every caller (the File
      * block, the v2 route) is confined by the same check. A folder tree
      * holding one subtree per user makes this scope the isolation boundary,
      * not a convenience filter.
      */
-    /*
+    /**
      * `!== undefined`, not a length check: an explicitly empty list is a scope
      * that names no folder, which must match nothing. Treating it as "absent"
      * would answer a request for nothing with the whole workspace.
@@ -58,15 +58,15 @@ export const searchWorkspaceFileContent = defineAuthorizedWorkspaceFileUseCase({
             includeSubfolders: input.includeSubfolders,
           })
         : undefined
-    input.signal?.throwIfAborted()
+    signal?.throwIfAborted()
 
     try {
       return await searchWorkspaceFileIndex({
         workspaceId: context.workspaceId,
         pattern: compileFileSearchPattern(input.query, input.mode),
         maxResults: input.maxResults,
         folderScope,
-        signal: input.signal,
+        signal,
       })
     } catch (error) {
       /**

apps/sim/lib/workspace-files/search/README.md

@@ -22,7 +22,9 @@ Search joins the current file revision and resolved workspace/folder scope. A re
 
 For regular chunks, PostgreSQL checks the pattern with newline-aware semantics, then verifies individual logical lines. Long-line fragments use only necessary three-character literals as a conservative prefilter, including all required alternation branches. Two-code-point overlap preserves those literals at every boundary. PostgreSQL reconstructs the complete candidate line and evaluates the original regex, so anchors, word boundaries, repetitions, and arbitrarily long match spans retain line semantics. Fixed overlap alone is never treated as proof of a match. The supported regex grammar and minimum literal requirement are unchanged.
 
-Regular blocks are verified in batches of at most 16 (128 KiB of indexed text); long lines are reconstructed one at a time. Only bounded match-centered previews leave PostgreSQL: at most 201 rows to detect truncation, and at most 2 KiB per rendered result. A single search has a ten-second application deadline with per-statement guards. PostgreSQL 17 additionally enforces a total transaction timeout; PostgreSQL 16 uses the compatible idle-transaction guard. Transaction advisory locks admit at most two simultaneous searches per workspace and ten globally per database. Busy and timed-out searches fail explicitly; they never report an incomplete scan as an authoritative empty result. The reader uses the normal application database connection, so admission is coordinated on the same database as the index.
+Regular blocks are verified in batches of at most 16 (128 KiB of indexed text); long lines are reconstructed one at a time. Only bounded match-centered previews leave PostgreSQL: at most 201 rows to detect truncation, and at most 2 KiB per rendered result. A single search has a fifteen-second caller deadline covering queueing, connection acquisition, and execution; SQL runs for at most ten seconds within that budget, with per-statement guards. PostgreSQL 17 additionally enforces a total transaction timeout; PostgreSQL 16 uses the compatible idle-transaction guard. Transaction advisory locks admit at most 20 simultaneous searches per workspace and 5,000 globally per database. Search transactions use the dedicated `dbFor('search')` primary pool, with five connections per process, so they cannot occupy the application or execution client pools. Before acquiring a connection, a process admits five active searches and at most 100 waiting requests, capped at 20 waiting requests per workspace so one burst cannot fill the entire queue. Queued workspaces rotate after each grant; waiting requests expire after five seconds or leave immediately on cancellation. The local active budget comes from the same pool profile as the driver. A slot is released only when the transaction settles, including errors. Cancellation reaches this boundary from both HTTP requests and File-tool execution. The existing deadline helper ends the caller’s wait promptly even when connection acquisition stalls. A late transaction checks cancellation before running search SQL; its active slot remains reserved until the driver settles, so a timed-out request cannot start replacement work on top of a still-running query. The driver and upstream pooler still own physical connection cleanup; the application deadline does not cancel a queued PostgreSQL protocol command. Busy and timed-out searches fail explicitly; they never report an incomplete scan as an authoritative empty result. These bounds apply identically to exact and regex search and do not change result or line-number semantics.
+
+The 20/workspace and 5,000/global advisory ceilings bound admitted transactions across processes; they are not promises of simultaneous execution or throughput. Local queues absorb short bursts without holding database connections. They are not durable jobs or a fleet-wide fair scheduler. The dedicated client pool isolates connection ownership, not PostgreSQL CPU, memory, I/O, or an upstream PgBouncer server pool. Its default URL is the process primary URL; any `DATABASE_URL_SEARCH` override must target the same primary database so current revisions and advisory admission remain coherent. Independent PgBouncer server budgets require separate database/user pool configuration. Total client connections can increase by five per participating process. Before raising execution capacity, measure the number of processes, backend pool budget, queue wait/rejection rates, search latency, and database resource headroom under representative exact and broad-regex workloads. More queueing cannot increase sustained throughput.
 
 Arbitrary regex cannot have a fixed latency guarantee. Common terms, broad alternatives, and punctuation-only literals may require scanning significant scoped text. Larger capacity decisions need representative query plans and workload measurements; neither a per-file byte cap nor a PostgreSQL row-count claim establishes a total corpus capacity.
 

apps/sim/lib/workspace-files/search/admission.test.ts

@@ -0,0 +1,212 @@
+/** @vitest-environment node */
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { FileSearchAdmission } from '@/lib/workspace-files/search/admission'
+import { WorkspaceFileSearchUnavailableError } from '@/lib/workspace-files/search/errors'
+
+describe('FileSearchAdmission', () => {
+  beforeEach(() => vi.useFakeTimers())
+  afterEach(() => vi.useRealTimers())
+
+  function createAdmission(concurrency = 1, maxPending = 3, maxPendingPerWorkspace = maxPending) {
+    return new FileSearchAdmission({
+      concurrency,
+      maxPending,
+      maxPendingPerWorkspace,
+      timeoutMs: 5000,
+    })
+  }
+
+  it('queues a burst without exceeding the active budget and releases each slot once', async () => {
+    const admission = createAdmission(2)
+    const first = await admission.acquire('a')
+    const second = await admission.acquire('a')
+    const granted = vi.fn()
+    const waiting = admission.acquire('a').then((release) => {
+      granted()
+      return release
+    })
+    await vi.advanceTimersByTimeAsync(1)
+    expect(granted).not.toHaveBeenCalled()
+    first()
+    const third = await waiting
+    first()
+    const fourthGranted = vi.fn()
+    const fourth = admission.acquire('b').then((release) => {
+      fourthGranted()
+      return release
+    })
+    await vi.advanceTimersByTimeAsync(1)
+    expect(fourthGranted).not.toHaveBeenCalled()
+    second()
+    ;(await fourth)()
+    third()
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('rotates waiting workspaces instead of draining one burst first', async () => {
+    const admission = createAdmission()
+    const first = await admission.acquire('a')
+    const order: string[] = []
+    const request = (workspace: string) =>
+      admission.acquire(workspace).then((release) => {
+        order.push(workspace)
+        release()
+      })
+    const waiting = [request('a'), request('a'), request('b')]
+    first()
+    await Promise.all(waiting)
+    expect(order).toEqual(['a', 'b', 'a'])
+  })
+
+  it('rejects overflow and recovers when the queue drains', async () => {
+    const admission = createAdmission(1, 1)
+    const first = await admission.acquire('a')
+    const waiting = admission.acquire('a')
+    await expect(admission.acquire('b')).rejects.toBeInstanceOf(WorkspaceFileSearchUnavailableError)
+    first()
+    ;(await waiting)()
+    ;(await admission.acquire('b'))()
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('expires waiting requests without executing them or leaking queue capacity', async () => {
+    const admission = createAdmission(1, 1)
+    const first = await admission.acquire('a')
+    const waiting = expect(admission.acquire('b')).rejects.toBeInstanceOf(
+      WorkspaceFileSearchUnavailableError
+    )
+    await vi.advanceTimersByTimeAsync(5000)
+    await waiting
+    const next = admission.acquire('c')
+    first()
+    ;(await next)()
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('checks expiry when granting even if a busy event loop has delayed the timer', async () => {
+    const admission = createAdmission()
+    const first = await admission.acquire('a')
+    const waiting = expect(admission.acquire('b')).rejects.toBeInstanceOf(
+      WorkspaceFileSearchUnavailableError
+    )
+    vi.setSystemTime(Date.now() + 5000)
+    first()
+    await waiting
+    ;(await admission.acquire('c'))()
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('removes cancelled waiters and their listeners without consuming a connection', async () => {
+    const admission = createAdmission(1, 1)
+    const first = await admission.acquire('a')
+    const controller = new AbortController()
+    const remove = vi.spyOn(controller.signal, 'removeEventListener')
+    const reason = new Error('cancelled')
+    const waiting = expect(admission.acquire('b', controller.signal)).rejects.toBe(reason)
+    controller.abort(reason)
+    await waiting
+    expect(remove).toHaveBeenCalledWith('abort', expect.any(Function))
+    expect(vi.getTimerCount()).toBe(0)
+    const next = admission.acquire('c')
+    first()
+    ;(await next)()
+  })
+
+  it('rejects an already cancelled caller without occupying a slot', async () => {
+    const admission = createAdmission()
+    const reason = new Error('cancelled')
+    await expect(admission.acquire('a', AbortSignal.abort(reason))).rejects.toBe(reason)
+    ;(await admission.acquire('b'))()
+  })
+
+  it('does not recycle an active slot on cancellation until the database work settles', async () => {
+    const admission = createAdmission()
+    const controller = new AbortController()
+    const first = await admission.acquire('a', controller.signal)
+    controller.abort()
+    const granted = vi.fn()
+    const next = admission.acquire('b').then((release) => {
+      granted()
+      release()
+    })
+    await vi.advanceTimersByTimeAsync(1)
+    expect(granted).not.toHaveBeenCalled()
+    first()
+    await next
+  })
+  it('leaves waiting capacity for another workspace when one burst hits its own cap', async () => {
+    const admission = createAdmission(1, 4, 2)
+    const release = await admission.acquire('hot')
+    const hot = [admission.acquire('hot'), admission.acquire('hot')]
+    await expect(admission.acquire('hot')).rejects.toBeInstanceOf(
+      WorkspaceFileSearchUnavailableError
+    )
+    const quiet = admission.acquire('quiet')
+    release()
+    ;(await hot[0])()
+    ;(await quiet)()
+    ;(await hot[1])()
+    ;(await admission.acquire('hot'))()
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it.each(['timeout', 'abort'] as const)(
+    'keeps a stalled operation counted after caller %s until the operation settles',
+    async (reason) => {
+      const admission = createAdmission()
+      const controller = new AbortController()
+      const dependency = Promise.withResolvers<void>()
+      const lateWork = vi.fn()
+      const running = admission.run(
+        'a',
+        async (signal) => {
+          await dependency.promise
+          signal.throwIfAborted()
+          lateWork()
+        },
+        controller.signal
+      )
+      const rejected = expect(running).rejects.toThrow(
+        reason === 'timeout' ? /timed out/ : /cancelled/
+      )
+      await vi.advanceTimersByTimeAsync(1)
+      if (reason === 'timeout') await vi.advanceTimersByTimeAsync(15000)
+      else controller.abort(new Error('cancelled'))
+      await rejected
+      const nextWork = vi.fn().mockResolvedValue('done')
+      const next = admission.run('b', nextWork)
+      await vi.advanceTimersByTimeAsync(1)
+      expect(nextWork).not.toHaveBeenCalled()
+      dependency.resolve()
+      await expect(next).resolves.toBe('done')
+      expect(lateWork).not.toHaveBeenCalled()
+      expect(vi.getTimerCount()).toBe(0)
+    }
+  )
+
+  it('releases a lease after an operation throws synchronously', async () => {
+    const admission = createAdmission()
+    await expect(
+      admission.run('a', () => {
+        throw new Error('operation failed')
+      })
+    ).rejects.toThrow('operation failed')
+    await expect(admission.run('b', async () => 'done')).resolves.toBe('done')
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('does not start an operation cancelled while waiting', async () => {
+    const admission = createAdmission()
+    const release = await admission.acquire('a')
+    const controller = new AbortController()
+    const operation = vi.fn()
+    const waiting = expect(admission.run('b', operation, controller.signal)).rejects.toThrow(
+      'cancelled'
+    )
+    controller.abort(new Error('cancelled'))
+    await waiting
+    release()
+    expect(operation).not.toHaveBeenCalled()
+    expect(vi.getTimerCount()).toBe(0)
+  })
+})