fix(db): re-decide the deletion guard inside each deleting transaction and reset member relist state

Each deleting transaction share-locks the search index and its connectors and re-decides the guard, so a connector resumed between pages waits for the page in flight and the next page refuses. Every document-deleting transaction also resets the stopped connectors' listing state, now including the directory checkpoint and each member's retry time, so a run stopped partway never leaves a connector that would skip deleted documents on resume.

Author: waleedlatif1

apps/sim/scripts/dormant-org-search/README.md

@@ -152,12 +152,14 @@ Source-connected documents are not metered as uploaded storage, so there is no s
 
 The **knowledge base row and its paused connectors are kept**: their configuration, credentials, members, permission snapshots and sync history. Re-enabling is then a matter of resuming the connectors, with nothing to set up again.
 
-Once no connector-owned document remains, the run clears each stopped connector's listing state. These are the same columns the app clears when a connector must list everything again (an access-mode switch or a source change):
+Every page that deletes documents also clears each stopped connector's listing state, in the same transaction, so a run stopped partway never leaves a connector that would skip what was already deleted. These are the columns the app clears when a connector must list everything again (an access-mode switch or a source change), plus the directory checkpoint and each member's retry time:
 
-- `lastSyncAt`, `lastSyncDocCount`, `listingCheckpoint` and `memberTombstoneCursor` on `knowledge_connector`
-- `listingCheckpoint`, `changeCursor`, `memberSyncedThrough`, `lastCompleteListingAt` and `lastListedCount` on its `knowledge_connector_member` rows
+- `lastSyncAt`, `lastSyncDocCount`, `listingCheckpoint`, `directoryCheckpoint` and `memberTombstoneCursor` on `knowledge_connector`
+- `listingCheckpoint`, `changeCursor`, `memberSyncedThrough`, `lastCompleteListingAt`, `lastListedCount` and `nextAttemptAt` (so every member is due) on its `knowledge_connector_member` rows
 
-Without this, a resumed connector would sync incrementally from its old cursor ("changed since last sync"). It would never re-list the documents deleted here, and the index would stay silently incomplete. Partition work rows (`knowledge_connector_partition`) belong to the old listing generation, and the next full listing replaces them. `--no-connector-reset` skips the reset. The reset runs only when the run reaches the end with no connector documents left: a run resumed with `--after-id` checks for documents before its cursor and says so.
+Without this, a resumed connector would sync incrementally from its old cursor ("changed since last sync"), skip directory reconciliation behind a `complete` checkpoint, or leave members waiting on a future retry. It would never re-list the documents deleted here, and the index would stay silently incomplete. Partition work rows (`knowledge_connector_partition`) belong to the old listing generation, and the next full listing replaces them. `--no-connector-reset` skips the reset. When a run finishes with no connector documents left, it resets once more to catch rows a page could not (for example a member added mid-run).
+
+Each deleting transaction also re-decides the guard while holding the knowledge base and its connectors `FOR SHARE`. Resuming a connector or claiming a sync updates the connector row, so it waits for the page in flight to commit, and the next page refuses.
 
 ### Duration and monitoring
 

apps/sim/scripts/dormant-org-search/dormant-org-search.integration.ts

@@ -262,13 +262,25 @@ describe('dormant organization search runbook in PostgreSQL', () => {
         sleep,
       })
     ).rejects.toThrow('not an organization search index')
