improvement(knowledge): stop consuming access batches once both overview probes saturate

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.

Author: waleedlatif1

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)
+  })
 })

apps/sim/lib/knowledge/application/search-source-overview.ts

@@ -107,26 +107,27 @@ export const readSearchSourceOverview = instrumentSourceOverviewUseCase(
       const indexingTypes = new Set<string>()
       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 {