From 698df08754f553815e0cf07bd2ccd5604dd9549f Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 18 Sep 2026 01:06:41 -0700 Subject: [PATCH] improvement(knowledge): stop consuming access batches once both overview probes saturate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The source overview loops over the access batches the read-access generator yields. The first batch is free, but every later one costs a connector discovery query plus a per-connector live source proof over the network. Both probes in the loop already stop issuing queries once they saturate — the searchable probe on its first hit, the indexing probe once every configured provider type is accounted for — but the loop kept pulling batches afterwards, paying the producer's cost for no probe at all. Hoist the two probe guards into closures so the loop body and the exit share one definition of each, and break once neither can change the result. Output is unchanged; only the work is dropped. --- .../search-source-overview.test.ts | 52 +++++++++++++++++-- .../application/search-source-overview.ts | 31 ++++++----- 2 files changed, 68 insertions(+), 15 deletions(-) diff --git a/apps/sim/lib/knowledge/application/search-source-overview.test.ts b/apps/sim/lib/knowledge/application/search-source-overview.test.ts index 08b28960654..b135f453550 100644 --- a/apps/sim/lib/knowledge/application/search-source-overview.test.ts +++ b/apps/sim/lib/knowledge/application/search-source-overview.test.ts @@ -45,10 +45,16 @@ const indexingProbeCount = () => dbChainMockFns.limit.mock.calls.filter(([rows]) => rows === MAX_SEARCH_SOURCE_PROVIDER_TYPES) .length - CONFIGURED_PROVIDER_READS +/** Counted at the yield, so batches the use case never asks for stay uncounted. */ function yieldBatches(count: number) { + const consumed = { batches: 0 } mocks.batches.mockImplementation(async function* () { - for (let index = 0; index < count; index += 1) yield sql`batch-${sql.raw(String(index))}` + for (let index = 0; index < count; index += 1) { + consumed.batches += 1 + yield sql`batch-${sql.raw(String(index))}` + } }) + return consumed } beforeEach(() => { @@ -84,15 +90,20 @@ describe('readSearchSourceOverview', () => { }) it('stops probing for indexing once every configured provider type is known', async () => { - yieldBatches(3) + const consumed = yieldBatches(3) queueTableRows(member, [{ role: 'owner' }]) queueTableRows(knowledgeConnector, [{ connectorType: 'gmail' }]) queueTableRows(knowledgeConnector, [{ connectorType: 'gmail' }]) const result = await readSearchSourceOverview.execute({ principal, input }) - expect(result.providers).toEqual([{ connectorType: 'gmail', isSyncing: true }]) + expect(result).toEqual({ + providers: [{ connectorType: 'gmail', isSyncing: true }], + hasSearchableDocuments: false, + }) expect(indexingProbeCount()).toBe(1) + /** The searchable probe is still unsatisfied, so the batches keep being consumed. */ + expect(consumed.batches).toBe(3) }) it('keeps probing every batch while a configured provider type is still unaccounted for', async () => { @@ -109,4 +120,39 @@ describe('readSearchSourceOverview', () => { ]) expect(indexingProbeCount()).toBe(3) }) + + it('stops consuming access batches once neither probe can change the result', async () => { + const consumed = yieldBatches(3) + queueTableRows(member, [{ role: 'owner' }]) + queueTableRows(knowledgeConnector, [{ connectorType: 'gmail' }]) + queueTableRows(knowledgeConnector, [{ connectorType: 'gmail' }]) + queueTableRows(document, [{ id: 'doc-1' }]) + + const result = await readSearchSourceOverview.execute({ principal, input }) + + expect(result).toEqual({ + providers: [{ connectorType: 'gmail', isSyncing: true }], + hasSearchableDocuments: true, + }) + expect(consumed.batches).toBe(1) + }) + + it('keeps consuming access batches for a provider type still unaccounted for', async () => { + const consumed = yieldBatches(3) + queueTableRows(member, [{ role: 'owner' }]) + queueTableRows(knowledgeConnector, [{ connectorType: 'gmail' }, { connectorType: 'notion' }]) + queueTableRows(knowledgeConnector, [{ connectorType: 'gmail' }]) + queueTableRows(document, [{ id: 'doc-1' }]) + + const result = await readSearchSourceOverview.execute({ principal, input }) + + expect(result).toEqual({ + providers: [ + { connectorType: 'gmail', isSyncing: true }, + { connectorType: 'notion', isSyncing: false }, + ], + hasSearchableDocuments: true, + }) + expect(consumed.batches).toBe(3) + }) }) diff --git a/apps/sim/lib/knowledge/application/search-source-overview.ts b/apps/sim/lib/knowledge/application/search-source-overview.ts index c0892fcb856..dbc4f92ec0a 100644 --- a/apps/sim/lib/knowledge/application/search-source-overview.ts +++ b/apps/sim/lib/knowledge/application/search-source-overview.ts @@ -107,26 +107,27 @@ export const readSearchSourceOverview = instrumentSourceOverviewUseCase( const indexingTypes = new Set() let searchableProbes = 0 let hasSearchableDocuments = false + const probesSources: boolean = availability.memberScoped || availability.sourceMirrored + /** One searchable document is the whole answer, so later batches skip the probe entirely. */ + const probesSearchable = (): boolean => probesSources && !hasSearchableDocuments + /** + * A provider type is only read back as membership of `indexingTypes`, so once every + * configured type is in the set no later batch can change the answer. + */ + const probesIndexing = (): boolean => + probesSources && providers.some(({ connectorType }) => !indexingTypes.has(connectorType)) for await (const accessCondition of knowledgeReadAccessBatches(access, [ configured, available, documentConditions, ])) { const readableDocument = and(documentConditions, accessCondition) - const probesSources: boolean = availability.memberScoped || availability.sourceMirrored - /** One searchable document is the whole answer, so later batches skip the probe entirely. */ - const probesSearchable: boolean = probesSources && !hasSearchableDocuments - if (probesSearchable) searchableProbes += 1 - /** - * A provider type is only read back as membership of `indexingTypes`, so once every - * configured type is in the set no later batch can change the answer. - */ - const probesIndexing: boolean = - probesSources && providers.some(({ connectorType }) => !indexingTypes.has(connectorType)) + const probesSearchableNow = probesSearchable() + if (probesSearchableNow) searchableProbes += 1 /** Annotated so the searchable probe's guard does not infer through its own result. */ const [indexing, searchable]: [{ connectorType: string }[], { id: string }[]] = await Promise.all([ - probesIndexing + probesIndexing() ? measureSearchStage('source_overview.indexing', () => configuredProvidersQuery() .where( @@ -164,7 +165,7 @@ export const readSearchSourceOverview = instrumentSourceOverviewUseCase( .limit(MAX_SEARCH_SOURCE_PROVIDER_TYPES) ) : [], - probesSearchable + probesSearchableNow ? measureSearchStage('source_overview.searchable', () => db .select({ id: document.id }) @@ -199,6 +200,12 @@ export const readSearchSourceOverview = instrumentSourceOverviewUseCase( ]) for (const provider of indexing) indexingTypes.add(provider.connectorType) hasSearchableDocuments ||= searchable.length > 0 + /** + * Both probes are saturated, so every remaining batch would be discovered and live-proved + * for no probe. `accessBatchCount` and `liveProofConnectorCount` stay what they document: + * the batches and proofs this read actually spent, not the batches the owner could produce. + */ + if (!probesSearchable() && !probesIndexing()) break } annotateSearchDiagnostics({ searchableProbeCount: searchableProbes }) return {