From f828ac45cc632376162f085026b8b5edbe8d013e Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 17 Sep 2026 13:54:29 -0700 Subject: [PATCH 1/2] fix(knowledge): write external group membership as a diff MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every directory sync deleted a group's entire membership and reinserted it, whether or not anything had changed. That took the membership table to roughly 55M lifetime inserts and 55M deletes against ~127k live rows — about 430x write amplification — holding it near 91% dead tuples across 2,447 autovacuum cycles, an order of magnitude more churn than any comparable table. The vacuum load that generates competes for the same I/O as every other query on the instance. Membership is now written as a difference: read the group's current tokens, delete only those the enumeration no longer lists, insert only those it newly lists. An unchanged group writes nothing. The group row is locked before that read. The two callers fence on different leases — the directory lease and the connector sync lease — so neither excludes the other, and on the directory path the group upsert commits in a separate transaction, so its row lock is already released. Without the lock, the removal set is computed against a snapshot a concurrent pass may have moved past, and a subject that pass inserted would survive an enumeration that never observed it: membership retained rather than revoked. Locking also closes the narrower window the blind delete already had between its delete and its commit. `created_at` on a member row now means first-observed rather than last-observed. No reader projects it, and the group's `lastSyncedAt` remains the freshness signal and is still written every pass. Also corrects the processing-queue TSDoc, which argued the per-tenant lane limits were sized rather than inherited and cited a concurrency figure derived from an unsound measurement. --- apps/sim/background/knowledge-processing.ts | 28 ++++--- .../connectors/external-group-sync.test.ts | 82 ++++++++++++++++++- .../connectors/external-group-sync.ts | 60 +++++++++++++- 3 files changed, 154 insertions(+), 16 deletions(-) diff --git a/apps/sim/background/knowledge-processing.ts b/apps/sim/background/knowledge-processing.ts index 8ca8526ae26..3e8a6c2f81e 100644 --- a/apps/sim/background/knowledge-processing.ts +++ b/apps/sim/background/knowledge-processing.ts @@ -211,20 +211,24 @@ export async function runDocumentProcessing( /** * Both lanes are keyed by tenant at dispatch, so `concurrencyLimit` is the - * ceiling one tenant may hold in that lane, not a ceiling for the fleet. The - * shared bound is the Trigger.dev environment concurrency limit, which is where - * a global ceiling belongs; observed peak there is ~97 across every task. + * ceiling one tenant may hold in that lane, not a ceiling for the fleet. There + * is no longer a fleet-wide ceiling for document processing: the aggregate is + * active tenants times the lane limit, bounded only by the Trigger.dev + * environment concurrency limit, which every other task shares. * - * Both default to the limit the single shared queue carried, which is what - * keeps this split from ever draining slower than the queue it replaces: the - * busiest case it has to beat is one tenant alone, and one tenant alone still - * gets the same slots it used to get for backfill plus a separate allowance for - * work someone is waiting on. Any second tenant is pure gain, because under the - * shared queue it got whatever the first one left. + * Both carry 20 because that is the number the single shared queue carried, not + * because 20 was derived for a per-tenant ceiling — it has been the default + * since the queue was introduced and the split changed its unit rather than its + * value. One tenant alone therefore still gets what it used to for backfill, + * plus a separate allowance for work someone is waiting on; two tenants draw + * twice the aggregate the shared queue ever allowed. * - * Splitting the two into separate variables is for operating them, not for - * sizing them: backfill is the one to lower when the environment ceiling is the - * binding constraint, and lowering it must not slow down a person's upload. + * So backfill is the one to lower, and the database is what decides when: it is + * the resource the aggregate actually lands on, and the per-document embedding + * writes are the load. Lower it when their latency climbs, not when the + * Trigger.dev environment limit is approached. The queue concurrency override + * API applies a new value without a redeploy; this variable is read when the + * worker deploy registers the queue, so changing it here needs one. */ export const interactiveProcessingQueue = queue({ name: INTERACTIVE_PROCESSING_QUEUE_NAME, diff --git a/apps/sim/lib/knowledge/connectors/external-group-sync.test.ts b/apps/sim/lib/knowledge/connectors/external-group-sync.test.ts index 82082479983..3213b0164a1 100644 --- a/apps/sim/lib/knowledge/connectors/external-group-sync.test.ts +++ b/apps/sim/lib/knowledge/connectors/external-group-sync.test.ts @@ -124,6 +124,78 @@ describe('syncExternalDirectoryGroups', () => { ) }) + /** + * The reason membership writes are a diff: a directory sync overwhelmingly + * re-observes membership that has not changed, and rewriting the group would + * charge two row writes per member to autovacuum for no change at all. + */ + it('writes nothing when the observed membership already matches', async () => { + queueTableRows(schemaMock.knowledgeExternalGroup, []) + queueTableRows(schemaMock.knowledgeExternalGroupMember, [{ subjectToken: 'u:alice@corp.com' }]) + const dir = directory({ + listGroups: vi.fn(async () => [{ id: 'eng@corp.com' }]), + listGroupMembers: vi.fn(async (group) => ({ + group, + memberTokens: ['u:alice@corp.com'], + complete: true, + })), + }) + + await syncExternalDirectoryGroups({ workspaceId: 'ws-1', directory: dir }) + + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + expect(dbChainMockFns.values).not.toHaveBeenCalledWith( + expect.arrayContaining([expect.objectContaining({ subjectToken: 'u:alice@corp.com' })]) + ) + expect(dbChainMockFns.set.mock.calls.some(([value]) => 'lastSyncedAt' in value)).toBe(true) + }) + + /** + * The removal set is computed from a read, so that read has to be serialized + * against the other writer. Both callers fence on different leases, and the + * directory path commits its group upsert in a separate transaction, so the + * lock has to be taken here. + */ + it('locks the group row before reading the membership it will diff against', async () => { + queueTableRows(schemaMock.knowledgeExternalGroup, []) + queueTableRows(schemaMock.knowledgeExternalGroupMember, [{ subjectToken: 'u:alice@corp.com' }]) + const dir = directory({ + listGroups: vi.fn(async () => [{ id: 'eng@corp.com' }]), + listGroupMembers: vi.fn(async (group) => ({ + group, + memberTokens: ['u:alice@corp.com'], + complete: true, + })), + }) + + await syncExternalDirectoryGroups({ workspaceId: 'ws-1', directory: dir }) + + expect(dbChainMockFns.for).toHaveBeenCalledWith('update') + }) + + it('writes only the difference when membership changed', async () => { + queueTableRows(schemaMock.knowledgeExternalGroup, []) + queueTableRows(schemaMock.knowledgeExternalGroupMember, [ + { subjectToken: 'u:alice@corp.com' }, + { subjectToken: 'u:bob@corp.com' }, + ]) + const dir = directory({ + listGroups: vi.fn(async () => [{ id: 'eng@corp.com' }]), + listGroupMembers: vi.fn(async (group) => ({ + group, + memberTokens: ['u:alice@corp.com', 'u:carol@corp.com'], + complete: true, + })), + }) + + await syncExternalDirectoryGroups({ workspaceId: 'ws-1', directory: dir }) + + expect(dbChainMockFns.values).toHaveBeenCalledWith([ + { groupId: expect.any(String), subjectToken: 'u:carol@corp.com' }, + ]) + expect(dbChainMockFns.delete).toHaveBeenCalledTimes(1) + }) + it.each(['ws', 'pub', 'link', 'g:confluence:cloud:group', 'alice@corp.com', 'u:Alice@corp.com'])( 'rejects invalid member %s before replacing membership or updating freshness', async (invalid) => { @@ -370,7 +442,15 @@ describe('refreshConnectorDirectory', () => { expect(dbChainMockFns.set.mock.calls.some(([value]) => 'lastCompleteSyncAt' in value)).toBe( false ) - expect(dbChainMockFns.delete).toHaveBeenCalledTimes(1) + /** + * The accessible group's membership is written; the denied one is left + * alone. Nothing is deleted because the diff found no member to remove — + * membership writes are the difference, not a full rewrite. + */ + expect(dbChainMockFns.values).toHaveBeenCalledWith([ + { groupId: expect.any(String), subjectToken: 'u:alice@corp.com' }, + ]) + expect(dbChainMockFns.delete).not.toHaveBeenCalled() }) it('keeps unknown failures blocking even when other group memberships refreshed', async () => { diff --git a/apps/sim/lib/knowledge/connectors/external-group-sync.ts b/apps/sim/lib/knowledge/connectors/external-group-sync.ts index a5b24273f58..9313a3551a2 100644 --- a/apps/sim/lib/knowledge/connectors/external-group-sync.ts +++ b/apps/sim/lib/knowledge/connectors/external-group-sync.ts @@ -353,7 +353,23 @@ export async function persistExternalGroupMembership( ) } -/** Membership replacement and its freshness watermark commit together. */ +/** + * Membership replacement and its freshness watermark commit together. + * + * Writes only the difference. Rewriting a whole group per sync costs two row + * writes per member every time, and a directory sync overwhelmingly re-observes + * membership that has not changed. Unconditional replacement had taken this + * table to ~55M lifetime inserts and ~55M deletes against ~127k live rows — + * roughly 430x write amplification, holding it at ~91% dead tuples through + * 2,447 autovacuum cycles, an order of magnitude more than any comparable + * table. That vacuum load is charged to the same I/O every other query on the + * instance competes for. The extra read is one index scan of the group's + * primary-key prefix, and it is what lets an unchanged group write nothing. + * + * `created_at` therefore becomes first-observed rather than last-observed. No + * reader projects it; the group's own `lastSyncedAt` below is the freshness + * signal, and it is still written every pass. + */ async function replaceGroupMembers( groupId: string, memberTokens: string[], @@ -364,13 +380,51 @@ async function replaceGroupMembers( if (memberTokens.some((token) => !isDirectoryMemberToken(token, directory))) { throw new Error('Directory membership contains an invalid identity token') } + /** + * Serializes membership writes for this group before the set is read. + * + * The two callers fence on different leases — the directory lease and the + * connector sync lease — so neither excludes the other, and on the directory + * path the group upsert commits in a separate transaction from this one, so + * its row lock is already gone. Without this, the removal set is computed + * from a snapshot a concurrent pass may have moved past, and a subject that + * pass inserted would survive a complete enumeration that did not observe it: + * membership retained rather than revoked. The blind delete this replaced was + * immune because it never read first. + */ await tx - .delete(knowledgeExternalGroupMember) + .select({ id: knowledgeExternalGroup.id }) + .from(knowledgeExternalGroup) + .where(eq(knowledgeExternalGroup.id, groupId)) + .for('update') + const desired = new Set(memberTokens) + const existing = await tx + .select({ subjectToken: knowledgeExternalGroupMember.subjectToken }) + .from(knowledgeExternalGroupMember) .where(eq(knowledgeExternalGroupMember.groupId, groupId)) - for (const batch of chunkArray([...new Set(memberTokens)], MEMBER_WRITE_BATCH_SIZE)) { + const retained = new Set() + const removed: string[] = [] + for (const row of existing) { + if (desired.has(row.subjectToken)) retained.add(row.subjectToken) + else removed.push(row.subjectToken) + } + const added = [...desired].filter((subjectToken) => !retained.has(subjectToken)) + + for (const batch of chunkArray(removed, MEMBER_WRITE_BATCH_SIZE)) { + await tx + .delete(knowledgeExternalGroupMember) + .where( + and( + eq(knowledgeExternalGroupMember.groupId, groupId), + inArray(knowledgeExternalGroupMember.subjectToken, batch) + ) + ) + } + for (const batch of chunkArray(added, MEMBER_WRITE_BATCH_SIZE)) { await tx .insert(knowledgeExternalGroupMember) .values(batch.map((subjectToken) => ({ groupId, subjectToken }))) + .onConflictDoNothing() } await tx .update(knowledgeExternalGroup) From 9f2e322237086c521a57736350e63015595f1a43 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 17 Sep 2026 14:00:31 -0700 Subject: [PATCH 2/2] test(knowledge): assert the group lock precedes the membership read Presence of the lock was the only thing asserted, so moving it below the read would have kept the test green while reopening the stale-snapshot race the lock exists to close. The assertion is now the relative call order of the lock and the member-table read. --- .../knowledge/connectors/external-group-sync.test.ts | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/apps/sim/lib/knowledge/connectors/external-group-sync.test.ts b/apps/sim/lib/knowledge/connectors/external-group-sync.test.ts index 3213b0164a1..d95ff2551b0 100644 --- a/apps/sim/lib/knowledge/connectors/external-group-sync.test.ts +++ b/apps/sim/lib/knowledge/connectors/external-group-sync.test.ts @@ -171,6 +171,17 @@ describe('syncExternalDirectoryGroups', () => { await syncExternalDirectoryGroups({ workspaceId: 'ws-1', directory: dir }) expect(dbChainMockFns.for).toHaveBeenCalledWith('update') + const memberReadIndex = dbChainMockFns.from.mock.calls.findIndex( + ([table]) => table === schemaMock.knowledgeExternalGroupMember + ) + expect(memberReadIndex).toBeGreaterThanOrEqual(0) + /** + * Ordering is the property, not the presence: a lock taken after the read + * leaves exactly the stale-snapshot race it exists to close. + */ + expect(dbChainMockFns.for.mock.invocationCallOrder[0]).toBeLessThan( + dbChainMockFns.from.mock.invocationCallOrder[memberReadIndex] + ) }) it('writes only the difference when membership changed', async () => {