Skip to content

Commit 09627c5

Browse files
authored
improvement(search): overlap live search stages and report every unsearched source (#9514)
* improvement(search): overlap live search stages and report every unsearched source * fix(search): defer previews when verification shares their budget, and bound session opening by the account deadline
1 parent 22bf85a commit 09627c5

10 files changed

Lines changed: 552 additions & 123 deletions

File tree

‎apps/sim/lib/core/utils/concurrency.ts‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,27 @@ export function mapSettledWithConcurrency<T, R>(
5656
})
5757
}
5858

59+
/**
60+
* A gate that runs at most `limit` of the tasks passed to it at once, starting the rest in call
61+
* order as slots free. Bounds one phase of otherwise concurrent work, such as the part that holds
62+
* database connections, without bounding the rest of it.
63+
*/
64+
export function createConcurrencyLimit(limit: number) {
65+
let active = 0
66+
const waiting: (() => void)[] = []
67+
return async <R>(task: () => Promise<R>): Promise<R> => {
68+
if (active >= limit) await new Promise<void>((resolve) => waiting.push(resolve))
69+
else active++
70+
try {
71+
return await task()
72+
} finally {
73+
const next = waiting.shift()
74+
if (next) next()
75+
else active--
76+
}
77+
}
78+
}
79+
5980
/** Default bound for per-row object-storage materialization fan-out. */
6081
export const MATERIALIZE_CONCURRENCY = 20
6182

‎apps/sim/lib/sim-search/live/account-session.ts‎

