fix(knowledge): tombstone and resurrect only under the lease

The document lifecycle's tombstone and resurrection writes ran outside the
lease guard, so a run reclaimed after its ACL transaction could hide a
document the replacement restored or expose one it removed. They now run
in the same lease-guarded transaction as every other member write, and a
run whose lease is lost while it disables itself ends as superseded rather
than rejecting.

Author: waleedlatif1

apps/sim/lib/knowledge/connectors/member-observations.ts

@@ -254,6 +254,8 @@ export async function applyMemberDocumentLifecycle(input: {
   knowledgeBaseId: string
   runId: string
   lease: Pick<SyncRunLease, 'beatIfDue'>
+  /** Runs the tombstone and resurrection writes only while the run still holds its lease. */
+  withLease: <T>(fn: (tx: DbOrTx) => Promise<T>) => Promise<T>
   /** External ids whose refresh did not land this run; withheld from resurrection. */
   failedExternalIds: ReadonlySet<string>
   /**
@@ -266,52 +268,55 @@ export async function applyMemberDocumentLifecycle(input: {
   const { connectorId, knowledgeBaseId, runId } = input
   const now = new Date()
 
-  const tombstoned = !input.allowRemoval
-    ? []
-    : await db
-        .update(document)
-        .set({ deletedAt: now })
-        .where(
-          and(
-            eq(document.connectorId, connectorId),
-            eq(document.userExcluded, false),
-            isNull(document.archivedAt),
-            isNull(document.deletedAt),
-            hasNoObservation()
-          )
-        )
-        .returning({ id: document.id })
-
-  const resurrectionCandidates = await db
-    .select({ id: document.id, externalId: document.externalId })
-    .from(document)
-    .where(
-      and(
-        eq(document.connectorId, connectorId),
-        isNull(document.archivedAt),
-        isNotNull(document.deletedAt),
-        hasObservation()
-      )
-    )
-  const resurrectIds = resurrectionCandidates
-    .filter((row) => !row.externalId || !input.failedExternalIds.has(row.externalId))
-    .map((row) => row.id)
-  const resurrected =
-    resurrectIds.length === 0
+  const { tombstoned, resurrected } = await input.withLease(async (tx) => {
+    const tombstoned = !input.allowRemoval
       ? []
-      : await db
+      : await tx
           .update(document)
-          .set({ deletedAt: null })
+          .set({ deletedAt: now })
           .where(
             and(
-              inArray(document.id, resurrectIds),
               eq(document.connectorId, connectorId),
+              eq(document.userExcluded, false),
               isNull(document.archivedAt),
-              isNotNull(document.deletedAt)
+              isNull(document.deletedAt),
+              hasNoObservation()
             )
           )
           .returning({ id: document.id })
 
+    const resurrectionCandidates = await tx
+      .select({ id: document.id, externalId: document.externalId })
+      .from(document)
+      .where(
+        and(
+          eq(document.connectorId, connectorId),
+          isNull(document.archivedAt),
+          isNotNull(document.deletedAt),
+          hasObservation()
+        )
+      )
+    const resurrectIds = resurrectionCandidates
+      .filter((row) => !row.externalId || !input.failedExternalIds.has(row.externalId))
+      .map((row) => row.id)
+    const resurrected =
+      resurrectIds.length === 0
+        ? []
+        : await tx
+            .update(document)
+            .set({ deletedAt: null })
+            .where(
+              and(
+                inArray(document.id, resurrectIds),
+                eq(document.connectorId, connectorId),
+                isNull(document.archivedAt),
+                isNotNull(document.deletedAt)
+              )
+            )
+            .returning({ id: document.id })
+    return { tombstoned, resurrected }
+  })
+
   const purgeCutoff = new Date(now.getTime() - MEMBER_TOMBSTONE_PURGE_DAYS * 24 * 60 * 60 * 1000)
   const purgeCandidates = input.allowRemoval
     ? await db

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

@@ -1539,6 +1539,7 @@ export async function executeMemberSync(
       knowledgeBaseId: connector.knowledgeBaseId,
       runId,
       lease: run.lease,
+      withLease: (fn) => withMemberLease(run, fn),
       failedExternalIds: state.failedExternalIds,
       allowRemoval: (listed?.count ?? 0) > 0,
     })
@@ -1590,7 +1591,16 @@ export async function executeMemberSync(
       return skipped(result, 'connector_deleted_during_sync')
     }
     if (error instanceof MemberBindingGoneError) {
-      await disableMemberSync(run, error.message)
+      try {
+        await disableMemberSync(run, error.message)
+      } catch (disableError) {
+        if (!(disableError instanceof SyncLockLostException)) throw disableError
+        logger.warn('Member sync abandoned — lock was reclaimed before it could be disabled', {
+          connectorId,
+          runId,
+        })
+        return skipped(result, 'sync_superseded')
+      }
       return { ...skipped(result, 'connector_not_syncable'), error: error.message }
     }