opensource/sim
3
0
Fork 0
mirror of https://github.com/simstudioai/sim.git synced 2026-09-24 15:45:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(knowledge): replace deletion-safety heuristics with a two-phase tombstone (#5884)

Browse Source 
* fix(knowledge): replace deletion-safety heuristics with a two-phase tombstone

PR #5883 merged an interim fix (sourceConfirmedEmpty bypassing two
static safety heuristics) before this follow-up redesign was ready.
This supersedes that approach with a properly general fix, matching
how production sync systems (Entra Connect, SCIM/Entra ID
deprovisioning, Cassandra/Couchbase tombstones) handle this exact
problem: never let a single observation trigger an irreversible mass
deletion, no matter how confident the signal looks.

A document missing from a normal sync's listing is now soft-deleted
(marked pending-removal) rather than hard-deleted immediately. It's
only actually purged once a *later* sync confirms it's still absent.
If it reappears in between, it's resurrected automatically — this
self-heals a transient outage or a bad API response without needing
to distinguish 'real' emptiness from 'ambiguous' emptiness at all,
which is what the removed heuristics were trying (and failing) to do
from a single observation.

This removes shouldSkipEmptyListing, exceedsDeletionSafetyThreshold,
and sourceConfirmedEmpty entirely — Google Sheets no longer needs a
connector-specific bypass flag, since a genuinely trashed spreadsheet
now reconciles through the exact same general path as every other
connector, with no special-casing and no new misuse surface for
future connectors. A forced fullSync still purges everything absent
in one pass, preserving the existing 'trigger a full sync to force
cleanup' escape hatch.

Uses the existing (previously unused for individual documents)
document.deletedAt column as the tombstone marker — no schema
migration required. shouldReconcileDeletions (the isIncremental /
listingCapped / listingTruncated gate) is unchanged; it still governs
whether reconciliation may run at all. Resurrection runs
unconditionally even when that gate is closed, since presence is
trustworthy evidence regardless of whether the listing was complete.

* fix(knowledge): resurrect atomically on content update, sweep tombstoned docs on connector teardown

Two real gaps found by review, both stemming from the same root
cause: other code paths assumed deletedAt IS NULL means 'the only
real rows' and were never updated for the new tombstone semantics.

- updateDocument's guard required deletedAt IS NULL, so a tombstoned
  document reappearing with CHANGED content failed its content
  update (rejected by the guard) while the separate resurrect step
  still cleared deletedAt regardless — the document became active
  again but kept serving stale pre-tombstone content. Fixed by
  clearing deletedAt as part of the same update statement and
  dropping the guard, so content and resurrection land atomically.
- Both connector-teardown cleanup paths (the ConnectorDeletedException
  handler in sync-engine.ts, and the connector DELETE API route) only
  swept documents with deletedAt IS NULL, so pending-removal documents
  escaped cleanup entirely and were orphaned once their connector was
  gone. Fixed by including tombstoned docs in both sweeps — there's no
  future sync left to confirm or resurrect them once the connector
  itself is deleted.

* fix(knowledge): don't resurrect a tombstoned document whose content refresh failed

Critical race found by adversarial audit: resurrectIds was derived
purely from 'externalId was seen in the listing', independent of
whether the paired content update actually succeeded. If updateDocument
threw (hydration failure, storage upload failure, or any other
transient error) or the deferred-hydration fetch itself failed, the
document's row was never touched — yet the separate unconditional
resurrect step still cleared deletedAt for it anyway, reproducing the
exact bug this PR fixes (visible again, serving stale pre-tombstone
content), just triggered by a failed write instead of a gated one.

Track externalIds whose refresh attempt failed (hydration rejection or
write rejection) and exclude them from partitionSyncReconciliation's
resurrectIds. A failed refresh leaves the document tombstoned as-is —
not soft-deleted, not hard-deleted, not resurrected — so a later sync
gets a clean retry instead of the row landing in an inconsistent state
either way.

