fix(knowledge): settle detach reservations when a knowledge base is purged

.github/workflows/test-build.yml

@@ -278,6 +278,7 @@ jobs:
           lib/knowledge/__integration__/slack-empty-threads.integration.ts
           lib/knowledge/__integration__/kb-block-search.integration.ts
           lib/knowledge/__integration__/unfilled-projection-source.integration.ts
+          lib/knowledge/__integration__/purged-detach-reservation.integration.ts
           lib/core/outbox/service.integration.ts
           lib/knowledge/__integration__/connector-upload.integration.ts
           lib/uploads/contexts/organization-logo/application.integration.ts

apps/sim/background/cleanup-soft-deletes.test.ts

@@ -26,6 +26,7 @@ const {
   mockPrepareChatCleanup,
   mockResolveStorageBillingContext,
   mockSelectRowsByIdChunks,
+  mockSettleDetachedConnectorReservations,
   mockDeduplicateWorkflowName,
   mockAllocateUniqueWorkspaceFileName,
   mockDeduplicateFolderName,
@@ -46,6 +47,7 @@ const {
   mockPrepareChatCleanup: vi.fn(async () => ({ execute: vi.fn(async () => undefined) })),
   mockResolveStorageBillingContext: vi.fn(),
   mockSelectRowsByIdChunks: vi.fn(async () => [] as unknown[]),
+  mockSettleDetachedConnectorReservations: vi.fn(async () => undefined),
 }))
 
 vi.mock('@/lib/billing/cleanup-dispatcher', () => ({ runCleanupWithLimits: vi.fn() }))
@@ -71,6 +73,10 @@ vi.mock('@/lib/billing/storage', () => ({
   resolveStorageBillingContext: mockResolveStorageBillingContext,
 }))
 
+vi.mock('@/lib/knowledge/connectors/detachment', () => ({
+  settleDetachedConnectorReservations: mockSettleDetachedConnectorReservations,
+}))
+
 vi.mock('@/lib/knowledge/documents/service', () => ({
   hardDeleteDocuments: mockHardDeleteDocuments,
 }))
@@ -303,6 +309,32 @@ describe('cleanup soft deletes', () => {
     )
   })
 
+  it('settles overdrawn reservations before the documents and the rest before the base delete', async () => {
+    mockChunkedBatchDelete.mockImplementationOnce(
+      async (options: { onBatch?: (rows: Array<{ id: string }>) => Promise<void> }) => {
+        await options.onBatch?.([{ id: 'kb-1' }, { id: 'kb-2' }])
+        mockKnowledgeBaseContainerDelete()
+        return { deleted: 2, failed: 0 }
+      }
+    )
+    dbChainMockFns.limit
+      .mockResolvedValueOnce([{ id: 'doc-1' }])
+      .mockResolvedValueOnce([])
+      .mockResolvedValueOnce([])
+
+    await runCleanupSoftDeletes(basePayload)
+
+    expect(mockSettleDetachedConnectorReservations.mock.calls).toEqual([
+      [['kb-1', 'kb-2'], 'overdrawn'],
+      [['kb-1', 'kb-2'], 'remaining'],
+    ])
+    const [overdrawn, remaining] = mockSettleDetachedConnectorReservations.mock.invocationCallOrder
+    const [deletedDocuments] = mockHardDeleteDocuments.mock.invocationCallOrder
+    expect(overdrawn).toBeLessThan(deletedDocuments)
+    expect(deletedDocuments).toBeLessThan(remaining)
+    expect(remaining).toBeLessThan(mockKnowledgeBaseContainerDelete.mock.invocationCallOrder[0])
+  })
+
   it('soft-deletes abandoned KB bindings and removes their storage objects', async () => {
     dbChainMockFns.limit
       .mockResolvedValueOnce([{ key: 'kb/orphan-1' }, { key: 'kb/orphan-2' }])

apps/sim/background/cleanup-soft-deletes.ts

@@ -40,6 +40,7 @@ import {
   resolveCleanupOwnerScope,
 } from '@/lib/cleanup/resource-scope'
 import { deduplicateFolderName } from '@/lib/folders/naming'
+import { settleDetachedConnectorReservations } from '@/lib/knowledge/connectors/detachment'
 import { hardDeleteDocuments } from '@/lib/knowledge/documents/service'
 import type { StorageContext } from '@/lib/uploads'
 import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
@@ -445,11 +446,17 @@ async function cleanupExpiredKnowledgeBases(
       isNotNull(knowledgeBase.deletedAt),
       lt(knowledgeBase.deletedAt, retentionDate)
     ),
-    onBatch: (rows: { id: string }[]) =>
-      hardDeleteKnowledgeBaseDocuments(
-        rows.map(({ id }) => id),
-        label
-      ),
+    /**
+     * The bases' DELETE cascades their connectors away, so a detached connector's reservation is
+     * settled here: an overdrawn one before the documents (their deletion floors usage at zero),
+     * the rest after them, so a deletion that fails partway leaves every step's ledger consistent.
+     */
+    onBatch: async (rows: { id: string }[]) => {
+      const knowledgeBaseIds = rows.map(({ id }) => id)
+      await settleDetachedConnectorReservations(knowledgeBaseIds, 'overdrawn')
+      await hardDeleteKnowledgeBaseDocuments(knowledgeBaseIds, label)
+      await settleDetachedConnectorReservations(knowledgeBaseIds, 'remaining')
+    },
   }
   return scope.kind === 'workspace'
     ? chunkedBatchDelete({ ...options, workspaceIds: scope.ids })

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

@@ -1,6 +1,6 @@
 import { db } from '@sim/db'
 import { outboxEvent } from '@sim/db/schema'
-import { and, eq, sql } from 'drizzle-orm'
+import { and, asc, 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'
@@ -13,9 +13,11 @@ export async function drainConnectorEvent(connectorId: string, eventType: string
     .where(
       and(
         eq(outboxEvent.eventType, eventType),
+        eq(outboxEvent.status, 'pending'),
         sql`${outboxEvent.payload}->>'connectorId' = ${connectorId}`
       )
     )
+    .orderBy(asc(outboxEvent.availableAt), asc(outboxEvent.id))
     .limit(1)
   expect(job).toBeDefined()
   let status = await processOutboxEventById(job.id, knowledgeDocumentProcessingOutboxHandlers)