Lines changed: 17 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -11,14 +11,15 @@ import {
1111
createNativeClient,
1212
NATIVE_SEARCH_REQUEST_BUDGET,
1313
NativeSearchError,
14+
withRequestSignal,
1415
} from '@/lib/sim-search/live/http'
1516
import { readHubSpotMcp, searchHubSpotMcp } from '@/lib/sim-search/live/hubspot-mcp'
1617
import { readLucidMcp, searchLucidMcp } from '@/lib/sim-search/live/lucid-mcp'
1718
import { createManagedSearchMcpClient } from '@/lib/sim-search/live/managed-mcp'
1819
import { isManagedSearchMcpProvider } from '@/lib/sim-search/live/managed-mcp-config'
1920
import { readNotionMcp, searchNotionMcp } from '@/lib/sim-search/live/notion-mcp'
2021
import { createPolicyVerifier } from '@/lib/sim-search/live/policy'
21-
import type { LiveSearchPolicy } from '@/lib/sim-search/live/policy-schema'
22+
import { type LiveSearchPolicy, requiresScopedRetrieval } from '@/lib/sim-search/live/policy-schema'
2223
import { livePolicyFor, loadLiveSearchPolicies } from '@/lib/sim-search/live/policy-store'
2324
import { LIVE_SEARCH_PROVIDER_CATALOG } from '@/lib/sim-search/live/provider-catalog'
2425
import { isSearchProviderEnabled } from '@/lib/sim-search/live/provider-rollout'
@@ -51,8 +52,16 @@ export interface LiveAccountSession {
5152
/** True only when the document is inside the source boundary and the member may read it. */
5253
verify(document: Reference): Promise<boolean>
5354
read(reference: Reference, filters?: WorkspaceSearchFilters): Promise<NativeDocument>
54-
/** Full text of a verified search result whose provider returned only an excerpt or metadata. */
55-
previewText(document: NativeDocument): Promise<string | undefined>
55+
/**
56+
* Full text of a search result whose provider returned only an excerpt or metadata; `signal`
57+
* cancels the read when its preview is no longer wanted.
58+
*/
59+
previewText(document: NativeDocument, signal: AbortSignal): Promise<string | undefined>
60+
/**
61+
* True when `verify` spends requests from the member budget `previewText` also draws on, so a
62+
* preview read before verification finishes could leave a later result unverified.
63+
*/
64+
verifySharesPreviewBudget: boolean
5665
/** Checks a fetched document against the source settings saved now, not at session start. */
5766
verifyCurrent(document: NativeDocument): Promise<boolean>
5867
}
@@ -174,14 +183,15 @@ export async function openLiveAccountSession(
174183
signal,
175184
pool: input.pool,
176185
})
177-
if (service) return service
186+
if (service) return { ...service, spendsMemberBudget: service.spendsMemberBudget === true }
178187
const verifyPolicy = createPolicyVerifier(provider, memberPolicy, client, origin, mcp, {
179188
fresh,
180189
})
181190
return {
182191
policy: memberPolicy,
183192
partial: false,
184193
scopeSearch: undefined,
194+
spendsMemberBudget: requiresScopedRetrieval(provider, memberPolicy),
185195
verify: (document: Reference) => verifyPolicy(document, document.accessMetadata),
186196
}
187197
}
@@ -196,6 +206,7 @@ export async function openLiveAccountSession(
196206
},
197207
policy: boundary.policy,
198208
servicePartial: boundary.partial,
209+
verifySharesPreviewBudget: boundary.spendsMemberBudget,
199210
async search(search) {
200211
const searchOnce = async (request: NativeSearchInput): Promise<NativePage> => {
201212
const scoped = boundary.scopeSearch ? boundary.scopeSearch(request) : request
@@ -226,9 +237,9 @@ export async function openLiveAccountSession(
226237
})
227238
return readMcp(reference)
228239
},
229-
async previewText(document) {
240+
async previewText(document, previewSignal) {
230241
if (admin || !client) return undefined
231-
return previewNativeText(provider, client, document)
242+
return previewNativeText(provider, withRequestSignal(client, previewSignal), document)
232243
},
233244
async verifyCurrent(document) {
234245
if (!(await isSearchProviderEnabled(provider, resourceScopeFromOwner(owner)))) return false

‎apps/sim/lib/sim-search/live/application.test.ts‎

Lines changed: 172 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { ErrorCode, McpError } from '@modelcontextprotocol/sdk/types.js'
2-
import { dbChainMockFns, resetDbChainMock } from '@sim/testing'
2+
import { createDeferred, dbChainMockFns, resetDbChainMock } from '@sim/testing'
33
import { createSessionPrincipal } from '@sim/testing/factories/principal.factory'
44
import { setEnv } from '@sim/testing/mocks/env.mock'
55
import {
@@ -740,6 +740,76 @@ describe('authorized live retrieval', () => {
740740
})
741741
expect(mocks.preview).not.toHaveBeenCalled()
742742
})
743+
744+
it('reads a permitted result’s preview while others verify, and never a denied one’s', async () => {
745+
mocks.search.mockResolvedValue({ documents: excerpts })
746+
const pendingB = createDeferred<boolean>()
747+
mocks.service.mockResolvedValue({
748+
policy: defaultLiveSearchPolicy(),
749+
partial: false,
750+
verify: async ({ id }: { id: string }) => (id === 'b' ? pendingB.promise : id !== 'c'),
751+
})
752+
mocks.preview.mockImplementation(
753+
async (_provider, _client, { id }: { id: string }) => `Full ${id} launch text`
754+
)
755+
const previewedIds = () => mocks.preview.mock.calls.map(([, , { id }]) => id)
756+
const pending = searchLiveKnowledge.execute({ principal, input })
757+
await vi.waitFor(() => expect(previewedIds()).toEqual(['a', 'd']))
758+
pendingB.resolve(true)
759+
const result = await pending
760+
expect(contents(result)).toEqual([
761+
'Full a launch text',
762+
'Full b launch text',
763+
'Full d launch text',
764+
])
765+
expect(previewedIds()).toEqual(['a', 'd', 'b'])
766+
})
767+
768+
it('reads no preview until a verification that spends the same request budget finishes', async () => {
769+
const googleDoc = 'application/vnd.google-apps.document'
770+
vi.mocked(livePolicyFor).mockReturnValue({
771+
...defaultLiveSearchPolicy(),
772+
fileTypes: [googleDoc],
773+
})
774+
mocks.search.mockResolvedValue({ documents: excerpts })
775+
const pendingB = createDeferred<unknown>()
776+
mocks.json.mockReset().mockImplementation(async (path: string) => {
777+
const id = path.split('/').at(-1)
778+
return id === 'b' ? pendingB.promise : { id, mimeType: googleDoc }
779+
})
780+
mocks.preview.mockImplementation(
781+
async (_provider, _client, { id }: { id: string }) => `Full ${id} launch text`
782+
)
783+
const pending = searchLiveKnowledge.execute({ principal, input })
784+
await vi.waitFor(() => expect(mocks.json).toHaveBeenCalledTimes(4))
785+
expect(mocks.preview).not.toHaveBeenCalled()
786+
pendingB.resolve({ id: 'b', mimeType: googleDoc })
787+
expect(contents(await pending)).toEqual([
788+
'Full a launch text',
789+
'Full b launch text',
790+
'Full c launch text',
791+
'Excerpt d',
792+
])
793+
})
794+
795+
it('never surfaces the preview text of a candidate verification denies', async () => {
796+
mocks.search.mockResolvedValue({ documents: excerpts })
797+
mocks.service.mockResolvedValue({
798+
policy: defaultLiveSearchPolicy(),
799+
partial: false,
800+
verify: async ({ id }: { id: string }) => id !== 'a',
801+
})
802+
mocks.preview.mockImplementation(async (_provider, _client, { id }: { id: string }) =>
803+
id === 'a' ? 'Private a launch text' : `Full ${id} launch text`
804+
)
805+
const result = await searchLiveKnowledge.execute({ principal, input })
806+
expect(contents(result)).toEqual([
807+
'Full b launch text',
808+
'Full c launch text',
809+
'Full d launch text',
810+
])
811+
expect(JSON.stringify(result)).not.toContain('Private a')
812+
})
743813
})
744814

