From 1dddb8bd662f0e3e82c8873aad71f3e2027b9e39 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 20 Sep 2026 10:05:55 -0700 Subject: [PATCH 1/6] fix(knowledge): run the projection source/ACL backfill in the background instead of the deploy Script migration 0021 filled the ranking projections' connector_id and acl columns synchronously, 500 rows per page under a 60s statement timeout. On embedding_search every filled row is re-inserted into each HNSW index, so a page's cost is index maintenance rather than the plan: one page ran past the timeout and the migration, and the deploy, failed. The migration now installs the triggers and builds the indexes only, both idempotent, and the backfill runs from a Trigger.dev task in keyset pages of unfilled rows, paced with a pause between pages and chained across bounded runs. Search does not depend on it: an unfilled row passes the on-row candidate predicate and is decided at hydration under the full document predicate, exactly as every candidate was before the columns existed. Claude-Session: https://claude.ai/code/session_01XU6c7pKRpa5CMoMHKDdqxX --- .../projection-source-acl-backfill.ts | 35 ++++ .../access/predicate.postgres.test.ts | 14 ++ .../lib/knowledge/access/predicate.test.ts | 40 ++++- apps/sim/lib/knowledge/access/predicate.ts | 9 +- .../projection-source-acl-backfill.test.ts | 118 +++++++++++++ .../search/projection-source-acl-backfill.ts | 96 +++++++++++ apps/sim/lib/knowledge/search/queries.test.ts | 2 + .../scripts/backfill-projection-source-acl.ts | 34 ++++ ...mbedding_search_connector.postgres.test.ts | 128 ++++++++++++++ .../0021_embedding_search_connector.ts | 158 +++++++++++++----- 10 files changed, 589 insertions(+), 45 deletions(-) create mode 100644 apps/sim/background/projection-source-acl-backfill.ts create mode 100644 apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts create mode 100644 apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts create mode 100644 apps/sim/scripts/backfill-projection-source-acl.ts create mode 100644 packages/db/script-migrations/0021_embedding_search_connector.postgres.test.ts diff --git a/apps/sim/background/projection-source-acl-backfill.ts b/apps/sim/background/projection-source-acl-backfill.ts new file mode 100644 index 00000000000..e6b75c0794a --- /dev/null +++ b/apps/sim/background/projection-source-acl-backfill.ts @@ -0,0 +1,35 @@ +import { task, tasks } from '@trigger.dev/sdk' +import { resolveTriggerRegion } from '@/lib/core/async-jobs/region' +import { + PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, + type ProjectionSourceAclBackfillPayload, + runProjectionSourceAclBackfill, +} from '@/lib/knowledge/search/projection-source-acl-backfill' + +/** One run's share of the backfill, inside the worker's run ceiling with room to end its page. */ +const RUN_BUDGET_MS = 60 * 60 * 1000 + +/** + * Trigger.dev wrapper around `runProjectionSourceAclBackfill`. A run fills unset rows for up to + * {@link RUN_BUDGET_MS}, then triggers its continuation from the cursor it reached, so the whole + * projection is filled across as many bounded runs as it takes. Retry-safe: every run writes only + * rows still unset, so a retried or restarted run repeats no write. The queue admits one run at a + * time, so two starts never fill the same pages against each other. + */ +export const projectionSourceAclBackfillTask = task({ + id: PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, + machine: 'small-1x', + retry: { maxAttempts: 3 }, + queue: { + name: PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, + concurrencyLimit: 1, + }, + run: async (payload: ProjectionSourceAclBackfillPayload) => { + const cursor = await runProjectionSourceAclBackfill(payload, { budgetMs: RUN_BUDGET_MS }) + if (!cursor) return + const continuation: ProjectionSourceAclBackfillPayload = { ...payload, cursor } + await tasks.trigger(PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, continuation, { + region: await resolveTriggerRegion(), + }) + }, +}) diff --git a/apps/sim/lib/knowledge/access/predicate.postgres.test.ts b/apps/sim/lib/knowledge/access/predicate.postgres.test.ts index 028bb285eae..7d04a832913 100644 --- a/apps/sim/lib/knowledge/access/predicate.postgres.test.ts +++ b/apps/sim/lib/knowledge/access/predicate.postgres.test.ts @@ -622,6 +622,20 @@ describe.runIf(Boolean(databaseUrl))('knowledge ACLs in PostgreSQL', () => { ) expect(await admits(perRow, 'members-other')).toBe(false) expect(await onRowAdmits('members-other')).toBe(false) + /** + * A chunk the backfill has not reached carries no source or ACL yet. It is ranked, and decided + * at hydration under the full predicate, exactly as every candidate was before the columns + * existed — search must not depend on the backfill, and must not lose a document to it. + */ + await connection.unsafe( + "INSERT INTO document(id, connector_id, acl, acl_verified_at) VALUES ('unfilled-other', 'members', ARRAY['s:slack:-:bob'], statement_timestamp())" + ) + await connection.unsafe( + "INSERT INTO embedding_search(id, document_id) VALUES ('unfilled-other-chunk', 'unfilled-other'), ('admin-current-unfilled', 'admin-current')" + ) + expect(await onRowAdmits('unfilled-other')).toBe(true) + expect(await admits(perRow, 'unfilled-other')).toBe(false) + expect(await onRowAdmits('admin-current')).toBe(true) /** * Candidate ranking defers the live source proof, as the per-row candidate predicate does: a * caller holds those grants only after authorization, so applying the clause during ranking diff --git a/apps/sim/lib/knowledge/access/predicate.test.ts b/apps/sim/lib/knowledge/access/predicate.test.ts index 11582cc360e..8de16218f92 100644 --- a/apps/sim/lib/knowledge/access/predicate.test.ts +++ b/apps/sim/lib/knowledge/access/predicate.test.ts @@ -17,13 +17,51 @@ vi.unmock('@sim/db/schema') process.env.DATABASE_URL ??= 'postgresql://user:pass@localhost:5432/test' const { PgDialect } = await import('drizzle-orm/pg-core') -const { knowledgeAccessCondition } = await import('@/lib/knowledge/access/predicate') +const { embeddingSearch } = await import('@sim/db/schema') +const { knowledgeAccessCondition, projectionCandidateAccessCondition } = await import( + '@/lib/knowledge/access/predicate' +) const { SYSTEM_ACCESS_SCOPE } = await import('@/lib/knowledge/access/types') function render(condition: ReturnType) { return new PgDialect().sqlToQuery(condition) } +describe('projectionCandidateAccessCondition', () => { + const plan = { + connectors: { workspace: ['ws-src'], admin: [], members: [], liveProofRequired: [] }, + observers: { confirmed: [], observed: [] }, + memberSources: [], + } + + it('admits a row the backfill has not filled, and decides a filled row on its mirrored columns', () => { + const { sql, params } = render( + projectionCandidateAccessCondition( + embeddingSearch, + { kind: 'user', userId: 'user-1', tokens: ['ws', 'u:alice'] }, + plan + ) + ) + expect(sql).toBe( + '("embedding_search"."acl" IS NULL OR ("embedding_search"."acl" && ARRAY[$1, $2]::text[]\n' + + ' AND ("embedding_search"."connector_id" IS NULL OR "embedding_search"."connector_id" = ANY(ARRAY[$3]::text[]))))' + ) + expect(params).toEqual(['ws', 'u:alice', 'ws-src']) + }) + + it('still denies everything for an empty token set', () => { + expect( + render( + projectionCandidateAccessCondition( + embeddingSearch, + { kind: 'user', userId: 'user-1', tokens: [] }, + plan + ) + ).sql + ).toBe('false') + }) +}) + describe('knowledgeAccessCondition', () => { it('overlaps the ACL with the tokens as a literal array of scalar binds', () => { const { sql, params } = render( diff --git a/apps/sim/lib/knowledge/access/predicate.ts b/apps/sim/lib/knowledge/access/predicate.ts index a6a988c723e..203d47f93a3 100644 --- a/apps/sim/lib/knowledge/access/predicate.ts +++ b/apps/sim/lib/knowledge/access/predicate.ts @@ -312,6 +312,11 @@ export function knowledgeCandidateAccessConditionForConnectors( * observation's freshness, and requirement clauses live on the document. Both are refused there, * under the full predicate, before content is returned — this predicate only decides what is worth * ranking. + * + * A row the backfill has not reached yet carries no ACL (`acl IS NULL`) and passes: it is decided at + * hydration under the full document predicate, exactly as every candidate was before the columns + * existed. The backfill runs in the background, so search never waits on it and never loses a + * document to it. */ export function projectionCandidateAccessCondition( projection: { connectorId: AnyPgColumn | SQL; acl: AnyPgColumn | SQL }, @@ -330,8 +335,8 @@ export function projectionCandidateAccessCondition( ...plan.connectors.admin, ...plan.connectors.members, ] - return sql`(${projection.acl} && ${tokens} - AND (${projection.connectorId} IS NULL OR ${inSources(mirrored)}))` + return sql`(${projection.acl} IS NULL OR (${projection.acl} && ${tokens} + AND (${projection.connectorId} IS NULL OR ${inSources(mirrored)})))` } /** diff --git a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts new file mode 100644 index 00000000000..4277086242b --- /dev/null +++ b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts @@ -0,0 +1,118 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockBackfill, mockEnd, mockPostgres, mockTasksTrigger } = vi.hoisted(() => ({ + mockBackfill: vi.fn(), + mockEnd: vi.fn(async () => undefined), + mockPostgres: vi.fn(), + mockTasksTrigger: vi.fn(async () => ({ id: 'run-1' })), +})) + +vi.mock('@sim/db', () => ({ resolveDbUrl: () => 'postgres://localhost:5432/sim' })) +vi.mock('@sim/db/script-migrations/0021_embedding_search_connector', () => ({ + PROJECTION_SOURCE_ACL_TABLES: ['embedding_search', 'embedding_keyword_tin'], + backfillProjectionSourceAcl: mockBackfill, +})) +vi.mock('postgres', () => ({ default: mockPostgres })) +vi.mock('@trigger.dev/sdk', () => ({ tasks: { trigger: mockTasksTrigger } })) +vi.mock('@/lib/core/async-jobs/region', () => ({ resolveTriggerRegion: async () => 'us-east-1' })) + +import { + enqueueProjectionSourceAclBackfill, + runProjectionSourceAclBackfill, +} from '@/lib/knowledge/search/projection-source-acl-backfill' + +const connection = { end: mockEnd } + +describe('runProjectionSourceAclBackfill', () => { + beforeEach(() => { + vi.clearAllMocks() + mockPostgres.mockReturnValue(connection) + mockBackfill.mockImplementation(async (_sql, projection, options) => ({ + projection, + scanned: 0, + written: 0, + afterId: options?.afterId ?? '', + done: true, + })) + }) + + it('fills both projections in order on its own connection and closes it', async () => { + await expect(runProjectionSourceAclBackfill({ pageSize: 50, pauseMs: 10 })).resolves.toBeNull() + expect(mockBackfill.mock.calls.map(([, projection]) => projection)).toEqual([ + 'embedding_search', + 'embedding_keyword_tin', + ]) + for (const [sql, , options] of mockBackfill.mock.calls) { + expect(sql).toBe(connection) + expect(options).toMatchObject({ afterId: undefined, pageSize: 50, pauseMs: 10 }) + } + expect(mockEnd).toHaveBeenCalledTimes(1) + }) + + it('resumes after the cursor in its projection and from the start of the next', async () => { + await runProjectionSourceAclBackfill({ + cursor: { projection: 'embedding_keyword_tin', afterId: 'chunk-9' }, + }) + expect(mockBackfill).toHaveBeenCalledTimes(1) + expect(mockBackfill.mock.calls[0][1]).toBe('embedding_keyword_tin') + expect(mockBackfill.mock.calls[0][2]).toMatchObject({ afterId: 'chunk-9' }) + }) + + it('returns where a budgeted run stopped so the next run can carry on', async () => { + mockBackfill.mockResolvedValueOnce({ + projection: 'embedding_search', + scanned: 100, + written: 100, + afterId: 'chunk-100', + done: false, + }) + await expect(runProjectionSourceAclBackfill({}, { budgetMs: 1000 })).resolves.toEqual({ + projection: 'embedding_search', + afterId: 'chunk-100', + }) + expect(mockBackfill).toHaveBeenCalledTimes(1) + expect(mockBackfill.mock.calls[0][2].budgetMs).toBeLessThanOrEqual(1000) + expect(mockEnd).toHaveBeenCalledTimes(1) + }) + + it('closes the connection when a page fails', async () => { + mockBackfill.mockRejectedValueOnce(new Error('canceling statement due to statement timeout')) + await expect(runProjectionSourceAclBackfill({})).rejects.toThrow('statement timeout') + expect(mockEnd).toHaveBeenCalledTimes(1) + }) +}) + +describe('enqueueProjectionSourceAclBackfill', () => { + beforeEach(() => { + vi.clearAllMocks() + mockPostgres.mockReturnValue(connection) + mockBackfill.mockResolvedValue({ + projection: 'embedding_search', + scanned: 0, + written: 0, + afterId: '', + done: true, + }) + }) + + it('hands the backfill to the Trigger.dev worker when there is one', async () => { + await expect(enqueueProjectionSourceAclBackfill({ pageSize: 25 }, true)).resolves.toEqual({ + runId: 'run-1', + }) + expect(mockTasksTrigger).toHaveBeenCalledWith( + 'projection-source-acl-backfill', + { pageSize: 25 }, + { region: 'us-east-1' } + ) + expect(mockBackfill).not.toHaveBeenCalled() + }) + + it('fills the projections inline without one', async () => { + await expect(enqueueProjectionSourceAclBackfill({}, false)).resolves.toBeNull() + expect(mockTasksTrigger).not.toHaveBeenCalled() + expect(mockBackfill).toHaveBeenCalledTimes(2) + }) +}) diff --git a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts new file mode 100644 index 00000000000..ee1f07529a0 --- /dev/null +++ b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts @@ -0,0 +1,96 @@ +import { resolveDbUrl } from '@sim/db' +import { + backfillProjectionSourceAcl, + PROJECTION_SOURCE_ACL_TABLES, + type ProjectionSourceAclTable, +} from '@sim/db/script-migrations/0021_embedding_search_connector' +import { createLogger } from '@sim/logger' +import postgres from 'postgres' +import { resolveTriggerRegion } from '@/lib/core/async-jobs/region' +import { env } from '@/lib/core/config/env' +import { isTriggerDevEnabled } from '@/lib/core/config/env-flags' + +const logger = createLogger('ProjectionSourceAclBackfill') + +export const PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID = 'projection-source-acl-backfill' + +/** Where a run stopped, so the next one carries on from there instead of rescanning. */ +export interface ProjectionSourceAclBackfillCursor { + projection: ProjectionSourceAclTable + afterId: string +} + +export interface ProjectionSourceAclBackfillPayload { + /** The first projection's first page when absent. */ + cursor?: ProjectionSourceAclBackfillCursor + pageSize?: number + pauseMs?: number +} + +export interface ProjectionSourceAclBackfillRunOptions { + /** Stop once this much time has passed and return where to resume; unbounded otherwise. */ + budgetMs?: number +} + +/** + * Fills the ranking projections' source and ACL columns from the cursor onwards, one projection + * after the other, on a connection of its own: the page statement binds the keyset cursor as a + * scalar and needs no array parameter, so the pool's options would serve, but a run this long + * should not hold one of the worker's pooled connections. Returns the cursor to continue from when + * the budget ran out, `null` once both projections are filled. + */ +export async function runProjectionSourceAclBackfill( + payload: ProjectionSourceAclBackfillPayload, + options: ProjectionSourceAclBackfillRunOptions = {} +): Promise { + const url = resolveDbUrl('DATABASE_URL', process.env.SIM_DB_ROLE?.trim() || 'web') + if (!url) throw new Error('DATABASE_URL is required to backfill the projection source and ACL') + const sql = postgres(url, { max: 1, max_lifetime: null, onnotice: () => undefined }) + const startedAt = Date.now() + try { + const start = payload.cursor + ? PROJECTION_SOURCE_ACL_TABLES.indexOf(payload.cursor.projection) + : 0 + if (start < 0) throw new Error(`Unknown projection ${payload.cursor?.projection}`) + for (const projection of PROJECTION_SOURCE_ACL_TABLES.slice(start)) { + const budgetMs = + options.budgetMs === undefined + ? undefined + : Math.max(0, options.budgetMs - (Date.now() - startedAt)) + const progress = await backfillProjectionSourceAcl(sql, projection, { + afterId: payload.cursor?.projection === projection ? payload.cursor.afterId : undefined, + pageSize: payload.pageSize, + pauseMs: payload.pauseMs, + budgetMs, + }) + if (!progress.done) return { projection, afterId: progress.afterId } + } + logger.info('Projection source and ACL backfill complete', { + elapsedMs: Date.now() - startedAt, + }) + return null + } finally { + await sql.end() + } +} + +/** + * Starts the backfill the way the table backfill is started: on the deployment's Trigger.dev + * worker when there is one, where bounded runs chain until both projections are filled, and inline + * otherwise. Safe to call again at any time — a run only fills rows still unset. + */ +export async function enqueueProjectionSourceAclBackfill( + payload: ProjectionSourceAclBackfillPayload = {}, + useTrigger = Boolean(isTriggerDevEnabled && env.TRIGGER_SECRET_KEY) +): Promise<{ runId: string } | null> { + if (useTrigger) { + const { tasks } = await import('@trigger.dev/sdk') + const handle = await tasks.trigger(PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, payload, { + region: await resolveTriggerRegion(), + }) + logger.info('Projection source and ACL backfill enqueued', { runId: handle.id }) + return { runId: handle.id } + } + await runProjectionSourceAclBackfill(payload) + return null +} diff --git a/apps/sim/lib/knowledge/search/queries.test.ts b/apps/sim/lib/knowledge/search/queries.test.ts index 8137a007d31..f76f18bc09e 100644 --- a/apps/sim/lib/knowledge/search/queries.test.ts +++ b/apps/sim/lib/knowledge/search/queries.test.ts @@ -1527,6 +1527,8 @@ describe('permitted-document planner', () => { /** The ranked CTE carries the mirrored source and ACL the predicate tests. */ expect(statement).toContain('AS connector_id') expect(statement).toContain('ranked_tin_chunks.acl') + /** A row the backfill has not filled (`acl IS NULL`) is ranked and decided at hydration. */ + expect(statement).toContain('" IS NULL OR ("') }) it('widens the window for a broad resolved scope whose first page came back short', async () => { diff --git a/apps/sim/scripts/backfill-projection-source-acl.ts b/apps/sim/scripts/backfill-projection-source-acl.ts new file mode 100644 index 00000000000..7df5b91fdc6 --- /dev/null +++ b/apps/sim/scripts/backfill-projection-source-acl.ts @@ -0,0 +1,34 @@ +#!/usr/bin/env bun + +/** + * Starts the projection source and ACL backfill that script migration + * `0021_embedding_search_connector` leaves to the background: on a deployment with Trigger.dev it + * enqueues the `projection-source-acl-backfill` task, which chains bounded runs until both ranking + * projections are filled; without one it fills them here, paced the same way. Safe to run again at + * any time — a run only fills rows still unset. + * + * Usage: + * bun apps/sim/scripts/backfill-projection-source-acl.ts + */ + +import { createLogger } from '@sim/logger' +import { toError } from '@sim/utils/errors' +import { enqueueProjectionSourceAclBackfill } from '@/lib/knowledge/search/projection-source-acl-backfill' + +const logger = createLogger('BackfillProjectionSourceAcl') + +if (import.meta.main) { + enqueueProjectionSourceAclBackfill().then( + (handle) => { + logger.info( + handle ? 'Backfill enqueued on the Trigger.dev worker' : 'Backfill complete', + handle ?? {} + ) + process.exit(0) + }, + (error) => { + logger.error('Backfill failed', toError(error)) + process.exit(1) + } + ) +} diff --git a/packages/db/script-migrations/0021_embedding_search_connector.postgres.test.ts b/packages/db/script-migrations/0021_embedding_search_connector.postgres.test.ts new file mode 100644 index 00000000000..f0ab90c3a7c --- /dev/null +++ b/packages/db/script-migrations/0021_embedding_search_connector.postgres.test.ts @@ -0,0 +1,128 @@ +import { + backfillProjectionSourceAcl, + embeddingSearchConnectorMigration, +} from '@sim/db/script-migrations/0021_embedding_search_connector' +import { generateId } from '@sim/utils/id' +import postgres, { type Sql } from 'postgres' +import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest' + +const databaseUrl = process.env.KNOWLEDGE_ACL_TEST_DATABASE_URL + +/** + * The projections here carry only the columns the source and ACL triggers and backfill touch; the + * vector and lexeme columns, and their indexes, are what make the backfill slow, not what decides + * which rows it writes. + */ +describe.runIf(Boolean(databaseUrl))('projection source and ACL backfill in PostgreSQL', () => { + let admin: Sql + let sql: Sql + const schemaName = `projection_acl_${generateId().replaceAll('-', '')}` + + const projected = (projection: 'embedding_search' | 'embedding_keyword_tin') => + sql<{ id: string; connector_id: string | null; acl: string[] | null }[]>` + SELECT id, connector_id, acl FROM ${sql(projection)} ORDER BY id` + + beforeAll(async () => { + const url = new URL(databaseUrl!) + if ( + !['localhost', '127.0.0.1'].includes(url.hostname) || + !url.pathname.startsWith('/sim_acl_test') + ) { + throw new Error('Projection tests require a disposable local integration database') + } + admin = postgres(url.toString(), { max: 1, onnotice: () => undefined }) + await admin.unsafe(`CREATE SCHEMA "${schemaName}"`) + sql = postgres(url.toString(), { + max: 1, + onnotice: () => undefined, + connection: { search_path: schemaName }, + }) + await sql`CREATE TABLE document ( + id text PRIMARY KEY, connector_id text, acl text[] NOT NULL DEFAULT '{ws}' + )` + for (const projection of ['embedding_search', 'embedding_keyword_tin']) { + await sql`CREATE TABLE ${sql(projection)} ( + id text PRIMARY KEY, document_id text NOT NULL, enabled boolean NOT NULL DEFAULT true, + connector_id text, acl text[] + )` + } + await embeddingSearchConnectorMigration.up(sql) + }, 60_000) + + afterAll(async () => { + await sql?.end() + await admin?.unsafe(`DROP SCHEMA IF EXISTS "${schemaName}" CASCADE`) + await admin?.end() + }) + + beforeEach(async () => { + await sql`TRUNCATE embedding_search, embedding_keyword_tin, document` + await sql`ALTER TABLE embedding_search DISABLE TRIGGER embedding_search_source_acl_set` + await sql`ALTER TABLE embedding_keyword_tin DISABLE TRIGGER embedding_keyword_tin_source_acl_set` + }) + + it('installs its triggers and indexes again without failing, so a cut-short deploy completes', async () => { + await expect(embeddingSearchConnectorMigration.up(sql)).resolves.toBeUndefined() + const indexes = await sql<{ indexname: string }[]>` + SELECT indexname FROM pg_indexes WHERE schemaname = ${schemaName} ORDER BY indexname` + expect(indexes.map((row) => row.indexname)).toEqual( + expect.arrayContaining([ + 'embedding_keyword_tin_acl_gin_idx', + 'embedding_search_acl_gin_idx', + 'embedding_search_source_idx', + ]) + ) + }) + + it('fills only the rows still unset, in pages, and leaves a chunk that changed documents to its trigger', async () => { + await sql`INSERT INTO document (id, connector_id, acl) VALUES + ('doc-a', 'src-a', ARRAY['u:alice']), ('doc-b', NULL, ARRAY['ws']), ('doc-c', 'src-c', ARRAY['u:carol'])` + await sql`INSERT INTO embedding_search (id, document_id, connector_id, acl) VALUES + ('c1', 'doc-a', NULL, NULL), ('c2', 'doc-b', NULL, NULL), + ('c3', 'doc-c', 'src-old', ARRAY['u:stale']), ('c4', 'doc-a', NULL, NULL), ('c5', 'missing', NULL, NULL)` + const progress = await backfillProjectionSourceAcl(sql, 'embedding_search', { + pageSize: 2, + pauseMs: 0, + }) + expect(progress).toEqual({ + projection: 'embedding_search', + scanned: 3, + written: 3, + afterId: 'c4', + done: true, + }) + expect(await projected('embedding_search')).toEqual([ + { id: 'c1', connector_id: 'src-a', acl: ['u:alice'] }, + { id: 'c2', connector_id: null, acl: ['ws'] }, + { id: 'c3', connector_id: 'src-old', acl: ['u:stale'] }, + { id: 'c4', connector_id: 'src-a', acl: ['u:alice'] }, + { id: 'c5', connector_id: null, acl: null }, + ]) + expect(await projected('embedding_keyword_tin')).toEqual([]) + }) + + it('stops at its budget with the cursor to resume from, and resumes after it', async () => { + await sql`INSERT INTO document (id, connector_id, acl) VALUES ('doc', 'src', ARRAY['u:alice'])` + await sql`INSERT INTO embedding_keyword_tin (id, document_id) VALUES + ('k1', 'doc'), ('k2', 'doc'), ('k3', 'doc')` + const paused = await backfillProjectionSourceAcl(sql, 'embedding_keyword_tin', { + pageSize: 1, + pauseMs: 0, + budgetMs: 0, + }) + expect(paused).toMatchObject({ scanned: 1, written: 1, afterId: 'k1', done: false }) + const resumed = await backfillProjectionSourceAcl(sql, 'embedding_keyword_tin', { + afterId: paused.afterId, + pageSize: 1, + pauseMs: 0, + }) + expect(resumed).toMatchObject({ scanned: 2, written: 2, afterId: 'k3', done: true }) + expect((await projected('embedding_keyword_tin')).map((row) => row.acl)).toEqual([ + ['u:alice'], + ['u:alice'], + ['u:alice'], + ]) + const again = await backfillProjectionSourceAcl(sql, 'embedding_keyword_tin', { pauseMs: 0 }) + expect(again).toMatchObject({ scanned: 0, written: 0, afterId: '', done: true }) + }) +}) diff --git a/packages/db/script-migrations/0021_embedding_search_connector.ts b/packages/db/script-migrations/0021_embedding_search_connector.ts index 3ddb1ad9430..cc004a81e4d 100644 --- a/packages/db/script-migrations/0021_embedding_search_connector.ts +++ b/packages/db/script-migrations/0021_embedding_search_connector.ts @@ -1,15 +1,29 @@ import type { ScriptMigration } from '@sim/db/script-migrations/types' import { createLogger } from '@sim/logger' +import { sleep } from '@sim/utils/helpers' import postgres, { type Sql } from 'postgres' const logger = createLogger('ProjectionSourceAcl') -/** Chunks read per page; each page commits on its own, as the other projection backfills do. */ -const BATCH_SIZE = 500 +/** + * Chunks filled per page. Every write to `embedding_search` re-inserts the row into each of its + * HNSW indexes, so a page's cost is index maintenance rather than the plan, and a small page keeps + * each transaction short and its locks brief. + */ +export const PROJECTION_SOURCE_ACL_PAGE_SIZE = 100 + +/** Pause between pages, so the backfill shares the database with the search it serves. */ +export const PROJECTION_SOURCE_ACL_PAGE_PAUSE_MS = 250 + +/** Longest a page may run before the database cancels it; the run then fails and resumes. */ +const PAGE_STATEMENT_TIMEOUT = '60s' + +/** Pages between progress log lines. */ +const PROGRESS_EVERY_PAGES = 100 /** The projections that carry their document's source and ACL. */ -const PROJECTIONS = ['embedding_search', 'embedding_keyword_tin'] as const -type Projection = (typeof PROJECTIONS)[number] +export const PROJECTION_SOURCE_ACL_TABLES = ['embedding_search', 'embedding_keyword_tin'] as const +export type ProjectionSourceAclTable = (typeof PROJECTION_SOURCE_ACL_TABLES)[number] /** * Carries a chunk's source and ACL onto the ranking projections and keeps them there. @@ -50,7 +64,7 @@ export async function installProjectionSourceAcl(sql: Sql): Promise { RETURN NEW; END; $$`) - for (const projection of PROJECTIONS) { + for (const projection of PROJECTION_SOURCE_ACL_TABLES) { await tx.unsafe(`CREATE OR REPLACE TRIGGER ${projection}_source_acl_set BEFORE INSERT OR UPDATE OF document_id, enabled ON ${projection} FOR EACH ROW EXECUTE FUNCTION set_projection_source_acl()`) @@ -63,84 +77,127 @@ export async function installProjectionSourceAcl(sql: Sql): Promise { }) } +export interface ProjectionSourceAclBackfillOptions { + /** Resume after this chunk id; the projection's first page otherwise. */ + afterId?: string + pageSize?: number + pauseMs?: number + /** Stop once this much time has passed and report where to resume; unbounded otherwise. */ + budgetMs?: number +} + +export interface ProjectionSourceAclBackfillProgress { + projection: ProjectionSourceAclTable + /** Unfilled chunks this run read, including any a concurrent write filled first. */ + scanned: number + written: number + /** The last chunk id this run reached; the next run resumes after it while `done` is false. */ + afterId: string + done: boolean +} + /** * Fills a projection's source and ACL for chunks written before the trigger existed, in - * independently committed keyset pages. Each page bounds its own locking and runtime and writes only - * rows still unset, so an interrupted run resumes by rerunning and a row the trigger has since - * written is left alone. The documents are share-locked before their values are copied, so a change - * in flight waits for the page and then fans its own values out; and a chunk that moved to another - * document meanwhile is left to that document's trigger, since the write requires the document the - * values were read from. + * independently committed keyset pages of unfilled rows. Each page is one statement that bounds + * its own locking and runtime and writes only rows still unset, so an interrupted run resumes by + * rerunning and a row the trigger has since written is left alone. The documents are share-locked + * before their values are copied, so a change in flight waits for the page and then fans its own + * values out; and a chunk that moved to another document meanwhile is left to that document's + * trigger, since the write requires the document the values were read from. + * + * On `embedding_search` every filled row is re-inserted into each HNSW index, which is the whole + * cost of a page and far more than a deploy can wait for; the run paces itself with a pause between + * pages and stops at its budget so a background task can chain runs until the projection is filled. + * Search does not wait: an unfilled row passes the on-row candidate predicate and is decided at + * hydration under the full document predicate. */ export async function backfillProjectionSourceAcl( sql: Sql, - projection: Projection -): Promise { + projection: ProjectionSourceAclTable, + options: ProjectionSourceAclBackfillOptions = {} +): Promise { + const pageSize = options.pageSize ?? PROJECTION_SOURCE_ACL_PAGE_SIZE + const pauseMs = options.pauseMs ?? PROJECTION_SOURCE_ACL_PAGE_PAUSE_MS const startedAt = Date.now() - let afterId = '' + const deadline = + options.budgetMs === undefined ? Number.POSITIVE_INFINITY : startedAt + options.budgetMs + let afterId = options.afterId ?? '' let scanned = 0 let written = 0 + let pages = 0 + let done = false for (;;) { const page = await sql.begin(async (tx) => { await tx.unsafe("SET LOCAL lock_timeout = '5s'") - await tx.unsafe("SET LOCAL statement_timeout = '60s'") - const rows = await tx.unsafe>( - `SELECT id FROM ${projection} WHERE id > $1 ORDER BY id LIMIT ${BATCH_SIZE}`, - [afterId] - ) - if (rows.length === 0) return null - const ids = rows.map((row) => row.id) - const [{ filled }] = await tx.unsafe>( + await tx.unsafe(`SET LOCAL statement_timeout = '${PAGE_STATEMENT_TIMEOUT}'`) + const [row] = await tx.unsafe< + Array<{ scanned: number; filled: number; last_id: string | null }> + >( `WITH page AS ( SELECT s.id, s.document_id, d.connector_id, d.acl FROM ${projection} s JOIN document d ON d.id = s.document_id - WHERE s.id = ANY($1::text[]) AND s.acl IS NULL + WHERE s.id > $1 AND s.acl IS NULL + ORDER BY s.id LIMIT ${pageSize} FOR SHARE OF d ), updated AS ( UPDATE ${projection} s SET connector_id = page.connector_id, acl = page.acl FROM page WHERE s.id = page.id AND s.document_id = page.document_id AND s.acl IS NULL RETURNING s.id - ) SELECT count(*)::int AS filled FROM updated`, - [ids] + ) + SELECT (SELECT count(*)::int FROM page) AS scanned, + (SELECT count(*)::int FROM updated) AS filled, + (SELECT max(id) FROM page) AS last_id`, + [afterId] ) - return { afterId: ids[ids.length - 1], scanned: ids.length, filled } + return row }) - if (!page) break - afterId = page.afterId + if (page.last_id === null) { + done = true + break + } + afterId = page.last_id scanned += page.scanned written += page.filled - if (scanned % (BATCH_SIZE * 100) === 0) { + pages += 1 + if (pages % PROGRESS_EVERY_PAGES === 0) { logger.info('Projection source and ACL backfill progress', { projection, scanned, written, + afterId, elapsedMs: Date.now() - startedAt, }) } + if (Date.now() >= deadline) break + if (pauseMs > 0) await sleep(pauseMs) } - logger.info('Projection source and ACL backfilled', { - projection, - scanned, - written, - elapsedMs: Date.now() - startedAt, - }) - return written + logger.info( + done ? 'Projection source and ACL backfilled' : 'Projection source and ACL backfill paused', + { + projection, + scanned, + written, + afterId, + elapsedMs: Date.now() - startedAt, + } + ) + return { projection, scanned, written, afterId, done } } /** * The indexes exact ranking of a readable set needs: the ACL index on each projection, and on the * vector projection the source index that lets the planner lead with a few sources when the - * caller's tokens alone would match most of the index. Built after the bulk load and concurrently, - * so the triggers keep writing. `CONCURRENTLY` cannot run in a transaction, and the pool's lock - * timeout would cancel a build that merely waits for a long transaction to finish. + * caller's tokens alone would match most of the index. Built concurrently, so the triggers and the + * backfill keep writing. `CONCURRENTLY` cannot run in a transaction, and the pool's lock timeout + * would cancel a build that merely waits for a long transaction to finish. */ export async function indexProjectionAcl(sql: Sql): Promise { const [{ timeout }] = await sql`SELECT current_setting('lock_timeout') AS timeout` await sql.unsafe('SET lock_timeout = 0') try { const builds: Array<[name: string, definition: string]> = [ - ...PROJECTIONS.map((projection): [string, string] => [ + ...PROJECTION_SOURCE_ACL_TABLES.map((projection): [string, string] => [ `${projection}_acl_gin_idx`, `ON ${projection} USING gin (acl) WHERE enabled`, ]), @@ -159,27 +216,44 @@ export async function indexProjectionAcl(sql: Sql): Promise { await sql.unsafe(`CREATE INDEX CONCURRENTLY IF NOT EXISTS ${name} ${definition}`) } /** The new columns carry no statistics until analyzed; the reach and source predicates plan on them. */ - for (const projection of PROJECTIONS) await sql.unsafe(`ANALYZE ${projection}`) + for (const projection of PROJECTION_SOURCE_ACL_TABLES) await sql.unsafe(`ANALYZE ${projection}`) } finally { await sql`SELECT set_config('lock_timeout', ${timeout}, false)` } } +/** + * Installs the triggers and builds the indexes; both are idempotent, so a run that was cut short + * completes on the next deploy. The columns are not filled here: on `embedding_search` every + * filled row is re-inserted into each HNSW index, which puts the full projection far beyond what a + * deploy job can wait for. The backfill runs afterwards from the `projection-source-acl-backfill` + * Trigger.dev task, started once after the deploy with + * `bun apps/sim/scripts/backfill-projection-source-acl.ts` (or a test run of the task from the + * Trigger.dev dashboard); it chains bounded runs until both projections are filled and is safe to + * start again at any time. Until it completes, an unfilled row passes the on-row candidate + * predicate and is decided at hydration under the full document predicate. + */ export const embeddingSearchConnectorMigration: ScriptMigration = { name: '0021_embedding_search_connector', async up(sql) { await installProjectionSourceAcl(sql) - for (const projection of PROJECTIONS) await backfillProjectionSourceAcl(sql, projection) await indexProjectionAcl(sql) }, } +/** + * Run directly — `db:push`, or a deployment without Trigger.dev — the projections are filled here + * as well, paced the same way, since there is no task to hand the backfill to. + */ if (import.meta.main) { const url = process.env.MIGRATION_DATABASE_URL ?? process.env.DATABASE_URL if (!url) throw new Error('DATABASE_URL is required to backfill the projection source and ACL') const sql = postgres(url, { max: 1, max_lifetime: null, onnotice: () => undefined }) try { await embeddingSearchConnectorMigration.up(sql) + for (const projection of PROJECTION_SOURCE_ACL_TABLES) { + await backfillProjectionSourceAcl(sql, projection) + } } finally { await sql.end() } From 8a64dd1cb83041963f4f7b4532490093ab079798 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 20 Sep 2026 10:19:15 -0700 Subject: [PATCH 2/6] fix(knowledge): decide unfilled projection rows on their document and start the backfill from the outbox An unfilled row no longer passes the on-row candidate predicate outright: it is decided on its document under the resolved candidate predicate, the join per candidate every row paid before the columns existed, so the bounded candidate pools and the exact slice hold only rows that hydration will keep. A partial index on the unfilled rows keeps that branch, and the backfill's keyset pages, an index probe. The migration also leaves one outbox event whose handler starts the backfill task, so it runs after every deploy without an operator. Claude-Session: https://claude.ai/code/session_01XU6c7pKRpa5CMoMHKDdqxX --- apps/sim/lib/core/outbox/processor.ts | 2 + .../access/predicate.postgres.test.ts | 16 +++-- .../lib/knowledge/access/predicate.test.ts | 15 ++-- apps/sim/lib/knowledge/access/predicate.ts | 22 ++++-- .../projection-source-acl-backfill.test.ts | 17 ++++- .../search/projection-source-acl-backfill.ts | 20 +++++- apps/sim/lib/knowledge/search/queries.test.ts | 5 +- apps/sim/lib/knowledge/search/queries.ts | 6 +- .../scripts/backfill-projection-source-acl.ts | 36 ++++++---- ...mbedding_search_connector.postgres.test.ts | 16 ++++- .../0021_embedding_search_connector.ts | 69 ++++++++++++++----- 11 files changed, 167 insertions(+), 57 deletions(-) diff --git a/apps/sim/lib/core/outbox/processor.ts b/apps/sim/lib/core/outbox/processor.ts index 41e625b118f..5183914d723 100644 --- a/apps/sim/lib/core/outbox/processor.ts +++ b/apps/sim/lib/core/outbox/processor.ts @@ -18,6 +18,7 @@ import { slackSearchOutboxHandlers } from '@/lib/knowledge/application/slack-sea import { getConnectorFailureDiagnostic } from '@/lib/knowledge/connectors/connector-error' import { knowledgeDocumentProcessingOutboxHandlers } from '@/lib/knowledge/documents/processing-outbox-handler' import { recoverKnowledgeDocumentProcessing } from '@/lib/knowledge/documents/processing-recovery' +import { projectionSourceAclBackfillOutboxHandlers } from '@/lib/knowledge/search/projection-source-acl-backfill' import { inboxCleanupOutboxHandlers } from '@/lib/mothership/inbox/cleanup-outbox' import { organizationResourceCleanupOutboxHandlers } from '@/lib/organizations/resource-cleanup' import { workspaceFileLiveDocOutboxHandlers } from '@/lib/uploads/contexts/workspace/workspace-file-live-doc-outbox' @@ -42,6 +43,7 @@ const handlers = { ...invitationMigrationOutboxHandlers, ...directGrantOutboxHandlers, ...knowledgeDocumentProcessingOutboxHandlers, + ...projectionSourceAclBackfillOutboxHandlers, ...organizationResourceCleanupOutboxHandlers, ...inboxCleanupOutboxHandlers, ...permissionAccessRequestOutboxHandlers, diff --git a/apps/sim/lib/knowledge/access/predicate.postgres.test.ts b/apps/sim/lib/knowledge/access/predicate.postgres.test.ts index 7d04a832913..11a1d8559df 100644 --- a/apps/sim/lib/knowledge/access/predicate.postgres.test.ts +++ b/apps/sim/lib/knowledge/access/predicate.postgres.test.ts @@ -623,19 +623,23 @@ describe.runIf(Boolean(databaseUrl))('knowledge ACLs in PostgreSQL', () => { expect(await admits(perRow, 'members-other')).toBe(false) expect(await onRowAdmits('members-other')).toBe(false) /** - * A chunk the backfill has not reached carries no source or ACL yet. It is ranked, and decided - * at hydration under the full predicate, exactly as every candidate was before the columns - * existed — search must not depend on the backfill, and must not lose a document to it. + * A chunk the backfill has not reached carries no source or ACL yet and is decided on its + * document, as every candidate was before the columns existed: search must not depend on the + * backfill, must not lose a document to it, and must not rank an unreadable one because of it. */ await connection.unsafe( "INSERT INTO document(id, connector_id, acl, acl_verified_at) VALUES ('unfilled-other', 'members', ARRAY['s:slack:-:bob'], statement_timestamp())" ) + await connection.unsafe(`INSERT INTO embedding_search(id, document_id) VALUES + ('unfilled-other-chunk', 'unfilled-other'), ('unfilled-admin-chunk', 'admin-current'), + ('unfilled-members-chunk', 'members-current'), ('unfilled-gone-chunk', 'deleted-connector')`) await connection.unsafe( - "INSERT INTO embedding_search(id, document_id) VALUES ('unfilled-other-chunk', 'unfilled-other'), ('admin-current-unfilled', 'admin-current')" + "DELETE FROM embedding_search WHERE id IN ('admin-current-chunk', 'members-current-chunk', 'deleted-connector-chunk')" ) - expect(await onRowAdmits('unfilled-other')).toBe(true) - expect(await admits(perRow, 'unfilled-other')).toBe(false) + expect(await onRowAdmits('unfilled-other')).toBe(false) expect(await onRowAdmits('admin-current')).toBe(true) + expect(await onRowAdmits('members-current')).toBe(true) + expect(await onRowAdmits('deleted-connector')).toBe(false) /** * Candidate ranking defers the live source proof, as the per-row candidate predicate does: a * caller holds those grants only after authorization, so applying the clause during ranking diff --git a/apps/sim/lib/knowledge/access/predicate.test.ts b/apps/sim/lib/knowledge/access/predicate.test.ts index 8de16218f92..4e2e7327e71 100644 --- a/apps/sim/lib/knowledge/access/predicate.test.ts +++ b/apps/sim/lib/knowledge/access/predicate.test.ts @@ -34,7 +34,7 @@ describe('projectionCandidateAccessCondition', () => { memberSources: [], } - it('admits a row the backfill has not filled, and decides a filled row on its mirrored columns', () => { + it('decides a filled row on its mirrored columns and an unfilled row on its document', () => { const { sql, params } = render( projectionCandidateAccessCondition( embeddingSearch, @@ -42,11 +42,16 @@ describe('projectionCandidateAccessCondition', () => { plan ) ) - expect(sql).toBe( - '("embedding_search"."acl" IS NULL OR ("embedding_search"."acl" && ARRAY[$1, $2]::text[]\n' + - ' AND ("embedding_search"."connector_id" IS NULL OR "embedding_search"."connector_id" = ANY(ARRAY[$3]::text[]))))' + expect(sql).toContain( + '("embedding_search"."acl" IS NULL AND EXISTS (\n SELECT 1 FROM "document"\n WHERE "document"."id" = "embedding_search"."document_id"\n AND (' ) - expect(params).toEqual(['ws', 'u:alice', 'ws-src']) + expect(sql).toContain('"document"."acl" && ARRAY[$1, $2]::text[]') + expect(sql).toMatch( + /OR \("embedding_search"\."acl" && ARRAY\[\$\d+, \$\d+\]::text\[\]\n {4}AND \("embedding_search"\."connector_id" IS NULL OR "embedding_search"\."connector_id" = ANY\(ARRAY\[\$\d+\]::text\[\]\)\)\)\)$/ + ) + expect(params.slice(0, 2)).toEqual(['ws', 'u:alice']) + expect(params.slice(-3)).toEqual(['ws', 'u:alice', 'ws-src']) + for (const param of params) expect(Array.isArray(param)).toBe(false) }) it('still denies everything for an empty token set', () => { diff --git a/apps/sim/lib/knowledge/access/predicate.ts b/apps/sim/lib/knowledge/access/predicate.ts index 203d47f93a3..9c8cd680c32 100644 --- a/apps/sim/lib/knowledge/access/predicate.ts +++ b/apps/sim/lib/knowledge/access/predicate.ts @@ -313,13 +313,18 @@ export function knowledgeCandidateAccessConditionForConnectors( * under the full predicate, before content is returned — this predicate only decides what is worth * ranking. * - * A row the backfill has not reached yet carries no ACL (`acl IS NULL`) and passes: it is decided at - * hydration under the full document predicate, exactly as every candidate was before the columns - * existed. The backfill runs in the background, so search never waits on it and never loses a - * document to it. + * A row the backfill has not reached yet carries no ACL (`acl IS NULL`) and is decided on its + * document instead, under {@link knowledgeCandidateAccessConditionForConnectors} — the join per + * candidate that every row paid before the columns existed. The backfill runs in the background, + * so search never waits on it, never loses a document to it, and never ranks an unreadable one + * into a bounded candidate pool because of it. */ export function projectionCandidateAccessCondition( - projection: { connectorId: AnyPgColumn | SQL; acl: AnyPgColumn | SQL }, + projection: { + connectorId: AnyPgColumn | SQL + acl: AnyPgColumn | SQL + documentId: AnyPgColumn | SQL + }, scope: KnowledgeAccessScope | SystemAccessScope, plan: SearchAccessPlan ): SQL { @@ -335,7 +340,12 @@ export function projectionCandidateAccessCondition( ...plan.connectors.admin, ...plan.connectors.members, ] - return sql`(${projection.acl} IS NULL OR (${projection.acl} && ${tokens} + const unfilled = sql`(${projection.acl} IS NULL AND EXISTS ( + SELECT 1 FROM ${document} + WHERE ${document.id} = ${projection.documentId} + AND ${knowledgeCandidateAccessConditionForConnectors(scope, plan)} + ))` + return sql`(${unfilled} OR (${projection.acl} && ${tokens} AND (${projection.connectorId} IS NULL OR ${inSources(mirrored)})))` } diff --git a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts index 4277086242b..512304b2e2e 100644 --- a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts +++ b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts @@ -13,14 +13,21 @@ const { mockBackfill, mockEnd, mockPostgres, mockTasksTrigger } = vi.hoisted(() vi.mock('@sim/db', () => ({ resolveDbUrl: () => 'postgres://localhost:5432/sim' })) vi.mock('@sim/db/script-migrations/0021_embedding_search_connector', () => ({ PROJECTION_SOURCE_ACL_TABLES: ['embedding_search', 'embedding_keyword_tin'], + PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT: 'knowledge.projection.source_acl.backfill', backfillProjectionSourceAcl: mockBackfill, })) vi.mock('postgres', () => ({ default: mockPostgres })) vi.mock('@trigger.dev/sdk', () => ({ tasks: { trigger: mockTasksTrigger } })) vi.mock('@/lib/core/async-jobs/region', () => ({ resolveTriggerRegion: async () => 'us-east-1' })) +vi.mock('@/lib/core/utils/background', () => ({ + runDetached: (_label: string, work: () => Promise) => { + void work() + }, +})) import { enqueueProjectionSourceAclBackfill, + projectionSourceAclBackfillOutboxHandlers, runProjectionSourceAclBackfill, } from '@/lib/knowledge/search/projection-source-acl-backfill' @@ -110,9 +117,15 @@ describe('enqueueProjectionSourceAclBackfill', () => { expect(mockBackfill).not.toHaveBeenCalled() }) - it('fills the projections inline without one', async () => { + it('fills the projections detached in this process without one', async () => { await expect(enqueueProjectionSourceAclBackfill({}, false)).resolves.toBeNull() expect(mockTasksTrigger).not.toHaveBeenCalled() - expect(mockBackfill).toHaveBeenCalledTimes(2) + await vi.waitFor(() => expect(mockBackfill).toHaveBeenCalledTimes(2)) + }) + + it('starts from the outbox event the migration leaves behind', () => { + expect(Object.keys(projectionSourceAclBackfillOutboxHandlers)).toEqual([ + 'knowledge.projection.source_acl.backfill', + ]) }) }) diff --git a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts index ee1f07529a0..d09b121556a 100644 --- a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts +++ b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts @@ -1,6 +1,7 @@ import { resolveDbUrl } from '@sim/db' import { backfillProjectionSourceAcl, + PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT, PROJECTION_SOURCE_ACL_TABLES, type ProjectionSourceAclTable, } from '@sim/db/script-migrations/0021_embedding_search_connector' @@ -9,6 +10,8 @@ import postgres from 'postgres' import { resolveTriggerRegion } from '@/lib/core/async-jobs/region' import { env } from '@/lib/core/config/env' import { isTriggerDevEnabled } from '@/lib/core/config/env-flags' +import type { OutboxHandlerRegistry } from '@/lib/core/outbox/service' +import { runDetached } from '@/lib/core/utils/background' const logger = createLogger('ProjectionSourceAclBackfill') @@ -76,8 +79,9 @@ export async function runProjectionSourceAclBackfill( /** * Starts the backfill the way the table backfill is started: on the deployment's Trigger.dev - * worker when there is one, where bounded runs chain until both projections are filled, and inline - * otherwise. Safe to call again at any time — a run only fills rows still unset. + * worker when there is one, where bounded runs chain until both projections are filled, and + * detached in this process otherwise. Safe to call again at any time — a run only fills rows still + * unset. */ export async function enqueueProjectionSourceAclBackfill( payload: ProjectionSourceAclBackfillPayload = {}, @@ -91,6 +95,16 @@ export async function enqueueProjectionSourceAclBackfill( logger.info('Projection source and ACL backfill enqueued', { runId: handle.id }) return { runId: handle.id } } - await runProjectionSourceAclBackfill(payload) + runDetached(PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, () => runProjectionSourceAclBackfill(payload)) return null } + +/** + * The event script migration `0021_embedding_search_connector` leaves behind, so the backfill it + * hands off starts once the app that ships this handler is up, on every deployment. + */ +export const projectionSourceAclBackfillOutboxHandlers: OutboxHandlerRegistry = { + [PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT]: async () => { + await enqueueProjectionSourceAclBackfill() + }, +} diff --git a/apps/sim/lib/knowledge/search/queries.test.ts b/apps/sim/lib/knowledge/search/queries.test.ts index f76f18bc09e..76ca0005a6a 100644 --- a/apps/sim/lib/knowledge/search/queries.test.ts +++ b/apps/sim/lib/knowledge/search/queries.test.ts @@ -1527,8 +1527,9 @@ describe('permitted-document planner', () => { /** The ranked CTE carries the mirrored source and ACL the predicate tests. */ expect(statement).toContain('AS connector_id') expect(statement).toContain('ranked_tin_chunks.acl') - /** A row the backfill has not filled (`acl IS NULL`) is ranked and decided at hydration. */ - expect(statement).toContain('" IS NULL OR ("') + /** A row the backfill has not filled (`acl IS NULL`) is decided on its document instead. */ + expect(statement).toContain('IS NULL AND EXISTS (') + expect(statement).toContain('ranked_tin_chunks.document_id') }) it('widens the window for a broad resolved scope whose first page came back short', async () => { diff --git a/apps/sim/lib/knowledge/search/queries.ts b/apps/sim/lib/knowledge/search/queries.ts index ac7174e6082..abb9aa76113 100644 --- a/apps/sim/lib/knowledge/search/queries.ts +++ b/apps/sim/lib/knowledge/search/queries.ts @@ -1819,7 +1819,11 @@ export async function executeKeywordSearch(params: KeywordSearchParams): Promise const onRowKeywordVisibility = (excludedSources: readonly string[]) => and( projectionCandidateAccessCondition( - { connectorId: sql`ranked_tin_chunks.connector_id`, acl: sql`ranked_tin_chunks.acl` }, + { + connectorId: sql`ranked_tin_chunks.connector_id`, + acl: sql`ranked_tin_chunks.acl`, + documentId: sql`ranked_tin_chunks.document_id`, + }, access, accessPlan! ), diff --git a/apps/sim/scripts/backfill-projection-source-acl.ts b/apps/sim/scripts/backfill-projection-source-acl.ts index 7df5b91fdc6..9e8179d49b4 100644 --- a/apps/sim/scripts/backfill-projection-source-acl.ts +++ b/apps/sim/scripts/backfill-projection-source-acl.ts @@ -2,10 +2,11 @@ /** * Starts the projection source and ACL backfill that script migration - * `0021_embedding_search_connector` leaves to the background: on a deployment with Trigger.dev it - * enqueues the `projection-source-acl-backfill` task, which chains bounded runs until both ranking - * projections are filled; without one it fills them here, paced the same way. Safe to run again at - * any time — a run only fills rows still unset. + * `0021_embedding_search_connector` leaves to the background, by hand. The migration's outbox + * event already starts it after a deploy; this is for starting it again — on a deployment with + * Trigger.dev it enqueues the `projection-source-acl-backfill` task, which chains bounded runs + * until both ranking projections are filled; without one it fills them here, paced the same way. + * Safe to run at any time — a run only fills rows still unset. * * Usage: * bun apps/sim/scripts/backfill-projection-source-acl.ts @@ -13,19 +14,28 @@ import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' -import { enqueueProjectionSourceAclBackfill } from '@/lib/knowledge/search/projection-source-acl-backfill' +import { env } from '@/lib/core/config/env' +import { isTriggerDevEnabled } from '@/lib/core/config/env-flags' +import { + enqueueProjectionSourceAclBackfill, + runProjectionSourceAclBackfill, +} from '@/lib/knowledge/search/projection-source-acl-backfill' const logger = createLogger('BackfillProjectionSourceAcl') +async function main(): Promise { + if (isTriggerDevEnabled && env.TRIGGER_SECRET_KEY) { + const handle = await enqueueProjectionSourceAclBackfill({}, true) + logger.info('Backfill enqueued on the Trigger.dev worker', handle ?? {}) + return + } + await runProjectionSourceAclBackfill({}) + logger.info('Backfill complete') +} + if (import.meta.main) { - enqueueProjectionSourceAclBackfill().then( - (handle) => { - logger.info( - handle ? 'Backfill enqueued on the Trigger.dev worker' : 'Backfill complete', - handle ?? {} - ) - process.exit(0) - }, + main().then( + () => process.exit(0), (error) => { logger.error('Backfill failed', toError(error)) process.exit(1) diff --git a/packages/db/script-migrations/0021_embedding_search_connector.postgres.test.ts b/packages/db/script-migrations/0021_embedding_search_connector.postgres.test.ts index f0ab90c3a7c..c7364f2aee8 100644 --- a/packages/db/script-migrations/0021_embedding_search_connector.postgres.test.ts +++ b/packages/db/script-migrations/0021_embedding_search_connector.postgres.test.ts @@ -40,6 +40,10 @@ describe.runIf(Boolean(databaseUrl))('projection source and ACL backfill in Post await sql`CREATE TABLE document ( id text PRIMARY KEY, connector_id text, acl text[] NOT NULL DEFAULT '{ws}' )` + await sql`CREATE TABLE outbox_event ( + id text PRIMARY KEY, event_type text NOT NULL, payload json NOT NULL, + status text NOT NULL DEFAULT 'pending' + )` for (const projection of ['embedding_search', 'embedding_keyword_tin']) { await sql`CREATE TABLE ${sql(projection)} ( id text PRIMARY KEY, document_id text NOT NULL, enabled boolean NOT NULL DEFAULT true, @@ -61,17 +65,27 @@ describe.runIf(Boolean(databaseUrl))('projection source and ACL backfill in Post await sql`ALTER TABLE embedding_keyword_tin DISABLE TRIGGER embedding_keyword_tin_source_acl_set` }) - it('installs its triggers and indexes again without failing, so a cut-short deploy completes', async () => { + it('installs its triggers, indexes and start event again without failing, so a cut-short deploy completes', async () => { await expect(embeddingSearchConnectorMigration.up(sql)).resolves.toBeUndefined() const indexes = await sql<{ indexname: string }[]>` SELECT indexname FROM pg_indexes WHERE schemaname = ${schemaName} ORDER BY indexname` expect(indexes.map((row) => row.indexname)).toEqual( expect.arrayContaining([ 'embedding_keyword_tin_acl_gin_idx', + 'embedding_keyword_tin_acl_unfilled_idx', 'embedding_search_acl_gin_idx', + 'embedding_search_acl_unfilled_idx', 'embedding_search_source_idx', ]) ) + /** Two runs of the migration leave the app one event, so the backfill is started once. */ + expect(await sql`SELECT id, event_type, status FROM outbox_event`).toEqual([ + { + id: 'projection-source-acl-backfill:0021', + event_type: 'knowledge.projection.source_acl.backfill', + status: 'pending', + }, + ]) }) it('fills only the rows still unset, in pages, and leaves a chunk that changed documents to its trigger', async () => { diff --git a/packages/db/script-migrations/0021_embedding_search_connector.ts b/packages/db/script-migrations/0021_embedding_search_connector.ts index cc004a81e4d..96b6353f29e 100644 --- a/packages/db/script-migrations/0021_embedding_search_connector.ts +++ b/packages/db/script-migrations/0021_embedding_search_connector.ts @@ -21,6 +21,15 @@ const PAGE_STATEMENT_TIMEOUT = '60s' /** Pages between progress log lines. */ const PROGRESS_EVERY_PAGES = 100 +/** + * The outbox event the migration leaves for the app, whose handler starts the backfill on the + * deployment's worker. Written once, under a fixed id, so a rerun of the migration does not start + * it twice. + */ +export const PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT = + 'knowledge.projection.source_acl.backfill' +const PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_ID = 'projection-source-acl-backfill:0021' + /** The projections that carry their document's source and ACL. */ export const PROJECTION_SOURCE_ACL_TABLES = ['embedding_search', 'embedding_keyword_tin'] as const export type ProjectionSourceAclTable = (typeof PROJECTION_SOURCE_ACL_TABLES)[number] @@ -108,8 +117,8 @@ export interface ProjectionSourceAclBackfillProgress { * On `embedding_search` every filled row is re-inserted into each HNSW index, which is the whole * cost of a page and far more than a deploy can wait for; the run paces itself with a pause between * pages and stops at its budget so a background task can chain runs until the projection is filled. - * Search does not wait: an unfilled row passes the on-row candidate predicate and is decided at - * hydration under the full document predicate. + * Search does not wait: an unfilled row is decided on its document by the on-row candidate + * predicate, the join per candidate every row paid before the columns existed. */ export async function backfillProjectionSourceAcl( sql: Sql, @@ -154,6 +163,8 @@ export async function backfillProjectionSourceAcl( }) if (page.last_id === null) { done = true + /** The planner last saw every row unfilled; it should see the finished projection. */ + await sql.unsafe(`ANALYZE ${projection}`) break } afterId = page.last_id @@ -191,16 +202,23 @@ export async function backfillProjectionSourceAcl( * caller's tokens alone would match most of the index. Built concurrently, so the triggers and the * backfill keep writing. `CONCURRENTLY` cannot run in a transaction, and the pool's lock timeout * would cancel a build that merely waits for a long transaction to finish. + * + * The unfilled index on each projection lists the rows the backfill has not reached: each page + * reads its rows from it instead of walking past every filled one, and the on-row predicate's + * unfilled branch, an `OR` beside the ACL overlap, stays an index probe for the planner — once the + * projection is filled, a probe of an empty index. */ export async function indexProjectionAcl(sql: Sql): Promise { const [{ timeout }] = await sql`SELECT current_setting('lock_timeout') AS timeout` await sql.unsafe('SET lock_timeout = 0') try { const builds: Array<[name: string, definition: string]> = [ - ...PROJECTION_SOURCE_ACL_TABLES.map((projection): [string, string] => [ - `${projection}_acl_gin_idx`, - `ON ${projection} USING gin (acl) WHERE enabled`, - ]), + ...PROJECTION_SOURCE_ACL_TABLES.flatMap( + (projection): Array<[string, string]> => [ + [`${projection}_acl_gin_idx`, `ON ${projection} USING gin (acl) WHERE enabled`], + [`${projection}_acl_unfilled_idx`, `ON ${projection} (id) WHERE acl IS NULL`], + ] + ), ['embedding_search_source_idx', 'ON embedding_search (connector_id) WHERE enabled'], ] for (const [name, definition] of builds) { @@ -223,34 +241,49 @@ export async function indexProjectionAcl(sql: Sql): Promise { } /** - * Installs the triggers and builds the indexes; both are idempotent, so a run that was cut short - * completes on the next deploy. The columns are not filled here: on `embedding_search` every - * filled row is re-inserted into each HNSW index, which puts the full projection far beyond what a - * deploy job can wait for. The backfill runs afterwards from the `projection-source-acl-backfill` - * Trigger.dev task, started once after the deploy with - * `bun apps/sim/scripts/backfill-projection-source-acl.ts` (or a test run of the task from the - * Trigger.dev dashboard); it chains bounded runs until both projections are filled and is safe to - * start again at any time. Until it completes, an unfilled row passes the on-row candidate - * predicate and is decided at hydration under the full document predicate. + * Leaves the app one outbox event to start the backfill from. The outbox processor runs on every + * deployment, so the backfill starts on its own once the app that ships the handler is up — + * on the Trigger.dev worker where there is one, detached in the app otherwise — without an + * operator remembering to. Idempotent under its fixed id. + */ +export async function enqueueProjectionSourceAclBackfillEvent(sql: Sql): Promise { + await sql` + INSERT INTO outbox_event (id, event_type, payload) + VALUES (${PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_ID}, ${PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT}, '{}'::json) + ON CONFLICT (id) DO NOTHING` +} + +/** + * Installs the triggers, builds the indexes and leaves the outbox event that starts the backfill; + * all three are idempotent, so a run that was cut short completes on the next deploy. The columns + * are not filled here: on `embedding_search` every filled row is re-inserted into each HNSW index, + * which puts the full projection far beyond what a deploy job can wait for. The backfill runs + * afterwards from the `projection-source-acl-backfill` Trigger.dev task, which the outbox handler + * starts and which chains bounded runs until both projections are filled. It is safe to start again + * at any time, with `bun apps/sim/scripts/backfill-projection-source-acl.ts` or a test run of the + * task from the Trigger.dev dashboard. Until it completes, an unfilled row is decided on its + * document, the join per candidate every row paid before the columns existed. */ export const embeddingSearchConnectorMigration: ScriptMigration = { name: '0021_embedding_search_connector', async up(sql) { await installProjectionSourceAcl(sql) await indexProjectionAcl(sql) + await enqueueProjectionSourceAclBackfillEvent(sql) }, } /** - * Run directly — `db:push`, or a deployment without Trigger.dev — the projections are filled here - * as well, paced the same way, since there is no task to hand the backfill to. + * Run directly — `db:push`, or an operator filling a database by hand — the projections are filled + * here, paced the same way, rather than left to the app. */ if (import.meta.main) { const url = process.env.MIGRATION_DATABASE_URL ?? process.env.DATABASE_URL if (!url) throw new Error('DATABASE_URL is required to backfill the projection source and ACL') const sql = postgres(url, { max: 1, max_lifetime: null, onnotice: () => undefined }) try { - await embeddingSearchConnectorMigration.up(sql) + await installProjectionSourceAcl(sql) + await indexProjectionAcl(sql) for (const projection of PROJECTION_SOURCE_ACL_TABLES) { await backfillProjectionSourceAcl(sql, projection) } From ecbda392048c5e43920751f33a409f37f1bcda46 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 20 Sep 2026 10:28:32 -0700 Subject: [PATCH 3/6] fix(knowledge): keep the outbox event pending until a deployment without a worker fills the projections Without a Trigger.dev worker the outbox handler no longer detaches the backfill and completes the event; it runs one bounded slice per outbox run and yields with continueOutboxHandler until both projections are filled, so a restart loses at most one slice and the event is never marked done ahead of the work. The index builds run on one reserved connection, so the session-scoped lock timeout covers every build and its reset. Claude-Session: https://claude.ai/code/session_01XU6c7pKRpa5CMoMHKDdqxX --- apps/sim/lib/core/outbox/processor.test.ts | 3 + .../projection-source-acl-backfill.test.ts | 63 ++++++++++++---- .../search/projection-source-acl-backfill.ts | 73 +++++++++++++------ .../scripts/backfill-projection-source-acl.ts | 9 +-- .../0021_embedding_search_connector.ts | 12 ++- 5 files changed, 113 insertions(+), 47 deletions(-) diff --git a/apps/sim/lib/core/outbox/processor.test.ts b/apps/sim/lib/core/outbox/processor.test.ts index 6ad93cc80fa..d678e5b461a 100644 --- a/apps/sim/lib/core/outbox/processor.test.ts +++ b/apps/sim/lib/core/outbox/processor.test.ts @@ -33,6 +33,9 @@ vi.mock('@/lib/knowledge/application/slack-search/outbox', () => ({ vi.mock('@/lib/knowledge/documents/processing-outbox-handler', () => ({ knowledgeDocumentProcessingOutboxHandlers: {}, })) +vi.mock('@/lib/knowledge/search/projection-source-acl-backfill', () => ({ + projectionSourceAclBackfillOutboxHandlers: {}, +})) vi.mock('@/lib/mothership/inbox/cleanup-outbox', () => ({ inboxCleanupOutboxHandlers: {} })) vi.mock('@/lib/organizations/resource-cleanup', () => ({ organizationResourceCleanupOutboxHandlers: {}, diff --git a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts index 512304b2e2e..527d5f5a953 100644 --- a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts +++ b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts @@ -3,11 +3,12 @@ */ import { beforeEach, describe, expect, it, vi } from 'vitest' -const { mockBackfill, mockEnd, mockPostgres, mockTasksTrigger } = vi.hoisted(() => ({ +const { mockBackfill, mockEnd, mockPostgres, mockTasksTrigger, envState } = vi.hoisted(() => ({ mockBackfill: vi.fn(), mockEnd: vi.fn(async () => undefined), mockPostgres: vi.fn(), mockTasksTrigger: vi.fn(async () => ({ id: 'run-1' })), + envState: { triggerEnabled: false, secret: undefined as string | undefined }, })) vi.mock('@sim/db', () => ({ resolveDbUrl: () => 'postgres://localhost:5432/sim' })) @@ -19,11 +20,27 @@ vi.mock('@sim/db/script-migrations/0021_embedding_search_connector', () => ({ vi.mock('postgres', () => ({ default: mockPostgres })) vi.mock('@trigger.dev/sdk', () => ({ tasks: { trigger: mockTasksTrigger } })) vi.mock('@/lib/core/async-jobs/region', () => ({ resolveTriggerRegion: async () => 'us-east-1' })) -vi.mock('@/lib/core/utils/background', () => ({ - runDetached: (_label: string, work: () => Promise) => { - void work() +vi.mock('@/lib/core/config/env', () => ({ + env: { + get TRIGGER_SECRET_KEY() { + return envState.secret + }, }, })) +vi.mock('@/lib/core/config/env-flags', () => ({ + get isTriggerDevEnabled() { + return envState.triggerEnabled + }, +})) +vi.mock('@/lib/core/outbox/service', () => ({ + continueOutboxHandler: (reason: string) => ({ + outcome: 'deferred', + reason, + consumeAttempt: false, + }), + withOutboxHandlerTimeout: (handler: unknown, timeoutMs: number) => + Object.assign(handler as object, { timeoutMs }), +})) import { enqueueProjectionSourceAclBackfill, @@ -32,6 +49,9 @@ import { } from '@/lib/knowledge/search/projection-source-acl-backfill' const connection = { end: mockEnd } +const handler = + projectionSourceAclBackfillOutboxHandlers['knowledge.projection.source_acl.backfill'] +const context = { eventId: 'e', eventType: 'knowledge.projection.source_acl.backfill' } describe('runProjectionSourceAclBackfill', () => { beforeEach(() => { @@ -92,7 +112,7 @@ describe('runProjectionSourceAclBackfill', () => { }) }) -describe('enqueueProjectionSourceAclBackfill', () => { +describe('the outbox event the migration leaves behind', () => { beforeEach(() => { vi.clearAllMocks() mockPostgres.mockReturnValue(connection) @@ -106,7 +126,9 @@ describe('enqueueProjectionSourceAclBackfill', () => { }) it('hands the backfill to the Trigger.dev worker when there is one', async () => { - await expect(enqueueProjectionSourceAclBackfill({ pageSize: 25 }, true)).resolves.toEqual({ + envState.triggerEnabled = true + envState.secret = 'tr_secret' + await expect(enqueueProjectionSourceAclBackfill({ pageSize: 25 })).resolves.toEqual({ runId: 'run-1', }) expect(mockTasksTrigger).toHaveBeenCalledWith( @@ -114,18 +136,29 @@ describe('enqueueProjectionSourceAclBackfill', () => { { pageSize: 25 }, { region: 'us-east-1' } ) + await expect(handler({}, context as never)).resolves.toBeUndefined() + expect(mockTasksTrigger).toHaveBeenCalledTimes(2) expect(mockBackfill).not.toHaveBeenCalled() }) - it('fills the projections detached in this process without one', async () => { - await expect(enqueueProjectionSourceAclBackfill({}, false)).resolves.toBeNull() + it('runs a bounded slice per outbox run without one, and stays pending until it is filled', async () => { + envState.triggerEnabled = false + envState.secret = undefined + mockBackfill.mockResolvedValueOnce({ + projection: 'embedding_search', + scanned: 100, + written: 100, + afterId: 'chunk-100', + done: false, + }) + await expect(handler({}, context as never)).resolves.toMatchObject({ + outcome: 'deferred', + consumeAttempt: false, + }) expect(mockTasksTrigger).not.toHaveBeenCalled() - await vi.waitFor(() => expect(mockBackfill).toHaveBeenCalledTimes(2)) - }) - - it('starts from the outbox event the migration leaves behind', () => { - expect(Object.keys(projectionSourceAclBackfillOutboxHandlers)).toEqual([ - 'knowledge.projection.source_acl.backfill', - ]) + expect(mockBackfill).toHaveBeenCalledTimes(1) + expect(mockBackfill.mock.calls[0][2].budgetMs).toBeLessThanOrEqual(handler.timeoutMs!) + await expect(handler({}, context as never)).resolves.toBeUndefined() + expect(mockBackfill).toHaveBeenCalledTimes(3) }) }) diff --git a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts index d09b121556a..55c2ef87214 100644 --- a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts +++ b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts @@ -10,8 +10,13 @@ import postgres from 'postgres' import { resolveTriggerRegion } from '@/lib/core/async-jobs/region' import { env } from '@/lib/core/config/env' import { isTriggerDevEnabled } from '@/lib/core/config/env-flags' -import type { OutboxHandlerRegistry } from '@/lib/core/outbox/service' -import { runDetached } from '@/lib/core/utils/background' +import { + continueOutboxHandler, + type DeferredOutboxHandlerResult, + type OutboxEventContext, + type OutboxHandlerRegistry, + withOutboxHandlerTimeout, +} from '@/lib/core/outbox/service' const logger = createLogger('ProjectionSourceAclBackfill') @@ -77,34 +82,56 @@ export async function runProjectionSourceAclBackfill( } } +/** Whether the deployment has a Trigger.dev worker to hand the backfill to. */ +export function projectionSourceAclBackfillUsesTrigger(): boolean { + return Boolean(isTriggerDevEnabled && env.TRIGGER_SECRET_KEY) +} + /** * Starts the backfill the way the table backfill is started: on the deployment's Trigger.dev - * worker when there is one, where bounded runs chain until both projections are filled, and - * detached in this process otherwise. Safe to call again at any time — a run only fills rows still - * unset. + * worker, where bounded runs chain until both projections are filled. Safe to call again at any + * time — a run only fills rows still unset. */ export async function enqueueProjectionSourceAclBackfill( - payload: ProjectionSourceAclBackfillPayload = {}, - useTrigger = Boolean(isTriggerDevEnabled && env.TRIGGER_SECRET_KEY) -): Promise<{ runId: string } | null> { - if (useTrigger) { - const { tasks } = await import('@trigger.dev/sdk') - const handle = await tasks.trigger(PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, payload, { - region: await resolveTriggerRegion(), - }) - logger.info('Projection source and ACL backfill enqueued', { runId: handle.id }) - return { runId: handle.id } - } - runDetached(PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, () => runProjectionSourceAclBackfill(payload)) - return null + payload: ProjectionSourceAclBackfillPayload = {} +): Promise<{ runId: string }> { + const { tasks } = await import('@trigger.dev/sdk') + const handle = await tasks.trigger(PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, payload, { + region: await resolveTriggerRegion(), + }) + logger.info('Projection source and ACL backfill enqueued', { runId: handle.id }) + return { runId: handle.id } } +/** One outbox run's share of the backfill on a deployment without a worker, inside its window. */ +const OUTBOX_RUN_BUDGET_MS = 4 * 60 * 1000 +const OUTBOX_HANDLER_TIMEOUT_MS = 5 * 60 * 1000 + /** - * The event script migration `0021_embedding_search_connector` leaves behind, so the backfill it - * hands off starts once the app that ships this handler is up, on every deployment. + * Handles the event script migration `0021_embedding_search_connector` leaves behind, so the + * backfill starts once the app that ships this handler is up, on every deployment. With a + * Trigger.dev worker the event is done once the task is enqueued: the task owns its own retries + * and continuations. Without one the backfill runs here, one bounded slice per outbox run, and the + * event stays pending until both projections are filled — every slice's writes are durable, and a + * slice that starts over after a restart skips the filled rows through the unfilled index, so an + * interrupted deployment loses nothing but the time of one slice. */ export const projectionSourceAclBackfillOutboxHandlers: OutboxHandlerRegistry = { - [PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT]: async () => { - await enqueueProjectionSourceAclBackfill() - }, + [PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT]: withOutboxHandlerTimeout( + async ( + _payload: unknown, + _context: OutboxEventContext + ): Promise => { + if (projectionSourceAclBackfillUsesTrigger()) { + await enqueueProjectionSourceAclBackfill() + return undefined + } + const cursor = await runProjectionSourceAclBackfill({}, { budgetMs: OUTBOX_RUN_BUDGET_MS }) + if (!cursor) return undefined + return continueOutboxHandler( + `projection source and ACL backfill paused at ${cursor.projection} after ${cursor.afterId}` + ) + }, + OUTBOX_HANDLER_TIMEOUT_MS + ), } diff --git a/apps/sim/scripts/backfill-projection-source-acl.ts b/apps/sim/scripts/backfill-projection-source-acl.ts index 9e8179d49b4..804c37d0968 100644 --- a/apps/sim/scripts/backfill-projection-source-acl.ts +++ b/apps/sim/scripts/backfill-projection-source-acl.ts @@ -14,19 +14,18 @@ import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' -import { env } from '@/lib/core/config/env' -import { isTriggerDevEnabled } from '@/lib/core/config/env-flags' import { enqueueProjectionSourceAclBackfill, + projectionSourceAclBackfillUsesTrigger, runProjectionSourceAclBackfill, } from '@/lib/knowledge/search/projection-source-acl-backfill' const logger = createLogger('BackfillProjectionSourceAcl') async function main(): Promise { - if (isTriggerDevEnabled && env.TRIGGER_SECRET_KEY) { - const handle = await enqueueProjectionSourceAclBackfill({}, true) - logger.info('Backfill enqueued on the Trigger.dev worker', handle ?? {}) + if (projectionSourceAclBackfillUsesTrigger()) { + const handle = await enqueueProjectionSourceAclBackfill() + logger.info('Backfill enqueued on the Trigger.dev worker', handle) return } await runProjectionSourceAclBackfill({}) diff --git a/packages/db/script-migrations/0021_embedding_search_connector.ts b/packages/db/script-migrations/0021_embedding_search_connector.ts index 96b6353f29e..7e80b5df02a 100644 --- a/packages/db/script-migrations/0021_embedding_search_connector.ts +++ b/packages/db/script-migrations/0021_embedding_search_connector.ts @@ -201,14 +201,17 @@ export async function backfillProjectionSourceAcl( * vector projection the source index that lets the planner lead with a few sources when the * caller's tokens alone would match most of the index. Built concurrently, so the triggers and the * backfill keep writing. `CONCURRENTLY` cannot run in a transaction, and the pool's lock timeout - * would cancel a build that merely waits for a long transaction to finish. + * would cancel a build that merely waits for a long transaction to finish. The timeout is a + * session setting, so one connection is reserved for it, the builds, and the reset — a pool + * would otherwise hand the builds to connections that never saw the setting. * * The unfilled index on each projection lists the rows the backfill has not reached: each page * reads its rows from it instead of walking past every filled one, and the on-row predicate's * unfilled branch, an `OR` beside the ACL overlap, stays an index probe for the planner — once the * projection is filled, a probe of an empty index. */ -export async function indexProjectionAcl(sql: Sql): Promise { +export async function indexProjectionAcl(pool: Sql): Promise { + const sql = await pool.reserve() const [{ timeout }] = await sql`SELECT current_setting('lock_timeout') AS timeout` await sql.unsafe('SET lock_timeout = 0') try { @@ -237,14 +240,15 @@ export async function indexProjectionAcl(sql: Sql): Promise { for (const projection of PROJECTION_SOURCE_ACL_TABLES) await sql.unsafe(`ANALYZE ${projection}`) } finally { await sql`SELECT set_config('lock_timeout', ${timeout}, false)` + sql.release() } } /** * Leaves the app one outbox event to start the backfill from. The outbox processor runs on every * deployment, so the backfill starts on its own once the app that ships the handler is up — - * on the Trigger.dev worker where there is one, detached in the app otherwise — without an - * operator remembering to. Idempotent under its fixed id. + * enqueued on the Trigger.dev worker where there is one, run in bounded slices by the outbox + * itself otherwise — without an operator remembering to. Idempotent under its fixed id. */ export async function enqueueProjectionSourceAclBackfillEvent(sql: Sql): Promise { await sql` From 526f659f2d55299b7ba3884b7f3a83820cd07df5 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 20 Sep 2026 10:40:30 -0700 Subject: [PATCH 4/6] fix(knowledge): start the projection backfill the way the table backfill is started The outbox event and its slice-and-defer handler are gone; nothing else in the repo starts background work that way. The fill is started from the app side as the table backfill is: tasks.trigger on the Trigger.dev worker when one is configured, runDetached in-process otherwise, with the manual script as the operator path after a deploy. The registered migration is now 0022_projection_source_acl_backfill, which supersedes 0021 so a database that already recorded the synchronous shape still gains the unfilled indexes. The page statement timeout is exported so a caller bounding a run can leave it as headroom. Claude-Session: https://claude.ai/code/session_01XU6c7pKRpa5CMoMHKDdqxX --- apps/sim/lib/core/outbox/processor.test.ts | 3 - apps/sim/lib/core/outbox/processor.ts | 2 - .../projection-source-acl-backfill.test.ts | 61 +++------------- .../search/projection-source-acl-backfill.ts | 73 +++++-------------- .../scripts/backfill-projection-source-acl.ts | 19 ++--- ...rations-paused-billing-attribution.test.ts | 2 +- ...6_backfill_search_vectors.postgres.test.ts | 1 + ...mbedding_search_connector.postgres.test.ts | 20 +---- .../0021_embedding_search_connector.ts | 56 +++----------- .../0022_projection_source_acl_backfill.ts | 30 ++++++++ packages/db/script-migrations/index.ts | 5 +- 11 files changed, 84 insertions(+), 188 deletions(-) create mode 100644 packages/db/script-migrations/0022_projection_source_acl_backfill.ts diff --git a/apps/sim/lib/core/outbox/processor.test.ts b/apps/sim/lib/core/outbox/processor.test.ts index d678e5b461a..6ad93cc80fa 100644 --- a/apps/sim/lib/core/outbox/processor.test.ts +++ b/apps/sim/lib/core/outbox/processor.test.ts @@ -33,9 +33,6 @@ vi.mock('@/lib/knowledge/application/slack-search/outbox', () => ({ vi.mock('@/lib/knowledge/documents/processing-outbox-handler', () => ({ knowledgeDocumentProcessingOutboxHandlers: {}, })) -vi.mock('@/lib/knowledge/search/projection-source-acl-backfill', () => ({ - projectionSourceAclBackfillOutboxHandlers: {}, -})) vi.mock('@/lib/mothership/inbox/cleanup-outbox', () => ({ inboxCleanupOutboxHandlers: {} })) vi.mock('@/lib/organizations/resource-cleanup', () => ({ organizationResourceCleanupOutboxHandlers: {}, diff --git a/apps/sim/lib/core/outbox/processor.ts b/apps/sim/lib/core/outbox/processor.ts index 5183914d723..41e625b118f 100644 --- a/apps/sim/lib/core/outbox/processor.ts +++ b/apps/sim/lib/core/outbox/processor.ts @@ -18,7 +18,6 @@ import { slackSearchOutboxHandlers } from '@/lib/knowledge/application/slack-sea import { getConnectorFailureDiagnostic } from '@/lib/knowledge/connectors/connector-error' import { knowledgeDocumentProcessingOutboxHandlers } from '@/lib/knowledge/documents/processing-outbox-handler' import { recoverKnowledgeDocumentProcessing } from '@/lib/knowledge/documents/processing-recovery' -import { projectionSourceAclBackfillOutboxHandlers } from '@/lib/knowledge/search/projection-source-acl-backfill' import { inboxCleanupOutboxHandlers } from '@/lib/mothership/inbox/cleanup-outbox' import { organizationResourceCleanupOutboxHandlers } from '@/lib/organizations/resource-cleanup' import { workspaceFileLiveDocOutboxHandlers } from '@/lib/uploads/contexts/workspace/workspace-file-live-doc-outbox' @@ -43,7 +42,6 @@ const handlers = { ...invitationMigrationOutboxHandlers, ...directGrantOutboxHandlers, ...knowledgeDocumentProcessingOutboxHandlers, - ...projectionSourceAclBackfillOutboxHandlers, ...organizationResourceCleanupOutboxHandlers, ...inboxCleanupOutboxHandlers, ...permissionAccessRequestOutboxHandlers, diff --git a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts index 527d5f5a953..a314999a4bf 100644 --- a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts +++ b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.test.ts @@ -3,55 +3,33 @@ */ import { beforeEach, describe, expect, it, vi } from 'vitest' -const { mockBackfill, mockEnd, mockPostgres, mockTasksTrigger, envState } = vi.hoisted(() => ({ +const { mockBackfill, mockEnd, mockPostgres, mockTasksTrigger } = vi.hoisted(() => ({ mockBackfill: vi.fn(), mockEnd: vi.fn(async () => undefined), mockPostgres: vi.fn(), mockTasksTrigger: vi.fn(async () => ({ id: 'run-1' })), - envState: { triggerEnabled: false, secret: undefined as string | undefined }, })) vi.mock('@sim/db', () => ({ resolveDbUrl: () => 'postgres://localhost:5432/sim' })) vi.mock('@sim/db/script-migrations/0021_embedding_search_connector', () => ({ PROJECTION_SOURCE_ACL_TABLES: ['embedding_search', 'embedding_keyword_tin'], - PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT: 'knowledge.projection.source_acl.backfill', backfillProjectionSourceAcl: mockBackfill, })) vi.mock('postgres', () => ({ default: mockPostgres })) vi.mock('@trigger.dev/sdk', () => ({ tasks: { trigger: mockTasksTrigger } })) vi.mock('@/lib/core/async-jobs/region', () => ({ resolveTriggerRegion: async () => 'us-east-1' })) -vi.mock('@/lib/core/config/env', () => ({ - env: { - get TRIGGER_SECRET_KEY() { - return envState.secret - }, +vi.mock('@/lib/core/utils/background', () => ({ + runDetached: (_label: string, work: () => Promise) => { + void work() }, })) -vi.mock('@/lib/core/config/env-flags', () => ({ - get isTriggerDevEnabled() { - return envState.triggerEnabled - }, -})) -vi.mock('@/lib/core/outbox/service', () => ({ - continueOutboxHandler: (reason: string) => ({ - outcome: 'deferred', - reason, - consumeAttempt: false, - }), - withOutboxHandlerTimeout: (handler: unknown, timeoutMs: number) => - Object.assign(handler as object, { timeoutMs }), -})) import { enqueueProjectionSourceAclBackfill, - projectionSourceAclBackfillOutboxHandlers, runProjectionSourceAclBackfill, } from '@/lib/knowledge/search/projection-source-acl-backfill' const connection = { end: mockEnd } -const handler = - projectionSourceAclBackfillOutboxHandlers['knowledge.projection.source_acl.backfill'] -const context = { eventId: 'e', eventType: 'knowledge.projection.source_acl.backfill' } describe('runProjectionSourceAclBackfill', () => { beforeEach(() => { @@ -112,7 +90,7 @@ describe('runProjectionSourceAclBackfill', () => { }) }) -describe('the outbox event the migration leaves behind', () => { +describe('enqueueProjectionSourceAclBackfill', () => { beforeEach(() => { vi.clearAllMocks() mockPostgres.mockReturnValue(connection) @@ -125,10 +103,8 @@ describe('the outbox event the migration leaves behind', () => { }) }) - it('hands the backfill to the Trigger.dev worker when there is one', async () => { - envState.triggerEnabled = true - envState.secret = 'tr_secret' - await expect(enqueueProjectionSourceAclBackfill({ pageSize: 25 })).resolves.toEqual({ + it('hands the backfill to the Trigger.dev worker when one is configured', async () => { + await expect(enqueueProjectionSourceAclBackfill({ pageSize: 25 }, true)).resolves.toEqual({ runId: 'run-1', }) expect(mockTasksTrigger).toHaveBeenCalledWith( @@ -136,29 +112,12 @@ describe('the outbox event the migration leaves behind', () => { { pageSize: 25 }, { region: 'us-east-1' } ) - await expect(handler({}, context as never)).resolves.toBeUndefined() - expect(mockTasksTrigger).toHaveBeenCalledTimes(2) expect(mockBackfill).not.toHaveBeenCalled() }) - it('runs a bounded slice per outbox run without one, and stays pending until it is filled', async () => { - envState.triggerEnabled = false - envState.secret = undefined - mockBackfill.mockResolvedValueOnce({ - projection: 'embedding_search', - scanned: 100, - written: 100, - afterId: 'chunk-100', - done: false, - }) - await expect(handler({}, context as never)).resolves.toMatchObject({ - outcome: 'deferred', - consumeAttempt: false, - }) + it('fills the projections detached in this process without one', async () => { + await expect(enqueueProjectionSourceAclBackfill({}, false)).resolves.toBeNull() expect(mockTasksTrigger).not.toHaveBeenCalled() - expect(mockBackfill).toHaveBeenCalledTimes(1) - expect(mockBackfill.mock.calls[0][2].budgetMs).toBeLessThanOrEqual(handler.timeoutMs!) - await expect(handler({}, context as never)).resolves.toBeUndefined() - expect(mockBackfill).toHaveBeenCalledTimes(3) + await vi.waitFor(() => expect(mockBackfill).toHaveBeenCalledTimes(2)) }) }) diff --git a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts index 55c2ef87214..fec8a6f489a 100644 --- a/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts +++ b/apps/sim/lib/knowledge/search/projection-source-acl-backfill.ts @@ -1,7 +1,6 @@ import { resolveDbUrl } from '@sim/db' import { backfillProjectionSourceAcl, - PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT, PROJECTION_SOURCE_ACL_TABLES, type ProjectionSourceAclTable, } from '@sim/db/script-migrations/0021_embedding_search_connector' @@ -10,13 +9,7 @@ import postgres from 'postgres' import { resolveTriggerRegion } from '@/lib/core/async-jobs/region' import { env } from '@/lib/core/config/env' import { isTriggerDevEnabled } from '@/lib/core/config/env-flags' -import { - continueOutboxHandler, - type DeferredOutboxHandlerResult, - type OutboxEventContext, - type OutboxHandlerRegistry, - withOutboxHandlerTimeout, -} from '@/lib/core/outbox/service' +import { runDetached } from '@/lib/core/utils/background' const logger = createLogger('ProjectionSourceAclBackfill') @@ -82,56 +75,24 @@ export async function runProjectionSourceAclBackfill( } } -/** Whether the deployment has a Trigger.dev worker to hand the backfill to. */ -export function projectionSourceAclBackfillUsesTrigger(): boolean { - return Boolean(isTriggerDevEnabled && env.TRIGGER_SECRET_KEY) -} - /** * Starts the backfill the way the table backfill is started: on the deployment's Trigger.dev - * worker, where bounded runs chain until both projections are filled. Safe to call again at any - * time — a run only fills rows still unset. + * worker when one is configured, where bounded runs chain until both projections are filled, and + * detached in this process otherwise. Safe to call again at any time — a run only fills rows still + * unset. */ export async function enqueueProjectionSourceAclBackfill( - payload: ProjectionSourceAclBackfillPayload = {} -): Promise<{ runId: string }> { - const { tasks } = await import('@trigger.dev/sdk') - const handle = await tasks.trigger(PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, payload, { - region: await resolveTriggerRegion(), - }) - logger.info('Projection source and ACL backfill enqueued', { runId: handle.id }) - return { runId: handle.id } -} - -/** One outbox run's share of the backfill on a deployment without a worker, inside its window. */ -const OUTBOX_RUN_BUDGET_MS = 4 * 60 * 1000 -const OUTBOX_HANDLER_TIMEOUT_MS = 5 * 60 * 1000 - -/** - * Handles the event script migration `0021_embedding_search_connector` leaves behind, so the - * backfill starts once the app that ships this handler is up, on every deployment. With a - * Trigger.dev worker the event is done once the task is enqueued: the task owns its own retries - * and continuations. Without one the backfill runs here, one bounded slice per outbox run, and the - * event stays pending until both projections are filled — every slice's writes are durable, and a - * slice that starts over after a restart skips the filled rows through the unfilled index, so an - * interrupted deployment loses nothing but the time of one slice. - */ -export const projectionSourceAclBackfillOutboxHandlers: OutboxHandlerRegistry = { - [PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT]: withOutboxHandlerTimeout( - async ( - _payload: unknown, - _context: OutboxEventContext - ): Promise => { - if (projectionSourceAclBackfillUsesTrigger()) { - await enqueueProjectionSourceAclBackfill() - return undefined - } - const cursor = await runProjectionSourceAclBackfill({}, { budgetMs: OUTBOX_RUN_BUDGET_MS }) - if (!cursor) return undefined - return continueOutboxHandler( - `projection source and ACL backfill paused at ${cursor.projection} after ${cursor.afterId}` - ) - }, - OUTBOX_HANDLER_TIMEOUT_MS - ), + payload: ProjectionSourceAclBackfillPayload = {}, + useTrigger = Boolean(isTriggerDevEnabled && env.TRIGGER_SECRET_KEY) +): Promise<{ runId: string } | null> { + if (useTrigger) { + const { tasks } = await import('@trigger.dev/sdk') + const handle = await tasks.trigger(PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, payload, { + region: await resolveTriggerRegion(), + }) + logger.info('Projection source and ACL backfill enqueued', { runId: handle.id }) + return { runId: handle.id } + } + runDetached(PROJECTION_SOURCE_ACL_BACKFILL_TASK_ID, () => runProjectionSourceAclBackfill(payload)) + return null } diff --git a/apps/sim/scripts/backfill-projection-source-acl.ts b/apps/sim/scripts/backfill-projection-source-acl.ts index 804c37d0968..def5181c49c 100644 --- a/apps/sim/scripts/backfill-projection-source-acl.ts +++ b/apps/sim/scripts/backfill-projection-source-acl.ts @@ -2,11 +2,10 @@ /** * Starts the projection source and ACL backfill that script migration - * `0021_embedding_search_connector` leaves to the background, by hand. The migration's outbox - * event already starts it after a deploy; this is for starting it again — on a deployment with - * Trigger.dev it enqueues the `projection-source-acl-backfill` task, which chains bounded runs - * until both ranking projections are filled; without one it fills them here, paced the same way. - * Safe to run at any time — a run only fills rows still unset. + * `0022_projection_source_acl_backfill` leaves to the background. On a deployment with Trigger.dev + * it enqueues the `projection-source-acl-backfill` task, which chains bounded runs until both + * ranking projections are filled; without one it fills them here, paced the same way. Safe to run + * again at any time — a run only fills rows still unset. * * Usage: * bun apps/sim/scripts/backfill-projection-source-acl.ts @@ -14,18 +13,20 @@ import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' +import { env } from '@/lib/core/config/env' +import { isTriggerDevEnabled } from '@/lib/core/config/env-flags' import { enqueueProjectionSourceAclBackfill, - projectionSourceAclBackfillUsesTrigger, runProjectionSourceAclBackfill, } from '@/lib/knowledge/search/projection-source-acl-backfill' const logger = createLogger('BackfillProjectionSourceAcl') +/** A script has no long-lived process to detach into, so without a worker it fills inline. */ async function main(): Promise { - if (projectionSourceAclBackfillUsesTrigger()) { - const handle = await enqueueProjectionSourceAclBackfill() - logger.info('Backfill enqueued on the Trigger.dev worker', handle) + if (isTriggerDevEnabled && env.TRIGGER_SECRET_KEY) { + const handle = await enqueueProjectionSourceAclBackfill({}, true) + logger.info('Backfill enqueued on the Trigger.dev worker', handle ?? {}) return } await runProjectionSourceAclBackfill({}) diff --git a/packages/db/script-migrations-paused-billing-attribution.test.ts b/packages/db/script-migrations-paused-billing-attribution.test.ts index bc9b2eb474f..cea6d576863 100644 --- a/packages/db/script-migrations-paused-billing-attribution.test.ts +++ b/packages/db/script-migrations-paused-billing-attribution.test.ts @@ -453,7 +453,7 @@ describe('script migration registry', () => { '0017_index_search_documents', '0018_repair_workspace_file_content_revision', '0019_tin_keyword_projection', - '0021_embedding_search_connector', + '0022_projection_source_acl_backfill', ]) }) }) diff --git a/packages/db/script-migrations/0016_backfill_search_vectors.postgres.test.ts b/packages/db/script-migrations/0016_backfill_search_vectors.postgres.test.ts index 3781437a856..6345cab6c3f 100644 --- a/packages/db/script-migrations/0016_backfill_search_vectors.postgres.test.ts +++ b/packages/db/script-migrations/0016_backfill_search_vectors.postgres.test.ts @@ -392,6 +392,7 @@ describe.runIf(Boolean(databaseUrl))('search projection upgrade in PostgreSQL', { name: '0018_repair_workspace_file_content_revision' }, { name: '0019_tin_keyword_projection' }, { name: '0021_embedding_search_connector' }, + { name: '0022_projection_source_acl_backfill' }, ]) const [{ complete }] = await sql`SELECT count(*)::int AS complete FROM embedding e JOIN embedding_search s ON s.id = e.id JOIN embedding_keyword_search k ON k.id = e.id diff --git a/packages/db/script-migrations/0021_embedding_search_connector.postgres.test.ts b/packages/db/script-migrations/0021_embedding_search_connector.postgres.test.ts index c7364f2aee8..b0e3ad37326 100644 --- a/packages/db/script-migrations/0021_embedding_search_connector.postgres.test.ts +++ b/packages/db/script-migrations/0021_embedding_search_connector.postgres.test.ts @@ -1,7 +1,5 @@ -import { - backfillProjectionSourceAcl, - embeddingSearchConnectorMigration, -} from '@sim/db/script-migrations/0021_embedding_search_connector' +import { backfillProjectionSourceAcl } from '@sim/db/script-migrations/0021_embedding_search_connector' +import { projectionSourceAclBackfillMigration as embeddingSearchConnectorMigration } from '@sim/db/script-migrations/0022_projection_source_acl_backfill' import { generateId } from '@sim/utils/id' import postgres, { type Sql } from 'postgres' import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest' @@ -40,10 +38,6 @@ describe.runIf(Boolean(databaseUrl))('projection source and ACL backfill in Post await sql`CREATE TABLE document ( id text PRIMARY KEY, connector_id text, acl text[] NOT NULL DEFAULT '{ws}' )` - await sql`CREATE TABLE outbox_event ( - id text PRIMARY KEY, event_type text NOT NULL, payload json NOT NULL, - status text NOT NULL DEFAULT 'pending' - )` for (const projection of ['embedding_search', 'embedding_keyword_tin']) { await sql`CREATE TABLE ${sql(projection)} ( id text PRIMARY KEY, document_id text NOT NULL, enabled boolean NOT NULL DEFAULT true, @@ -65,7 +59,7 @@ describe.runIf(Boolean(databaseUrl))('projection source and ACL backfill in Post await sql`ALTER TABLE embedding_keyword_tin DISABLE TRIGGER embedding_keyword_tin_source_acl_set` }) - it('installs its triggers, indexes and start event again without failing, so a cut-short deploy completes', async () => { + it('installs its triggers and indexes again without failing, so a cut-short deploy completes', async () => { await expect(embeddingSearchConnectorMigration.up(sql)).resolves.toBeUndefined() const indexes = await sql<{ indexname: string }[]>` SELECT indexname FROM pg_indexes WHERE schemaname = ${schemaName} ORDER BY indexname` @@ -78,14 +72,6 @@ describe.runIf(Boolean(databaseUrl))('projection source and ACL backfill in Post 'embedding_search_source_idx', ]) ) - /** Two runs of the migration leave the app one event, so the backfill is started once. */ - expect(await sql`SELECT id, event_type, status FROM outbox_event`).toEqual([ - { - id: 'projection-source-acl-backfill:0021', - event_type: 'knowledge.projection.source_acl.backfill', - status: 'pending', - }, - ]) }) it('fills only the rows still unset, in pages, and leaves a chunk that changed documents to its trigger', async () => { diff --git a/packages/db/script-migrations/0021_embedding_search_connector.ts b/packages/db/script-migrations/0021_embedding_search_connector.ts index 7e80b5df02a..86630e7d69a 100644 --- a/packages/db/script-migrations/0021_embedding_search_connector.ts +++ b/packages/db/script-migrations/0021_embedding_search_connector.ts @@ -1,4 +1,3 @@ -import type { ScriptMigration } from '@sim/db/script-migrations/types' import { createLogger } from '@sim/logger' import { sleep } from '@sim/utils/helpers' import postgres, { type Sql } from 'postgres' @@ -15,21 +14,16 @@ export const PROJECTION_SOURCE_ACL_PAGE_SIZE = 100 /** Pause between pages, so the backfill shares the database with the search it serves. */ export const PROJECTION_SOURCE_ACL_PAGE_PAUSE_MS = 250 -/** Longest a page may run before the database cancels it; the run then fails and resumes. */ -const PAGE_STATEMENT_TIMEOUT = '60s' +/** + * Longest a page may run before the database cancels it; the run then fails and resumes. A caller + * that bounds a run leaves at least this much headroom after its budget, since the budget is + * checked between pages and the page in flight runs to this limit. + */ +export const PROJECTION_SOURCE_ACL_PAGE_TIMEOUT_MS = 60_000 /** Pages between progress log lines. */ const PROGRESS_EVERY_PAGES = 100 -/** - * The outbox event the migration leaves for the app, whose handler starts the backfill on the - * deployment's worker. Written once, under a fixed id, so a rerun of the migration does not start - * it twice. - */ -export const PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT = - 'knowledge.projection.source_acl.backfill' -const PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_ID = 'projection-source-acl-backfill:0021' - /** The projections that carry their document's source and ACL. */ export const PROJECTION_SOURCE_ACL_TABLES = ['embedding_search', 'embedding_keyword_tin'] as const export type ProjectionSourceAclTable = (typeof PROJECTION_SOURCE_ACL_TABLES)[number] @@ -138,7 +132,7 @@ export async function backfillProjectionSourceAcl( for (;;) { const page = await sql.begin(async (tx) => { await tx.unsafe("SET LOCAL lock_timeout = '5s'") - await tx.unsafe(`SET LOCAL statement_timeout = '${PAGE_STATEMENT_TIMEOUT}'`) + await tx.unsafe(`SET LOCAL statement_timeout = ${PROJECTION_SOURCE_ACL_PAGE_TIMEOUT_MS}`) const [row] = await tx.unsafe< Array<{ scanned: number; filled: number; last_id: string | null }> >( @@ -244,42 +238,10 @@ export async function indexProjectionAcl(pool: Sql): Promise { } } -/** - * Leaves the app one outbox event to start the backfill from. The outbox processor runs on every - * deployment, so the backfill starts on its own once the app that ships the handler is up — - * enqueued on the Trigger.dev worker where there is one, run in bounded slices by the outbox - * itself otherwise — without an operator remembering to. Idempotent under its fixed id. - */ -export async function enqueueProjectionSourceAclBackfillEvent(sql: Sql): Promise { - await sql` - INSERT INTO outbox_event (id, event_type, payload) - VALUES (${PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_ID}, ${PROJECTION_SOURCE_ACL_BACKFILL_OUTBOX_EVENT}, '{}'::json) - ON CONFLICT (id) DO NOTHING` -} - -/** - * Installs the triggers, builds the indexes and leaves the outbox event that starts the backfill; - * all three are idempotent, so a run that was cut short completes on the next deploy. The columns - * are not filled here: on `embedding_search` every filled row is re-inserted into each HNSW index, - * which puts the full projection far beyond what a deploy job can wait for. The backfill runs - * afterwards from the `projection-source-acl-backfill` Trigger.dev task, which the outbox handler - * starts and which chains bounded runs until both projections are filled. It is safe to start again - * at any time, with `bun apps/sim/scripts/backfill-projection-source-acl.ts` or a test run of the - * task from the Trigger.dev dashboard. Until it completes, an unfilled row is decided on its - * document, the join per candidate every row paid before the columns existed. - */ -export const embeddingSearchConnectorMigration: ScriptMigration = { - name: '0021_embedding_search_connector', - async up(sql) { - await installProjectionSourceAcl(sql) - await indexProjectionAcl(sql) - await enqueueProjectionSourceAclBackfillEvent(sql) - }, -} - /** * Run directly — `db:push`, or an operator filling a database by hand — the projections are filled - * here, paced the same way, rather than left to the app. + * here, paced the same way, rather than left to the app. The registered migration is + * `0022_projection_source_acl_backfill`, which supersedes this file's earlier, synchronous shape. */ if (import.meta.main) { const url = process.env.MIGRATION_DATABASE_URL ?? process.env.DATABASE_URL diff --git a/packages/db/script-migrations/0022_projection_source_acl_backfill.ts b/packages/db/script-migrations/0022_projection_source_acl_backfill.ts new file mode 100644 index 00000000000..9d3718ec84f --- /dev/null +++ b/packages/db/script-migrations/0022_projection_source_acl_backfill.ts @@ -0,0 +1,30 @@ +import { + indexProjectionAcl, + installProjectionSourceAcl, +} from '@sim/db/script-migrations/0021_embedding_search_connector' +import type { ScriptMigration } from '@sim/db/script-migrations/types' + +/** + * Installs the projection source and ACL triggers and builds their indexes; both are idempotent, + * so a run that was cut short completes on the next deploy. The columns are not filled here: on + * `embedding_search` every filled row is re-inserted into each HNSW index, which puts the full + * projection far beyond what a deploy job can wait for. The backfill is started after the deploy + * with `bun apps/sim/scripts/backfill-projection-source-acl.ts`, which enqueues the + * `projection-source-acl-backfill` Trigger.dev task the way the table backfill is enqueued (or runs + * the same backfill detached in the app where no worker is configured); the task chains bounded + * runs until both projections are filled and is safe to start again at any time. Until it + * completes, an unfilled row is decided on its document, the join per candidate every row paid + * before the columns existed. + * + * Supersedes `0021_embedding_search_connector`, whose synchronous backfill this replaces: a + * database that recorded it still gains the unfilled indexes, and one where it was cut short + * records both names once this completes. + */ +export const projectionSourceAclBackfillMigration: ScriptMigration = { + name: '0022_projection_source_acl_backfill', + supersedes: ['0021_embedding_search_connector'], + async up(sql) { + await installProjectionSourceAcl(sql) + await indexProjectionAcl(sql) + }, +} diff --git a/packages/db/script-migrations/index.ts b/packages/db/script-migrations/index.ts index aab5835488a..85a3b2f099c 100644 --- a/packages/db/script-migrations/index.ts +++ b/packages/db/script-migrations/index.ts @@ -5,7 +5,7 @@ import { backfillSearchVectorsMigration } from '@sim/db/script-migrations/0016_b import { indexSearchDocumentsMigration } from '@sim/db/script-migrations/0017_index_search_documents' import { repairWorkspaceFileContentRevisionMigration } from '@sim/db/script-migrations/0018_repair_workspace_file_content_revision' import { tinKeywordProjectionMigration } from '@sim/db/script-migrations/0019_tin_keyword_projection' -import { embeddingSearchConnectorMigration } from '@sim/db/script-migrations/0021_embedding_search_connector' +import { projectionSourceAclBackfillMigration } from '@sim/db/script-migrations/0022_projection_source_acl_backfill' import type { Sql } from 'postgres' import { backfillTableOrderKeys } from './0001_backfill_table_order_keys' import { backfillPausedBillingAttribution } from './0002_backfill_paused_billing_attribution' @@ -44,7 +44,8 @@ export const scriptMigrations: readonly ScriptMigration[] = [ /** 0358 stops new sub-millisecond revisions; this retires the ones that predate it. */ repairWorkspaceFileContentRevisionMigration, tinKeywordProjectionMigration, - embeddingSearchConnectorMigration, + /** 0022 supersedes 0021, whose synchronous backfill could not finish inside a deploy. */ + projectionSourceAclBackfillMigration, ] /** From 9435d510fdd65fa8cec044626709b534bedfec02 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 20 Sep 2026 10:50:28 -0700 Subject: [PATCH 5/6] fix(knowledge): refuse a projection backfill page size that is not a positive integer The page size is interpolated into the page statement and a page of nothing would report the projection filled, so a payload that asks for either is refused before the first page rather than quietly reshaped. Claude-Session: https://claude.ai/code/session_01XU6c7pKRpa5CMoMHKDdqxX --- .../0021_embedding_search_connector.test.ts | 28 +++++++++++++++++++ .../0021_embedding_search_connector.ts | 10 +++++++ 2 files changed, 38 insertions(+) create mode 100644 packages/db/script-migrations/0021_embedding_search_connector.test.ts diff --git a/packages/db/script-migrations/0021_embedding_search_connector.test.ts b/packages/db/script-migrations/0021_embedding_search_connector.test.ts new file mode 100644 index 00000000000..d4e030b449a --- /dev/null +++ b/packages/db/script-migrations/0021_embedding_search_connector.test.ts @@ -0,0 +1,28 @@ +/** + * @vitest-environment node + */ +import { backfillProjectionSourceAcl } from '@sim/db/script-migrations/0021_embedding_search_connector' +import type { Sql } from 'postgres' +import { describe, expect, it, vi } from 'vitest' + +/** A session that must never be reached: every case below is refused before the first page. */ +const untouched = { begin: vi.fn() } as unknown as Sql + +describe('backfillProjectionSourceAcl', () => { + it.each([0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY])( + 'refuses a page size of %s instead of reporting the projection filled', + async (pageSize) => { + await expect( + backfillProjectionSourceAcl(untouched, 'embedding_search', { pageSize }) + ).rejects.toThrow('page size must be a positive integer') + expect(untouched.begin).not.toHaveBeenCalled() + } + ) + + it.each([-1, Number.NaN])('refuses a pause of %s', async (pauseMs) => { + await expect( + backfillProjectionSourceAcl(untouched, 'embedding_search', { pauseMs }) + ).rejects.toThrow('pause must be a non-negative number') + expect(untouched.begin).not.toHaveBeenCalled() + }) +}) diff --git a/packages/db/script-migrations/0021_embedding_search_connector.ts b/packages/db/script-migrations/0021_embedding_search_connector.ts index 86630e7d69a..5c68808208b 100644 --- a/packages/db/script-migrations/0021_embedding_search_connector.ts +++ b/packages/db/script-migrations/0021_embedding_search_connector.ts @@ -121,6 +121,16 @@ export async function backfillProjectionSourceAcl( ): Promise { const pageSize = options.pageSize ?? PROJECTION_SOURCE_ACL_PAGE_SIZE const pauseMs = options.pauseMs ?? PROJECTION_SOURCE_ACL_PAGE_PAUSE_MS + /** + * The page size is interpolated into the statement and a page of nothing would report the + * projection filled; a payload that asks for either is refused rather than quietly reshaped. + */ + if (!Number.isSafeInteger(pageSize) || pageSize < 1) { + throw new Error(`Projection backfill page size must be a positive integer, got ${pageSize}`) + } + if (!Number.isFinite(pauseMs) || pauseMs < 0) { + throw new Error(`Projection backfill pause must be a non-negative number, got ${pauseMs}`) + } const startedAt = Date.now() const deadline = options.budgetMs === undefined ? Number.POSITIVE_INFINITY : startedAt + options.budgetMs From ac644c06666acf4d53e494816d54284d87f29e33 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 20 Sep 2026 10:57:04 -0700 Subject: [PATCH 6/6] fix(knowledge): check the projection backfill budget after the pause between pages A pause that crossed the budget still let the next iteration open a page; the deadline is now checked after the pause, so a bounded run stops before its next page rather than after it. Claude-Session: https://claude.ai/code/session_01XU6c7pKRpa5CMoMHKDdqxX --- .../db/script-migrations/0021_embedding_search_connector.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/db/script-migrations/0021_embedding_search_connector.ts b/packages/db/script-migrations/0021_embedding_search_connector.ts index 5c68808208b..4fc87e10fe3 100644 --- a/packages/db/script-migrations/0021_embedding_search_connector.ts +++ b/packages/db/script-migrations/0021_embedding_search_connector.ts @@ -184,8 +184,9 @@ export async function backfillProjectionSourceAcl( elapsedMs: Date.now() - startedAt, }) } - if (Date.now() >= deadline) break if (pauseMs > 0) await sleep(pauseMs) + /** Checked after the pause, so the pause cannot carry a run past its budget into another page. */ + if (Date.now() >= deadline) break } logger.info( done ? 'Projection source and ACL backfilled' : 'Projection source and ACL backfill paused',