fix(knowledge): release kept connector documents in the background so removing a large source cannot time out

Author: waleedlatif1

apps/sim/ee/workspace-forking/lib/mapping/cascade.ts

@@ -165,7 +165,8 @@ export async function detectForkCascadeReferences(params: {
           inArray(knowledgeConnector.knowledgeBaseId, Array.from(knowledgeBaseIds)),
           eq(knowledgeBase.workspaceId, sourceWorkspaceId),
           isNull(knowledgeBase.deletedAt),
-          isNull(knowledgeConnector.deletedAt)
+          isNull(knowledgeConnector.deletedAt),
+          isNull(knowledgeConnector.detachedAt)
         )
       )
     for (const connector of connectors) {

apps/sim/lib/billing/storage/index.ts

@@ -12,6 +12,7 @@ export {
   applyStorageUsageDeltasInTx,
   checkAndIncrementStorageUsageInTx,
   decrementStorageUsageForBillingContextInTx,
+  incrementAdmittedStorageUsageForBillingContextInTx,
   incrementStorageUsageForBillingContextInTx,
   type LegacyStorageUsageDelta,
   maybeNotifyStorageLimitForBillingContext,

apps/sim/lib/billing/storage/tracking.ts

@@ -590,6 +590,29 @@ export async function incrementStorageUsageForBillingContextInTx(
   return result.updatedUsage
 }
 
+/**
+ * Increments one workspace and its current payer for bytes whose admission was
+ * already decided, such as documents a connector removal accepted and a
+ * background job now releases page by page. Never refuses: a page that could
+ * cross the limit after admission would otherwise leave the release half done.
+ */
+export async function incrementAdmittedStorageUsageForBillingContextInTx(
+  tx: DbOrTx,
+  context: StorageBillingContext,
+  bytes: number
+): Promise<number | undefined> {
+  if (bytes <= 0) return undefined
+  const result = await mutateWorkspaceStorageUsage(
+    tx,
+    context.workspaceId,
+    bytes,
+    'increment',
+    undefined,
+    context
+  )
+  return result.updatedUsage
+}
+
 /**
  * Atomically check quota and increment a user's (or their org's) storage
  * counter inside an existing transaction, using a pre-resolved subscription.

apps/sim/lib/copilot/chat/workspace-context.ts

@@ -24,6 +24,7 @@ import {
   getAccessibleOAuthCredentials,
 } from '@/lib/credentials/environment'
 import { listWorkspaceSandboxes } from '@/lib/execution/remote-sandbox/workspace-sandboxes'
+import { connectorIsLive } from '@/lib/knowledge/connectors/sync-lock'
 import { listCustomBlockSummariesForWorkspace } from '@/lib/workflows/custom-blocks/operations'
 import { listCustomTools } from '@/lib/workflows/custom-tools/operations'
 import { listSkillsForUser } from '@/lib/workflows/skills/operations'
@@ -457,13 +458,7 @@ async function buildWorkspaceMdData(
               connectorType: knowledgeConnector.connectorType,
             })
             .from(knowledgeConnector)
-            .where(
-              and(
-                inArray(knowledgeConnector.knowledgeBaseId, kbIds),
-                isNull(knowledgeConnector.archivedAt),
-                isNull(knowledgeConnector.deletedAt)
-              )
-            )
+            .where(and(inArray(knowledgeConnector.knowledgeBaseId, kbIds), connectorIsLive()))
         : []
     const connectorTypesByKb = new Map<string, string[]>()
     for (const row of connectorRows) {

apps/sim/lib/knowledge/__integration__/drain-connector-event.ts

@@ -0,0 +1,27 @@
+import { db } from '@sim/db'
+import { outboxEvent } from '@sim/db/schema'
+import { and, eq, sql } from 'drizzle-orm'
+import { expect } from 'vitest'
+import { processOutboxEventById } from '@/lib/core/outbox/service'
+import { knowledgeDocumentProcessingOutboxHandlers } from '@/lib/knowledge/documents/processing-outbox-handler'
+
+/** Runs a connector's queued removal event until it completes, as the outbox worker would. */
+export async function drainConnectorEvent(connectorId: string, eventType: string): Promise<void> {
+  const [job] = await db
+    .select()
+    .from(outboxEvent)
+    .where(
+      and(
+        eq(outboxEvent.eventType, eventType),
+        sql`${outboxEvent.payload}->>'connectorId' = ${connectorId}`
+      )
+    )
+    .limit(1)
+  expect(job).toBeDefined()
+  let status = await processOutboxEventById(job.id, knowledgeDocumentProcessingOutboxHandlers)
+  for (let attempt = 0; status === 'pending' && attempt < 20; attempt++) {
+    await db.update(outboxEvent).set({ availableAt: new Date() }).where(eq(outboxEvent.id, job.id))
+    status = await processOutboxEventById(job.id, knowledgeDocumentProcessingOutboxHandlers)
+  }
+  expect(status).toBe('completed')
+}

apps/sim/lib/knowledge/__integration__/search-source-progress.integration.ts

@@ -14,6 +14,7 @@ import { and, eq, inArray } from 'drizzle-orm'
 import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
 import type { ConnectorDocumentFilter } from '@/lib/api/contracts/knowledge/connectors'
 import * as embeddings from '@/lib/embeddings'
+import { drainConnectorEvent } from '@/lib/knowledge/__integration__/drain-connector-event'
 import {
   createKnowledgeAclFixtureIds,
   seedKnowledgeAclFixture,
@@ -29,6 +30,7 @@ import {
 } from '@/lib/knowledge/application/documents'
 import { readSearchSourceProgress } from '@/lib/knowledge/application/search-source-progress'
 import { listSearchSources } from '@/lib/knowledge/application/search-sources'
+import { KNOWLEDGE_CONNECTOR_DETACH_EVENT } from '@/lib/knowledge/connectors/detachment'
 import { createContentSyncLease } from '@/lib/knowledge/connectors/sync-lock'
 import { persistSkippedDocuments } from '@/lib/knowledge/connectors/sync-persistence'
 import * as documentProcessor from '@/lib/knowledge/documents/document-processor'
@@ -630,6 +632,7 @@ describe('intentional skips and genuine failures across document reads', () => {
       input: { ...scope, deleteDocuments: false },
     })
     expect(result).toMatchObject({ documentsDeleted: 0, documentsKept: 6 })
+    await drainConnectorEvent(fixture.connectorId, KNOWLEDGE_CONNECTOR_DETACH_EVENT)
     const retained = await db
       .select()
       .from(document)

apps/sim/lib/knowledge/__integration__/storage-accounting.integration.ts

@@ -6,6 +6,7 @@ import { db } from '@sim/db'
 import {
   document,
   embedding,
+  embeddingSearch,
   knowledgeBase,
   knowledgeConnector,
   organization,
@@ -14,19 +15,19 @@ import {
   workspace,
 } from '@sim/db/schema'
 import { generateId } from '@sim/utils/id'
-import { and, eq, inArray, isNull, sql } from 'drizzle-orm'
+import { and, eq, inArray, isNotNull, isNull, sql } from 'drizzle-orm'
 import { afterAll, describe, expect, it, vi } from 'vitest'
-import { processOutboxEventById } from '@/lib/core/outbox/service'
+import { drainConnectorEvent } from '@/lib/knowledge/__integration__/drain-connector-event'
 import {
   createKnowledgeAclFixtureIds,
   seedKnowledgeAclFixture,
 } from '@/lib/knowledge/__integration__/seed-source-access-fixture'
 import { WORKSPACE_ACCESS_SCOPE } from '@/lib/knowledge/access/scope'
 import { SYSTEM_ACCESS_SCOPE } from '@/lib/knowledge/access/types'
 import { KNOWLEDGE_CONNECTOR_CLEANUP_EVENT } from '@/lib/knowledge/connectors/deletion'
+import { KNOWLEDGE_CONNECTOR_DETACH_EVENT } from '@/lib/knowledge/connectors/detachment'
 import { createContentSyncLease, SyncLockLostException } from '@/lib/knowledge/connectors/sync-lock'
 import { persistSkippedDocuments } from '@/lib/knowledge/connectors/sync-persistence'
-import { knowledgeDocumentProcessingOutboxHandlers } from '@/lib/knowledge/documents/processing-outbox-handler'
 import {
   createSingleDocument,
   getKnowledgeDocument,
@@ -104,13 +105,23 @@ function disconnect(ids: Fixture, deleteDocuments = false) {
   })
 }
 
+/** Removes the source keeping its documents, then runs the background release to completion. */
+async function detach(ids: Fixture) {
+  const outcome = await disconnect(ids)
+  if (outcome.success) await drainConnectorEvent(ids.connectorId, KNOWLEDGE_CONNECTOR_DETACH_EVENT)
+  return outcome
+}
+
 afterAll(async () => {
   for (const ids of fixtures) {
     await db
       .delete(outboxEvent)
       .where(
         and(
-          eq(outboxEvent.eventType, KNOWLEDGE_CONNECTOR_CLEANUP_EVENT),
+          inArray(outboxEvent.eventType, [
+            KNOWLEDGE_CONNECTOR_CLEANUP_EVENT,
+            KNOWLEDGE_CONNECTOR_DETACH_EVENT,
+          ]),
           sql`${outboxEvent.payload}->>'knowledgeBaseId' = ${ids.knowledgeBaseId}`
         )
       )
@@ -130,7 +141,7 @@ describe('knowledge document storage ledgers', () => {
     const transaction = db.transaction.bind(db)
     /** Interleave real operations at the lock boundary; every query and commit still uses PostgreSQL. */
     const detachBeforeLock: typeof db.transaction = async (callback, config) => {
-      expect(await disconnect(ids)).toMatchObject({ success: true })
+      expect(await detach(ids)).toMatchObject({ success: true })
       expect(await ledger(ids)).toEqual({ workspaceBytes: 37, payerBytes: 37 })
       return transaction(callback, config)
     }
@@ -167,7 +178,19 @@ describe('knowledge document storage ledgers', () => {
       { success: true, documentsKept: 6, documentsDeleted: 0 },
     ])
     expect(outcomes.filter((result) => !result.success)).toHaveLength(1)
+    /** Detached documents stay attached, readable, and unbilled until the release runs. */
+    expect(await ledger(ids)).toEqual({ workspaceBytes: 29, payerBytes: 29 })
+    expect(
+      await getKnowledgeDocument(ids.knowledgeBaseId, source[0].id, WORKSPACE_ACCESS_SCOPE)
+    ).not.toBeNull()
+    await drainConnectorEvent(ids.connectorId, KNOWLEDGE_CONNECTOR_DETACH_EVENT)
     expect(await ledger(ids)).toEqual({ workspaceBytes: 70, payerBytes: 70 })
+    expect(
+      await db
+        .select({ id: knowledgeConnector.id })
+        .from(knowledgeConnector)
+        .where(eq(knowledgeConnector.id, ids.connectorId))
+    ).toHaveLength(0)
     const retained = await db
       .select({ id: document.id, connectorId: document.connectorId, deletedAt: document.deletedAt })
       .from(document)
@@ -197,6 +220,43 @@ describe('knowledge document storage ledgers', () => {
     expect(await ledger(ids)).toEqual({ workspaceBytes: 0, payerBytes: 0 })
   })
 
+  it('releases a document larger than one page without a search row still naming its source', async () => {
+    const ids = await seed()
+    const row = sourceDocument(ids, 5)
+    await db.insert(document).values(row)
+    await db.insert(embedding).values(
+      Array.from({ length: 600 }, (_, chunkIndex) => ({
+        id: generateId(),
+        knowledgeBaseId: ids.knowledgeBaseId,
+        documentId: row.id,
+        chunkIndex,
+        chunkHash: `hash-${chunkIndex}`,
+        content: 'test chunk',
+        contentLength: 10,
+        tokenCount: 2,
+        startOffset: 0,
+        endOffset: 10,
+        embedding384: Array(384).fill(0.1),
+      }))
+    )
+    const sourceRows = () =>
+      db
+        .select({ count: sql<number>`COUNT(*)::integer` })
+        .from(embeddingSearch)
+        .where(and(eq(embeddingSearch.documentId, row.id), isNotNull(embeddingSearch.connectorId)))
+    expect((await sourceRows())[0].count).toBe(600)
+
+    expect(await detach(ids)).toEqual({ success: true, documentsKept: 1, documentsDeleted: 0 })
+
+    expect((await sourceRows())[0].count).toBe(0)
+    const [released] = await db
+      .select({ connectorId: document.connectorId })
+      .from(document)
+      .where(eq(document.id, row.id))
+    expect(released.connectorId).toBeNull()
+    expect(await ledger(ids)).toEqual({ workspaceBytes: 5, payerBytes: 5 })
+  })
+
   it('hides a source immediately and cleans bounded batches without debiting manual storage', async () => {
     const ids = await seed()
     await manualDocument(ids, 31)
@@ -240,17 +300,6 @@ describe('knowledge document storage ledgers', () => {
     for (const access of [SYSTEM_ACCESS_SCOPE, WORKSPACE_ACCESS_SCOPE]) {
       expect(await getKnowledgeDocument(ids.knowledgeBaseId, rows[0].id, access)).toBeNull()
     }
-    const [job] = await db
-      .select()
-      .from(outboxEvent)
-      .where(
-        and(
-          eq(outboxEvent.eventType, KNOWLEDGE_CONNECTOR_CLEANUP_EVENT),
-          sql`${outboxEvent.payload}->>'connectorId' = ${ids.connectorId}`
-        )
-      )
-      .limit(1)
-    expect(job).toBeDefined()
     await expect(
       persistSkippedDocuments(
         ids.knowledgeBaseId,
@@ -274,17 +323,7 @@ describe('knowledge document storage ledgers', () => {
         createContentSyncLease(ids.connectorId, ids.lockId)
       )
     ).rejects.toBeInstanceOf(SyncLockLostException)
-    const handlers = knowledgeDocumentProcessingOutboxHandlers
-    let status = await processOutboxEventById(job.id, handlers)
-    expect(status).toBe('pending')
-    for (let attempt = 0; status === 'pending' && attempt < 5; attempt++) {
-      await db
-        .update(outboxEvent)
-        .set({ availableAt: new Date() })
-        .where(eq(outboxEvent.id, job.id))
-      status = await processOutboxEventById(job.id, handlers)
-    }
-    expect(status).toBe('completed')
+    await drainConnectorEvent(ids.connectorId, KNOWLEDGE_CONNECTOR_CLEANUP_EVENT)
     expect(
       await db
         .select({ id: knowledgeConnector.id })
@@ -398,7 +437,7 @@ describe('knowledge document storage ledgers', () => {
   it('serializes ordinary uploads against source detachment without losing either charge', async () => {
     const ids = await seed()
     await db.insert(document).values([sourceDocument(ids, 37)])
-    const [detached, manual] = await Promise.all([disconnect(ids), manualDocument(ids, 41)])
+    const [detached, manual] = await Promise.all([detach(ids), manualDocument(ids, 41)])
     expect(detached).toMatchObject({ success: true })
     expect(await ledger(ids)).toEqual({ workspaceBytes: 78, payerBytes: 78 })
     const [source] = await db

apps/sim/lib/knowledge/application/connectors.ts

@@ -74,6 +74,7 @@ import {
   readConnectorPermissionSummary,
 } from '@/lib/knowledge/connectors/permission-config.server'
 import { MEMBER_OBSERVATION_STALE_AFTER_HOURS } from '@/lib/knowledge/connectors/sync-limits'
+import { connectorIsLive } from '@/lib/knowledge/connectors/sync-lock'
 import {
   DEFAULT_KNOWLEDGE_CONNECTOR_DOCUMENT_PAGE_SIZE,
   MAX_KNOWLEDGE_CONNECTOR_DOCUMENT_MUTATION_ITEMS,
@@ -525,11 +526,7 @@ export const listKnowledgeConnectors = defineAuthorizedKnowledgeUseCase({
       .select()
       .from(knowledgeConnector)
       .where(
-        and(
-          eq(knowledgeConnector.knowledgeBaseId, context.knowledgeBaseId),
-          isNull(knowledgeConnector.archivedAt),
-          isNull(knowledgeConnector.deletedAt)
-        )
+        and(eq(knowledgeConnector.knowledgeBaseId, context.knowledgeBaseId), connectorIsLive())
       )
       .orderBy(sortOrder(sortColumn), sortOrder(knowledgeConnector.id))
     const offset = input.offset ?? 0

apps/sim/lib/knowledge/connectors/deletion.md

@@ -2,10 +2,12 @@
 
 All existing UI, API, and Copilot callers still enter `knowledge.connectors.delete` through its authorized application use case. Roles, scope checks, response shapes, audit attribution, and the workspace-only option to retain documents are unchanged.
 
-When documents are removed, a transaction locks the canonical knowledge base and connector, counts the attached documents, marks the connector deleted, invalidates both sync leases, disables scheduling, and inserts one `knowledge.connector.cleanup` outbox event. Failure rolls back both the deletion and the event. Returned deletion counts describe documents logically removed from Sim; physical deletion follows asynchronously. The keep-documents path still performs its storage quota check and detachment atomically.
+When documents are removed, a transaction locks the canonical knowledge base and connector, counts the attached documents, marks the connector deleted, invalidates both sync leases, disables scheduling, and inserts one `knowledge.connector.cleanup` outbox event. Failure rolls back both the deletion and the event. Returned deletion counts describe documents logically removed from Sim; physical deletion follows asynchronously.
 
 The shared document access predicate excludes deleted connectors, including public and workspace access and internal indexing reads. Ingestion also checks connector liveness before claiming or committing document processing. Existing source-write leases reject late sync writes. Documents keep their connector reference until physical deletion, so they cannot become standalone readable or billable documents during cleanup. Restoring a knowledge base does not restore a directly deleted connector.
 
+When documents are kept, the same transaction admits their storage against the payer's quota, sets the connector's `detached_at` instead of `deleted_at`, invalidates both sync leases, disables scheduling, and inserts one `knowledge.connector.detach` outbox event. The request writes no document: releasing a document nulls its `connector_id`, and the projection trigger then rewrites every enabled chunk of it in both search projections, each a fresh index entry, so an in-request release timed out on any sizeable source. A detached connector fails `connectorIsLive()`, so it cannot be synced, managed, or listed, while document visibility checks only `archived_at` and `deleted_at`, so its documents stay readable throughout. The detach worker releases at most 250 projection rows per table of the next 100 documents per transaction and flips those documents, resurrecting live tombstones and zeroing legacy skipped sizes, once none of their rows still names the connector; the trigger then has nothing to rewrite. Each flipped page is billed in its own transaction, so a document is billable exactly when it no longer names a connector. When no document remains, the worker drains the connector's history and member rows and deletes it.
+
 The existing outbox worker runs cleanup, with 48 failure attempts and bounded continuations that do not consume that retry budget. Each transaction removes at most 1,000 chunks, 250 documents, or 1,000 sync-history/member rows. A run does at most four batches and yields after its time budget. Transactions use lock and statement timeouts. The worker verifies the connector's deletion timestamp, locks documents against late indexing commits, and commits storage cleanup intents before deleting those documents. It resolves storage ownership from the currently locked knowledge base. Already committed batches survive worker restarts. Failures remain retryable; exhausted events remain as dead letters while the source stays inaccessible.
 
 After document and connector cleanup, credential grant revocation and unused-tag cleanup are retried as needed. Tag cleanup checks for existence rather than counting the entire remaining corpus. No provider credentials or document contents enter the connector cleanup payload.