745815
it('gives provider searches the account session cancellation signal', async () => {
@@ -1153,4 +1223,105 @@ describe('authorized live retrieval', () => {
11531223
})
11541224
).toBe(false)
11551225
})
1226+
1227+
const accounts = (count: number) =>
1228+
Array.from({ length: count }, (_, index) => ({
1229+
...account,
1230+
id: `account-${String(index).padStart(2, '0')}`,
1231+
displayName: `Drive ${String(index).padStart(2, '0')}`,
1232+
}))
1233+
1234+
it('reports each account beyond the search limit instead of omitting it', async () => {
1235+
mocks.accounts.mockResolvedValue(accounts(21))
1236+
const result = await searchLiveKnowledge.execute({ principal, input })
1237+
const statuses = result.live?.accounts ?? []
1238+
expect(statuses).toHaveLength(21)
1239+
expect(new Set(statuses.map(({ accountId }) => accountId)).size).toBe(21)
1240+
expect(statuses.at(-1)).toMatchObject({
1241+
accountId: 'account-20',
1242+
status: 'unavailable',
1243+
message: expect.stringContaining('Not searched'),
1244+
})
1245+
expect(result.retrieval.status).toBe('partial')
1246+
})
1247+
1248+
describe('latency', () => {
1249+
beforeEach(() => {
1250+
vi.useFakeTimers()
1251+
vi.spyOn(AbortSignal, 'timeout').mockImplementation((milliseconds) => {
1252+
const controller = new AbortController()
1253+
setTimeout(
1254+
() => controller.abort(new DOMException('deadline', 'TimeoutError')),
1255+
milliseconds
1256+
)
1257+
return controller.signal
1258+
})
1259+
})
1260+
afterEach(() => {
1261+
vi.mocked(AbortSignal.timeout).mockRestore()
1262+
vi.useRealTimers()
1263+
})
1264+
1265+
/** Settles after `ms` of fake time, like a provider round trip. */
1266+
const delayed = <T>(ms: number, value: T) =>
1267+
new Promise<T>((resolve) => setTimeout(() => resolve(value), ms))
1268+
1269+
it('searches every account at once while bounding concurrent account resolution', async () => {
1270+
mocks.accounts.mockResolvedValue(accounts(20))
1271+
let resolving = 0
1272+
let peak = 0
1273+
mocks.resolveAccount.mockImplementation(async () => {
1274+
peak = Math.max(peak, ++resolving)
1275+
await delayed(50, undefined)
1276+
resolving--
1277+
return { account, accessToken: 'secret' }
1278+
})
1279+
mocks.search.mockImplementation(() => delayed(3000, { documents: [document] }))
1280+
let result: Awaited<ReturnType<typeof searchLiveKnowledge.execute>> | undefined
1281+
const pending = searchLiveKnowledge.execute({ principal, input }).then((value) => {
1282+
result = value
1283+
})
1284+
await vi.advanceTimersByTimeAsync(3500)
1285+
expect(result?.live?.accounts.map(({ status }) => status)).toEqual(
1286+
Array.from({ length: 20 }, () => 'ok')
1287+
)
1288+
expect(peak).toBeLessThanOrEqual(6)
1289+
await pending
1290+
})
1291+
1292+
it('times an account out at its deadline when its session never opens', async () => {
1293+
mocks.service.mockReturnValue(new Promise(() => {}))
1294+
let result: Awaited<ReturnType<typeof searchLiveKnowledge.execute>> | undefined
1295+
void searchLiveKnowledge.execute({ principal, input }).then((value) => {
1296+
result = value
1297+
})
1298+
await vi.advanceTimersByTimeAsync(12_000)
1299+
expect(result?.live?.accounts[0]?.status).toBe('timeout')
1300+
expect(mocks.search).not.toHaveBeenCalled()
1301+
})
1302+
1303+
it('gives each account its own deadline from its slot, and skips one the search left no time', async () => {
1304+
mocks.accounts.mockResolvedValue(accounts(13))
1305+
mocks.resolveAccount.mockImplementation(() =>
1306+
delayed(9500, { account, accessToken: 'secret' })
1307+
)
1308+
mocks.search.mockImplementation(async (_provider, _client, search) => {
1309+
search.signal?.throwIfAborted()
1310+
return { documents: [document] }
1311+
})
1312+
const pending = searchLiveKnowledge.execute({ principal, input })
1313+
await vi.advanceTimersByTimeAsync(20_000)
1314+
const statuses = (await pending).live?.accounts ?? []
1315+
/** The second wave waits 9.5s for a slot, then still has its full account deadline. */
1316+
expect(statuses.slice(0, 12).map(({ status }) => status)).toEqual(
1317+
Array.from({ length: 12 }, () => 'ok')
1318+
)
1319+
expect(statuses[12]).toMatchObject({
1320+
accountId: 'account-12',
1321+
status: 'timeout',
1322+
message: expect.stringContaining('Not searched before the search deadline'),
1323+
})
1324+
expect(mocks.resolveAccount).toHaveBeenCalledTimes(12)
1325+
})
1326+
})
11561327
})

0 commit comments

Comments
 (0)