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

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 {