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

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)