* fix(knowledge): exclude fulfilled-but-unverified hydration outcomes from resurrection too

Both bots independently found a third instance of the same bug class:
when deferred hydration for an update fulfills but has no usable
content (skipped as oversized, or an empty re-fetch), the code falls
back to keeping the stored content as last-known-good and counts it
unchanged — correct for an already-visible document, but for a
tombstoned one it means content was never actually verified as
current. That fallback wasn't added to failedExternalIds, so
reconciliation still resurrected it with stale pre-tombstone content
despite hydration never actually confirming anything.

Fixed by adding both fallback branches to failedExternalIds, same as
the rejected-promise cases. Verified across every connector's
skippedReason call site that this only ever happens inside
getDocument (the deferred hydration path already covered here) —
no connector sets skippedReason directly in listDocuments, so there's
no equivalent listing-time gap to fix.

* fix(knowledge): force a full listing when pending-removal documents exist

A subtler instance of the same bug class, this time architectural
rather than a code-path gap: an incremental listing only includes
documents whose content changed since the last sync. A tombstoned
document that's still genuinely present at the source but unchanged
would never appear in an incremental delta at all, so it could never
be resurrected — and on a connector that runs incrementally from here
on (its normal syncMode), it would stay tombstoned indefinitely with
no self-correcting path, only a manual full resync.