+    /** A deleting transaction re-decides the guard itself, so a resume between pages cannot race it. */
+    await expect(
+      store.deleteChunkBatch(indexId, indexDocumentIds.slice(0, 1), 10)
+    ).rejects.toBeInstanceOf(SearchIndexDeletionRefused)
+    await expect(
+      store.deleteDocuments(indexId, indexDocumentIds.slice(0, 1), 'fixture', true)
+    ).rejects.toBeInstanceOf(SearchIndexDeletionRefused)
     expect((await rowsOf(indexId)).documents).toBe(DOCUMENTS)
   })
 
   it('deletes the search index in resumable pages, queues storage cleanup, and marks nothing', async () => {
     await db
       .update(knowledgeConnector)
-      .set({ status: 'paused' })
+      .set({
+        status: 'paused',
+        lastSyncAt: new Date(),
+        listingCheckpoint: { cursor: 'fixture' },
+        directoryCheckpoint: { phase: 'complete' },
+      })
       .where(eq(knowledgeConnector.id, indexConnectorId))
     const store = drizzleSearchIndexDeletionStore(timeouts)
     const options = {
@@ -295,6 +307,16 @@ describe('dormant organization search runbook in PostgreSQL', () => {
     })
     expect(first).toMatchObject({ pages: 1, documentsDeleted: 3, done: false })
     expect((await rowsOf(indexId)).documents).toBe(DOCUMENTS - 3)
+    /** A run stopped after one page already leaves the connector listing from scratch on resume. */
+    const [partial] = await db
+      .select()
+      .from(knowledgeConnector)
+      .where(eq(knowledgeConnector.id, indexConnectorId))
+    expect(partial).toMatchObject({
+      lastSyncAt: null,
+      listingCheckpoint: null,
+      directoryCheckpoint: null,
+    })
 
     const rest = await deleteSearchIndexDocuments(store, {
       ...options,
@@ -304,7 +326,7 @@ describe('dormant organization search runbook in PostgreSQL', () => {
     expect(rest).toMatchObject({
       documentsDeleted: DOCUMENTS - 3,
       done: true,
-      connectorsReset: { connectors: 1, members: 0 },
+      connectorsReset: { connectors: 0, members: 0 },
       standaloneDocumentsRemain: false,
       projectionMarks: { before: 0, after: 0 },
     })
@@ -343,6 +365,7 @@ describe('dormant organization search runbook in PostgreSQL', () => {
       status: 'paused',
       lastSyncAt: null,
       listingCheckpoint: null,
+      directoryCheckpoint: null,
     })
     const [kbRow] = await db.select().from(knowledgeBase).where(eq(knowledgeBase.id, indexId))
     expect(kbRow).toMatchObject({ isSearchIndex: true, deletedAt: null })

apps/sim/scripts/dormant-org-search/search-index-deletion-store.ts

@@ -8,13 +8,15 @@ import {
   knowledgeProjectionDirty,
   outboxEvent,
 } from '@sim/db/schema'
-import { and, asc, count, eq, gt, inArray, isNotNull, isNull, sql } from 'drizzle-orm'
+import { and, asc, count, eq, gt, inArray, isNotNull, isNull, or, sql } from 'drizzle-orm'
 import type { DbTransaction } from '@/lib/db/types'
 import {
   enqueueKnowledgeStorageCleanup,
   KNOWLEDGE_STORAGE_CLEANUP_EVENT,
 } from '@/lib/knowledge/documents/storage-cleanup'
 import {
+  evaluateDeletionGuard,
+  SearchIndexDeletionRefused,
   type SearchIndexDeletionStore,
   STOPPED_CONNECTOR_STATUSES,
 } from '@/scripts/dormant-org-search/search-index-deletion'
@@ -32,6 +34,129 @@ async function enterBoundedTransaction(tx: DbTransaction, timeouts: DeletionTime
   )
 }
 
+/**
+ * Re-decides the deletion guard inside a deleting transaction, holding the base and its
+ * connectors `FOR SHARE` until commit. A resume or a sync claim updates the connector row, so it
+ * waits for this page to commit and the next page's guard refuses it: no page can delete
+ * documents under a sync that started after the page-level guard ran.
+ */
+async function lockGuard(tx: DbTransaction, knowledgeBaseId: string) {
+  const [base] = await tx
+    .select({
+      id: knowledgeBase.id,
+      isSearchIndex: knowledgeBase.isSearchIndex,
+      deletedAt: knowledgeBase.deletedAt,
+      workspaceId: knowledgeBase.workspaceId,
+      organizationId: knowledgeBase.organizationId,
+      userId: knowledgeBase.userId,
+    })
+    .from(knowledgeBase)
+    .where(eq(knowledgeBase.id, knowledgeBaseId))
+    .for('share')
+    .limit(1)
+  const connectors = await tx
+    .select({
+      id: knowledgeConnector.id,
+      status: knowledgeConnector.status,
+      syncLockToken: knowledgeConnector.syncLockToken,
+      memberSyncLockToken: knowledgeConnector.memberSyncLockToken,
+      deletedAt: knowledgeConnector.deletedAt,
+      detachedAt: knowledgeConnector.detachedAt,
+    })
+    .from(knowledgeConnector)
+    .where(eq(knowledgeConnector.knowledgeBaseId, knowledgeBaseId))
+    .orderBy(asc(knowledgeConnector.id))
+    .for('share')
+  const reasons = evaluateDeletionGuard(
+    base ?? null,
+    connectors.map((row) => ({
+      id: row.id,
+      status: row.status,
+      syncLockHeld: row.syncLockToken !== null,
+      memberSyncLockHeld: row.memberSyncLockToken !== null,
+      deletedAt: row.deletedAt,
+      detachedAt: row.detachedAt,
+    }))
+  )
+  if (!base || reasons.length > 0) throw new SearchIndexDeletionRefused(reasons)
+  return base
+}
+
+/**
+ * Makes every stopped connector of the base list its sources from scratch when it resumes: the
+ * same columns the app clears when a connector must list everything again (an access mode
+ * switch, a source change), plus the directory checkpoint, and every member made due. Runs in
+ * each deleting transaction, so a run stopped partway never leaves a connector whose cursors
+ * would skip the documents already deleted; rows already reset are left alone.
+ */
+async function resetCursors(tx: DbTransaction, knowledgeBaseId: string, now: Date) {
+  const connectors = await tx
+    .update(knowledgeConnector)
+    .set({
+      lastSyncAt: null,
+      lastSyncDocCount: null,
+      listingCheckpoint: null,
+      directoryCheckpoint: null,
+      memberTombstoneCursor: null,
+      updatedAt: now,
+    })
+    .where(
+      and(
+        eq(knowledgeConnector.knowledgeBaseId, knowledgeBaseId),
+        isNull(knowledgeConnector.deletedAt),
+        isNull(knowledgeConnector.detachedAt),
+        inArray(knowledgeConnector.status, [...STOPPED_CONNECTOR_STATUSES]),
+        isNull(knowledgeConnector.syncLockToken),
+        isNull(knowledgeConnector.memberSyncLockToken),
+        or(
+          isNotNull(knowledgeConnector.lastSyncAt),
+          isNotNull(knowledgeConnector.lastSyncDocCount),
+          isNotNull(knowledgeConnector.listingCheckpoint),
+          isNotNull(knowledgeConnector.directoryCheckpoint),
+          isNotNull(knowledgeConnector.memberTombstoneCursor)
+        )
+      )
+    )
+    .returning({ id: knowledgeConnector.id })
+  const stopped = tx
+    .select({ id: knowledgeConnector.id })
+    .from(knowledgeConnector)
+    .where(
+      and(
+        eq(knowledgeConnector.knowledgeBaseId, knowledgeBaseId),
+        isNull(knowledgeConnector.deletedAt),
+        isNull(knowledgeConnector.detachedAt),
+        inArray(knowledgeConnector.status, [...STOPPED_CONNECTOR_STATUSES])
+      )
+    )
+  const members = await tx
+    .update(knowledgeConnectorMember)
+    .set({
+      listingCheckpoint: null,
+      changeCursor: null,
+      memberSyncedThrough: null,
+      lastCompleteListingAt: null,
+      lastListedCount: null,
+      nextAttemptAt: null,
+      updatedAt: now,
+    })
+    .where(
+      and(
+        inArray(knowledgeConnectorMember.connectorId, stopped),
+        or(
+          isNotNull(knowledgeConnectorMember.listingCheckpoint),
+          isNotNull(knowledgeConnectorMember.changeCursor),
+          isNotNull(knowledgeConnectorMember.memberSyncedThrough),
+          isNotNull(knowledgeConnectorMember.lastCompleteListingAt),
+          isNotNull(knowledgeConnectorMember.lastListedCount),
+          isNotNull(knowledgeConnectorMember.nextAttemptAt)
+        )
+      )
+    )
+    .returning({ id: knowledgeConnectorMember.id })
+  return { connectors: connectors.length, members: members.length }
+}
+
 /**
  * The deletion store on the app's database client, so storage cleanup intents are queued by the
  * app's own `enqueueKnowledgeStorageCleanup` in the deleting transaction, and deleted by the app's
@@ -101,9 +226,10 @@ export function drizzleSearchIndexDeletionStore(
       return Number(row?.chunks ?? 0)
     },
 
-    async deleteChunkBatch(documentIds, limit) {
+    async deleteChunkBatch(knowledgeBaseId, documentIds, limit) {
       return db.transaction(async (tx) => {
         await enterBoundedTransaction(tx, timeouts)
+        await lockGuard(tx, knowledgeBaseId)
         const batch = tx
           .select({ id: embedding.id })
           .from(embedding)
@@ -117,23 +243,10 @@ export function drizzleSearchIndexDeletionStore(
       })
     },
 
-    async deleteDocuments(knowledgeBaseId, documentIds, requestId) {
+    async deleteDocuments(knowledgeBaseId, documentIds, requestId, resetConnectors) {
       return db.transaction(async (tx) => {
         await enterBoundedTransaction(tx, timeouts)
-        const [owner] = await tx
-          .select({
-            workspaceId: knowledgeBase.workspaceId,
-            organizationId: knowledgeBase.organizationId,
-            userId: knowledgeBase.userId,
-            isSearchIndex: knowledgeBase.isSearchIndex,
-          })
-          .from(knowledgeBase)
-          .where(eq(knowledgeBase.id, knowledgeBaseId))
-          .for('share')
-          .limit(1)
-        if (!owner?.isSearchIndex) {
-          throw new Error('The knowledge base stopped being a search index during the run')
-        }
+        const owner = await lockGuard(tx, knowledgeBaseId)
         /** Locks the documents against a late indexing commit between the chunk check and the delete. */
         const docs = await tx
           .select({ id: document.id, fileUrl: document.fileUrl })
@@ -171,6 +284,7 @@ export function drizzleSearchIndexDeletionStore(
           .delete(document)
           .where(inArray(document.id, ids))
           .returning({ id: document.id })
+        if (resetConnectors) await resetCursors(tx, knowledgeBaseId, new Date())
         return {
           kind: 'deleted',
           deleted: deleted.length,
@@ -221,50 +335,7 @@ export function drizzleSearchIndexDeletionStore(
     async resetConnectorCursors(knowledgeBaseId) {
       return db.transaction(async (tx) => {
         await enterBoundedTransaction(tx, timeouts)
-        const now = new Date()
-        /**
-         * The same columns the app clears when a connector must list everything again (an access
-         * mode switch, a source change), limited to stopped connectors no sync holds.
-         */
-        const connectors = await tx
-          .update(knowledgeConnector)
-          .set({
-            lastSyncAt: null,
-            lastSyncDocCount: null,
-            listingCheckpoint: null,
-            memberTombstoneCursor: null,
-            updatedAt: now,
-          })
-          .where(
-            and(
-              eq(knowledgeConnector.knowledgeBaseId, knowledgeBaseId),
-              isNull(knowledgeConnector.deletedAt),
-              isNull(knowledgeConnector.detachedAt),
-              inArray(knowledgeConnector.status, [...STOPPED_CONNECTOR_STATUSES]),
-              isNull(knowledgeConnector.syncLockToken),
-              isNull(knowledgeConnector.memberSyncLockToken)
-            )
-          )
-          .returning({ id: knowledgeConnector.id })
-        if (connectors.length === 0) return { connectors: 0, members: 0 }
-        const members = await tx
-          .update(knowledgeConnectorMember)
-          .set({
-            listingCheckpoint: null,
-            changeCursor: null,
-            memberSyncedThrough: null,
-            lastCompleteListingAt: null,
-            lastListedCount: null,
-            updatedAt: now,
-          })
-          .where(
-            inArray(
-              knowledgeConnectorMember.connectorId,
-              connectors.map((connector) => connector.id)
-            )
-          )
-          .returning({ id: knowledgeConnectorMember.id })
-        return { connectors: connectors.length, members: members.length }
+        return resetCursors(tx, knowledgeBaseId, new Date())
       })
     },
   }

apps/sim/scripts/dormant-org-search/search-index-deletion.test.ts

@@ -65,7 +65,7 @@ class FakeStore implements SearchIndexDeletionStore {
   async countChunks(ids: readonly string[]) {
     return ids.reduce((sum, id) => sum + (this.documents.get(id)?.chunks ?? 0), 0)
   }
-  async deleteChunkBatch(ids: readonly string[], limit: number) {
+  async deleteChunkBatch(_knowledgeBaseId: string, ids: readonly string[], limit: number) {
     this.mutations.push('deleteChunkBatch')
     let deleted = 0
     for (const id of ids) {
@@ -78,7 +78,7 @@ class FakeStore implements SearchIndexDeletionStore {
     }
     return deleted
   }
-  async deleteDocuments(_kb: string, ids: readonly string[]) {
+  async deleteDocuments(_kb: string, ids: readonly string[], _requestId: string, _reset: boolean) {
     this.mutations.push('deleteDocuments')
     for (const id of ids) {
       const late = this.lateChunks.get(id)

apps/sim/scripts/dormant-org-search/search-index-deletion.ts

@@ -70,15 +70,22 @@ export interface SearchIndexDeletionStore {
   /** Chunks of these documents; read only by a dry run. */
   countChunks(documentIds: readonly string[]): Promise<number>
   /** Deletes at most `limit` chunks of these documents in one transaction; returns how many. */
-  deleteChunkBatch(documentIds: readonly string[], limit: number): Promise<number>
+  deleteChunkBatch(
+    knowledgeBaseId: string,
+    documentIds: readonly string[],
+    limit: number
+  ): Promise<number>
   /**
-   * In one transaction: share-locks the knowledge base and re-checks it is a search index, locks
-   * the documents, and, when none has a chunk left, queues their storage cleanup and deletes them.
+   * In one transaction: share-locks the knowledge base and its connectors and re-decides the
+   * guard, locks the documents, and, when none has a chunk left, queues their storage cleanup and
+   * deletes them. With `resetConnectors`, the same transaction resets the connectors' listing
+   * cursors, so a run stopped partway never leaves a connector that would skip deleted documents.
    */
   deleteDocuments(
     knowledgeBaseId: string,
     documentIds: readonly string[],
-    requestId: string
+    requestId: string,
+    resetConnectors: boolean
   ): Promise<DeleteDocumentsOutcome>
   /** Pending storage cleanup events, counted up to `cap`. */
   pendingStorageCleanup(cap: number): Promise<number>
@@ -320,15 +327,20 @@ export async function deleteSearchIndexDocuments(
     ) {
       for (;;) {
         const deleted = await retry('delete chunk batch', () =>
-          store.deleteChunkBatch(documentIds, chunkBatchSize)
+          store.deleteChunkBatch(knowledgeBaseId, documentIds, chunkBatchSize)
         )
         summary.chunksDeleted += deleted
         if (deleted === 0) break
         await pause()
         if (deleted < chunkBatchSize) break
       }
       outcome = await retry('delete documents', () =>
-        store.deleteDocuments(knowledgeBaseId, documentIds, options.requestId)
+        store.deleteDocuments(
+          knowledgeBaseId,
+          documentIds,
+          options.requestId,
+          options.resetConnectors !== false
+        )
       )
     }
     if (outcome.kind === 'chunks-remain') {