diff --git a/apps/sim/lib/workspace-files/application/search-workspace-file-content.test.ts b/apps/sim/lib/workspace-files/application/search-workspace-file-content.test.ts new file mode 100644 index 00000000000..0acaca86d86 --- /dev/null +++ b/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() + }) +}) diff --git a/apps/sim/lib/workspace-files/application/search-workspace-file-content.ts b/apps/sim/lib/workspace-files/application/search-workspace-file-content.ts index 2ff0ab35713..0c6e97ef267 100644 --- a/apps/sim/lib/workspace-files/application/search-workspace-file-content.ts +++ b/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,7 +58,7 @@ export const searchWorkspaceFileContent = defineAuthorizedWorkspaceFileUseCase({ includeSubfolders: input.includeSubfolders, }) : undefined - input.signal?.throwIfAborted() + signal?.throwIfAborted() try { return await searchWorkspaceFileIndex({ @@ -66,7 +66,7 @@ export const searchWorkspaceFileContent = defineAuthorizedWorkspaceFileUseCase({ pattern: compileFileSearchPattern(input.query, input.mode), maxResults: input.maxResults, folderScope, - signal: input.signal, + signal, }) } catch (error) { /** diff --git a/apps/sim/lib/workspace-files/search/README.md b/apps/sim/lib/workspace-files/search/README.md index 2d51795a2cf..e00b0318be8 100644 --- a/apps/sim/lib/workspace-files/search/README.md +++ b/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. diff --git a/apps/sim/lib/workspace-files/search/admission.test.ts b/apps/sim/lib/workspace-files/search/admission.test.ts new file mode 100644 index 00000000000..623c806a2aa --- /dev/null +++ b/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() + 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) + }) +}) diff --git a/apps/sim/lib/workspace-files/search/admission.ts b/apps/sim/lib/workspace-files/search/admission.ts new file mode 100644 index 00000000000..d00b24dcdeb --- /dev/null +++ b/apps/sim/lib/workspace-files/search/admission.ts @@ -0,0 +1,143 @@ +import { DB_POOL_PROFILES } from '@sim/db/pool-profiles' +import { DeadlineExceededError, withinDeadline } from '@/lib/core/utils/deadline' +import { + FILE_SEARCH_QUERY_WORKSPACE_CONCURRENCY, + FILE_SEARCH_QUEUE_MAX_PENDING, + FILE_SEARCH_QUEUE_TIMEOUT_MS, + FILE_SEARCH_STATEMENT_TIMEOUT_MS, +} from '@/lib/workspace-files/search/constants' +import { WorkspaceFileSearchUnavailableError } from '@/lib/workspace-files/search/errors' + +interface Waiter { + grant: () => void +} + +/** + * Bounds search work before it reaches the database pool. Waiting workspaces + * rotate after each grant; cancellation and expiry remove waiters immediately. + */ +export class FileSearchAdmission { + private active = 0 + private pending = 0 + private readonly workspaces = new Map>() + + constructor( + private readonly options: { + concurrency: number + maxPending: number + maxPendingPerWorkspace: number + timeoutMs: number + } + ) {} + + /** + * One caller deadline includes queueing, connection acquisition, and execution. + * The lease belongs to the underlying operation, even if its caller stops waiting. + */ + async run( + workspaceId: string, + operation: (signal: AbortSignal, deadlineAt: number) => Promise, + signal?: AbortSignal + ): Promise { + const deadlineAt = Date.now() + this.options.timeoutMs + FILE_SEARCH_STATEMENT_TIMEOUT_MS + try { + return await withinDeadline( + async (operationSignal) => { + const release = await this.acquire(workspaceId, operationSignal) + try { + operationSignal.throwIfAborted() + return await operation(operationSignal, deadlineAt) + } finally { + release() + } + }, + deadlineAt, + signal + ) + } catch (error) { + signal?.throwIfAborted() + if (error instanceof DeadlineExceededError) { + throw new WorkspaceFileSearchUnavailableError( + 'Workspace file search timed out. Retry shortly.' + ) + } + throw error + } + } + + async acquire(workspaceId: string, signal?: AbortSignal): Promise<() => void> { + signal?.throwIfAborted() + if (this.active < this.options.concurrency) return this.claim() + if ( + this.pending >= this.options.maxPending || + (this.workspaces.get(workspaceId)?.size ?? 0) >= this.options.maxPendingPerWorkspace + ) { + throw new WorkspaceFileSearchUnavailableError('Workspace file search is busy. Retry shortly.') + } + + return new Promise((resolve, reject) => { + const deadline = Date.now() + this.options.timeoutMs + let settled = false + const remove = () => { + settled = true + clearTimeout(timer) + signal?.removeEventListener('abort', abort) + const waiting = this.workspaces.get(workspaceId) + waiting?.delete(waiter) + if (waiting?.size === 0) this.workspaces.delete(workspaceId) + this.pending-- + } + const fail = (error: unknown) => { + if (settled) return + remove() + reject(error) + } + const expire = () => + fail( + new WorkspaceFileSearchUnavailableError('Workspace file search is busy. Retry shortly.') + ) + const abort = () => fail(signal?.reason) + const waiter: Waiter = { + grant: () => { + if (settled) return + if (signal?.aborted) return abort() + if (Date.now() >= deadline) return expire() + remove() + resolve(this.claim()) + }, + } + const timer = setTimeout(expire, this.options.timeoutMs) + const waiting = this.workspaces.get(workspaceId) ?? new Set() + waiting.add(waiter) + this.workspaces.set(workspaceId, waiting) + this.pending++ + signal?.addEventListener('abort', abort, { once: true }) + }) + } + + private claim(): () => void { + this.active++ + let released = false + return () => { + if (released) return + released = true + this.active-- + while (this.active < this.options.concurrency && this.workspaces.size > 0) { + const [workspaceId, waiters] = this.workspaces.entries().next().value! + const waiter = waiters.values().next().value! + waiter.grant() + if (this.workspaces.has(workspaceId)) { + this.workspaces.delete(workspaceId) + this.workspaces.set(workspaceId, waiters) + } + } + } + } +} + +export const fileSearchAdmission = new FileSearchAdmission({ + concurrency: DB_POOL_PROFILES.search.primaryMax, + maxPending: FILE_SEARCH_QUEUE_MAX_PENDING, + maxPendingPerWorkspace: FILE_SEARCH_QUERY_WORKSPACE_CONCURRENCY, + timeoutMs: FILE_SEARCH_QUEUE_TIMEOUT_MS, +}) diff --git a/apps/sim/lib/workspace-files/search/chunks.integration.ts b/apps/sim/lib/workspace-files/search/chunks.integration.ts index 80969864f3e..c5759de539a 100644 --- a/apps/sim/lib/workspace-files/search/chunks.integration.ts +++ b/apps/sim/lib/workspace-files/search/chunks.integration.ts @@ -1,6 +1,7 @@ import { createHash } from 'node:crypto' import { readFileSync, writeFileSync } from 'node:fs' import { resolve } from 'node:path' +import { DB_POOL_PROFILES } from '@sim/db/pool-profiles' import { withUtcTimestamps } from '@sim/db/timestamps' import { generateId } from '@sim/utils/id' import { sql } from 'drizzle-orm' @@ -9,8 +10,15 @@ import postgres from 'postgres' import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest' import { buildLiteralMatchStart } from '@/lib/workspace-files/search/sql-pattern' -const database = vi.hoisted(() => ({ current: undefined as PostgresJsDatabase | undefined })) +const database = vi.hoisted(() => ({ + current: undefined as PostgresJsDatabase | undefined, + search: undefined as PostgresJsDatabase | undefined, +})) vi.mock('@sim/db', () => ({ + dbFor: (role: string) => { + if (role !== 'search' || !database.search) throw new Error('Search database not initialized') + return database.search + }, get db() { if (!database.current) throw new Error('Test database not initialized') return database.current @@ -27,6 +35,8 @@ import { FILE_SEARCH_CLEANUP_BATCH_ROWS, FILE_SEARCH_CLEANUP_BUDGET_MS, FILE_SEARCH_CLEANUP_MAX_BATCHES, + FILE_SEARCH_QUERY_GLOBAL_CONCURRENCY, + FILE_SEARCH_QUERY_WORKSPACE_CONCURRENCY, } from '@/lib/workspace-files/search/constants' import { prepareWorkspaceFileSearchDispatch } from '@/lib/workspace-files/search/dispatcher' import { @@ -73,6 +83,16 @@ describe('chunked workspace file search on PostgreSQL', () => { onnotice: () => {}, }) ) + const searchConnection = postgres( + databaseUrl, + withUtcTimestamps({ + max: DB_POOL_PROFILES.search.primaryMax, + prepare: false, + fetch_types: false, + connection: { search_path: `${schema},public` }, + onnotice: () => {}, + }) + ) async function addFile( fileId: string, @@ -137,7 +157,8 @@ describe('chunked workspace file search on PostgreSQL', () => { for (const statement of source.split('--> statement-breakpoint')) if (statement.trim()) await connection.unsafe(statement) } - database.current = drizzle(connection, { + database.current = drizzle(connection) + database.search = drizzle(searchConnection, { logger: { logQuery(query, params) { if ( @@ -163,7 +184,8 @@ describe('chunked workspace file search on PostgreSQL', () => { await connection`DROP SCHEMA ${connection(schema)} CASCADE` } finally { database.current = undefined - await connection.end() + database.search = undefined + await Promise.all([connection.end(), searchConnection.end()]) } }) @@ -475,11 +497,75 @@ describe('chunked workspace file search on PostgreSQL', () => { ) }) - it('releases query admission slots after a busy response', async () => { + async function withOccupiedPool(client: postgres.Sql, count: number, run: () => Promise) { + let release!: () => void + const held = new Promise((resolve) => { + release = resolve + }) + let started = 0 + const transactions: Promise[] = [] + const ready = new Promise((resolve, reject) => { + for (let i = 0; i < count; i++) { + transactions.push( + client + .begin(async () => { + if (++started === count) resolve() + await held + }) + .catch(reject) + ) + } + }) + try { + await ready + await run() + } finally { + release() + await Promise.all(transactions) + } + } + + it('searches while every shared application connection is occupied', async () => { + await index('needle') + await withOccupiedPool(connection, 4, async () => { + expect((await search('needle')).results).toHaveLength(1) + }) + }) + + it('keeps application queries available while every search connection is occupied', async () => { + await withOccupiedPool(searchConnection, DB_POOL_PROFILES.search.primaryMax, async () => { + expect((await connection`SELECT 1 AS available`)[0].available).toBe(1) + }) + }) + + it('serves a twenty-search workspace burst through the bounded search pool', async () => { + await index('needle') + const results = await Promise.all(Array.from({ length: 20 }, () => search('needle'))) + expect(results).toHaveLength(20) + for (const result of results) expect(result.results).toHaveLength(1) + }) + + it('admits a search up to the workspace and global ceilings', async () => { + const held = await connection.reserve() + try { + await held`BEGIN` + await held`SELECT pg_advisory_xact_lock(hashtextextended('workspace-file-search-read:workspace:workspace-1:' || n::text, 0)) FROM generate_series(1, ${FILE_SEARCH_QUERY_WORKSPACE_CONCURRENCY - 1}) n` + await held`SELECT pg_advisory_xact_lock(hashtextextended('workspace-file-search-read:global:' || n::text, 0)) FROM generate_series(1, ${FILE_SEARCH_QUERY_GLOBAL_CONCURRENCY - 1}) n` + expect((await search('needle')).results).toEqual([]) + } finally { + await held`ROLLBACK` + held.release() + } + }) + + it.each([ + ['workspace:workspace-1', FILE_SEARCH_QUERY_WORKSPACE_CONCURRENCY], + ['global', FILE_SEARCH_QUERY_GLOBAL_CONCURRENCY], + ])('releases query admission slots after %s saturation', async (scope, capacity) => { const held = await connection.reserve() try { await held`BEGIN` - await held`SELECT pg_advisory_xact_lock(hashtextextended('workspace-file-search-read:workspace:workspace-1:' || n::text, 0)) FROM generate_series(1, 2) n` + await held`SELECT pg_advisory_xact_lock(hashtextextended('workspace-file-search-read:' || ${scope} || ':' || n::text, 0)) FROM generate_series(1, ${capacity}) n` await expect(search('needle')).rejects.toThrow('busy') } finally { await held`ROLLBACK` diff --git a/apps/sim/lib/workspace-files/search/constants.ts b/apps/sim/lib/workspace-files/search/constants.ts index bf4214fd636..ed26fd31a70 100644 --- a/apps/sim/lib/workspace-files/search/constants.ts +++ b/apps/sim/lib/workspace-files/search/constants.ts @@ -23,8 +23,10 @@ export const FILE_SEARCH_MAX_PREVIEW_BYTES = 2 * 1024 export const FILE_SEARCH_CHUNK_BYTES = 8 * 1024 export const FILE_SEARCH_CANDIDATE_PAGE_SIZE = 16 export const FILE_SEARCH_CANDIDATE_PROBE_SIZE = 256 -export const FILE_SEARCH_QUERY_GLOBAL_CONCURRENCY = 10 -export const FILE_SEARCH_QUERY_WORKSPACE_CONCURRENCY = 2 +export const FILE_SEARCH_QUERY_GLOBAL_CONCURRENCY = 5000 +export const FILE_SEARCH_QUERY_WORKSPACE_CONCURRENCY = 20 +export const FILE_SEARCH_QUEUE_MAX_PENDING = 100 +export const FILE_SEARCH_QUEUE_TIMEOUT_MS = 5000 export const FILE_SEARCH_CANDIDATE_LITERAL_CHARS = 3 export const FILE_SEARCH_BUILD_LEASE_MS = 20 * 60 * 1000 export const FILE_SEARCH_CLEANUP_BATCH_ROWS = 1000 diff --git a/apps/sim/lib/workspace-files/search/errors.ts b/apps/sim/lib/workspace-files/search/errors.ts new file mode 100644 index 00000000000..3c45e22f4a8 --- /dev/null +++ b/apps/sim/lib/workspace-files/search/errors.ts @@ -0,0 +1,7 @@ +/** A retryable search failure caused by contention or index maintenance. */ +export class WorkspaceFileSearchUnavailableError extends Error { + constructor(message: string) { + super(message) + this.name = 'WorkspaceFileSearchUnavailableError' + } +} diff --git a/apps/sim/lib/workspace-files/search/repository.test.ts b/apps/sim/lib/workspace-files/search/repository.test.ts index 2cb40bf2c2e..8c6f0c6081f 100644 --- a/apps/sim/lib/workspace-files/search/repository.test.ts +++ b/apps/sim/lib/workspace-files/search/repository.test.ts @@ -3,15 +3,13 @@ */ import { db } from '@sim/db' import { dbChainMockFns, resetDbChainMock } from '@sim/testing' -import { beforeEach, describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { WorkspaceFileSearchUnavailableError } from '@/lib/workspace-files/search/errors' import { compileFileSearchPattern, FileSearchPatternError, } from '@/lib/workspace-files/search/pattern' -import { - searchWorkspaceFileIndex, - WorkspaceFileSearchUnavailableError, -} from '@/lib/workspace-files/search/repository' +import { searchWorkspaceFileIndex } from '@/lib/workspace-files/search/repository' /** * The shape a failed query really arrives in, captured from PostgreSQL 17 @@ -24,6 +22,7 @@ function driverError(code: string): Error { } describe('searchWorkspaceFileIndex fault mapping', () => { + afterEach(() => vi.useRealTimers()) beforeEach(() => { resetDbChainMock() }) @@ -98,4 +97,26 @@ describe('searchWorkspaceFileIndex fault mapping', () => { expect(guards).toContain('lock_timeout') expect(db.transaction).toHaveBeenCalled() }) + it('bounds the caller while BEGIN is stalled and does not run late search SQL', async () => { + vi.useFakeTimers() + const acquired = Promise.withResolvers() + let transactionFinished!: Promise + dbChainMockFns.transaction.mockImplementationOnce((callback) => { + transactionFinished = acquired.promise.then(() => callback(db)) + return transactionFinished + }) + const waiting = expect( + searchWorkspaceFileIndex({ + workspaceId: 'workspace-1', + pattern: compileFileSearchPattern('needle', 'exact'), + maxResults: 50, + }) + ).rejects.toBeInstanceOf(WorkspaceFileSearchUnavailableError) + await vi.advanceTimersByTimeAsync(15000) + await waiting + expect(dbChainMockFns.execute).not.toHaveBeenCalled() + acquired.resolve() + await expect(transactionFinished).rejects.toThrow('Operation deadline expired') + expect(dbChainMockFns.execute).not.toHaveBeenCalled() + }) }) diff --git a/apps/sim/lib/workspace-files/search/repository.ts b/apps/sim/lib/workspace-files/search/repository.ts index c4315a64cc7..2c97222ca28 100644 --- a/apps/sim/lib/workspace-files/search/repository.ts +++ b/apps/sim/lib/workspace-files/search/repository.ts @@ -1,10 +1,11 @@ -import { db } from '@sim/db' +import { dbFor } from '@sim/db' import { workspaceFileSearchRevision, workspaceFiles } from '@sim/db/schema' import { getPostgresErrorCode } from '@sim/utils/errors' import { and, eq, inArray, isNull, or, type SQL, type SQLWrapper, sql } from 'drizzle-orm' import type { DbTransaction } from '@/lib/db/types' import type { FolderIdScope } from '@/lib/folders/scope' import type { WorkspaceFileSecretProvenanceIdentity } from '@/lib/uploads/contexts/workspace/workspace-file-secret-provenance' +import { fileSearchAdmission } from '@/lib/workspace-files/search/admission' import { probeFileSearchCandidates, readOrderedFileSearchCandidates, @@ -19,6 +20,7 @@ import { FILE_SEARCH_QUERY_WORKSPACE_CONCURRENCY, FILE_SEARCH_STATEMENT_TIMEOUT_MS, } from '@/lib/workspace-files/search/constants' +import { WorkspaceFileSearchUnavailableError } from '@/lib/workspace-files/search/errors' import { alignToCodePoints, type CompiledFileSearchPattern, @@ -71,18 +73,6 @@ const QUERY_CANCELED = '57014' const LOCK_NOT_AVAILABLE = '55P03' const INVALID_REGULAR_EXPRESSION = '2201B' -/** - * The search could not run, for a reason the caller did not cause and cannot fix - * by changing the query — distinct from {@link FileSearchPatternError}, so a - * surface reports "try again" rather than blaming the pattern. - */ -export class WorkspaceFileSearchUnavailableError extends Error { - constructor(message: string) { - super(message) - this.name = 'WorkspaceFileSearchUnavailableError' - } -} - /** Query deadlines cover expensive patterns; lock and transaction faults are retryable. */ function asFileSearchFault(error: unknown): Error | null { const sqlState = getPostgresErrorCode(error) @@ -237,166 +227,189 @@ export async function searchWorkspaceFileIndex({ */ const folderPredicate = folderScope ? buildFolderPredicate(folderScope) : undefined - try { - /** Metadata pages and line reads share one deadline and a consistent revision snapshot. */ - const { rows, coverageRows } = await db.transaction( - async (tx) => { - await configureFileSearchTransaction(tx) + return fileSearchAdmission.run( + workspaceId, + async (signal, deadlineAt) => { + try { + signal.throwIfAborted() + /** Metadata pages and line reads share one deadline and a consistent revision snapshot. */ + const { rows, coverageRows } = await dbFor('search').transaction( + async (tx) => { + signal?.throwIfAborted() + const deadline = Math.min(deadlineAt, Date.now() + FILE_SEARCH_STATEMENT_TIMEOUT_MS) + const remainingMs = deadline - Date.now() + if (remainingMs <= 0) { + throw new WorkspaceFileSearchUnavailableError( + 'Workspace file search timed out. Retry shortly.' + ) + } + await configureFileSearchTransaction(tx, { statementTimeout: remainingMs }) - /** Transaction-owned slots release on completion, cancellation, or connection loss. */ - for (const [scope, capacity] of [ - [`workspace:${workspaceId}`, FILE_SEARCH_QUERY_WORKSPACE_CONCURRENCY], - ['global', FILE_SEARCH_QUERY_GLOBAL_CONCURRENCY], - ] as const) { - const slots = await tx.execute(sql`SELECT slot FROM generate_series(1, ${capacity}) slot + /** Transaction-owned slots release on completion, cancellation, or connection loss. */ + for (const [scope, capacity] of [ + [`workspace:${workspaceId}`, FILE_SEARCH_QUERY_WORKSPACE_CONCURRENCY], + ['global', FILE_SEARCH_QUERY_GLOBAL_CONCURRENCY], + ] as const) { + const slots = + await tx.execute(sql`SELECT slot FROM generate_series(1, ${capacity}) slot WHERE pg_try_advisory_xact_lock(hashtextextended('workspace-file-search-read:' || ${scope} || ':' || slot::text, 0)) LIMIT 1`) - if (!slots.length) - throw new WorkspaceFileSearchUnavailableError( - 'Workspace file search is busy. Retry shortly.' - ) - } + if (!slots.length) + throw new WorkspaceFileSearchUnavailableError( + 'Workspace file search is busy. Retry shortly.' + ) + } - const deadline = Date.now() + FILE_SEARCH_STATEMENT_TIMEOUT_MS - const guardRemainingTime = async () => { - signal?.throwIfAborted() - const remaining = deadline - Date.now() - if (remaining <= 0) - throw new WorkspaceFileSearchUnavailableError( - 'Search timed out. Narrow the query or folder scope.' - ) - await tx.execute(sql`SELECT set_config('statement_timeout', ${`${remaining}ms`}, true)`) - } - /** Probe without a global sort. Rare queries finish here; broad queries scan files in order. */ - const probed = await probeFileSearchCandidates(tx, { - workspaceId, - pattern, - folderPredicate, - }) - const broad = probed.length > FILE_SEARCH_CANDIDATE_PROBE_SIZE - const matchedRows: SearchRow[] = [] - let after: { name: string; id: string; lineStart: number } | undefined - while (matchedRows.length <= maxResults) { - await guardRemainingTime() - let candidates = probed - if (broad) { - candidates = await readOrderedFileSearchCandidates( - tx, - { workspaceId, pattern, folderPredicate }, - after - ) - } - for ( - let position = 0; - position < candidates.length && matchedRows.length <= maxResults; - ) { - const first = candidates[position++] - const batch = [first] - if (first.fragment) { - while ( - position < candidates.length && - candidates[position].buildId === first.buildId && - candidates[position].lineStart === first.lineStart + const guardRemainingTime = async () => { + signal?.throwIfAborted() + const remaining = deadline - Date.now() + if (remaining <= 0) + throw new WorkspaceFileSearchUnavailableError( + 'Search timed out. Narrow the query or folder scope.' + ) + await tx.execute( + sql`SELECT set_config('statement_timeout', ${`${remaining}ms`}, true)` ) - position++ - } else { - while ( - position < candidates.length && - batch.length < FILE_SEARCH_CANDIDATE_PAGE_SIZE && - !candidates[position].fragment - ) - batch.push(candidates[position++]) } await guardRemainingTime() - matchedRows.push( - ...(await readCandidateLines(tx, batch, pattern, maxResults + 1 - matchedRows.length)) - ) - } - if (!broad || candidates.length < FILE_SEARCH_CANDIDATE_PAGE_SIZE) break - const last = candidates.at(-1)! - after = { name: last.fileName, id: last.fileId, lineStart: last.lineStart } - } - await guardRemainingTime() - signal?.throwIfAborted() - const coverage = await tx - .select({ - readyFiles: sql`count(*) filter (where ${workspaceFileSearchRevision.status} = 'ready' AND ${workspaceFileSearchRevision.buildId} IS NOT NULL)::int`, - pendingFiles: sql`count(*) filter (where ${workspaceFileSearchRevision.status} is null or ${workspaceFileSearchRevision.status} = 'pending' or (${workspaceFileSearchRevision.status} = 'ready' AND ${workspaceFileSearchRevision.buildId} IS NULL))::int`, - failedFiles: sql`count(*) filter (where ${workspaceFileSearchRevision.status} = 'failed')::int`, - skippedFiles: sql`count(*) filter (where ${workspaceFileSearchRevision.status} = 'skipped')::int`, - partialFiles: sql`0::int`, - }) - .from(workspaceFiles) - .leftJoin( - workspaceFileSearchRevision, - and( - eq(workspaceFileSearchRevision.fileId, workspaceFiles.id), - eq( - workspaceFileSearchRevision.sourceContentUpdatedAt, - workspaceFiles.contentUpdatedAt + /** Probe without a global sort. Rare queries finish here; broad queries scan files in order. */ + const probed = await probeFileSearchCandidates(tx, { + workspaceId, + pattern, + folderPredicate, + }) + const broad = probed.length > FILE_SEARCH_CANDIDATE_PROBE_SIZE + const matchedRows: SearchRow[] = [] + let after: { name: string; id: string; lineStart: number } | undefined + while (matchedRows.length <= maxResults) { + await guardRemainingTime() + let candidates = probed + if (broad) { + candidates = await readOrderedFileSearchCandidates( + tx, + { workspaceId, pattern, folderPredicate }, + after + ) + } + for ( + let position = 0; + position < candidates.length && matchedRows.length <= maxResults; + ) { + const first = candidates[position++] + const batch = [first] + if (first.fragment) { + while ( + position < candidates.length && + candidates[position].buildId === first.buildId && + candidates[position].lineStart === first.lineStart + ) + position++ + } else { + while ( + position < candidates.length && + batch.length < FILE_SEARCH_CANDIDATE_PAGE_SIZE && + !candidates[position].fragment + ) + batch.push(candidates[position++]) + } + await guardRemainingTime() + matchedRows.push( + ...(await readCandidateLines( + tx, + batch, + pattern, + maxResults + 1 - matchedRows.length + )) + ) + } + if (!broad || candidates.length < FILE_SEARCH_CANDIDATE_PAGE_SIZE) break + const last = candidates.at(-1)! + after = { name: last.fileName, id: last.fileId, lineStart: last.lineStart } + } + await guardRemainingTime() + signal?.throwIfAborted() + const coverage = await tx + .select({ + readyFiles: sql`count(*) filter (where ${workspaceFileSearchRevision.status} = 'ready' AND ${workspaceFileSearchRevision.buildId} IS NOT NULL)::int`, + pendingFiles: sql`count(*) filter (where ${workspaceFileSearchRevision.status} is null or ${workspaceFileSearchRevision.status} = 'pending' or (${workspaceFileSearchRevision.status} = 'ready' AND ${workspaceFileSearchRevision.buildId} IS NULL))::int`, + failedFiles: sql`count(*) filter (where ${workspaceFileSearchRevision.status} = 'failed')::int`, + skippedFiles: sql`count(*) filter (where ${workspaceFileSearchRevision.status} = 'skipped')::int`, + partialFiles: sql`0::int`, + }) + .from(workspaceFiles) + .leftJoin( + workspaceFileSearchRevision, + and( + eq(workspaceFileSearchRevision.fileId, workspaceFiles.id), + eq( + workspaceFileSearchRevision.sourceContentUpdatedAt, + workspaceFiles.contentUpdatedAt + ) + ) + ) + .where( + and( + eq(workspaceFiles.workspaceId, workspaceId), + eq(workspaceFiles.context, 'workspace'), + isNull(workspaceFiles.deletedAt), + folderPredicate + ) ) - ) - ) - .where( - and( - eq(workspaceFiles.workspaceId, workspaceId), - eq(workspaceFiles.context, 'workspace'), - isNull(workspaceFiles.deletedAt), - folderPredicate - ) - ) - return { rows: matchedRows, coverageRows: coverage } - }, - { isolationLevel: 'repeatable read', accessMode: 'read only' } - ) + return { rows: matchedRows, coverageRows: coverage } + }, + { isolationLevel: 'repeatable read', accessMode: 'read only' } + ) - signal?.throwIfAborted() - const resultRows = rows.slice(0, maxResults) - const indexStatus = coverageRows[0] ?? { - readyFiles: 0, - pendingFiles: 0, - failedFiles: 0, - skippedFiles: 0, - partialFiles: 0, - } - const sourcesByFileId = new Map() - for (const row of resultRows) { - sourcesByFileId.set(row.fileId, { - identity: { - fileId: row.fileId, - key: row.fileKey, - context: 'workspace', - contentUpdatedAt: row.contentUpdatedAt, - }, - ownerUserId: row.ownerUserId, - }) - } + signal?.throwIfAborted() + const resultRows = rows.slice(0, maxResults) + const indexStatus = coverageRows[0] ?? { + readyFiles: 0, + pendingFiles: 0, + failedFiles: 0, + skippedFiles: 0, + partialFiles: 0, + } + const sourcesByFileId = new Map() + for (const row of resultRows) { + sourcesByFileId.set(row.fileId, { + identity: { + fileId: row.fileId, + key: row.fileKey, + context: 'workspace', + contentUpdatedAt: row.contentUpdatedAt, + }, + ownerUserId: row.ownerUserId, + }) + } - const results = resultRows.map((row) => ({ - fileId: row.fileId, - lineNumber: row.lineNumber, - text: createFileSearchPreview(row.content, pattern, undefined, { - prefixOmitted: row.prefixOmitted, - suffixOmitted: row.suffixOmitted, - matchRange: - pattern.mode === 'regex' - ? toPreviewRange(row.content, row.matchStart, row.matchEnd) - : undefined, - }), - })) - signal?.throwIfAborted() - return { - results, - count: results.length, - truncated: rows.length > maxResults, - complete: indexStatus.pendingFiles === 0 && indexStatus.failedFiles === 0, - indexStatus, - sources: [...sourcesByFileId.values()], - } - } catch (error) { - signal?.throwIfAborted() - const fault = asFileSearchFault(error) - if (fault) throw fault - throw error - } + const results = resultRows.map((row) => ({ + fileId: row.fileId, + lineNumber: row.lineNumber, + text: createFileSearchPreview(row.content, pattern, undefined, { + prefixOmitted: row.prefixOmitted, + suffixOmitted: row.suffixOmitted, + matchRange: + pattern.mode === 'regex' + ? toPreviewRange(row.content, row.matchStart, row.matchEnd) + : undefined, + }), + })) + signal?.throwIfAborted() + return { + results, + count: results.length, + truncated: rows.length > maxResults, + complete: indexStatus.pendingFiles === 0 && indexStatus.failedFiles === 0, + indexStatus, + sources: [...sourcesByFileId.values()], + } + } catch (error) { + signal?.throwIfAborted() + const fault = asFileSearchFault(error) + if (fault) throw fault + throw error + } + }, + signal + ) } diff --git a/packages/db/db.ts b/packages/db/db.ts index 9b80dc8f60b..7176f92c6f1 100644 --- a/packages/db/db.ts +++ b/packages/db/db.ts @@ -1,3 +1,4 @@ +import { DB_POOL_PROFILES } from '@sim/db/pool-profiles' import { createLogger } from '@sim/logger' import { drizzle } from 'drizzle-orm/postgres-js' import postgres from 'postgres' @@ -8,21 +9,6 @@ import { instrumentPoolClient } from './tx-tripwire' const logger = createLogger('Db') -/** - * Per-role pool profiles. Starting numbers — validate against real per-role - * process counts (PgBouncer transaction mode, max_connections=200). - */ -export const DB_POOL_PROFILES = { - web: { primaryMax: 10, replicaMax: 4, appName: 'sim-app' }, - // 5, not 3 — one run can need 3+ simultaneous connections (parallel queries + - // overlapping logging writes); 3 risks intra-run deadlock. - trigger: { primaryMax: 5, replicaMax: 2, appName: 'sim-trigger' }, - realtime: { primaryMax: 5, replicaMax: 3, appName: 'sim-realtime' }, - // Sub-process pools, selected per call-site via dbFor() — never via SIM_DB_ROLE. - cleanup: { primaryMax: 5, replicaMax: 2, appName: 'sim-cleanup' }, - exec: { primaryMax: 10, replicaMax: 4, appName: 'sim-exec' }, -} as const - /** Roles a whole process runs as (via SIM_DB_ROLE). */ const PROCESS_ROLES = ['web', 'trigger', 'realtime'] as const @@ -129,7 +115,8 @@ const processUrlEnvVar = process.env[`DATABASE_URL_${role.toUpperCase()}`] * cached per role. Unlike the process-wide `db` (selected by `SIM_DB_ROLE`), * these are selected per call-site so a workload running inside an existing * process — cleanup jobs in the trigger worker, inline execution log writes in - * the web server — gets its own connection budget and PgBouncer pool. + * the web server — gets its own client connection budget. A separate PgBouncer + * server pool additionally requires a distinct database/user pool configuration. * * Resolves `DATABASE_URL_` with fallback to the URL the process itself * resolved (`DATABASE_URL_`, then base `DATABASE_URL`), so an diff --git a/packages/db/index.ts b/packages/db/index.ts index a472c56fa07..cd24c65a935 100644 --- a/packages/db/index.ts +++ b/packages/db/index.ts @@ -1,5 +1,6 @@ export * from './connection-url' export * from './db' +export * from './pool-profiles' export * from './schema' export * from './triggers' export { instrumentPoolClient, runOutsideTransactionContext } from './tx-tripwire' diff --git a/packages/db/package.json b/packages/db/package.json index df53ffeeaad..aab3d147679 100644 --- a/packages/db/package.json +++ b/packages/db/package.json @@ -21,6 +21,10 @@ "types": "./timestamps.ts", "default": "./timestamps.ts" }, + "./pool-profiles": { + "types": "./pool-profiles.ts", + "default": "./pool-profiles.ts" + }, "./sso-primary-provider": { "types": "./sso-primary-provider.ts", "default": "./sso-primary-provider.ts" diff --git a/packages/db/pool-profiles.ts b/packages/db/pool-profiles.ts new file mode 100644 index 00000000000..7ffb19bd73a --- /dev/null +++ b/packages/db/pool-profiles.ts @@ -0,0 +1,11 @@ +/** Per-process budgets; total database connections also depend on the number of processes. */ +export const DB_POOL_PROFILES = { + web: { primaryMax: 10, replicaMax: 4, appName: 'sim-app' }, + /** One run can need parallel queries and overlapping logging writes. */ + trigger: { primaryMax: 5, replicaMax: 2, appName: 'sim-trigger' }, + realtime: { primaryMax: 5, replicaMax: 3, appName: 'sim-realtime' }, + /** Sub-process pools are selected per call site through dbFor(). */ + cleanup: { primaryMax: 5, replicaMax: 2, appName: 'sim-cleanup' }, + exec: { primaryMax: 10, replicaMax: 4, appName: 'sim-exec' }, + search: { primaryMax: 5, replicaMax: 0, appName: 'sim-search' }, +} as const