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..d95ff2551b0 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,89 @@ 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') + 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 () => { + 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 +453,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)