fix(knowledge): grant access after dispatch, include skipped rows, and keep the grant best-effort

Author: waleedlatif1

apps/sim/lib/knowledge/connectors/member-sync-engine.ts

@@ -59,7 +59,10 @@ import {
   SyncLockLostException,
   stillHoldsMemberSyncLock,
 } from '@/lib/knowledge/connectors/sync-lock'
-import type { KnowledgeBaseOwner } from '@/lib/knowledge/connectors/sync-persistence'
+import type {
+  KnowledgeBaseOwner,
+  PersistedDocument,
+} from '@/lib/knowledge/connectors/sync-persistence'
 import {
   addSourcePagePayloadBytes,
   ConnectorDeletedException,
@@ -69,7 +72,6 @@ import {
   classifySuspectListing,
   createSyncRunState,
   loadOwnedCorpus,
-  type PersistedDocument,
   processDocOps,
   RETRY_WINDOW_DAYS,
   runChangeFeedPass,

apps/sim/lib/knowledge/connectors/sync-engine.test.ts

@@ -789,7 +789,7 @@ describe('persistSkippedDocuments', () => {
         'workspace',
         lease
       )
-    ).resolves.toBe(1)
+    ).resolves.toHaveLength(1)
 
     expect(dbChainMockFns.values).toHaveBeenCalledWith([
       expect.objectContaining({
@@ -835,7 +835,7 @@ describe('persistSkippedDocuments', () => {
         'workspace',
         lease
       )
-    ).resolves.toBe(1)
+    ).resolves.toHaveLength(1)
 
     expect(dbChainMockFns.set).toHaveBeenCalledWith(
       expect.objectContaining({

apps/sim/lib/knowledge/connectors/sync-persistence.ts

@@ -247,6 +247,12 @@ function buildSkippedDocumentRow(
  *
  * Returns the number of rows recorded.
  */
+/** A document a sync wrote, by the id the source knows it by and the id the row has. */
+export interface PersistedDocument {
+  externalId: string
+  documentId: string
+}
+
 export async function persistSkippedDocuments(
   knowledgeBaseId: string,
   connectorId: string,
@@ -259,9 +265,9 @@ export async function persistSkippedDocuments(
   sourceConfig: Record<string, unknown> | undefined,
   access: SyncDocumentAccess,
   lease: SyncWriteLease
-): Promise<number> {
+): Promise<PersistedDocument[]> {
   if (skipOps.length === 0) {
-    return 0
+    return []
   }
   const inserts = skipOps
     .filter((op) => !op.existingId)
@@ -275,6 +281,12 @@ export async function persistSkippedDocuments(
         access
       )
     )
+  const persisted: PersistedDocument[] = [
+    ...inserts.map((row) => ({ externalId: row.externalId, documentId: row.id })),
+    ...skipOps
+      .filter((op): op is typeof op & { existingId: string } => Boolean(op.existingId))
+      .map((op) => ({ externalId: op.extDoc.externalId, documentId: op.existingId })),
+  ]
   const replacements = skipOps.filter((op): op is typeof op & { existingId: string } =>
     Boolean(op.existingId)
   )
@@ -363,7 +375,7 @@ export async function persistSkippedDocuments(
     }
   }
 
-  return skipOps.length
+  return persisted
 }
 
 /**

apps/sim/lib/knowledge/connectors/sync-primitives.ts

@@ -16,6 +16,7 @@ import { SyncLockLostException, type SyncRunLease } from '@/lib/knowledge/connec
 import {
   addDocument,
   type KnowledgeBaseOwner,
+  type PersistedDocument,
   persistSkippedDocuments,
   persistSkippedRetryHashes,
   type SyncDocumentAccess,
@@ -1497,9 +1498,13 @@ export interface ProcessDocOpsInput {
   hydration: DocOpHydration
   lease: Pick<SyncRunLease, 'beatIfDue' | 'beatLive' | 'stillHeld'>
   /**
-   * Runs after each batch's rows are written, with the documents that landed.
-   * A members-mode run grants access here, batch by batch, so a long first
-   * crawl becomes searchable as it goes rather than all at once at the end.
+   * Runs after each batch is written and dispatched, with every row that
+   * landed: hydrated documents and skipped ones alike. A members-mode run
+   * grants access here, batch by batch, so a long first crawl becomes
+   * searchable as it goes rather than all at once at the end. Best effort: a
+   * failure here is logged and the run continues, because the listing's own
+   * pass at the end of the run writes the same grants authoritatively; only a
+   * lost lease ends the run.
    */
   onBatchPersisted?: (persisted: readonly PersistedDocument[]) => Promise<void>
   /** Who may read the documents this pass writes. */
@@ -1513,12 +1518,6 @@ export interface ProcessDocOpsInput {
  * run so the connector backs off, and a lost lease, which ends it so no
  * further write lands beside the replacement run's.
  */
-/** A document a batch wrote, by the id the source knows it by and the id the row has. */
-export interface PersistedDocument {
-  externalId: string
-  documentId: string
-}
-
 export async function processDocOps(input: ProcessDocOpsInput): Promise<void> {
   const {
     connectorId,
@@ -1689,6 +1688,7 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise<void> {
       }
     }
 
+    const skippedPersisted: PersistedDocument[] = []
     if (skipOps.length > 0) {
       try {
         const recorded = await persistSkippedDocuments(
@@ -1700,7 +1700,8 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise<void> {
           documentAccess,
           input.lease
         )
-        result.docsSkipped += recorded
+        result.docsSkipped += recorded.length
+        skippedPersisted.push(...recorded)
       } catch (error) {
         if (error instanceof SyncLockLostException) throw error
         /**
@@ -1756,7 +1757,7 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise<void> {
     if (leaseLost) throw leaseLost.reason
 
     const batchDocs: DocumentData[] = []
-    const persisted: PersistedDocument[] = []
+    const persisted: PersistedDocument[] = [...skippedPersisted]
     for (let j = 0; j < settled.length; j++) {
       const outcome = settled[j]
       if (outcome.status === 'fulfilled') {
@@ -1778,8 +1779,6 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise<void> {
       }
     }
 
-    if (persisted.length > 0) await input.onBatchPersisted?.(persisted)
-
     if (batchDocs.length > 0) {
       result.processingDispatch.requested += batchDocs.length
       try {
@@ -1803,6 +1802,19 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise<void> {
         })
       }
     }
+
+    if (persisted.length > 0 && input.onBatchPersisted) {
+      try {
+        await input.onBatchPersisted(persisted)
+      } catch (error) {
+        if (error instanceof SyncLockLostException) throw error
+        logger.warn('Failed to grant access for a persisted batch — the run end will retry', {
+          connectorId,
+          count: persisted.length,
+          error: toError(error).message,
+        })
+      }
+    }
   }
 }