fix(knowledge): write external group membership as a diff (#7943)

* fix(knowledge): write external group membership as a diff

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.

* 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.

Author: waleedlatif1

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,

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 () => {

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<string>()
+  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)