fix(knowledge): keep projection guards out of historical migrations

Restores 0016, 0019 and 0021 to their staging bodies; 0024 alone
re-creates the projection triggers with the mode guard, in the same
transaction that installs the marks. The fill starts only inside the
pass budget and reports what it marked as remaining. Passes dispatch
to Trigger.dev by the rule document processing uses, and the inline
coalescer clears its running flag in the same step it reads the owed
flag. Bumps the Helm chart for the projection cron job.

Author: waleedlatif1

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

@@ -548,6 +548,67 @@ describe('the projector', () => {
     expect((await rowAcl(embeddingSearch))?.acl).toEqual(aclOf('alice'))
   })
 
+  it('keeps the mark when a synchronous ACL change commits under a page that then writes stale values', async () => {
+    /** A pending revocation of Bob: the rows still name both until a pass. */
+    await write('async', (tx) =>
+      tx
+        .update(document)
+        .set({ acl: aclOf('alice') })
+        .where(eq(document.id, documentId))
+    )
+    const before = await markOf()
+    const [{ pid }] = await projector<Array<{ pid: number }>>`SELECT pg_backend_pid() AS pid`
+    /**
+     * A synchronous writer revokes Alice too and holds its transaction open: its fan-out has
+     * locked the projection rows and its mark bump is not yet visible.
+     */
+    const writer = postgres(process.env.DATABASE_URL!, { max: 1, onnotice: () => undefined })
+    let release: () => void = () => {}
+    const held = new Promise<void>((resolve) => {
+      release = resolve
+    })
+    let locked: () => void = () => {}
+    const fannedOut = new Promise<void>((resolve) => {
+      locked = resolve
+    })
+    try {
+      const writing = writer.begin(async (tx) => {
+        await tx`UPDATE document SET acl = ${aclOf('bob')} WHERE id = ${documentId}`
+        locked()
+        await held
+      })
+      await fannedOut
+      /**
+       * The pass reads the committed document, Alice alone, and its page blocks on the rows the
+       * writer holds. Once the writer commits, the page rewrites those rows from what it read.
+       */
+      const pass = project()
+      await vi.waitFor(async () => {
+        const [activity] = await db.execute<{ waiting: boolean }>(
+          sql`SELECT wait_event_type = 'Lock' AS waiting FROM pg_stat_activity WHERE pid = ${pid}`
+        )
+        expect(activity?.waiting).toBe(true)
+      })
+      release()
+      await writing
+      await pass
+    } finally {
+      release()
+      await writer.end()
+    }
+    /** The page wrote the value it read over the writer's newer one. */
+    expect((await rowAcl(embeddingSearch))?.acl).toEqual(aclOf('alice'))
+    /** The writer's bump is what the pass could not settle over: the rows stay decided on the document. */
+    expect((await markOf())?.generation).toBe((before?.generation ?? 0) + 1)
+    expect(await admitted()).toEqual({ vector: [], keyword: [] })
+    expect(await vectorIds()).toEqual([])
+    await project()
+    expect(await markOf()).toBeUndefined()
+    expect((await rowAcl(embeddingSearch))?.acl).toEqual(aclOf('bob'))
+    expect((await rowAcl(embeddingKeywordTin))?.acl).toEqual(aclOf('bob'))
+    expect(await admitted()).toEqual({ vector: [], keyword: [] })
+  })
+
   it('passes over a document deleted while it is being projected', async () => {
     const doomed = generateId()
     await db.insert(document).values({

apps/sim/lib/knowledge/projection/enqueue-inline.test.ts

@@ -1,6 +1,7 @@
 /**
  * @vitest-environment node
  */
+import { sleep } from '@sim/utils/helpers'
 import { describe, expect, it, vi } from 'vitest'
 
 const mocks = vi.hoisted(() => ({ runPass: vi.fn(), trigger: vi.fn() }))
@@ -10,6 +11,7 @@ vi.mock('@/lib/core/config/feature-flags', () => ({ isFeatureEnabled: async () =
 
 vi.mock('@trigger.dev/sdk', () => ({ tasks: { trigger: mocks.trigger } }))
 vi.mock('@/lib/core/config/env-flags', () => ({ isTriggerDevEnabled: false }))
+vi.mock('@/lib/core/config/trigger-runtime', () => ({ isInsideTriggerRun: () => false }))
 vi.mock('@/lib/knowledge/projection/run', () => ({ runKnowledgeProjectionPass: mocks.runPass }))
 
 import {
@@ -48,6 +50,28 @@ describe('knowledge projection without a Trigger.dev worker', () => {
     expect(mocks.trigger).not.toHaveBeenCalled()
   })
 
+  it('runs a pass for a request that arrives at any point while the last pass is finishing', async () => {
+    for (let hops = 0; hops < 8; hops++) {
+      mocks.runPass.mockReset()
+      /** The pass asks for another once it has settled, `hops` microtasks later. */
+      mocks.runPass
+        .mockImplementationOnce(() => {
+          void (async () => {
+            for (let hop = 0; hop < hops; hop++) await Promise.resolve()
+            await requestKnowledgeProjection()
+          })()
+          return Promise.resolve()
+        })
+        .mockResolvedValue(undefined)
+      await requestKnowledgeProjection()
+      await vi.waitFor(() => expect(mocks.runPass.mock.calls.length).toBeGreaterThanOrEqual(2), {
+        timeout: 200,
+      })
+      /** Lets the loop finish before the next interleaving starts. */
+      await sleep(1)
+    }
+  })
+
   it('logs a failed pass and still runs the pass owed after it', async () => {
     mocks.runPass.mockReset()
     const failFirst = heldPass()

apps/sim/lib/knowledge/projection/enqueue.test.ts

@@ -9,6 +9,8 @@ const mocks = vi.hoisted(() => ({
   trigger: vi.fn(),
   execute: vi.fn(),
   isFeatureEnabled: vi.fn(),
+  env: { TRIGGER_SECRET_KEY: 'fixture-key' as string | undefined },
+  insideRun: vi.fn(),
 }))
 
 vi.mock('@sim/db', () => ({ db: { execute: mocks.execute } }))
@@ -17,6 +19,8 @@ vi.mock('@/lib/core/config/feature-flags', () => ({ isFeatureEnabled: mocks.isFe
 vi.mock('@trigger.dev/sdk', () => ({ tasks: { trigger: mocks.trigger } }))
 vi.mock('@/lib/core/async-jobs/region', () => ({ resolveTriggerRegion: mocks.resolveRegion }))
 vi.mock('@/lib/core/config/env-flags', () => ({ isTriggerDevEnabled: true }))
+vi.mock('@/lib/core/config/env', () => ({ env: mocks.env }))
+vi.mock('@/lib/core/config/trigger-runtime', () => ({ isInsideTriggerRun: mocks.insideRun }))
 vi.mock('@/lib/knowledge/projection/run', () => ({ runKnowledgeProjectionPass: mocks.runPass }))
 
 import {
@@ -33,6 +37,8 @@ describe('knowledge projection enqueue', () => {
     mocks.trigger.mockResolvedValue({ id: 'run-1' })
     mocks.execute.mockResolvedValue([{ pending: true }])
     mocks.isFeatureEnabled.mockResolvedValue(false)
+    mocks.env.TRIGGER_SECRET_KEY = 'fixture-key'
+    mocks.insideRun.mockReturnValue(false)
   })
 
   afterEach(() => {
@@ -90,6 +96,26 @@ describe('knowledge projection enqueue', () => {
     expect(mocks.trigger).toHaveBeenCalledTimes(2)
   })
 
+  it('runs the pass in this process where Trigger.dev is enabled without its secret key', async () => {
+    mocks.env.TRIGGER_SECRET_KEY = undefined
+    mocks.runPass.mockResolvedValue(undefined)
+    await expect(enqueueKnowledgeProjectionSweep()).resolves.toEqual({
+      triggered: true,
+      backend: 'inline',
+      jobId: null,
+    })
+    expect(mocks.trigger).not.toHaveBeenCalled()
+    await vi.waitFor(() => expect(mocks.runPass).toHaveBeenCalledTimes(1))
+  })
+
+  it('enqueues from inside a Trigger.dev run whatever the environment holds', async () => {
+    mocks.env.TRIGGER_SECRET_KEY = undefined
+    mocks.insideRun.mockReturnValue(true)
+    await expect(enqueueKnowledgeProjectionSweep()).resolves.toMatchObject({
+      backend: 'trigger-dev',
+    })
+  })
+
   it('never fails the write that asked when the request is refused', async () => {
     vi.advanceTimersByTime(60_000)
     mocks.trigger.mockRejectedValueOnce(new Error('trigger unavailable'))

apps/sim/lib/knowledge/projection/enqueue.ts

@@ -3,8 +3,10 @@ import { SOURCE_ACL_PROJECTIONS } from '@sim/db/knowledge-projection'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
 import { sql } from 'drizzle-orm'
+import { env } from '@/lib/core/config/env'
 import { isTriggerDevEnabled } from '@/lib/core/config/env-flags'
 import { isFeatureEnabled } from '@/lib/core/config/feature-flags'
+import { isInsideTriggerRun } from '@/lib/core/config/trigger-runtime'
 
 const logger = createLogger('KnowledgeProjectionEnqueue')
 
@@ -31,35 +33,48 @@ const PROMPT_REQUEST_INTERVAL_MS = 5_000
 
 let lastPromptAt = 0
 
-/** The inline pass in flight when no Trigger.dev worker is configured, and whether another is owed. */
-let inlinePass: Promise<void> | undefined
+/** Whether an inline pass is running when no Trigger.dev worker is configured, and whether another is owed. */
+let inlineRunning = false
 let inlinePassOwed = false
 
 /**
  * Starts a pass in this process without waiting for it: at most one runs at a time, a request while
  * one runs is folded into a single pass after it, and a failed pass is logged without dropping one
- * owed after it. The
- * pass module loads on first use. For deployments without a Trigger.dev worker, whose sweep and
- * writes run passes here, as their document processing does.
+ * owed after it. The loop reads the owed flag and clears the running flag in the same synchronous
+ * step, so a request can never land between the two and be dropped. The pass module loads on first
+ * use. For deployments without a Trigger.dev worker, whose sweep and writes run passes here, as
+ * their document processing does.
  */
 function runInline(): void {
-  if (inlinePass) {
+  if (inlineRunning) {
     inlinePassOwed = true
     return
   }
-  inlinePass = (async () => {
-    do {
+  inlineRunning = true
+  void (async () => {
+    for (;;) {
       inlinePassOwed = false
       try {
         const { runKnowledgeProjectionPass } = await import('@/lib/knowledge/projection/run')
         await runKnowledgeProjectionPass({ budgetMs: KNOWLEDGE_PROJECTION_PASS_BUDGET_MS })
       } catch (error) {
         logger.error('Inline knowledge projection pass failed', { error: getErrorMessage(error) })
       }
-    } while (inlinePassOwed)
-  })().finally(() => {
-    inlinePass = undefined
-  })
+      if (!inlinePassOwed) {
+        inlineRunning = false
+        return
+      }
+    }
+  })()
+}
+
+/**
+ * Whether passes run on Trigger.dev, by the rule document processing dispatches with: inside a
+ * Trigger.dev run always, and otherwise only where Trigger.dev is enabled and the secret key the
+ * SDK authenticates with is set.
+ */
+function projectsOnTrigger(): boolean {
+  return isInsideTriggerRun() || Boolean(isTriggerDevEnabled && env.TRIGGER_SECRET_KEY)
 }
 
 /**
@@ -70,7 +85,7 @@ function runInline(): void {
  * leaves the marks to the sweep, which keeps enqueueing a pass every minute until one runs.
  */
 export async function requestKnowledgeProjection(): Promise<void> {
-  if (!isTriggerDevEnabled) {
+  if (!projectsOnTrigger()) {
     runInline()
     return
   }
@@ -125,7 +140,7 @@ export async function enqueueKnowledgeProjectionSweep(): Promise<KnowledgeProjec
   if (!(await hasKnowledgeProjectionWork())) {
     return { triggered: false, backend: null, jobId: null }
   }
-  if (!isTriggerDevEnabled) {
+  if (!projectsOnTrigger()) {
     runInline()
     return { triggered: true, backend: 'inline', jobId: null }
   }

apps/sim/lib/knowledge/projection/run.test.ts

@@ -1,7 +1,7 @@
 /**
  * @vitest-environment node
  */
-import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const mocks = vi.hoisted(() => ({
   runProjection: vi.fn(),
@@ -106,6 +106,40 @@ describe('runKnowledgeProjectionPass', () => {
     })
   })
 
+  describe('at the budget', () => {
+    beforeEach(() => {
+      vi.useFakeTimers()
+      mocks.isFeatureEnabled.mockResolvedValue(true)
+    })
+
+    afterEach(() => {
+      vi.useRealTimers()
+    })
+
+    it('does not start the fill once the budget is spent', async () => {
+      mocks.runProjection.mockImplementation(async () => {
+        vi.advanceTimersByTime(60_001)
+        return drained
+      })
+      await expect(runKnowledgeProjectionPass({ budgetMs: 60_000 })).resolves.toMatchObject({
+        remaining: false,
+        filled: 0,
+      })
+      expect(mocks.markUnfilled).not.toHaveBeenCalled()
+    })
+
+    it('reports documents the fill marked but no round settled as remaining', async () => {
+      mocks.markUnfilled.mockImplementation(async () => {
+        vi.advanceTimersByTime(60_001)
+        return { marked: 2, cursor: { projection: 0, afterId: 'row-2' } }
+      })
+      await expect(runKnowledgeProjectionPass({ budgetMs: 60_000 })).resolves.toMatchObject({
+        remaining: true,
+        filled: 2,
+      })
+    })
+  })
+
   it('does not fill while marks remain, so writers are converged first', async () => {
     mocks.isFeatureEnabled.mockResolvedValue(true)
     mocks.runProjection.mockResolvedValue({ ...drained, settled: 0, remaining: true })

apps/sim/lib/knowledge/projection/run.ts

@@ -106,10 +106,12 @@ export async function runKnowledgeProjectionPass(options: {
         result.deferred = round.reduce((most, progress) => Math.max(most, progress.deferred), 0)
         break
       }
-      if (fillCursor === null) break
+      /** The fill starts only inside the budget, and what it marks is left for a round to settle. */
+      if (fillCursor === null || Date.now() >= deadline) break
       const fill = await markUnfilledProjectionDocuments(sessions[0], fillCursor)
       result.filled += fill.marked
       fillCursor = fill.cursor
+      if (fill.marked > 0) result.remaining = true
       if (fill.marked === 0 && fillCursor === null) break
     }
     logger.info('Knowledge projection pass finished', result)

helm/sim/Chart.yaml

@@ -2,7 +2,7 @@ apiVersion: v2
 name: sim
 description: A Helm chart for Sim - the open-source AI workspace where teams build, deploy, and manage AI agents
 type: application
-version: 1.11.3
+version: 1.11.4
 appVersion: "v0.8.26"
 kubeVersion: ">=1.25.0-0"
 home: https://sim.ai

packages/db/knowledge-projection.ts

@@ -60,18 +60,18 @@ const SOURCE_VECTOR_WIDTHS = [1536, 384, 768, 1024, 3072] as const
 const widthColumn = (name: string, width: number) => (width === 1536 ? name : `${name}_${width}`)
 
 /** The halfvec columns of `embedding_search`, in width order. */
-export const SEARCH_VECTOR_COLUMNS = SEARCH_VECTOR_WIDTHS.map((width) =>
-  widthColumn('vector', width)
-)
+const SEARCH_VECTOR_COLUMNS = SEARCH_VECTOR_WIDTHS.map((width) => widthColumn('vector', width))
 
 /** The bit columns of `embedding_search`, whose width check requires exactly one. */
 const SEARCH_BINARY_COLUMNS = ['"binary"', 'binary_384', 'binary_768', 'binary_1024', 'binary_3072']
 
 /**
  * The halfvec projections of an embedding row, in {@link SEARCH_VECTOR_COLUMNS} order. Shortening
- * is valid only for the two OpenAI models trained for prefix retrieval.
+ * is valid only for the two OpenAI models trained for prefix retrieval. These and the bit
+ * projections match what `sync_embedding_search()` from `0016_backfill_search_vectors` writes, so
+ * a pass over rows a synchronous writer wrote finds nothing to rewrite.
  */
-export function searchVectorProjections(prefix: string, shortened: string): string {
+function searchVectorProjections(prefix: string, shortened: string): string {
   return SEARCH_VECTOR_WIDTHS.map((width) =>
     width === 512
       ? `CASE WHEN ${shortened} THEN subvector(coalesce(${SOURCE_VECTOR_WIDTHS.map((size) => `${prefix}.${widthColumn('embedding', size)}`).join(', ')}), 1, 512)::halfvec(512) END`
@@ -80,14 +80,14 @@ export function searchVectorProjections(prefix: string, shortened: string): stri
 }
 
 /** The bit projections of an embedding row, in {@link SEARCH_BINARY_COLUMNS} order. */
-export function searchBinaryProjections(prefix: string): string {
+function searchBinaryProjections(prefix: string): string {
   return `binary_quantize(${prefix}.embedding)::bit(1536), binary_quantize(${prefix}.embedding_384)::bit(384),
     binary_quantize(${prefix}.embedding_768)::bit(768), binary_quantize(${prefix}.embedding_1024)::bit(1024),
     binary_quantize(${prefix}.embedding_3072)::bit(3072)`
 }
 
 /** Whether a knowledge base's model is shortened for the 512 projection, for an embedding row. */
-export function searchVectorShortened(model: string, prefix: string): string {
+function searchVectorShortened(model: string, prefix: string): string {
   return `${model} IN ${SHORTENED_EMBEDDING_MODELS} AND ${prefix}.embedding_384 IS NULL`
 }