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

Author: waleedlatif1

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

apps/sim/lib/knowledge/__integration__/connector-lease-pages.integration.ts

@@ -9,6 +9,8 @@ import {
   document,
   knowledgeConnector,
   knowledgeConnectorMember,
+  knowledgeConnectorMemberSyncLog,
+  knowledgeDocumentObservation,
   organization,
   resourcePolicy,
   user,
@@ -47,6 +49,7 @@ import {
   materializeDocumentAcls,
   recordMemberObservations,
   rewriteConnectorAcls,
+  sweepStaleMemberObservations,
 } from '@/lib/knowledge/connectors/member-observations'
 import {
   executeMemberSync,
@@ -395,6 +398,83 @@ describe('connector lease ACL pages in PostgreSQL', () => {
     })
   })
 
+  describe('stale member sweep', () => {
+    it('defers a connector whose row a member run holds and still sweeps the others', async () => {
+      const busy = members
+      const idle = await seedKnowledgeMemberFixture(ids)
+      for (const fixture of [busy, idle]) {
+        await db
+          .update(knowledgeConnector)
+          .set({
+            status: 'active',
+            memberSyncStatus: 'idle',
+            memberSyncLockToken: null,
+            syncIntervalMinutes: 60,
+            lastMemberSyncAt: new Date(),
+          })
+          .where(eq(knowledgeConnector.id, fixture.connectorId))
+        /** Enrolled long enough ago that never having listed makes them stale. */
+        await db
+          .update(knowledgeConnectorMember)
+          .set({ createdAt: new Date(Date.now() - 3 * 24 * 60 * 60 * 1000) })
+          .where(eq(knowledgeConnectorMember.connectorId, fixture.connectorId))
+        const seeded = await seedDocuments(
+          fixture.connectorId,
+          fixture.members.map((member) => member.subjectToken).sort(),
+          3
+        )
+        for (const member of fixture.members)
+          await recordMemberObservations(
+            db,
+            member.id,
+            seeded.map((row) => row.id),
+            fixture.runId
+          )
+      }
+      const observed = async (connectorId: string) =>
+        (
+          await db
+            .select({ id: knowledgeDocumentObservation.memberId })
+            .from(knowledgeDocumentObservation)
+            .innerJoin(
+              knowledgeConnectorMember,
+              eq(knowledgeConnectorMember.id, knowledgeDocumentObservation.memberId)
+            )
+            .where(eq(knowledgeConnectorMember.connectorId, connectorId))
+        ).length
+
+      /** A member page of a running sync holds the busy connector's row for longer than a sweep page waits. */
+      let release!: () => void
+      let locked!: () => void
+      const held = new Promise<void>((resolve) => {
+        locked = resolve
+      })
+      const holder = db.transaction(async (tx) => {
+        await tx
+          .select({ id: knowledgeConnector.id })
+          .from(knowledgeConnector)
+          .where(eq(knowledgeConnector.id, busy.connectorId))
+          .for('update')
+        locked()
+        await new Promise<void>((resolve) => {
+          release = resolve
+        })
+      })
+      await held
+      try {
+        const result = await sweepStaleMemberObservations(new Date(Date.now() + 1_000))
+        expect(result.members).toBe(idle.members.length)
+      } finally {
+        release()
+        await holder
+      }
+
+      expect(await observed(busy.connectorId)).toBe(busy.members.length * 3)
+      expect(await observed(idle.connectorId)).toBe(0)
+      expect((await storedAcls(idle.connectorId)).every((acl) => acl.length === 0)).toBe(true)
+    }, 30_000)
+  })
+
   describe('resumeMembershipRewrites', () => {
     it('rematerialises a changed member one page per lease transaction', async () => {
       const seeded = await seedDocuments(members.connectorId, [])
@@ -649,5 +729,45 @@ describe('connector lease ACL pages in PostgreSQL', () => {
         )
       expect(granted).toBe(0)
     })
+
+    it('closes a run whose disable of a removed option fails as an ordinary failure', async () => {
+      /** The option id is set but no longer exists, which membership reconciliation reports. */
+      await db
+        .update(knowledgeConnector)
+        .set({ credentialGroupOptionId: generateId() })
+        .where(eq(knowledgeConnector.id, members.connectorId))
+      await seedDocuments(
+        members.connectorId,
+        members.members.map((member) => member.subjectToken).sort()
+      )
+      await db.execute(
+        sql.raw(`CREATE OR REPLACE FUNCTION fail_after_acl_writes() RETURNS trigger
+        LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'fixture statement failure'; END $$`)
+      )
+      await db.execute(
+        sql`CREATE TRIGGER fail_after_acl_writes BEFORE UPDATE OF acl ON document FOR EACH ROW
+          WHEN (NEW.connector_id = ${sql.raw(`'${members.connectorId}'`)})
+          EXECUTE FUNCTION fail_after_acl_writes()`
+      )
+
+      const failed = await executeMemberSync(members.connectorId, {
+        billingAttribution: await billing(),
+      })
+
+      expect(failed.error).toBeTruthy()
+      const state = await connectorState()
+      expect(state?.memberSyncStatus).toBe('error')
+      expect(state?.memberSyncLockToken).toBeNull()
+      const [log] = await db
+        .select({ status: knowledgeConnectorMemberSyncLog.status })
+        .from(knowledgeConnectorMemberSyncLog)
+        .where(eq(knowledgeConnectorMemberSyncLog.connectorId, members.connectorId))
+      expect(log?.status).toBe('failed')
+
+      await db.execute(sql`DROP TRIGGER fail_after_acl_writes ON document`)
+      await executeMemberSync(members.connectorId, { billingAttribution: await billing() })
+      expect((await connectorState())?.memberSyncStatus).toBe('disabled')
+      expect((await storedAcls(members.connectorId)).every((acl) => acl.length === 0)).toBe(true)
+    })
   })
 })

