fix(knowledge): bound every connector-lease ACL transaction to one short page (#8195)

* fix(knowledge): bound every connector-lease ACL transaction to one short page

* fix(knowledge): isolate stale sweep lock failures, close failed disables, update ACL test callers

* fix(knowledge): prove connector leases last, page ACL writes by projection rows, and skip unchanged work

* fix(knowledge): row-bound every observation page, size ACL pages from locked chunk counts, and check the sweep budget per page

* fix(knowledge): stop every ACL page walker at the run budget and keep pending rewrites until they finish

* test(knowledge): seed filled projection rows so the fan-out test holds on any provisioned database

* fix(knowledge): check the page deadline after the lease heartbeat

* test(knowledge): expire the restore budget at the heartbeat before the next window

Author: waleedlatif1

.github/workflows/test-build.yml

@@ -277,6 +277,7 @@ jobs:
           lib/knowledge/__integration__/listing-continuation.integration.ts
           lib/knowledge/__integration__/member-scope-renewal.integration.ts
           lib/knowledge/__integration__/member-document-lifecycle.integration.ts
+          lib/knowledge/__integration__/connector-lease-pages.integration.ts
           lib/knowledge/__integration__/slack-empty-threads.integration.ts
           lib/knowledge/__integration__/kb-block-search.integration.ts
           lib/knowledge/__integration__/gitlab-workspace.integration.ts

apps/sim/app/api/knowledge/connectors/member-sync/route.test.ts

@@ -101,6 +101,21 @@ describe('member sync scheduler owner routing', () => {
     ).toBe(true)
   })
 
+  it('still dispatches due connectors when the stale observation sweep fails', async () => {
+    mocks.sweep.mockRejectedValue(
+      Object.assign(new Error('canceling statement due to lock timeout'), { code: '55P03' })
+    )
+    queueTableRows(schemaMock.knowledgeConnector, [
+      { id: 'workspace-source', workspaceId: 'workspace-a', organizationId: null },
+    ])
+    const response = await GET(createMockRequest('GET'))
+    expect(response.status).toBe(200)
+    expect(mocks.dispatch).toHaveBeenCalledExactlyOnceWith(
+      'workspace-source',
+      expect.objectContaining({ requireRunnable: true })
+    )
+  })
+
   it('preserves workspace dispatch and refuses absent or ambiguous ownership', async () => {
     queueTableRows(schemaMock.knowledgeConnector, [
       { id: 'missing', workspaceId: null, organizationId: null },

apps/sim/app/api/knowledge/connectors/member-sync/route.ts

@@ -1,6 +1,7 @@
 import { db } from '@sim/db'
 import { knowledgeBase, knowledgeConnector, knowledgeConnectorMemberSyncLog } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
 import { and, asc, eq, inArray, isNull, lte, type SQL, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
@@ -157,9 +158,16 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       logger.warn(`[${requestId}] Closed ${closedLogs.length} orphaned member sync log(s)`)
     }
 
-    const sweep = await sweepStaleMemberObservations(now)
-    if (sweep.members > 0) {
-      logger.warn(`[${requestId}] Swept observations of ${sweep.members} stale member(s)`, sweep)
+    /** Observation hygiene never holds back dispatch; an unfinished sweep resumes next tick. */
+    try {
+      const sweep = await sweepStaleMemberObservations(now)
+      if (sweep.members > 0) {
+        logger.warn(`[${requestId}] Swept observations of ${sweep.members} stale member(s)`, sweep)
+      }
+    } catch (error) {
+      logger.error(`[${requestId}] Stale member observation sweep failed`, {
+        error: getErrorMessage(error),
+      })
     }
 
     const dueConnectors = await db