fix(knowledge): close remaining hard-delete race window, guard listing-time skip/drop resurrection

Two more real findings, both closing gaps in the previous round's fixes:

- The FOR UPDATE lock protected resurrect/soft-delete (applied inside
  the same transaction) but hardDeleteDocuments still ran after that
  transaction committed, using IDs snapshotted under the lock. A
  concurrent 'delete connector, keep documents' request could still
  detach those same documents in the gap between commit and the
  hardDeleteDocuments call. Added an optional expectedConnectorId
  parameter to hardDeleteDocuments/hardDeleteDocumentBatch — when
  provided, it re-verifies connectorId at the moment of the actual
  delete query, not just the caller's earlier snapshot. Every other
  caller is unaffected (parameter is optional, defaults to no filter).

- Two more listing-time paths could resurrect a tombstoned document
  without ever verifying its content: a listing-time skippedReason
  short-circuits classification straight to 'unchanged' before the
  hash comparison ever runs, and empty non-deferred content classifies
  as 'drop' unconditionally regardless of hash. Both are now added to
  failedExternalIds when reappearing on an existing (possibly
  tombstoned) document, same treatment as the deferred-hydration
  equivalents from the prior round.

Author: waleedlatif1

apps/sim/lib/knowledge/connectors/sync-engine.ts

@@ -689,14 +689,16 @@ export async function executeSync(
     const seenExternalIds = new Set<string>()
     /**
      * externalIds whose content was never verified as current: a hydration
-     * error, a rejected write, or a fulfilled-but-unusable hydration (skipped
-     * as oversized, or an empty re-fetch) that falls back to keeping the
-     * stored content as last-known-good. That fallback is fine for an
-     * already-visible document, but for a tombstoned one it means we still
-     * don't have confirmed-current content — so this excludes them from
-     * resurrection below: a tombstoned document whose refresh didn't actually
-     * land must stay tombstoned rather than come back visible while still
-     * serving stale pre-tombstone content.
+     * error, a rejected write, a fulfilled-but-unusable hydration (skipped as
+     * oversized, or an empty re-fetch), a listing-time skippedReason
+     * short-circuit, or empty non-deferred content (`drop`) — all fall back to
+     * either keeping the stored content as last-known-good or discarding the
+     * listing entry outright, without ever comparing or refreshing content.
+     * That's fine for an already-visible document, but for a tombstoned one it
+     * means we still don't have confirmed-current content — so this excludes
+     * them from resurrection below: a tombstoned document whose refresh didn't
+     * actually land must stay tombstoned rather than come back visible while
+     * still serving stale pre-tombstone content.
      */
     const failedExternalIds = new Set<string>()
 
@@ -718,6 +720,10 @@ export async function executeSync(
           pendingOps.push({ type: 'skip', extDoc })
           break
         case 'drop':
+          // Empty, non-deferred content is never usable. If this was a
+          // reappearing tombstoned document, its content was never verified as
+          // current — see failedExternalIds below.
+          if (existing) failedExternalIds.add(extDoc.externalId)
           logger.info(`Skipping empty document: ${extDoc.title}`, {
             externalId: extDoc.externalId,
           })
@@ -729,6 +735,12 @@ export async function executeSync(
           pendingOps.push({ type: 'update', existingId: classification.existingId, extDoc })
           break
         case 'unchanged':
+          // A listing-time skippedReason short-circuits classification before
+          // the hash comparison, so this is "kept as last-known-good", not a
+          // verified-unchanged match — same as the deferred-hydration
+          // equivalent above. A genuine hash match never sets skippedReason,
+          // so this only fires for the short-circuited case.
+          if (extDoc.skippedReason && existing) failedExternalIds.add(extDoc.externalId)
           result.docsUnchanged++
           break
       }
@@ -1019,7 +1031,11 @@ export async function executeSync(
       )
     }
     if (safeHardDeleteIds.length > 0) {
-      await hardDeleteDocuments(safeHardDeleteIds, syncLogId)
+      // Re-verifies connectorId once more at the moment of the actual delete
+      // query — the FOR UPDATE lock above only covers the window up to its
+      // own commit; this closes the remaining gap between that commit and
+      // this call.
+      await hardDeleteDocuments(safeHardDeleteIds, syncLogId, connectorId)
       result.docsDeleted += safeHardDeleteIds.length
     }
 
@@ -1166,7 +1182,8 @@ export async function executeSync(
 
         await hardDeleteDocuments(
           connectorDocs.map((doc) => doc.id),
-          syncLogId
+          syncLogId,
+          connectorId
         )
 
         await completeSyncLog(syncLogId, 'failed', result, 'Connector deleted during sync')

apps/sim/lib/knowledge/documents/service.ts

@@ -2325,7 +2325,17 @@ async function deleteDocumentsByLifecyclePolicy(
 
 export async function hardDeleteDocuments(
   documentIds: string[],
-  requestId: string
+  requestId: string,
+  /**
+   * When provided, re-verifies each document's connectorId still matches at
+   * the moment of the actual delete query — not just the caller's earlier
+   * snapshot. A caller (e.g. connector sync reconciliation) can compute this
+   * ID list well before the delete runs; a concurrent request that detaches
+   * these same documents from the connector in between (e.g. "delete
+   * connector, keep documents") would otherwise still have them purged here
+   * despite no longer belonging to the connector the caller reasoned about.
+   */
+  expectedConnectorId?: string
 ): Promise<number> {
   const ids = [...new Set(documentIds)]
   if (ids.length === 0) {
@@ -2336,7 +2346,8 @@ export async function hardDeleteDocuments(
   for (let offset = 0; offset < ids.length; offset += HARD_DELETE_DOCUMENT_BATCH_SIZE) {
     deletedCount += await hardDeleteDocumentBatch(
       ids.slice(offset, offset + HARD_DELETE_DOCUMENT_BATCH_SIZE),
-      requestId
+      requestId,
+      expectedConnectorId
     )
   }
   return deletedCount
@@ -2346,7 +2357,11 @@ export async function hardDeleteDocuments(
  * Hard-deletes one bounded metadata batch and applies every associated ledger
  * delta atomically.
  */
-async function hardDeleteDocumentBatch(documentIds: string[], requestId: string): Promise<number> {
+async function hardDeleteDocumentBatch(
+  documentIds: string[],
+  requestId: string,
+  expectedConnectorId?: string
+): Promise<number> {
   const ids = [...new Set(documentIds)]
   const documentsToDelete = await db
     .select({
@@ -2361,7 +2376,11 @@ async function hardDeleteDocumentBatch(documentIds: string[], requestId: string)
     })
     .from(document)
     .innerJoin(knowledgeBase, eq(document.knowledgeBaseId, knowledgeBase.id))
-    .where(inArray(document.id, ids))
+    .where(
+      expectedConnectorId
+        ? and(inArray(document.id, ids), eq(document.connectorId, expectedConnectorId))
+        : inArray(document.id, ids)
+    )
 
   if (documentsToDelete.length === 0) {
     return 0