apps/sim/lib/knowledge/access/predicate.postgres.test.ts

@@ -22,6 +22,7 @@ vi.mock('@/connectors/registry.server', () => ({ CONNECTOR_REGISTRY: {} }))
 const { drizzle } = await import('drizzle-orm/postgres-js')
 const schema = await import('@sim/db/schema')
 const { persistDocumentAcls } = await import('@/lib/knowledge/connectors/sync-persistence')
+const { leaseTransaction } = await import('@/lib/knowledge/connectors/sync-lock')
 const { mergeMirroredAcls, hideUnlistedDocuments } = await import(
   '@/lib/knowledge/connectors/mirrored-acls'
 )
@@ -750,7 +751,9 @@ describe.runIf(Boolean(databaseUrl))('knowledge ACLs in PostgreSQL', () => {
     const space = 'g:confluence:tenant:space'
     const page = 'g:confluence:tenant:page'
     const input = new Map([['page', { acl: [space], requirements: [[page]] }]])
-    expect(await persistDocumentAcls('admin', input, executor)).toEqual({ updated: 1, rejected: 0 })
+    expect(
+      await persistDocumentAcls('admin', input, leaseTransaction('admin', undefined, executor))
+    ).toEqual({ updated: 1, rejected: 0 })
     expect(await readable([space, page], 'persisted')).toBe(true)
     expect(await readable([page], 'persisted')).toBe(false)
     await connection.unsafe("UPDATE document SET acl = string_to_array($1, E'\\n')", [page])
@@ -760,7 +763,7 @@ describe.runIf(Boolean(databaseUrl))('knowledge ACLs in PostgreSQL', () => {
       "UPDATE document SET acl_verified_at = statement_timestamp() - interval '25 hours'"
     )
     expect(await readable([space, page], 'persisted')).toBe(false)
-    await persistDocumentAcls('admin', input, executor)
+    await persistDocumentAcls('admin', input, leaseTransaction('admin', undefined, executor))
     expect(await readable([space, page], 'persisted')).toBe(true)
     const [stored] = await connection.unsafe(
       "SELECT jsonb_typeof(acl_requirements) AS shape, acl_requirements FROM document WHERE id = 'persisted'"
@@ -844,10 +847,15 @@ describe.runIf(Boolean(databaseUrl))('knowledge ACLs in PostgreSQL', () => {
         {}
       )
       if (step === 'unlisted') hideUnlistedDocuments(merged.acls, ['shared-file'])
-      await persistDocumentAcls('admin', merged.acls, executor, {
-        unresolvedExternalIds: merged.unresolvedExternalIds,
-        generationStartedAt,
-      })
+      await persistDocumentAcls(
+        'admin',
+        merged.acls,
+        leaseTransaction('admin', undefined, executor),
+        {
+          unresolvedExternalIds: merged.unresolvedExternalIds,
+          generationStartedAt,
+        }
+      )
       const [stored] = await connection.unsafe(
         "SELECT to_jsonb(acl) AS acl, acl_verified_at FROM document WHERE id = 'shared'"
       )
@@ -882,10 +890,15 @@ describe.runIf(Boolean(databaseUrl))('knowledge ACLs in PostgreSQL', () => {
       [verifiedAt]
     )
     const executor = drizzle(connection, { schema })
-    const result = await persistDocumentAcls('admin', new Map([['file', []]]), executor, {
-      unresolvedExternalIds: new Set(['file']),
-      generationStartedAt,
-    })
+    const result = await persistDocumentAcls(
+      'admin',
+      new Map([['file', []]]),
+      leaseTransaction('admin', undefined, executor),
+      {
+        unresolvedExternalIds: new Set(['file']),
+        generationStartedAt,
+      }
+    )
     expect(result.updated).toBe(preserved ? 0 : 1)
     const [stored] = await connection.unsafe(
       "SELECT to_jsonb(acl) AS acl, acl_verified_at FROM document WHERE id = 'boundary'"

apps/sim/lib/knowledge/connectors/member-observations.test.ts

@@ -203,12 +203,44 @@ describe('sweepStaleMemberObservations', () => {
 
     expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(2)
     expect(dbChainMockFns.limit).toHaveBeenCalledWith(25)
-    const bounds = dbChainMockFns.execute.mock.calls.filter(([query]) =>
-      JSON.stringify(query).includes('lock_timeout')
+    const bounds = dbChainMockFns.execute.mock.calls.filter((call: unknown[]) =>
+      JSON.stringify(call).includes('lock_timeout')
     )
     expect(bounds).toHaveLength(2)
   })
 
+  /** A connector whose running member page holds its row must not stall the other connectors. */
+  it('defers a member whose page hits a lock timeout and still sweeps the next one', async () => {
+    queueTableRows(schemaMock.knowledgeConnectorMember, [
+      STALE_MEMBER,
+      { ...STALE_MEMBER, id: 'm-2', connectorId: 'c-2' },
+    ])
+    dbChainMockFns.transaction.mockRejectedValueOnce(
+      Object.assign(new Error('canceling statement due to lock timeout'), { code: '55P03' })
+    )
+    queueTableRows(schemaMock.knowledgeConnector, [{ id: 'c-2' }])
+    queueTableRows(schemaMock.knowledgeConnectorMember, [{ id: 'm-2' }])
+    dbChainMockFns.returning
+      .mockResolvedValueOnce([{ documentId: 'd-1' }])
+      .mockResolvedValueOnce([{ id: 'd-1' }])
+      .mockResolvedValueOnce([])
+
+    await expect(sweepStaleMemberObservations(NOW)).resolves.toEqual({
+      members: 1,
+      observationsRemoved: 1,
+      documentsRematerialized: 1,
+      docsTombstoned: 0,
+    })
+    expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(2)
+  })
+
+  it('fails the sweep on an error that is not a lock or capacity failure', async () => {
+    queueTableRows(schemaMock.knowledgeConnectorMember, [STALE_MEMBER])
+    dbChainMockFns.transaction.mockRejectedValueOnce(new TypeError('broken'))
+
+    await expect(sweepStaleMemberObservations(NOW)).rejects.toThrow('broken')
+  })
+
   /**
    * A run that claimed the member between the selection and the lock moved
    * `lastStartedAt` forward, so the re-check under `FOR UPDATE` finds nothing
@@ -336,8 +368,8 @@ describe('rewriteConnectorAcls', () => {
     expect(assignedPages()).toHaveLength(3)
     expect(pageSizes()).toEqual([25, 25, 10])
     expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(3)
-    const bounds = dbChainMockFns.execute.mock.calls.filter(([query]) =>
-      JSON.stringify(query).includes('lock_timeout')
+    const bounds = dbChainMockFns.execute.mock.calls.filter((call: unknown[]) =>
+      JSON.stringify(call).includes('lock_timeout')
     )
     expect(bounds).toHaveLength(3)
   })

apps/sim/lib/knowledge/connectors/member-observations.ts

@@ -5,6 +5,8 @@ import {
   knowledgeConnectorMember,
   knowledgeDocumentObservation,
 } from '@sim/db/schema'
+import { createLogger } from '@sim/logger'
+import { getPostgresErrorCode } from '@sim/utils/errors'
 import { chunkArray } from '@sim/utils/helpers'
 import {
   and,
@@ -49,6 +51,8 @@ import {
   hardDeleteDocuments,
 } from '@/lib/knowledge/documents/service'
 
+const logger = createLogger('MemberObservations')
+
 /**
  * Documents per tombstone or resurrection page. Those writes set `deleted_at` alone and fire no
  * projection fan-out, so they page wider than an ACL write.
@@ -870,6 +874,9 @@ function memberStillStale(memberId: string, cutoff: Date) {
   )
 }
 
+/** Lock, statement-timeout and serialization failures a sweep page defers instead of failing the tick. */
+const SWEEP_DEFERRABLE_CODES = new Set(['55P03', '57014', '40P01', '40001'])
+
 /**
  * Stale-member observations one sweep tick removes per member: the same
  * {@link OBSERVATION_BATCH_SIZE} as before, now in pages of
@@ -1044,14 +1051,28 @@ export async function sweepStaleMemberObservations(now: Date): Promise<StaleMemb
   for (const member of staleMembers) {
     const memberCutoff = new Date(now.getTime() - staleMemberWindowMs(member.syncIntervalMinutes))
     let sweptAny = false
-    for (let page = 0; page < STALE_MEMBER_PAGES_PER_TICK; page++) {
-      const swept = await sweepStaleMemberPage(member, memberCutoff, now)
-      if (!swept) break
-      sweptAny = true
-      result.observationsRemoved += swept.observationsRemoved
-      result.documentsRematerialized += swept.documentsRematerialized
-      result.docsTombstoned += swept.docsTombstoned
-      if (swept.observationsRemoved < ACL_CHANGE_BATCH_SIZE) break
+    try {
+      for (let page = 0; page < STALE_MEMBER_PAGES_PER_TICK; page++) {
+        const swept = await sweepStaleMemberPage(member, memberCutoff, now)
+        if (!swept) break
+        sweptAny = true
+        result.observationsRemoved += swept.observationsRemoved
+        result.documentsRematerialized += swept.documentsRematerialized
+        result.docsTombstoned += swept.docsTombstoned
+        if (swept.observationsRemoved < ACL_CHANGE_BATCH_SIZE) break
+      }
+    } catch (error) {
+      /**
+       * A connector whose member run holds its row, or whose page outruns the bounds, is left
+       * for the next tick: its committed pages stand, and the other members are still swept.
+       */
+      const code = getPostgresErrorCode(error)
+      if (!code || !SWEEP_DEFERRABLE_CODES.has(code)) throw error
+      logger.warn('Deferred a stale member sweep to the next tick', {
+        connectorId: member.connectorId,
+        memberId: member.id,
+        code,
+      })
     }
     if (sweptAny) result.members += 1
   }