Added shouldRunIncrementalSync (extracted as a testable pure function,
matching shouldReconcileDeletions' existing pattern) and a cheap
existence check for any pending-removal document on this connector.
Whenever one exists, this sync forces a full listing instead of an
incremental one, guaranteeing every tombstoned document gets a real
resurrect-or-confirm decision. This only affects which listing mode
runs — it doesn't touch options.fullSync, so the deletion-safety grace
period for other, unrelated documents in the same sync is unaffected.

* fix(knowledge): bound tombstone-forced full syncs, resurrect kept docs on connector delete

Two more real findings from this review round:

- A document whose refresh keeps failing every sync (e.g. permanently
  oversized) never resurrects and never hard-deletes (it's present in
  the listing, just unreadable) — correct on its own, but it also never
  stops being counted by hasTombstonedDocs, so it would force a full
  listing for this connector forever, permanently disabling incremental
  sync on account of one stuck document. Bounded the check to the same
  RETRY_WINDOW_DAYS already used for the stuck-document retry sweep
  below: past the window, this connector stops forcing full syncs on
  the stuck document's account. The document itself is unaffected —
  it stays tombstoned either way, matching the existing 'last-known-good
  forever' tolerance this codebase already accepts for any document
  whose hydration keeps failing.

- The connector-DELETE route's deleteDocuments=false path (kept docs)
  counted tombstoned documents but never resolved them one way or the
  other. With the connector gone, there's no future sync left to ever
  confirm or resurrect them, so they'd become permanent invisible
  orphans holding storage forever. Since 'kept' documents become normal
  standalone KB entries once detached from their connector, resurrect
  any pending-removal ones as part of that transition — consistent
  with what happens to their non-tombstoned sibling documents.

* fix(knowledge): serialize reconciliation writes against a concurrent connector delete

An independent adversarial audit (not just Greptile/Cursor) found the
one genuinely critical gap 6 rounds of bot review missed: resurrect/
soft-delete/hard-delete writes applied raw document IDs snapshotted at
the top of the sync, with no re-check immediately before the write. A
connector-DELETE request choosing to keep documents detaches them
(connectorId set to NULL) via the exact same FOR UPDATE lock on the
connector row that this fix now also takes before applying any
reconciliation write — serializing the two: whichever transaction
commits first wins, and the loser's re-check sees the up-to-date
connectorId and skips any document the other request already claimed.
Without this, a sync racing a 'delete connector, keep documents'
request could silently resurrect-then-strand or soft/hard-delete a
document the user explicitly chose to keep, with a secondary effect of
misclassifying it for storage-billing decrement (which keys off
whether connectorId is still set).

Also tightened the excludedDocs query: it previously required
deletedAt IS NULL, so a document that was both userExcluded and
tombstoned (reachable via excludeConnectorDocuments, which has no
deletedAt filter) fell out of the exclusion set and could be silently
un-excluded and re-indexed on reappearing. Dropped that requirement so
userExcluded is honored regardless of tombstone state, consistent with
how existingDocs/tombstonedDocs are already merged for classification.

Documented (not code-changed) the remaining lower-severity finding: a
document that outlives the 7-day hasTombstonedDocs bound on a
persistently-incremental connector can stay unresolved indefinitely.
Deliberately not hard-deleting it after the window expires — that
would delete a document with no positive evidence it's actually gone,
reintroducing the exact risk this whole design exists to avoid. It's
already fully excluded from search/billing/listings either way, so
this is an accepted, bounded, orphaned-row trade-off, not a
correctness or security issue.

* 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.

* refactor(knowledge): dedupe retry-cutoff computation, extract ownership-filter helper

/simplify pass: hoist the shared RETRY_WINDOW_DAYS cutoff into one computation
reused by both the tombstone-retry bound and the stuck-document retry query,
and pull the FOR-UPDATE-lock reconciliation's ID filtering into a pure,
directly-unit-tested filterStillOwnedReconciliationIds function matching this
file's existing convention for its other decision logic.

* fix(knowledge): re-verify connectorId at the actual hard-delete, fix docsDeleted count

Cursor findings: expectedConnectorId was only checked on the pre-transaction
SELECT in hardDeleteDocumentBatch, not on the DELETE itself — the billing
lookups and KB locking in between are async and can span a concurrent
"delete connector, keep documents" request, so the delete (and its embedding
cleanup) now re-verifies against a fresh in-transaction snapshot instead of
the stale existingIds. Also fixed docsDeleted to use hardDeleteDocuments'
actual returned count instead of the pre-filter candidate count, so a sync
log no longer overreports deletions that expectedConnectorId skipped.

/cleanup: dropped two comments that only restated the line below them.

* fix(knowledge): re-verify connectorId in updateDocument's atomic write

Greptile P1: updateDocument's content-update/resurrect write only checked
document.id and archivedAt, never connectorId — despite connectorId being
a parameter — so a document a concurrent "delete connector, keep documents"
request already detached could still be matched, resurrected, and
overwritten with connector-sourced content after the connector was deleted.
Adds the same connectorId ownership check already used by the reconciliation
transaction and hardDeleteDocuments in this PR.

* fix(knowledge): close connectorId race in the stuck-document retry path

Final independent audit: the stuck-document retry block selected candidate
IDs filtered by connectorId, then reset their processing state and deleted
their embeddings using that stale ID set with no re-check — the same
SELECT-then-write race already patched in updateDocument and
hardDeleteDocumentBatch. A concurrent "delete connector, keep documents"
request could null out connectorId in between, so this now re-verifies
ownership immediately before the embedding delete/document update/re-enqueue
and only acts on documents still owned by this connector.

* fix(knowledge): lock the connector row for stuck-doc retry ownership too

Cursor + Greptile (same root cause, two reports): the previous round's fix
re-checked connectorId via a separate SELECT before the embedding delete and
processing-state reset, but a bare re-SELECT only narrows a TOCTOU window, it
never closes it — a concurrent "delete connector, keep documents" request
could still commit its detach in between. Wraps the ownership re-check and
both writes in a transaction that takes the same knowledge_connector FOR
UPDATE lock the DELETE route takes before nulling connectorId, so the two
requests serialize instead of racing, matching the pattern already used by
the reconciliation transaction elsewhere in this PR.
This commit is contained in:
Waleed
2026-07-23 11:59:24 -07:00
committed by GitHub
parent 5adf68f332
commit c8fea6bd3b
6 changed files with 666 additions and 281 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts
+10 -8
View File
@@ -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
apps/sim/connectors/google-sheets/google-sheets.test.ts
+4 -33
View File
@@ -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<string, unknown> = {}
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<string, unknown> = {}
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<string, unknown> = {}
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', () => {
apps/sim/connectors/google-sheets/google-sheets.ts
+6 -10
View File
@@ -253,7 +253,7 @@ export const googleSheetsConnector: ConnectorConfig = {
accessToken: string,
sourceConfig: Record<string, unknown>,
_cursor?: string,
syncContext?: Record<string, unknown>
_syncContext?: Record<string, unknown>
): Promise<ExternalDocumentList> => {
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 }
}
apps/sim/lib/knowledge/connectors/sync-engine.test.ts
+215 -77
View File
@@ -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)
})
it('skips by default on an empty listing with existing documents', async () => {
const { shouldSkipEmptyListing } = 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)
})
it('does not skip when the connector confirms the empty result against the source', async () => {
const { shouldSkipEmptyListing } = await import('@/lib/knowledge/connectors/sync-engine')
expect(shouldSkipEmptyListing(0, 5, undefined, { sourceConfirmedEmpty: true })).toBe(false)
})
it('still skips when sourceConfirmedEmpty is falsy', async () => {
const { shouldSkipEmptyListing } = await import('@/lib/knowledge/connectors/sync-engine')
expect(shouldSkipEmptyListing(0, 5, undefined, { sourceConfirmedEmpty: false })).toBe(true)
})
})
describe('exceedsDeletionSafetyThreshold', () => {
it('does not block a small deletion, even above 50%', async () => {
const { exceedsDeletionSafetyThreshold } = await import(
'@/lib/knowledge/connectors/sync-engine'
)
expect(exceedsDeletionSafetyThreshold(4, 5, undefined, {})).toBe(false)
})
it('does not block a large deletion below 50%', async () => {
const { exceedsDeletionSafetyThreshold } = await import(
'@/lib/knowledge/connectors/sync-engine'
)
expect(exceedsDeletionSafetyThreshold(6, 20, undefined, {})).toBe(false)
})
it('blocks a deletion above both the ratio and count thresholds by default', async () => {
const { exceedsDeletionSafetyThreshold } = await import(
'@/lib/knowledge/connectors/sync-engine'
)
expect(exceedsDeletionSafetyThreshold(10, 10, undefined, {})).toBe(true)
expect(exceedsDeletionSafetyThreshold(10, 10, undefined, undefined)).toBe(true)
})
it('does not block on a forced fullSync', async () => {
const { exceedsDeletionSafetyThreshold } = await import(
'@/lib/knowledge/connectors/sync-engine'
)
expect(exceedsDeletionSafetyThreshold(10, 10, true, {})).toBe(false)
})
it('does not block when the connector confirms the deletion against the source', async () => {
const { exceedsDeletionSafetyThreshold } = await import(
'@/lib/knowledge/connectors/sync-engine'
)
expect(exceedsDeletionSafetyThreshold(10, 10, undefined, { sourceConfirmedEmpty: true })).toBe(
expect(shouldRunIncrementalSync(true, 'full', undefined, undefined, false, lastSyncAt)).toBe(
false
)
})
it('still blocks when sourceConfirmedEmpty is falsy', async () => {
const { exceedsDeletionSafetyThreshold } = await import(
it('never runs incrementally on a forced fullSync or rehydrate', async () => {
const { shouldRunIncrementalSync } = await import('@/lib/knowledge/connectors/sync-engine')
expect(shouldRunIncrementalSync(true, 'incremental', true, undefined, false, lastSyncAt)).toBe(
false
)
expect(shouldRunIncrementalSync(true, 'incremental', undefined, true, false, lastSyncAt)).toBe(
false
)
})
it('never runs incrementally before the first sync', async () => {
const { shouldRunIncrementalSync } = await import('@/lib/knowledge/connectors/sync-engine')
expect(shouldRunIncrementalSync(true, 'incremental', undefined, undefined, false, null)).toBe(
false
)
})
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(
shouldRunIncrementalSync(true, 'incremental', undefined, undefined, true, lastSyncAt)
).toBe(false)
})
})
describe('partitionSyncReconciliation', () => {
const live = (id: string, externalId: string | null = id) => ({ id, externalId })
const noFailures = new Set<string>()
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(result).toEqual({ resurrectIds: ['a'], softDeleteIds: [], hardDeleteIds: [] })
})
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(result).toEqual({ resurrectIds: [], softDeleteIds: [], hardDeleteIds: [] })
})
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(result.softDeleteIds).toEqual([])
expect(result.hardDeleteIds.sort()).toEqual(['a', 'b'])
})
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(result).toEqual({
resurrectIds: ['resurrected'],
softDeleteIds: ['newly-missing'],
hardDeleteIds: ['confirmed-gone'],
})
})
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(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 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: [] })
})
})
apps/sim/lib/knowledge/connectors/sync-engine.ts
+385 -147
View File
@@ -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<string, unknown> | 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 }
/**
* Partitions a connector's stored documents against the current listing into
* the three reconciliation actions.
*
* 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 partitionSyncReconciliation(
existingDocs: ReconciliationDoc[],
tombstonedDocs: ReconciliationDoc[],
seenExternalIds: Set<string>,
failedExternalIds: Set<string>,
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 }
}
/**
* Decides whether a deletion should be blocked by the mass-deletion safety
* threshold (more than half of existing documents, over 5) instead of proceeding.
*
* 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.
* 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 exceedsDeletionSafetyThreshold(
removedCount: number,
existingCount: number,
fullSync: boolean | undefined,
syncContext: Record<string, unknown> | undefined
): boolean {
if (fullSync || syncContext?.sourceConfirmedEmpty) return false
const deletionRatio = existingCount > 0 ? removedCount / existingCount : 0
return deletionRatio > 0.5 && removedCount > 5
export function filterStillOwnedReconciliationIds(
resurrectIds: string[],
softDeleteIds: string[],
hardDeleteIds: string[],
stillOwnedIds: Set<string>
): { 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<string, unknown> = { 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,6 +665,31 @@ 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({
id: document.id,
externalId: document.externalId,
contentHash: document.contentHash,
deletedAt: document.deletedAt,
})
.from(document)
.where(
and(
eq(document.connectorId, connectorId),
isNull(document.archivedAt),
isNotNull(document.deletedAt)
)
),
// 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)
@@ -570,56 +697,34 @@ export async function executeSync(
and(
eq(document.connectorId, connectorId),
eq(document.userExcluded, true),
isNull(document.archivedAt),
isNull(document.deletedAt)
isNull(document.archivedAt)
)
),
])
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,
})
.where(
and(
eq(knowledgeConnector.id, connectorId),
isNull(knowledgeConnector.archivedAt),
isNull(knowledgeConnector.deletedAt)
)
)
return result
}
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<string>()
/**
* 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<string>()
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)
const { resurrectIds, softDeleteIds, hardDeleteIds } = partitionSyncReconciliation(
existingDocs,
tombstonedDocs,
seenExternalIds,
failedExternalIds,
options?.fullSync
)
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 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 = []
await db.delete(embedding).where(inArray(embedding.documentId, stuckDocIds))
/**
* 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
.update(document)
.set({
processingStatus: 'pending',
processingStartedAt: null,
processingCompletedAt: null,
processingError: null,
chunkCount: 0,
tokenCount: 0,
characterCount: 0,
})
.where(inArray(document.id, 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))
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) {
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))
}
})
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<string, unknown>
): Promise<DocumentData> {
// 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 })
apps/sim/lib/knowledge/documents/service.ts
+46 -6
View File
@@ -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
@@ -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))