fix(knowledge): address release review follow-ups for ACL pages, projection fill, and sync retries (#8202)

* fix(knowledge): address release review follow-ups for ACL pages, projection fill, and sync retries

- Scope the observed-ACL aggregate to members of the document's own connector
- Advance a member's listing checkpoint only after the ACLs it decides are written,
  and rematerialise every document a replayed feed removal names
- Plan ACL pages from an unlocked read and lock only the page that will be written
- Lock fill documents FOR KEY SHARE SKIP LOCKED so a concurrent delete cannot fail
  the mark's foreign key
- Report an unfinished projection fill as remaining work
- Record the database failure class on failed sync-log rows and count only those
  toward the database retry streak (expand-only migration 0379)
- Heartbeat the lease between revokeDocumentAcls transactions
- Extract isTriggerAvailable to lib/core/config/trigger-availability and use it
  for every dispatch decision
- Exercise a concurrently committed deletion in the processing lock-scope test

* fix(knowledge): constrain the recorded database failure class

Author: waleedlatif1

apps/sim/lib/billing/cleanup-dispatcher.test.ts

@@ -28,7 +28,7 @@ vi.mock('@/lib/core/async-jobs', () => ({
 }))
 vi.mock('@/lib/core/async-jobs/config', () => ({ shouldExecuteInline: vi.fn(() => false) }))
 vi.mock('@/lib/core/async-jobs/region', () => ({ resolveTriggerRegion: vi.fn() }))
-vi.mock('@/lib/knowledge/documents/service', () => ({
+vi.mock('@/lib/core/config/trigger-availability', () => ({
   isTriggerAvailable: mockIsTriggerAvailable,
 }))
 vi.mock('@/lib/workspaces/policy', () => ({

apps/sim/lib/billing/cleanup-dispatcher.ts

@@ -16,7 +16,7 @@ import { shouldExecuteInline } from '@/lib/core/async-jobs/config'
 import { resolveTriggerRegion } from '@/lib/core/async-jobs/region'
 import type { EnqueueOptions } from '@/lib/core/async-jobs/types'
 import { isBillingEnabled, isDataRetentionEnabled } from '@/lib/core/config/env-flags'
-import { isTriggerAvailable } from '@/lib/knowledge/documents/service'
+import { isTriggerAvailable } from '@/lib/core/config/trigger-availability'
 import { isOrganizationWorkspace, WORKSPACE_MODE } from '@/lib/workspaces/policy'
 
 const logger = createLogger('RetentionDispatcher')

apps/sim/lib/core/config/trigger-availability.test.ts

@@ -0,0 +1,43 @@
+/**
+ * @vitest-environment node
+ */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({
+  env: { TRIGGER_SECRET_KEY: undefined as string | undefined },
+  flags: { isTriggerDevEnabled: false },
+  insideRun: vi.fn(() => false),
+}))
+
+vi.mock('@/lib/core/config/env', () => ({ env: mocks.env }))
+vi.mock('@/lib/core/config/env-flags', () => ({
+  get isTriggerDevEnabled() {
+    return mocks.flags.isTriggerDevEnabled
+  },
+}))
+vi.mock('@/lib/core/config/trigger-runtime', () => ({ isInsideTriggerRun: mocks.insideRun }))
+
+import { isTriggerAvailable } from '@/lib/core/config/trigger-availability'
+
+describe('isTriggerAvailable', () => {
+  beforeEach(() => {
+    mocks.env.TRIGGER_SECRET_KEY = undefined
+    mocks.flags.isTriggerDevEnabled = false
+    mocks.insideRun.mockReturnValue(false)
+  })
+
+  it('is available inside a Trigger.dev run whatever the environment says', () => {
+    mocks.insideRun.mockReturnValue(true)
+    expect(isTriggerAvailable()).toBe(true)
+  })
+
+  it('needs both the enable flag and the secret key outside a run', () => {
+    mocks.flags.isTriggerDevEnabled = true
+    expect(isTriggerAvailable()).toBe(false)
+    mocks.flags.isTriggerDevEnabled = false
+    mocks.env.TRIGGER_SECRET_KEY = 'fixture-key'
+    expect(isTriggerAvailable()).toBe(false)
+    mocks.flags.isTriggerDevEnabled = true
+    expect(isTriggerAvailable()).toBe(true)
+  })
+})

apps/sim/lib/core/config/trigger-availability.ts

@@ -0,0 +1,44 @@
+import { createLogger } from '@sim/logger'
+import { env } from '@/lib/core/config/env'
+import { isTriggerDevEnabled } from '@/lib/core/config/env-flags'
+import { isInsideTriggerRun } from '@/lib/core/config/trigger-runtime'
+
+const logger = createLogger('TriggerAvailability')
+
+let triggerAvailabilityLogged = false
+
+/**
+ * Whether background work may be dispatched to Trigger.dev rather than run
+ * in-process.
+ *
+ * Inside a Trigger.dev run the answer is unconditionally yes: the platform is
+ * what is executing this process, so no environment guess can be more reliable
+ * than the run marker. Outside a run the deployment must both enable
+ * Trigger.dev and hold the secret key the SDK authenticates with.
+ *
+ * Resolving `true` inside a run is safe even if the run process turns out not
+ * to expose `TRIGGER_SECRET_KEY`: the SDK would then reject the trigger and a
+ * caller that falls back to in-process work lands exactly where a `false`
+ * predicate lands anyway.
+ *
+ * The first evaluation in a process logs the resolved inputs. That is once per
+ * worker process rather than once per dispatch, and it is the signal that makes
+ * an app-vs-worker asymmetry visible without reading a crashed run's spans.
+ */
+export function isTriggerAvailable(): boolean {
+  const insideRun = isInsideTriggerRun()
+  const hasSecretKey = Boolean(env.TRIGGER_SECRET_KEY)
+  const available = insideRun || (hasSecretKey && isTriggerDevEnabled)
+
+  if (!triggerAvailabilityLogged) {
+    triggerAvailabilityLogged = true
+    logger.info('Resolved Trigger.dev dispatch availability', {
+      available,
+      insideTriggerRun: insideRun,
+      triggerDevEnabled: isTriggerDevEnabled,
+      hasSecretKey,
+    })
+  }
+
+  return available
+}

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

@@ -21,6 +21,7 @@ import { installProjectionSourceAcl } from '@sim/db/script-migrations/0021_embed
 import { installKnowledgeProjectionAsync } from '@sim/db/script-migrations/0024_knowledge_projection_async'
 import { generateId } from '@sim/utils/id'
 import { and, eq, inArray, sql } from 'drizzle-orm'
+import postgres from 'postgres'
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const provider = vi.hoisted(() => ({ list: vi.fn(), get: vi.fn(), changes: vi.fn() }))
@@ -1050,6 +1051,69 @@ describe('connector lease ACL pages in PostgreSQL', () => {
     })
   })
 
+  describe('lockProjectionPage', () => {
+    const withChunks = async (rows: { id: string }[], chunkCount: number) =>
+      db
+        .update(document)
+        .set({ chunkCount })
+        .where(
+          inArray(
+            document.id,
+            rows.map((row) => row.id)
+          )
+        )
+    const lockPage = (documentIds: string[]) =>
+      db.transaction(async (tx) => {
+        await tx.execute(sql`SET LOCAL lock_timeout = '500ms'`)
+        return memberObservations.lockProjectionPage(tx, documentIds)
+      })
+    const waitingOnLock = async () => {
+      const [row] = await db.execute<{ waiting: boolean }>(
+        sql`SELECT EXISTS (SELECT 1 FROM pg_locks WHERE NOT granted) AS waiting`
+      )
+      return Boolean(row?.waiting)
+    }
+
+    it('locks only the page it will write, so a held document past it stalls nothing', async () => {
+      const [first, held, last] = await seedDocuments(members.connectorId, [], 3)
+      await withChunks([first, held, last], PROJECTION_ROW_BATCH_SIZE)
+      const holder = postgres(process.env.DATABASE_URL!, { max: 1, onnotice: () => undefined })
+      try {
+        await holder.begin(async (tx) => {
+          await tx`SELECT id FROM document WHERE id = ${held.id} FOR UPDATE`
+          await expect(lockPage([first.id, held.id, last.id])).resolves.toEqual({
+            page: [first.id],
+            rest: [held.id, last.id],
+          })
+        })
+      } finally {
+        await holder.end()
+      }
+    })
+
+    it('cuts the page to what still fits once a concurrent commit grows its chunks', async () => {
+      const [first, grown] = await seedDocuments(members.connectorId, [], 2)
+      await withChunks([first, grown], PROJECTION_ROW_BATCH_SIZE / 2)
+      const holder = postgres(process.env.DATABASE_URL!, { max: 1, onnotice: () => undefined })
+      try {
+        let page: Promise<{ page: string[]; rest: string[] }> | undefined
+        await holder.begin(async (tx) => {
+          await tx`UPDATE document SET chunk_count = ${PROJECTION_ROW_BATCH_SIZE} WHERE id = ${grown.id}`
+          page = db.transaction(async (lockTx) =>
+            memberObservations.lockProjectionPage(lockTx, [first.id, grown.id])
+          )
+          await vi.waitFor(async () => expect(await waitingOnLock()).toBe(true), {
+            timeout: 5_000,
+            interval: 10,
+          })
+        })
+        await expect(page).resolves.toEqual({ page: [first.id], rest: [grown.id] })
+      } finally {
+        await holder.end()
+      }
+    })
+  })
+
   describe('member listing materialisation', () => {
     beforeEach(async () => {
       provider.list.mockReset()

apps/sim/lib/knowledge/__integration__/knowledge-projection.integration.ts

@@ -815,6 +815,73 @@ describe('the projector', () => {
     await db.delete(document).where(inArray(document.id, documents))
   })
 
+  it('marks what it can while a document it chose is deleted under it', async () => {
+    const [deleted, kept] = [generateId(), generateId()]
+    await db.insert(document).values(
+      [deleted, kept].map((id, index) => ({
+        id,
+        connectorId,
+        knowledgeBaseId: ids.knowledgeBaseId,
+        externalId: `fill-race-${index}`,
+        filename: `fill-race-${index}.md`,
+        fileUrl: `https://fixture.test/fill-race-${index}`,
+        fileSize: 12,
+        mimeType: 'text/plain',
+        processingStatus: 'completed' as const,
+        acl: aclOf('alice', 'bob'),
+      }))
+    )
+    await write('async', (tx) =>
+      tx
+        .insert(embedding)
+        .values([deleted, kept].map((id) => ({ ...chunkRow(generateId(), 0), documentId: id })))
+    )
+    await project()
+    for (const table of [embeddingSearch, embeddingKeywordTin]) {
+      await db
+        .update(table)
+        .set({ connectorId: null, acl: null })
+        .where(inArray(table.documentId, [deleted, kept]))
+    }
+    const deleter = postgres(process.env.DATABASE_URL!, { max: 1, onnotice: () => undefined })
+    try {
+      /** The deletion is under way when the fill reads, and commits while the fill still runs. */
+      let fill: ReturnType<typeof markUnfilledProjectionDocuments> | undefined
+      let settled = false
+      await deleter.begin(async (tx) => {
+        await tx`DELETE FROM document WHERE id = ${deleted}`
+        fill = markUnfilledProjectionDocuments(projector)
+        void fill.then(
+          () => {
+            settled = true
+          },
+          () => {
+            settled = true
+          }
+        )
+        await vi.waitFor(
+          async () => {
+            const [row] = await db.execute<{ waiting: boolean }>(
+              sql`SELECT EXISTS (SELECT 1 FROM pg_locks WHERE NOT granted) AS waiting`
+            )
+            expect(settled || Boolean(row?.waiting)).toBe(true)
+          },
+          { timeout: 5_000, interval: 10 }
+        )
+      })
+      await expect(fill).resolves.toMatchObject({ marked: expect.any(Number) })
+      const marks = await db
+        .select({ documentId: knowledgeProjectionDirty.documentId })
+        .from(knowledgeProjectionDirty)
+        .where(inArray(knowledgeProjectionDirty.documentId, [deleted, kept]))
+      expect(marks.map((mark) => mark.documentId)).toEqual([kept])
+    } finally {
+      await deleter.end()
+      await project()
+      await db.delete(document).where(inArray(document.id, [deleted, kept]))
+    }
+  })
+
   it.each(['sync', 'async'] as const)(
     'writes %s projection rows from a chunk commit only when the writer did not defer them',
     async (mode) => {

apps/sim/lib/knowledge/__integration__/member-document-lifecycle.integration.ts

@@ -19,7 +19,9 @@ import {
 } from '@/lib/knowledge/__integration__/seed-source-access-fixture'
 import {
   applyMemberDocumentLifecycle,
+  materializeDocumentAcls,
   recordMemberObservations,
+  rematerializeDocumentAcls,
   removeMemberObservationsForDocuments,
 } from '@/lib/knowledge/connectors/member-observations'
 import { resumeMembershipRewrites } from '@/lib/knowledge/connectors/member-sync-engine'
@@ -141,6 +143,24 @@ describe('member document lifecycle in PostgreSQL', () => {
         .where(eq(knowledgeConnector.id, members.connectorId))
     )[0].cursor
 
+  it('never grants a re-owned document to observers of the connector it left', async () => {
+    const moved = row('re-owned')
+    await insertRows([moved])
+    await observe([moved.id])
+    const aclOf = async () =>
+      (await db.select({ acl: document.acl }).from(document).where(eq(document.id, moved.id)))[0]
+        ?.acl
+    expect(await materializeDocumentAcls(members.connectorId, [moved.id])).toBe(1)
+    expect(await aclOf()).toEqual([members.members[0].subjectToken])
+
+    await db.update(document).set({ connectorId: ids.connectorId }).where(eq(document.id, moved.id))
+    expect(
+      await rematerializeDocumentAcls(ids.connectorId, [moved.id], (write) => db.transaction(write))
+    ).toBe(1)
+    expect(await aclOf()).toEqual([])
+    expect(await materializeDocumentAcls(ids.connectorId, [moved.id])).toBe(0)
+  })
+
   it('tombstones what this run unobserved right away and leaves the rest of a large connector to later runs', async () => {
     const pageBudget = MEMBER_TOMBSTONE_RECONCILE_PAGES_PER_RUN * 500
     const unobserved = Array.from({ length: pageBudget + 20 }, (_, index) =>