diff --git a/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts b/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts index ea6bc0bc89a..33717fdcb39 100644 --- a/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts +++ b/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts @@ -328,23 +328,25 @@ export const DELETE = withRouteHandler(async (request: NextRequest, { params }: const { deletedDocs, docCount } = await db.transaction(async (tx) => { await tx.execute(sql`SELECT 1 FROM knowledge_connector WHERE id = ${connectorId} FOR UPDATE`) + // Includes pending-removal (tombstoned) docs — the connector is being + // deleted, so there's no future sync left to confirm or resurrect them. const docs = await tx .select({ id: document.id, fileUrl: document.fileUrl }) .from(document) - .where( - and( - eq(document.connectorId, connectorId), - isNull(document.archivedAt), - isNull(document.deletedAt) - ) - ) + .where(and(eq(document.connectorId, connectorId), isNull(document.archivedAt))) + const documentIds = docs.map((doc) => doc.id) if (deleteDocuments) { - const documentIds = docs.map((doc) => doc.id) if (documentIds.length > 0) { await tx.delete(embedding).where(inArray(embedding.documentId, documentIds)) await tx.delete(document).where(inArray(document.id, documentIds)) } + } else if (documentIds.length > 0) { + // Kept documents become normal standalone KB entries once their connector + // is gone — resurrect any pending-removal ones rather than leaving them + // invisible tombstones with no future sync left to ever confirm or + // resurrect them. + await tx.update(document).set({ deletedAt: null }).where(inArray(document.id, documentIds)) } const deletedConnectors = await tx diff --git a/apps/sim/connectors/google-sheets/google-sheets.test.ts b/apps/sim/connectors/google-sheets/google-sheets.test.ts index df1e6c898e4..66eb84aa434 100644 --- a/apps/sim/connectors/google-sheets/google-sheets.test.ts +++ b/apps/sim/connectors/google-sheets/google-sheets.test.ts @@ -116,40 +116,26 @@ describe('googleSheetsConnector trashed handling', () => { }) describe('listDocuments', () => { - it('returns an empty listing and confirms the empty result when the spreadsheet is trashed', async () => { + it('returns an empty listing when the spreadsheet is trashed', async () => { stubFetch({ drive: { status: 200, body: { trashed: true, modifiedTime: '2026-07-01T00:00:00.000Z' } }, }) - const syncContext: Record = {} - const result = await googleSheetsConnector.listDocuments( - ACCESS_TOKEN, - SOURCE_CONFIG, - undefined, - syncContext - ) + const result = await googleSheetsConnector.listDocuments(ACCESS_TOKEN, SOURCE_CONFIG) expect(result).toEqual({ documents: [], hasMore: false }) - expect(syncContext.sourceConfirmedEmpty).toBe(true) }) it('lists every tab when the trashed field is absent', async () => { stubFetch({ drive: { status: 200, body: { modifiedTime: '2026-07-01T00:00:00.000Z' } } }) - const syncContext: Record = {} - const result = await googleSheetsConnector.listDocuments( - ACCESS_TOKEN, - SOURCE_CONFIG, - undefined, - syncContext - ) + const result = await googleSheetsConnector.listDocuments(ACCESS_TOKEN, SOURCE_CONFIG) expect(result.documents.map((d) => d.externalId)).toEqual([ `${SPREADSHEET_ID}__sheet__0`, `${SPREADSHEET_ID}__sheet__7`, ]) expect(result.hasMore).toBe(false) - expect(syncContext.sourceConfirmedEmpty).toBeUndefined() }) it('lists every tab when trashed is explicitly false', async () => { @@ -162,20 +148,13 @@ describe('googleSheetsConnector trashed handling', () => { it('fails open and lists every tab when the Drive read fails', async () => { stubFetch({ drive: { status: 500, body: { error: 'backend error' } } }) - const syncContext: Record = {} - const result = await googleSheetsConnector.listDocuments( - ACCESS_TOKEN, - SOURCE_CONFIG, - undefined, - syncContext - ) + const result = await googleSheetsConnector.listDocuments(ACCESS_TOKEN, SOURCE_CONFIG) expect(result.documents.map((d) => d.externalId)).toEqual([ `${SPREADSHEET_ID}__sheet__0`, `${SPREADSHEET_ID}__sheet__7`, ]) - expect(syncContext.sourceConfirmedEmpty).toBeUndefined() }) it('fails open when the Drive body is not an object', async () => { @@ -185,14 +164,6 @@ describe('googleSheetsConnector trashed handling', () => { expect(result.documents).toHaveLength(2) }) - - it('does not throw when trashed and no syncContext is passed', async () => { - stubFetch({ drive: { status: 200, body: { trashed: true } } }) - - const result = await googleSheetsConnector.listDocuments(ACCESS_TOKEN, SOURCE_CONFIG) - - expect(result).toEqual({ documents: [], hasMore: false }) - }) }) describe('getDocument', () => { diff --git a/apps/sim/connectors/google-sheets/google-sheets.ts b/apps/sim/connectors/google-sheets/google-sheets.ts index 11657e7842e..746bdb2bfa9 100644 --- a/apps/sim/connectors/google-sheets/google-sheets.ts +++ b/apps/sim/connectors/google-sheets/google-sheets.ts @@ -253,7 +253,7 @@ export const googleSheetsConnector: ConnectorConfig = { accessToken: string, sourceConfig: Record, _cursor?: string, - syncContext?: Record + _syncContext?: Record ): Promise => { const spreadsheetId = (sourceConfig.spreadsheetId as string)?.trim() if (!spreadsheetId) { @@ -269,18 +269,14 @@ export const googleSheetsConnector: ConnectorConfig = { /** * A trashed spreadsheet is no longer current content, so it drops out of the - * listing and stops being re-indexed. Unlike an ordinary empty listing page - * (which could equally mean the source is unreachable), this is a direct, - * single-resource confirmation from the Drive API that the spreadsheet itself - * is gone — so `sourceConfirmedEmpty` tells the sync engine's zero-document - * guard it's safe to reconcile (purge the stored tabs) on this sync, rather - * than requiring a forced full resync. `validateConfig` reports the trashed - * state so the connector does not look healthy while serving tabs from a file - * its owner has thrown away. + * listing and stops being re-indexed. The sync engine reconciles its absence + * the same way it does for every connector: pending-removal on the first + * sync that doesn't see it, purged once a later sync confirms it's still + * gone. `validateConfig` reports the trashed state so the connector does not + * look healthy while serving tabs from a file its owner has thrown away. */ if (isTrashedDriveFile(driveMetadata)) { logger.info('Spreadsheet is in the Drive trash; listing no documents', { spreadsheetId }) - if (syncContext) syncContext.sourceConfirmedEmpty = true return { documents: [], hasMore: false } } diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts index bdd197df0a3..540ae694595 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts @@ -76,98 +76,236 @@ describe('shouldReconcileDeletions', () => { }) }) -describe('shouldSkipEmptyListing', () => { - it('does not skip when the listing is non-empty', async () => { - const { shouldSkipEmptyListing } = await import('@/lib/knowledge/connectors/sync-engine') +describe('shouldRunIncrementalSync', () => { + const lastSyncAt = '2026-07-01T00:00:00.000Z' - expect(shouldSkipEmptyListing(1, 5, undefined, {})).toBe(false) + it('runs incrementally when everything is eligible', async () => { + const { shouldRunIncrementalSync } = await import('@/lib/knowledge/connectors/sync-engine') + + expect( + shouldRunIncrementalSync(true, 'incremental', undefined, undefined, false, lastSyncAt) + ).toBe(true) }) - it('does not skip when there are no existing documents to lose', async () => { - const { shouldSkipEmptyListing } = await import('@/lib/knowledge/connectors/sync-engine') + it('never runs incrementally when the connector does not support it', async () => { + const { shouldRunIncrementalSync } = await import('@/lib/knowledge/connectors/sync-engine') - expect(shouldSkipEmptyListing(0, 0, undefined, {})).toBe(false) + expect( + shouldRunIncrementalSync(false, 'incremental', undefined, undefined, false, lastSyncAt) + ).toBe(false) }) - it('does not skip on a forced fullSync', async () => { - const { shouldSkipEmptyListing } = await import('@/lib/knowledge/connectors/sync-engine') + it('never runs incrementally when the connector is configured for full syncs', async () => { + const { shouldRunIncrementalSync } = await import('@/lib/knowledge/connectors/sync-engine') - expect(shouldSkipEmptyListing(0, 5, true, {})).toBe(false) + expect(shouldRunIncrementalSync(true, 'full', undefined, undefined, false, lastSyncAt)).toBe( + false + ) }) - it('skips by default on an empty listing with existing documents', async () => { - const { shouldSkipEmptyListing } = await import('@/lib/knowledge/connectors/sync-engine') + it('never runs incrementally on a forced fullSync or rehydrate', async () => { + const { shouldRunIncrementalSync } = await import('@/lib/knowledge/connectors/sync-engine') - expect(shouldSkipEmptyListing(0, 5, undefined, {})).toBe(true) - expect(shouldSkipEmptyListing(0, 5, undefined, undefined)).toBe(true) - expect(shouldSkipEmptyListing(0, 5, false, {})).toBe(true) + expect(shouldRunIncrementalSync(true, 'incremental', true, undefined, false, lastSyncAt)).toBe( + false + ) + expect(shouldRunIncrementalSync(true, 'incremental', undefined, true, false, lastSyncAt)).toBe( + false + ) }) - it('does not skip when the connector confirms the empty result against the source', async () => { - const { shouldSkipEmptyListing } = await import('@/lib/knowledge/connectors/sync-engine') + it('never runs incrementally before the first sync', async () => { + const { shouldRunIncrementalSync } = await import('@/lib/knowledge/connectors/sync-engine') - expect(shouldSkipEmptyListing(0, 5, undefined, { sourceConfirmedEmpty: true })).toBe(false) + expect(shouldRunIncrementalSync(true, 'incremental', undefined, undefined, false, null)).toBe( + false + ) }) - it('still skips when sourceConfirmedEmpty is falsy', async () => { - const { shouldSkipEmptyListing } = await import('@/lib/knowledge/connectors/sync-engine') + it('forces a full listing whenever pending-removal documents exist, so they get a resurrect-or-confirm decision', async () => { + const { shouldRunIncrementalSync } = await import('@/lib/knowledge/connectors/sync-engine') - expect(shouldSkipEmptyListing(0, 5, undefined, { sourceConfirmedEmpty: false })).toBe(true) + expect( + shouldRunIncrementalSync(true, 'incremental', undefined, undefined, true, lastSyncAt) + ).toBe(false) }) }) -describe('exceedsDeletionSafetyThreshold', () => { - it('does not block a small deletion, even above 50%', async () => { - const { exceedsDeletionSafetyThreshold } = await import( - '@/lib/knowledge/connectors/sync-engine' +describe('partitionSyncReconciliation', () => { + const live = (id: string, externalId: string | null = id) => ({ id, externalId }) + const noFailures = new Set() + + it('marks a live document missing from the listing as pending removal, not hard-deleted', async () => { + const { partitionSyncReconciliation } = await import('@/lib/knowledge/connectors/sync-engine') + + const result = partitionSyncReconciliation([live('a')], [], new Set(), noFailures, undefined) + + expect(result).toEqual({ resurrectIds: [], softDeleteIds: ['a'], hardDeleteIds: [] }) + }) + + it('hard-deletes a document already pending removal that is still absent', async () => { + const { partitionSyncReconciliation } = await import('@/lib/knowledge/connectors/sync-engine') + + const result = partitionSyncReconciliation([], [live('a')], new Set(), noFailures, undefined) + + expect(result).toEqual({ resurrectIds: [], softDeleteIds: [], hardDeleteIds: ['a'] }) + }) + + it('resurrects a pending-removal document that reappears in the listing', async () => { + const { partitionSyncReconciliation } = await import('@/lib/knowledge/connectors/sync-engine') + + const result = partitionSyncReconciliation( + [], + [live('a')], + new Set(['a']), + noFailures, + undefined ) - expect(exceedsDeletionSafetyThreshold(4, 5, undefined, {})).toBe(false) + expect(result).toEqual({ resurrectIds: ['a'], softDeleteIds: [], hardDeleteIds: [] }) }) - it('does not block a large deletion below 50%', async () => { - const { exceedsDeletionSafetyThreshold } = await import( - '@/lib/knowledge/connectors/sync-engine' + it('leaves a document untouched when it is still present in the listing', async () => { + const { partitionSyncReconciliation } = await import('@/lib/knowledge/connectors/sync-engine') + + const result = partitionSyncReconciliation( + [live('a')], + [], + new Set(['a']), + noFailures, + undefined ) - expect(exceedsDeletionSafetyThreshold(6, 20, undefined, {})).toBe(false) + expect(result).toEqual({ resurrectIds: [], softDeleteIds: [], hardDeleteIds: [] }) }) - it('blocks a deletion above both the ratio and count thresholds by default', async () => { - const { exceedsDeletionSafetyThreshold } = await import( - '@/lib/knowledge/connectors/sync-engine' + it('resurrects even on a forced fullSync', async () => { + const { partitionSyncReconciliation } = await import('@/lib/knowledge/connectors/sync-engine') + + const result = partitionSyncReconciliation([], [live('a')], new Set(['a']), noFailures, true) + + expect(result.resurrectIds).toEqual(['a']) + }) + + it('hard-deletes both live and pending-removal documents immediately on a forced fullSync', async () => { + const { partitionSyncReconciliation } = await import('@/lib/knowledge/connectors/sync-engine') + + const result = partitionSyncReconciliation( + [live('a')], + [live('b')], + new Set(), + noFailures, + true ) - expect(exceedsDeletionSafetyThreshold(10, 10, undefined, {})).toBe(true) - expect(exceedsDeletionSafetyThreshold(10, 10, undefined, undefined)).toBe(true) + expect(result.softDeleteIds).toEqual([]) + expect(result.hardDeleteIds.sort()).toEqual(['a', 'b']) }) - it('does not block on a forced fullSync', async () => { - const { exceedsDeletionSafetyThreshold } = await import( - '@/lib/knowledge/connectors/sync-engine' + it('handles a mixed batch of every outcome in one pass', async () => { + const { partitionSyncReconciliation } = await import('@/lib/knowledge/connectors/sync-engine') + + const result = partitionSyncReconciliation( + [live('kept'), live('newly-missing')], + [live('resurrected'), live('confirmed-gone')], + new Set(['kept', 'resurrected']), + noFailures, + undefined ) - expect(exceedsDeletionSafetyThreshold(10, 10, true, {})).toBe(false) + expect(result).toEqual({ + resurrectIds: ['resurrected'], + softDeleteIds: ['newly-missing'], + hardDeleteIds: ['confirmed-gone'], + }) }) - it('does not block when the connector confirms the deletion against the source', async () => { - const { exceedsDeletionSafetyThreshold } = await import( - '@/lib/knowledge/connectors/sync-engine' + it('ignores documents with a null externalId', async () => { + const { partitionSyncReconciliation } = await import('@/lib/knowledge/connectors/sync-engine') + + const result = partitionSyncReconciliation( + [live('a', null)], + [live('b', null)], + new Set(), + noFailures, + undefined ) - expect(exceedsDeletionSafetyThreshold(10, 10, undefined, { sourceConfirmedEmpty: true })).toBe( - false + expect(result).toEqual({ resurrectIds: [], softDeleteIds: [], hardDeleteIds: [] }) + }) + + it('does not resurrect a reappearing document whose content refresh failed', async () => { + const { partitionSyncReconciliation } = await import('@/lib/knowledge/connectors/sync-engine') + + const result = partitionSyncReconciliation( + [], + [live('a')], + new Set(['a']), + new Set(['a']), + undefined ) + + expect(result).toEqual({ resurrectIds: [], softDeleteIds: [], hardDeleteIds: [] }) }) - it('still blocks when sourceConfirmedEmpty is falsy', async () => { - const { exceedsDeletionSafetyThreshold } = await import( + it('still refuses to resurrect a failed refresh even on a forced fullSync', async () => { + const { partitionSyncReconciliation } = await import('@/lib/knowledge/connectors/sync-engine') + + const result = partitionSyncReconciliation( + [], + [live('a')], + new Set(['a']), + new Set(['a']), + true + ) + + expect(result.resurrectIds).toEqual([]) + }) + + it('resurrects the ones that succeeded while excluding the one that failed', async () => { + const { partitionSyncReconciliation } = await import('@/lib/knowledge/connectors/sync-engine') + + const result = partitionSyncReconciliation( + [], + [live('ok'), live('failed')], + new Set(['ok', 'failed']), + new Set(['failed']), + undefined + ) + + expect(result.resurrectIds).toEqual(['ok']) + }) +}) + +describe('filterStillOwnedReconciliationIds', () => { + it('keeps ids present in the ownership snapshot', async () => { + const { filterStillOwnedReconciliationIds } = await import( '@/lib/knowledge/connectors/sync-engine' ) - expect(exceedsDeletionSafetyThreshold(10, 10, undefined, { sourceConfirmedEmpty: false })).toBe( - true + const result = filterStillOwnedReconciliationIds(['a'], ['b'], ['c'], new Set(['a', 'b', 'c'])) + + expect(result).toEqual({ resurrectIds: ['a'], softDeleteIds: ['b'], hardDeleteIds: ['c'] }) + }) + + it('drops ids a concurrent connector-delete already detached', async () => { + const { filterStillOwnedReconciliationIds } = await import( + '@/lib/knowledge/connectors/sync-engine' ) + + const result = filterStillOwnedReconciliationIds(['a'], ['b'], ['c'], new Set(['a'])) + + expect(result).toEqual({ resurrectIds: ['a'], softDeleteIds: [], hardDeleteIds: [] }) + }) + + it('returns all-empty lists when nothing is still owned', async () => { + const { filterStillOwnedReconciliationIds } = await import( + '@/lib/knowledge/connectors/sync-engine' + ) + + const result = filterStillOwnedReconciliationIds(['a'], ['b'], ['c'], new Set()) + + expect(result).toEqual({ resurrectIds: [], softDeleteIds: [], hardDeleteIds: [] }) }) }) diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index 67ecfe374b1..65b0d04542f 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -245,44 +245,107 @@ export function shouldReconcileDeletions( } /** - * Decides whether a zero-document listing should skip deletion reconciliation. + * Decides whether a sync should use the connector's incremental listing. * - * An empty listing is normally indistinguishable from a provider outage, so - * reconciliation is skipped by default rather than risk wiping a knowledge base on - * a bad response. A connector can set `sourceConfirmedEmpty` on `syncContext` to - * vouch that it verified the empty result directly against the source — not - * merely an empty listing page — e.g. a single-resource connector confirming its - * one source item was trashed/removed via a dedicated metadata lookup. + * A pending-removal document only surfaces in an incremental listing if its + * content changed since last sync — an unchanged-but-still-present document + * never appears in an incremental delta at all, so it could never be + * resurrected and would stay tombstoned indefinitely on a connector that runs + * incrementally from here on. `hasTombstonedDocs` forces a full listing + * whenever any pending-removal document exists for this connector, so every + * one of them gets a real resurrect-or-confirm decision on this sync. */ -export function shouldSkipEmptyListing( - externalDocsCount: number, - existingDocsCount: number, +export function shouldRunIncrementalSync( + supportsIncrementalSync: boolean | undefined, + syncMode: string | null | undefined, fullSync: boolean | undefined, - syncContext: Record | undefined + rehydrate: boolean | undefined, + hasTombstonedDocs: boolean, + lastSyncAt: string | Date | null | undefined ): boolean { - if (externalDocsCount !== 0 || existingDocsCount === 0 || fullSync) return false - return !syncContext?.sourceConfirmedEmpty + return Boolean( + supportsIncrementalSync && + syncMode !== 'full' && + !fullSync && + !hasTombstonedDocs && + !rehydrate && + lastSyncAt != null + ) } +/** A stored document's identity, as read back for reconciliation. */ +type ReconciliationDoc = { id: string; externalId: string | null } + /** - * Decides whether a deletion should be blocked by the mass-deletion safety - * threshold (more than half of existing documents, over 5) instead of proceeding. + * Partitions a connector's stored documents against the current listing into + * the three reconciliation actions. * - * This guards against a connector-side bug or transient glitch producing a - * listing that looks mostly empty. `sourceConfirmedEmpty` bypasses it the same - * way it bypasses `shouldSkipEmptyListing` — the connector positively verified - * the deletion against the source rather than merely inferring it from a - * listing, so the extra caution this threshold provides doesn't apply. + * A document absent from a normal (non-fullSync) listing is never purged + * immediately — an empty or shrunken listing can equally mean a transient + * source outage, and a single bad observation must never cause an + * irreversible mass deletion. It is instead marked pending-removal + * (`softDeleteIds`), and only becomes eligible for hard deletion + * (`hardDeleteIds`) once a *later* sync confirms it's still absent — i.e. it + * was already pending-removal (`tombstonedDocs`) coming into this sync. A + * document that reappears while pending-removal is resurrected + * (`resurrectIds`) regardless of `fullSync`, since presence — unlike absence — + * is trustworthy evidence even from a partial listing. A document whose + * content refresh was attempted but failed (`failedExternalIds`) is excluded + * from resurrection even though it was seen — surfacing it now would show + * known-stale pre-tombstone content; it stays tombstoned for a later sync to + * retry. + * + * A forced `fullSync` is an explicit request to reconcile right now: it skips + * the grace period and purges everything absent in one pass. */ -export function exceedsDeletionSafetyThreshold( - removedCount: number, - existingCount: number, - fullSync: boolean | undefined, - syncContext: Record | undefined -): boolean { - if (fullSync || syncContext?.sourceConfirmedEmpty) return false - const deletionRatio = existingCount > 0 ? removedCount / existingCount : 0 - return deletionRatio > 0.5 && removedCount > 5 +export function partitionSyncReconciliation( + existingDocs: ReconciliationDoc[], + tombstonedDocs: ReconciliationDoc[], + seenExternalIds: Set, + failedExternalIds: Set, + fullSync: boolean | undefined +): { resurrectIds: string[]; softDeleteIds: string[]; hardDeleteIds: string[] } { + const resurrectIds = tombstonedDocs + .filter( + (d) => + d.externalId && seenExternalIds.has(d.externalId) && !failedExternalIds.has(d.externalId) + ) + .map((d) => d.id) + const liveMissingIds = existingDocs + .filter((d) => d.externalId && !seenExternalIds.has(d.externalId)) + .map((d) => d.id) + const tombstonedStillMissingIds = tombstonedDocs + .filter((d) => d.externalId && !seenExternalIds.has(d.externalId)) + .map((d) => d.id) + + if (fullSync) { + return { + resurrectIds, + softDeleteIds: [], + hardDeleteIds: [...liveMissingIds, ...tombstonedStillMissingIds], + } + } + return { resurrectIds, softDeleteIds: liveMissingIds, hardDeleteIds: tombstonedStillMissingIds } +} + +/** + * Re-filters the three reconciliation ID lists against a fresh ownership + * snapshot taken under the connector's `FOR UPDATE` lock, dropping any + * document a concurrent "delete connector, keep documents" request already + * detached (its `connectorId` no longer matches) since the lists were first + * computed. + */ +export function filterStillOwnedReconciliationIds( + resurrectIds: string[], + softDeleteIds: string[], + hardDeleteIds: string[], + stillOwnedIds: Set +): { resurrectIds: string[]; softDeleteIds: string[]; hardDeleteIds: string[] } { + return { + resurrectIds: resurrectIds.filter((id) => stillOwnedIds.has(id)), + softDeleteIds: softDeleteIds.filter((id) => stillOwnedIds.has(id)), + hardDeleteIds: hardDeleteIds.filter((id) => stillOwnedIds.has(id)), + } } /** @@ -473,18 +536,57 @@ export async function executeSync( let hasMore = true const syncContext: Record = { syncRunId: generateId() } + // Shared cutoff for both the tombstone-retry bound below and the stuck-document + // retry near the end of this sync — same RETRY_WINDOW_DAYS window, one computation. + const retryCutoff = new Date(Date.now() - RETRY_WINDOW_DAYS * 24 * 60 * 60 * 1000) + + /** + * Bounded to the same retry window as the stuck-document retry below: a + * document whose refresh keeps failing every sync (e.g. permanently + * oversized) would otherwise be a tombstone that never resolves, forcing a + * full listing — and its listing-time overhead — for this connector + * forever. Past the window, this connector stops forcing full syncs on its + * account; the document itself is unaffected and stays tombstoned either way. + * + * Known accepted trade-off: once past the window, a still-tombstoned + * document that's unchanged-but-genuinely-present at the source can only + * be resurrected by a full listing — and nothing here forces one anymore. + * On a connector that never runs a full sync again (persistent incremental + * syncMode, no manual full resync), that document stays correctly + * invisible (excluded everywhere by `isNull(deletedAt)`, so no + * search/billing/listing leakage) but unresolved indefinitely. This is + * deliberately not "fixed" by hard-deleting it after the window expires — + * that would delete a document we have no positive evidence is actually + * gone, reintroducing the exact risk this whole design exists to avoid. + */ + const hasTombstonedDocs = await db + .select({ id: document.id }) + .from(document) + .where( + and( + eq(document.connectorId, connectorId), + isNull(document.archivedAt), + isNotNull(document.deletedAt), + gt(document.deletedAt, retryCutoff) + ) + ) + .limit(1) + .then((rows) => rows.length > 0) + /** * Determine if this sync should be incremental. A `rehydrate` request forces a * full listing too: re-hydration must see *every* document (a container page can * be unchanged itself yet transclude a page that changed), and an incremental * listing would omit those unchanged containers, so they'd never be re-fetched. */ - const isIncremental = - connectorConfig.supportsIncrementalSync && - connector.syncMode !== 'full' && - !options?.fullSync && - !options?.rehydrate && - connector.lastSyncAt != null + const isIncremental = shouldRunIncrementalSync( + connectorConfig.supportsIncrementalSync, + connector.syncMode, + options?.fullSync, + options?.rehydrate, + hasTombstonedDocs, + connector.lastSyncAt + ) const lastSyncAt = isIncremental && connector.lastSyncAt ? new Date(connector.lastSyncAt) : undefined @@ -548,7 +650,7 @@ export async function executeSync( connectorId, }) - const [existingDocs, excludedDocs] = await Promise.all([ + const [existingDocs, tombstonedDocs, excludedDocs] = await Promise.all([ db .select({ id: document.id, @@ -563,63 +665,66 @@ export async function executeSync( isNull(document.deletedAt) ) ), + // Docs already marked pending-removal by a prior sync's reconciliation (see + // shouldReconcileDeletions below): absent from the source once, not yet + // absent twice in a row. Included in classification so a document that + // reappears is recognized as existing (resurrected) rather than re-added + // as a duplicate. db - .select({ externalId: document.externalId }) + .select({ + id: document.id, + externalId: document.externalId, + contentHash: document.contentHash, + deletedAt: document.deletedAt, + }) .from(document) .where( and( eq(document.connectorId, connectorId), - eq(document.userExcluded, true), isNull(document.archivedAt), - isNull(document.deletedAt) + isNotNull(document.deletedAt) ) ), - ]) - - const excludedExternalIds = new Set(excludedDocs.map((d) => d.externalId).filter(Boolean)) - - if ( - shouldSkipEmptyListing( - externalDocs.length, - existingDocs.length, - options?.fullSync, - syncContext - ) - ) { - logger.warn( - `Source returned 0 documents but ${existingDocs.length} exist — skipping reconciliation`, - { connectorId } - ) - - await completeSyncLog(syncLogId, 'completed', result) - - const now = new Date() - await db - .update(knowledgeConnector) - .set({ - status: 'active', - lastSyncAt: now, - lastSyncError: null, - nextSyncAt: calculateNextSyncTime(connector.syncIntervalMinutes), - consecutiveFailures: 0, - updatedAt: now, - }) + // Not filtered on deletedAt: a document can be both userExcluded and + // tombstoned (e.g. excluded via a bulk request that raced a sync marking + // it pending-removal). Excluding it here regardless of tombstone state + // keeps it short-circuited in the classification loop below instead of + // silently reappearing through the normal update/resurrect path. + db + .select({ externalId: document.externalId }) + .from(document) .where( and( - eq(knowledgeConnector.id, connectorId), - isNull(knowledgeConnector.archivedAt), - isNull(knowledgeConnector.deletedAt) + eq(document.connectorId, connectorId), + eq(document.userExcluded, true), + isNull(document.archivedAt) ) - ) + ), + ]) - return result - } + const excludedExternalIds = new Set(excludedDocs.map((d) => d.externalId).filter(Boolean)) - const existingByExternalId = new Map( - existingDocs.filter((d) => d.externalId !== null).map((d) => [d.externalId!, d]) + const priorByExternalId = new Map( + [...existingDocs, ...tombstonedDocs] + .filter((d) => d.externalId !== null) + .map((d) => [d.externalId!, d]) ) const seenExternalIds = new Set() + /** + * externalIds whose content was never verified as current: a hydration + * 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() const pendingOps: DocOp[] = [] for (const extDoc of externalDocs) { @@ -631,7 +736,7 @@ export async function executeSync( continue } - const existing = existingByExternalId.get(extDoc.externalId) + const existing = priorByExternalId.get(extDoc.externalId) const classification = classifyExternalDoc(extDoc, existing, forceRehydrate) switch (classification.type) { @@ -639,6 +744,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, }) @@ -650,6 +759,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 } @@ -704,15 +819,21 @@ export async function executeSync( }) } else if (op.type === 'update') { // Already-indexed file is kept as last-known-good (not downgraded), so it - // counts as unchanged rather than slipping past every result counter. + // counts as unchanged rather than slipping past every result counter. Not a + // verified refresh, though — see failedExternalIds below. result.docsUnchanged++ + failedExternalIds.add(op.extDoc.externalId) } return null } if (!fullDoc?.content.trim()) { // An empty re-fetch leaves an already-indexed update as last-known-good; count - // it as unchanged so the totals still reconcile with documents seen. - if (op.type === 'update') result.docsUnchanged++ + // it as unchanged so the totals still reconcile with documents seen. Not a + // verified refresh, though — see failedExternalIds below. + if (op.type === 'update') { + result.docsUnchanged++ + failedExternalIds.add(op.extDoc.externalId) + } return null } const hydratedHash = fullDoc.contentHash ?? op.extDoc.contentHash @@ -725,7 +846,7 @@ export async function executeSync( if ( op.type === 'update' && !forceRehydrate && - existingByExternalId.get(op.extDoc.externalId)?.contentHash === hydratedHash + priorByExternalId.get(op.extDoc.externalId)?.contentHash === hydratedHash ) { result.docsUnchanged++ return null @@ -745,13 +866,16 @@ export async function executeSync( }) ) - for (const outcome of hydrated) { + for (let i = 0; i < hydrated.length; i++) { + const outcome = hydrated[i] if (outcome.status === 'fulfilled' && outcome.value) { readyOps.push(outcome.value) } else if (outcome.status === 'rejected') { result.docsFailed++ + failedExternalIds.add(deferredOps[i].extDoc.externalId) logger.error('Failed to hydrate deferred document', { connectorId, + externalId: deferredOps[i].extDoc.externalId, error: getErrorMessage(outcome.reason), }) } @@ -814,6 +938,7 @@ export async function executeSync( else result.docsUpdated++ } else { result.docsFailed++ + failedExternalIds.add(batch[j].extDoc.externalId) logger.error('Failed to process document', { connectorId, externalId: batch[j].extDoc.externalId, @@ -841,35 +966,108 @@ export async function executeSync( } } - if (shouldReconcileDeletions(isIncremental, syncContext, options?.fullSync)) { - const removedIds = existingDocs - .filter((d) => d.externalId && !seenExternalIds.has(d.externalId)) - .map((d) => d.id) - - if (removedIds.length > 0) { - if ( - exceedsDeletionSafetyThreshold( - removedIds.length, - existingDocs.length, - options?.fullSync, - syncContext - ) - ) { - logger.warn( - `Skipping deletion of ${removedIds.length}/${existingDocs.length} docs — exceeds safety threshold. Trigger a full sync to force cleanup.`, - { - connectorId, - deletionRatio: Math.round((removedIds.length / existingDocs.length) * 100), - } - ) - } else { - await hardDeleteDocuments(removedIds, syncLogId) - result.docsDeleted += removedIds.length + const { resurrectIds, softDeleteIds, hardDeleteIds } = partitionSyncReconciliation( + existingDocs, + tombstonedDocs, + seenExternalIds, + failedExternalIds, + options?.fullSync + ) + + const reconcileDeletionsAllowed = shouldReconcileDeletions( + isIncremental, + syncContext, + options?.fullSync + ) + const gatedSoftDeleteIds = reconcileDeletionsAllowed ? softDeleteIds : [] + const gatedHardDeleteIds = reconcileDeletionsAllowed ? hardDeleteIds : [] + + const candidateIds = [ + ...new Set([...resurrectIds, ...gatedSoftDeleteIds, ...gatedHardDeleteIds]), + ] + + let safeResurrectIds: string[] = [] + let safeSoftDeleteIds: string[] = [] + let safeHardDeleteIds: string[] = [] + + if (candidateIds.length > 0) { + /** + * A concurrent "delete connector, keep documents" request detaches these + * same documents (connectorId set to NULL) under the same FOR UPDATE lock + * the DELETE route takes on this connector row. Taking that lock here + * serializes the two requests: whichever commits first wins, and the + * loser's re-check below sees the up-to-date connectorId and skips any + * document the other request already claimed — instead of resurrecting or + * deleting a document that another request just detached (and possibly + * already billed) as a standalone KB entry. + */ + await db.transaction(async (tx) => { + await tx.execute( + sql`SELECT 1 FROM knowledge_connector WHERE id = ${connectorId} FOR UPDATE` + ) + + const stillOwned = new Set( + ( + await tx + .select({ id: document.id }) + .from(document) + .where(and(inArray(document.id, candidateIds), eq(document.connectorId, connectorId))) + ).map((d) => d.id) + ) + + const stillOwnedResult = filterStillOwnedReconciliationIds( + resurrectIds, + gatedSoftDeleteIds, + gatedHardDeleteIds, + stillOwned + ) + safeResurrectIds = stillOwnedResult.resurrectIds + safeSoftDeleteIds = stillOwnedResult.softDeleteIds + safeHardDeleteIds = stillOwnedResult.hardDeleteIds + + /** + * A document reappearing at the source is trustworthy evidence on its + * own — unlike absence, presence never depends on the listing being + * complete — so resurrection runs unconditionally, even on an + * incremental or otherwise gated sync. + */ + if (safeResurrectIds.length > 0) { + await tx + .update(document) + .set({ deletedAt: null }) + .where(inArray(document.id, safeResurrectIds)) } - } + if (safeSoftDeleteIds.length > 0) { + await tx + .update(document) + .set({ deletedAt: new Date() }) + .where(inArray(document.id, safeSoftDeleteIds)) + } + }) + } + + if (safeResurrectIds.length > 0) { + logger.info( + `Resurrected ${safeResurrectIds.length} documents that reappeared at the source`, + { + connectorId, + } + ) + } + if (safeSoftDeleteIds.length > 0) { + logger.info( + `Marked ${safeSoftDeleteIds.length} documents pending removal — absent from source, confirming on next sync`, + { connectorId } + ) + } + if (safeHardDeleteIds.length > 0) { + // 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. + result.docsDeleted += await hardDeleteDocuments(safeHardDeleteIds, syncLogId, connectorId) } - // Check if connector/KB were deleted before retrying stuck documents const postBatchLiveness = await checkSyncLiveness(connectorId, connector.knowledgeBaseId) if (postBatchLiveness.connectorDeleted) { throw new ConnectorDeletedException(connectorId) @@ -885,7 +1083,6 @@ export async function executeSync( // abandoned (e.g. the Trigger.dev task process exited before processing completed). // Documents uploaded more than RETRY_WINDOW_DAYS ago are not retried. const staleProcessingCutoff = new Date(Date.now() - STALE_PROCESSING_MINUTES * 60 * 1000) - const retryCutoff = new Date(Date.now() - RETRY_WINDOW_DAYS * 24 * 60 * 60 * 1000) const stuckDocs = await db .select({ id: document.id, @@ -923,35 +1120,71 @@ export async function executeSync( logger.info(`Retrying ${stuckDocs.length} stuck documents`, { connectorId }) try { const stuckDocIds = stuckDocs.map((doc) => doc.id) + let retryDocs: typeof stuckDocs = [] + + /** + * Takes the same `knowledge_connector` FOR UPDATE lock the DELETE route + * takes before nulling connectorId on detached documents, so the two + * requests serialize instead of racing — a plain re-SELECT only + * narrows the window between the ownership check and these writes, it + * never closes it, since a concurrent detach can still commit in + * between. Embedding cleanup and the processing-state reset happen + * inside the same locked transaction so a document already claimed by + * a detach never gets its embeddings wiped or is reprocessed as if + * still connector-owned. + */ + await db.transaction(async (tx) => { + await tx.execute( + sql`SELECT 1 FROM knowledge_connector WHERE id = ${connectorId} FOR UPDATE` + ) - await db.delete(embedding).where(inArray(embedding.documentId, stuckDocIds)) + const stillOwnedIds = new Set( + ( + await tx + .select({ id: document.id }) + .from(document) + .where( + and(inArray(document.id, stuckDocIds), eq(document.connectorId, connectorId)) + ) + ).map((d) => d.id) + ) + retryDocs = stuckDocs.filter((doc) => stillOwnedIds.has(doc.id)) + + if (retryDocs.length > 0) { + const retryDocIds = retryDocs.map((doc) => doc.id) + + await tx.delete(embedding).where(inArray(embedding.documentId, retryDocIds)) + + await tx + .update(document) + .set({ + processingStatus: 'pending', + processingStartedAt: null, + processingCompletedAt: null, + processingError: null, + chunkCount: 0, + tokenCount: 0, + characterCount: 0, + }) + .where(inArray(document.id, retryDocIds)) + } + }) - await db - .update(document) - .set({ - processingStatus: 'pending', - processingStartedAt: null, - processingCompletedAt: null, - processingError: null, - chunkCount: 0, - tokenCount: 0, - characterCount: 0, - }) - .where(inArray(document.id, stuckDocIds)) - - await processDocumentsWithQueue( - stuckDocs.map((doc) => ({ - documentId: doc.id, - filename: doc.filename ?? 'document.txt', - fileUrl: doc.fileUrl ?? '', - fileSize: doc.fileSize ?? 0, - mimeType: doc.mimeType ?? 'text/plain', - })), - connector.knowledgeBaseId, - {}, - generateId(), - billingAttribution - ) + if (retryDocs.length > 0) { + await processDocumentsWithQueue( + retryDocs.map((doc) => ({ + documentId: doc.id, + filename: doc.filename ?? 'document.txt', + fileUrl: doc.fileUrl ?? '', + fileSize: doc.fileSize ?? 0, + mimeType: doc.mimeType ?? 'text/plain', + })), + connector.knowledgeBaseId, + {}, + generateId(), + billingAttribution + ) + } } catch (error) { logger.warn('Failed to enqueue stuck documents for reprocessing', { connectorId, @@ -1003,20 +1236,17 @@ export async function executeSync( logger.info('Connector deleted during sync, cleaning up', { connectorId }) try { + // Includes pending-removal (tombstoned) docs — the connector is gone, so + // there's no future sync left to confirm or resurrect them. const connectorDocs = await db .select({ id: document.id }) .from(document) - .where( - and( - eq(document.connectorId, connectorId), - isNull(document.archivedAt), - isNull(document.deletedAt) - ) - ) + .where(and(eq(document.connectorId, connectorId), isNull(document.archivedAt))) await hardDeleteDocuments( connectorDocs.map((doc) => doc.id), - syncLogId + syncLogId, + connectorId ) await completeSyncLog(syncLogId, 'failed', result, 'Connector deleted during sync') @@ -1293,7 +1523,6 @@ async function updateDocument( kbOwner: KnowledgeBaseOwner, sourceConfig?: Record ): Promise { - // Fetch old file URL before uploading replacement const existingRows = await db .select({ fileUrl: document.fileUrl }) .from(document) @@ -1342,12 +1571,21 @@ async function updateDocument( ...tagValues, processingStatus: 'pending', uploadedAt: new Date(), + // A tombstoned document reappearing with changed content is resurrected + // in the same write as its content update — otherwise reconciliation's + // separate resurrect step would clear deletedAt while this update, gated + // on deletedAt IS NULL, rejects the row and leaves stale content active. + deletedAt: null, }) .where( and( eq(document.id, existingDocId), - isNull(document.archivedAt), - isNull(document.deletedAt) + // A concurrent "delete connector, keep documents" request can null out + // connectorId between this sync's liveness check and this write. Without + // this check, that now-standalone document would still match on id alone + // and get overwritten with connector-sourced content post-detachment. + eq(document.connectorId, connectorId), + isNull(document.archivedAt) ) ) .returning({ id: document.id }) diff --git a/apps/sim/lib/knowledge/documents/service.ts b/apps/sim/lib/knowledge/documents/service.ts index 1df4cd960a0..1b8651a8424 100644 --- a/apps/sim/lib/knowledge/documents/service.ts +++ b/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 { 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 { +async function hardDeleteDocumentBatch( + documentIds: string[], + requestId: string, + expectedConnectorId?: string +): Promise { 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 @@ -2432,10 +2451,31 @@ async function hardDeleteDocumentBatch(documentIds: string[], requestId: string) } } - await tx.delete(embedding).where(inArray(embedding.documentId, existingIds)) + /** + * Re-verify `expectedConnectorId` here too, not only on the pre-transaction + * SELECT above — the billing lookups and KB locking between that SELECT + * and this delete are async and can span a concurrent "delete connector, + * keep documents" request that clears these rows' `connectorId` in + * between. Deleting a detached document's embeddings would corrupt its + * search index even if the document row itself were spared, so both the + * embedding delete and the document delete are scoped to this re-verified + * ID set rather than the stale `existingIds`. + */ + const stillTargetedIds = expectedConnectorId + ? ( + await tx + .select({ id: document.id }) + .from(document) + .where( + and(inArray(document.id, existingIds), eq(document.connectorId, expectedConnectorId)) + ) + ).map((d) => d.id) + : existingIds + + await tx.delete(embedding).where(inArray(embedding.documentId, stillTargetedIds)) const deletedRows = await tx .delete(document) - .where(inArray(document.id, existingIds)) + .where(inArray(document.id, stillTargetedIds)) .returning({ id: document.id }) const deletedIds = new Set(deletedRows.map((row) => row.id))