From 5cb37e8693f3289229a687cc273bcf183c323ada Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 21 Aug 2026 19:50:38 -0700 Subject: [PATCH 1/5] fix(kb): let the server say a connector sync is queued MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The connector chip inferred "a sync is coming" from `createdAt` inside a 2-minute window, because nothing on the row distinguished a queued sync from an idle connector until a worker took the lock. The guess was wrong under queue backlog and under client clock skew, and it forced a pile of client state to stand in for it. Adds `pending`, written as the sync is handed to the queue and cleared when a worker takes the lock or the hand-off is found to have been lost. It is a phase of the same lock `syncing` holds, so it opens the lease and takes an ownership token the same way — the lease is what the scheduler ages a stranded queue entry against (`updatedAt` cannot serve: a pending connector is still editable, so any unrelated write would renew the recovery it should trigger), and the token is what proves a late release belongs to this dispatch. Deletes the 2-minute window, the in-flight id sets, the 5-minute cooldown timers and the forced re-render they needed. The cooldown lived in a ref inside a modal, so it evaporated whenever the modal closed; the disable now comes from durable server state and is shared across tabs. Also fixes, all found while tracing the lifecycle: - An on-demand sync on a paused or disabled connector silently resumed it for good. Nothing could put the pause back: success writes `active`, a lost queue entry writes `error`, and the due-sweep keeps syncing that. Refused. - A failed hand-off no longer advances the connector's auto-disable breaker. A queue outage would otherwise increment every connector in the fleet until they all disabled themselves for a fault that was never theirs. - Manual sync on an established connector gave no feedback at all: the poll only ran while the predicate matched, which it never did. - Four over-broad invalidations that refetched every cached chunk page and chunk search in a base when one connector document was excluded. - The dead-process reporter re-sent a PATCH per stale document on every poll. --- apps/docs/openapi-v2-knowledge.json | 8 +- .../knowledge/connectors/sync/route.test.ts | 73 +++++- .../api/knowledge/connectors/sync/route.ts | 198 ++++++++++----- .../[workspaceId]/knowledge/[id]/base.tsx | 84 +++++-- .../connectors-section/connectors-section.tsx | 124 +++------- apps/sim/hooks/kb/use-knowledge.ts | 22 +- apps/sim/hooks/queries/kb/connectors.test.ts | 227 +++++++++++++++++- apps/sim/hooks/queries/kb/connectors.ts | 215 +++++++++++++---- .../sim/hooks/queries/utils/knowledge-keys.ts | 10 +- .../lib/api/contracts/knowledge/connectors.ts | 3 +- apps/sim/lib/api/contracts/v2/knowledge.ts | 4 +- .../lib/knowledge/connectors/queue.test.ts | 150 +++++++++++- apps/sim/lib/knowledge/connectors/queue.ts | 144 ++++++++++- .../lib/knowledge/connectors/sync-engine.ts | 2 +- apps/sim/lib/knowledge/constants.ts | 9 + .../knowledge/documents/processing-claim.ts | 3 +- .../orchestration/connectors.test.ts | 27 +++ .../lib/knowledge/orchestration/connectors.ts | 50 +++- packages/db/schema.ts | 11 + packages/sim-cli/src/generated/v2-api.ts | 8 +- 20 files changed, 1109 insertions(+), 263 deletions(-) diff --git a/apps/docs/openapi-v2-knowledge.json b/apps/docs/openapi-v2-knowledge.json index 79634b38de6..40fc4e450f6 100644 --- a/apps/docs/openapi-v2-knowledge.json +++ b/apps/docs/openapi-v2-knowledge.json @@ -3450,8 +3450,8 @@ }, "status": { "type": "string", - "enum": ["active", "paused", "syncing", "error", "disabled"], - "description": "Current connector state." + "enum": ["active", "paused", "pending", "syncing", "error", "disabled"], + "description": "Current connector state. `pending` means a sync is queued but not yet running." }, "lastSyncAt": { "anyOf": [ @@ -3840,8 +3840,8 @@ }, "status": { "type": "string", - "enum": ["active", "paused", "syncing", "error", "disabled"], - "description": "Current connector state." + "enum": ["active", "paused", "pending", "syncing", "error", "disabled"], + "description": "Current connector state. `pending` means a sync is queued but not yet running." }, "lastSyncAt": { "anyOf": [ diff --git a/apps/sim/app/api/knowledge/connectors/sync/route.test.ts b/apps/sim/app/api/knowledge/connectors/sync/route.test.ts index a26b34469c2..2c8a6c2982b 100644 --- a/apps/sim/app/api/knowledge/connectors/sync/route.test.ts +++ b/apps/sim/app/api/knowledge/connectors/sync/route.test.ts @@ -109,6 +109,22 @@ function whereForUpdate(index: number): unknown { return dbChainMockFns.where.mock.calls[index][0] } +/** + * Position of the update targeting a given table, resolved by table rather than + * hardcoded: the tick runs several updates and a new one inserted between them + * would otherwise silently re-point every later assertion at the wrong chain. + */ +function updateIndexFor(table: unknown): number { + const index = dbChainMockFns.update.mock.calls.findIndex((call) => call[0] === table) + expect(index).toBeGreaterThanOrEqual(0) + return index +} + +/** The sync-log sweep's `.where()` condition, whichever chain it ran as. */ +function syncLogSweepWhere(): unknown { + return whereForUpdate(updateIndexFor(schemaMock.knowledgeConnectorSyncLog)) +} + beforeEach(() => { vi.clearAllMocks() resetDbChainMock() @@ -256,14 +272,14 @@ describe('connector sync scheduler stale-lock reaper', () => { it('closes orphaned sync-log rows still marked started', async () => { await runTickRecovering(['connector-1', 'connector-2']) - expect(dbChainMockFns.update.mock.calls[1][0]).toBe(schemaMock.knowledgeConnectorSyncLog) + const logUpdateIndex = updateIndexFor(schemaMock.knowledgeConnectorSyncLog) - const payload = setPayloadForUpdate(1) + const payload = setPayloadForUpdate(logUpdateIndex) expect(payload.status).toBe('failed') expect(renderedSql(payload.completedAt)).toContain('now()') expect(payload.errorMessage).toBe('Sync timed out (stale lock recovered)') - const where = dbChainMockFns.where.mock.calls[1][0] + const where = syncLogSweepWhere() expect( hasMockCondition( where, @@ -284,7 +300,7 @@ describe('connector sync scheduler stale-lock reaper', () => { /** The `NOT EXISTS` liveness fragment the sweep's WHERE carries. */ function sweepLivenessFragment(): MockSqlFragment { - const where = dbChainMockFns.where.mock.calls[1][0] + const where = syncLogSweepWhere() const fragment = flattenMockConditions(where).find( (node: MockCondition) => typeof node.toSQL === 'function' ) @@ -370,17 +386,56 @@ describe('connector sync scheduler stale-lock reaper', () => { expect(response.status).toBe(200) - const logUpdateIndex = dbChainMockFns.update.mock.calls.findIndex( - (call) => call[0] === schemaMock.knowledgeConnectorSyncLog + expect(setPayloadForUpdate(updateIndexFor(schemaMock.knowledgeConnectorSyncLog)).status).toBe( + 'failed' ) - expect(logUpdateIndex).toBeGreaterThanOrEqual(0) - expect(setPayloadForUpdate(logUpdateIndex).status).toBe('failed') + }) + + it('recovers connectors whose queued sync was never started', async () => { + await runTickRecovering(['connector-1']) + + /** Located by its `status = 'pending'` predicate, not by position in the tick. */ + const pendingIndex = dbChainMockFns.update.mock.calls.findIndex((call, index) => { + if (call[0] !== schemaMock.knowledgeConnector) return false + return hasMockCondition( + whereForUpdate(index), + (node: MockCondition) => + node.type === 'eq' && + node.left === schemaMock.knowledgeConnector.status && + node.right === 'pending' + ) + }) + expect(pendingIndex).toBeGreaterThanOrEqual(0) + + /** + * Ages against the lease, not `updatedAt`: a pending connector is still + * editable, and `updatedAt` moves on every unrelated write, so using it + * would let a config edit defer the recovery indefinitely — the bug the + * lease column was introduced to close for `syncing`. + */ + const pendingCutoff = flattenMockConditions(whereForUpdate(pendingIndex)).find( + (node: MockCondition) => typeof node.toSQL === 'function' + ) as unknown as MockSqlFragment | undefined + expect(pendingCutoff?.toSQL().sql).toBe('? <= ?') + expectLeaseExpression(pendingCutoff?.values[0]) + expect((pendingCutoff?.values[1] as { value: Date }).value).toEqual(EXPECTED_STALE_CUTOFF) + + /** Re-enters the shared failure ladder rather than re-queueing every tick. */ + const payload = setPayloadForUpdate(pendingIndex) + expect(renderedSql(payload.status)).toContain('disabled') + expect(renderedSql(payload.consecutiveFailures)).toBe('COALESCE(?, 0) + 1') + + /** + * Reports a lost hand-off, not a timeout: nothing ran, so the stale-lock + * wording would describe a run that never existed. + */ + expect(asFragment(payload.lastSyncError).values).toContain('Sync was queued but never started') }) it('never scopes the sync-log sweep to a connector id', async () => { await runTickRecovering(['connector-1']) - const where = dbChainMockFns.where.mock.calls[1][0] + const where = syncLogSweepWhere() /** * Checks every position, not just `column`. `eq()` builds `{left, right}` diff --git a/apps/sim/app/api/knowledge/connectors/sync/route.ts b/apps/sim/app/api/knowledge/connectors/sync/route.ts index c008f0fed73..f58f8f83bd0 100644 --- a/apps/sim/app/api/knowledge/connectors/sync/route.ts +++ b/apps/sim/app/api/knowledge/connectors/sync/route.ts @@ -32,6 +32,14 @@ const DISPATCH_CONCURRENCY = 10 const STALE_LOCK_ERROR_MESSAGE = 'Sync timed out (stale lock recovered)' +/** + * A connector left `pending` past the TTL — its sync was queued but no worker + * ever took the lock, so the hand-off was lost (the process died between the + * two writes, or the queued run was dropped). Distinct from the stale-lock + * message because nothing timed out: the sync never started. + */ +const LOST_DISPATCH_ERROR_MESSAGE = 'Sync was queued but never started' + /** * How long the connector holding the lock has gone without proving it is alive. * @@ -57,8 +65,8 @@ function syncLockLease(): SQL { * breaker and this SQL breaker cannot drift into two different messages for one * verdict. */ -function reclaimedError(): SQL { - return sql`CASE WHEN COALESCE(${knowledgeConnector.consecutiveFailures}, 0) + 1 >= ${MAX_CONSECUTIVE_FAILURES} THEN ${CONNECTOR_AUTO_DISABLED_ERROR} ELSE ${STALE_LOCK_ERROR_MESSAGE} END` +function reclaimedError(message: string): SQL { + return sql`CASE WHEN COALESCE(${knowledgeConnector.consecutiveFailures}, 0) + 1 >= ${MAX_CONSECUTIVE_FAILURES} THEN ${CONNECTOR_AUTO_DISABLED_ERROR} ELSE ${message} END` } /** @@ -123,6 +131,22 @@ function reclaimedNextSyncAt(): SQL { return sql`CASE WHEN COALESCE(${knowledgeConnector.consecutiveFailures}, 0) + 1 >= ${MAX_CONSECUTIVE_FAILURES} THEN NULL ELSE now() + LEAST((COALESCE(${knowledgeConnector.consecutiveFailures}, 0) + 1) * ${CONNECTOR_FAILURE_BACKOFF_STEP_MINUTES}, ${CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES}) * INTERVAL '1 minute' END` } +/** + * The write shared by both reclaims: a connector that stopped making progress + * re-enters the failure ladder. Factored so the two callers cannot drift into + * different ladders for the same verdict — the same reason + * {@link reclaimedError} takes the message rather than hardcoding it. + */ +function reclaimPayload(message: string) { + return { + status: reclaimedStatus(), + lastSyncError: reclaimedError(message), + nextSyncAt: reclaimedNextSyncAt(), + consecutiveFailures: reclaimedFailureCount(), + updatedAt: sql`now()`, + } +} + /** * Cron endpoint that checks for connectors due for sync and dispatches sync jobs. * Should be called every 5 minutes by an external cron service. @@ -141,29 +165,115 @@ export const GET = withRouteHandler(async (request: NextRequest) => { const staleCutoff = new Date(now.getTime() - CONNECTOR_SYNC_STALE_LOCK_TTL_MS) - const recoveredConnectors = await db - .update(knowledgeConnector) - .set({ - status: reclaimedStatus(), - lastSyncError: reclaimedError(), - nextSyncAt: reclaimedNextSyncAt(), - consecutiveFailures: reclaimedFailureCount(), - // Releases the reclaimed run's ownership token so its terminal write can - // no longer match, even before a replacement takes the lock, and closes - // its lease so a re-locked row starts from a fresh one. - syncLockToken: null, - syncLockLeaseAt: null, - updatedAt: sql`now()`, - }) - .where( - and( - eq(knowledgeConnector.status, 'syncing'), - sql`${syncLockLease()} <= ${sql.param(staleCutoff, knowledgeConnector.syncLockLeaseAt)}`, - isNull(knowledgeConnector.archivedAt), - isNull(knowledgeConnector.deletedAt) + /** + * The three recovery passes target disjoint row sets — a held-but-silent + * lock, a queue entry that never became one, and a sync-log row orphaned by + * a killed run — and none reads another's result, so they go out together + * rather than as three serialized round trips. + * + * `logRowNotHeldByLiveRun` is the one apparent coupling and it is benign: + * it spares a log row only while its connector's lease is still live, and + * every row the lock reclaim targets has an expired lease, so the sweep + * reaches the same verdict against either snapshot. + */ + const [recoveredConnectors, recoveredPendingConnectors, closedSyncLogs] = await Promise.all([ + db + .update(knowledgeConnector) + .set({ + ...reclaimPayload(STALE_LOCK_ERROR_MESSAGE), + /** + * Releases the reclaimed run's ownership token so its terminal write + * can no longer match, even before a replacement takes the lock, and + * closes its lease so a re-locked row starts from a fresh one. + */ + syncLockToken: null, + syncLockLeaseAt: null, + }) + .where( + and( + eq(knowledgeConnector.status, 'syncing'), + sql`${syncLockLease()} <= ${sql.param(staleCutoff, knowledgeConnector.syncLockLeaseAt)}`, + isNull(knowledgeConnector.archivedAt), + isNull(knowledgeConnector.deletedAt) + ) ) - ) - .returning({ id: knowledgeConnector.id }) + .returning({ id: knowledgeConnector.id }), + /** + * Recovers connectors whose queued sync was never picked up. + * + * `pending` is written just before the hand-off to the queue, so a row that + * is still `pending` past the TTL means no worker ever took the lock: the + * process died between the two writes, or the queued run was dropped. Left + * alone the connector would sit `pending` forever — the stale-lock reclaim + * above only looks at `syncing` rows, and the due-sweep below only at + * `active`/`error`. + * + * Flipped to `error` rather than straight back to `active` so it re-enters + * through the same failure ladder as any other unsuccessful sync: repeated + * lost dispatches back off and eventually disable, instead of re-queueing + * every tick forever. + * + * Ages against {@link syncLockLease}, the same expression the stale-lock + * pass reads, because `markSyncPending` opens the lease when it queues. + * `updatedAt` would be wrong here for exactly the reason the lease column + * exists: a `pending` connector is still editable, so every unrelated write + * to the row would renew the recovery it is meant to trigger — a config + * edit on a stuck connector could defer it forever. + */ + db + .update(knowledgeConnector) + .set({ + ...reclaimPayload(LOST_DISPATCH_ERROR_MESSAGE), + /** Releases the queue entry's token so a late hand-off cannot match it. */ + syncLockToken: null, + syncLockLeaseAt: null, + }) + .where( + and( + eq(knowledgeConnector.status, 'pending'), + sql`${syncLockLease()} <= ${sql.param(staleCutoff, knowledgeConnector.syncLockLeaseAt)}`, + isNull(knowledgeConnector.archivedAt), + isNull(knowledgeConnector.deletedAt) + ) + ) + .returning({ id: knowledgeConnector.id }), + /** + * Closes sync-log rows left `started` by a killed run. Nothing else ever + * reconciles them, and `loadPreviousListingObservation` reads only + * `completed` rows, so a never-closed run silently ages out the previous + * observation it should have provided. + * + * Deliberately independent of this tick's reclaims rather than scoped to + * them. A row orphaned before this shipped — or by a transient failure of + * this very statement — belongs to a connector already flipped out of + * `syncing`, so it would never appear in a future reclaim batch and would + * stay stranded forever. Keying off the row's own `startedAt` instead makes + * the sweep self-healing and lets it drain the existing backlog. + * + * Age alone does not prove a run is dead: the in-process fallback path has + * no duration cap, so a large self-hosted sync can genuinely still be + * working past the TTL. `logRowNotHeldByLiveRun` is what makes this safe — + * a run whose lock is still being heartbeated is spared regardless of age. + * The age predicate is also per-row on `startedAt`, so a fresh run's log row + * can never be caught by it, even on a connector whose previous run is being + * reclaimed in this same tick. + */ + db + .update(knowledgeConnectorSyncLog) + .set({ + status: 'failed', + completedAt: sql`now()`, + errorMessage: STALE_LOCK_ERROR_MESSAGE, + }) + .where( + and( + eq(knowledgeConnectorSyncLog.status, 'started'), + lte(knowledgeConnectorSyncLog.startedAt, staleCutoff), + logRowNotHeldByLiveRun(staleCutoff) + ) + ) + .returning({ id: knowledgeConnectorSyncLog.id }), + ]) if (recoveredConnectors.length > 0) { logger.warn( @@ -172,42 +282,12 @@ export const GET = withRouteHandler(async (request: NextRequest) => { ) } - /** - * Closes sync-log rows left `started` by a killed run. Nothing else ever - * reconciles them, and `loadPreviousListingObservation` reads only - * `completed` rows, so a never-closed run silently ages out the previous - * observation it should have provided. - * - * Deliberately independent of this tick's reclaims rather than scoped to - * them. A row orphaned before this shipped — or by a transient failure of - * this very statement — belongs to a connector already flipped out of - * `syncing`, so it would never appear in a future reclaim batch and would - * stay stranded forever. Keying off the row's own `startedAt` instead makes - * the sweep self-healing and lets it drain the existing backlog. - * - * Age alone does not prove a run is dead: the in-process fallback path has - * no duration cap, so a large self-hosted sync can genuinely still be - * working past the TTL. `logRowNotHeldByLiveRun` is what makes this safe — - * a run whose lock is still being heartbeated is spared regardless of age. - * The age predicate is also per-row on `startedAt`, so a fresh run's log row - * can never be caught by it, even on a connector whose previous run is being - * reclaimed in this same tick. - */ - const closedSyncLogs = await db - .update(knowledgeConnectorSyncLog) - .set({ - status: 'failed', - completedAt: sql`now()`, - errorMessage: STALE_LOCK_ERROR_MESSAGE, - }) - .where( - and( - eq(knowledgeConnectorSyncLog.status, 'started'), - lte(knowledgeConnectorSyncLog.startedAt, staleCutoff), - logRowNotHeldByLiveRun(staleCutoff) - ) + if (recoveredPendingConnectors.length > 0) { + logger.warn( + `[${requestId}] Recovered ${recoveredPendingConnectors.length} connectors whose queued sync was never started`, + { ids: recoveredPendingConnectors.map((c) => c.id) } ) - .returning({ id: knowledgeConnectorSyncLog.id }) + } if (closedSyncLogs.length > 0) { logger.warn(`[${requestId}] Closed ${closedSyncLogs.length} orphaned connector sync log(s)`) diff --git a/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/base.tsx b/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/base.tsx index 27ea4a08bed..1a29f058115 100644 --- a/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/base.tsx +++ b/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/base.tsx @@ -41,7 +41,12 @@ import { format } from 'date-fns' import { useParams, useRouter } from 'next/navigation' import { useQueryState, useQueryStates } from 'nuqs' import { usePostHog } from 'posthog-js/react' -import { ALL_TAG_SLOTS, type AllTagSlot, getFieldTypeForSlot } from '@/lib/knowledge/constants' +import { + ALL_TAG_SLOTS, + type AllTagSlot, + getFieldTypeForSlot, + KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS, +} from '@/lib/knowledge/constants' import type { DocumentSortField, SortOrder } from '@/lib/knowledge/documents/types' import { type FilterFieldType, getOperatorsForFieldType } from '@/lib/knowledge/filters/types' import type { DocumentData } from '@/lib/knowledge/types' @@ -95,11 +100,16 @@ import { useUserPermissionsContext } from '@/app/workspace/[workspaceId]/provide import { useContextMenu } from '@/app/workspace/[workspaceId]/w/components/sidebar/hooks' import { BrandIcon } from '@/blocks/brand-icon' import { CONNECTOR_META_REGISTRY } from '@/connectors/registry' -import { useKnowledgeBase, useKnowledgeBaseDocuments } from '@/hooks/kb/use-knowledge' +import { + hasProcessingDocuments, + useKnowledgeBase, + useKnowledgeBaseDocuments, +} from '@/hooks/kb/use-knowledge' import { type TagDefinition, useKnowledgeBaseTagDefinitions, } from '@/hooks/kb/use-knowledge-base-tag-definitions' +import type { ConnectorData } from '@/hooks/queries/kb/connectors' import { isConnectorSyncingOrPending, useConnectorList } from '@/hooks/queries/kb/connectors' import type { DocumentTagFilter } from '@/hooks/queries/kb/knowledge' import { @@ -117,8 +127,28 @@ import { useUrlSort } from '@/hooks/use-url-sort' const logger = createLogger('KnowledgeBase') +/** + * Identifies one processing *run*, not one document. + * + * Keying on the attempt's start time makes the reported-set self-invalidating: + * a document that is retried gets a new `processingStartedAt`, so a later stall + * is reportable again without the set needing to be pruned. + */ +function deadProcessKey(doc: Pick) { + return `${doc.id}:${doc.processingStartedAt ?? ''}` +} + const DOCUMENTS_PER_PAGE = 50 +/** Stable identity so an absent connector list does not re-fire list-dependent effects. */ +const EMPTY_CONNECTORS: ConnectorData[] = [] + +/** Cadence while a document is still indexing — its own status is what moves. */ +const PROCESSING_POLL_INTERVAL_MS = 3000 + +/** Slower cadence while only a connector sync is running: rows arrive in batches. */ +const CONNECTOR_SYNC_DOCUMENT_POLL_INTERVAL_MS = 5000 + const STATUS_FILTER_OPTIONS: ChipDropdownOption[] = [ { value: 'all', label: 'All' }, { value: 'enabled', label: 'Enabled' }, @@ -397,7 +427,8 @@ export function KnowledgeBase({ refresh: refreshKnowledgeBase, } = useKnowledgeBase(id) - const { data: connectors = [], isLoading: isLoadingConnectors } = useConnectorList(id) + const { data: connectors = EMPTY_CONNECTORS, isLoading: isLoadingConnectors } = + useConnectorList(id) const hasSyncingConnectors = connectors.some(isConnectorSyncingOrPending) const hasSyncingConnectorsRef = useRef(hasSyncingConnectors) hasSyncingConnectorsRef.current = hasSyncingConnectors @@ -408,7 +439,6 @@ export function KnowledgeBase({ isLoading: isLoadingDocuments, isPlaceholderData: isPlaceholderDocuments, error: documentsError, - hasProcessingDocuments, updateDocument, refreshDocuments, } = useKnowledgeBaseDocuments(id, { @@ -419,11 +449,8 @@ export function KnowledgeBase({ sortOrder: sortDirection as SortOrder, refetchInterval: (data) => { if (isDeleting) return false - const hasPending = data?.documents?.some( - (doc) => doc.processingStatus === 'pending' || doc.processingStatus === 'processing' - ) - if (hasPending) return 3000 - if (hasSyncingConnectorsRef.current) return 5000 + if (hasProcessingDocuments(data?.documents ?? [])) return PROCESSING_POLL_INTERVAL_MS + if (hasSyncingConnectorsRef.current) return CONNECTOR_SYNC_DOCUMENT_POLL_INTERVAL_MS return false }, enabledFilter: enabledFilter, @@ -479,20 +506,27 @@ export function KnowledgeBase({ const totalPages = Math.ceil(pagination.total / pagination.limit) /** - * Checks for documents with stale processing states and marks them as failed + * Processing runs already reported as timed out. + * + * The list below polls every few seconds while anything is processing, and + * each poll hands this effect a new array. Without this the same stale + * document is re-reported on every tick until the server's new status comes + * back — one redundant write per poll, per open tab. */ + const reportedDeadProcessesRef = useRef | null>(null) + const checkForDeadProcesses = useCallback( (docsToCheck: DocumentData[]) => { - const now = new Date() - const DEAD_PROCESS_THRESHOLD_MS = 600 * 1000 // 10 minutes + const reported = (reportedDeadProcessesRef.current ??= new Set()) + const nowMs = Date.now() const staleDocuments = docsToCheck.filter((doc) => { - if (doc.processingStatus !== 'processing' || !doc.processingStartedAt) { - return false - } - - const processingDuration = now.getTime() - new Date(doc.processingStartedAt).getTime() - return processingDuration > DEAD_PROCESS_THRESHOLD_MS + if (doc.processingStatus !== 'processing' || !doc.processingStartedAt) return false + if (reported.has(deadProcessKey(doc))) return false + return ( + nowMs - new Date(doc.processingStartedAt).getTime() > + KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS + ) }) if (staleDocuments.length === 0) return @@ -500,6 +534,7 @@ export function KnowledgeBase({ logger.warn(`Found ${staleDocuments.length} documents with dead processes`) staleDocuments.forEach((doc) => { + reported.add(deadProcessKey(doc)) updateDocumentMutation( { knowledgeBaseId: id, @@ -512,6 +547,8 @@ export function KnowledgeBase({ `Successfully marked dead process as failed for document: ${doc.filename}` ) }, + /** Retried on the next poll rather than left silently unreported. */ + onError: () => reported.delete(deadProcessKey(doc)), } ) }) @@ -520,10 +557,8 @@ export function KnowledgeBase({ ) useEffect(() => { - if (hasProcessingDocuments) { - checkForDeadProcesses(documents) - } - }, [hasProcessingDocuments, documents, checkForDeadProcesses]) + checkForDeadProcesses(documents) + }, [documents, checkForDeadProcesses]) const handleToggleEnabled = (docId: string) => { const document = documents.find((doc) => doc.id === docId) @@ -1073,6 +1108,7 @@ export function KnowledgeBase({ {connectors.map((connector) => { const def = CONNECTOR_META_REGISTRY[connector.connectorType] const ConnectorIcon = def?.icon + const syncInFlight = isConnectorSyncingOrPending(connector) return ( @@ -464,12 +410,7 @@ function ConnectorCard({ disabled={syncDisabled} onClick={() => onSync(false)} > - + @@ -496,11 +437,8 @@ function ConnectorCard({ variant='ghost' className={CONNECTOR_ACTION_BUTTON_CLASSES} onClick={onTogglePause} - disabled={isUpdating} > - {isUpdating ? ( - - ) : connector.status === 'paused' || connector.status === 'disabled' ? ( + {connector.status === 'paused' || connector.status === 'disabled' ? ( ) : ( diff --git a/apps/sim/hooks/kb/use-knowledge.ts b/apps/sim/hooks/kb/use-knowledge.ts index cb20889e09f..30f94c98691 100644 --- a/apps/sim/hooks/kb/use-knowledge.ts +++ b/apps/sim/hooks/kb/use-knowledge.ts @@ -57,6 +57,21 @@ export function useDocument(knowledgeBaseId: string, documentId: string) { } } +/** Stable identity so an absent page does not re-fire callers' document effects. */ +const EMPTY_DOCUMENTS: DocumentData[] = [] + +/** + * Whether any of these documents still has indexing work outstanding. + * + * Exported so a caller driving its own poll cadence reads the same rule this + * hook does rather than repeating the status literals. + */ +export function hasProcessingDocuments(documents: Pick[]) { + return documents.some( + (doc) => doc.processingStatus === 'pending' || doc.processingStatus === 'processing' + ) +} + /** * Hook to fetch and manage documents for a knowledge base * Uses React Query as single source of truth @@ -118,7 +133,7 @@ export function useKnowledgeBaseDocuments( } ) - const documents = query.data?.documents ?? [] + const documents = query.data?.documents ?? EMPTY_DOCUMENTS const pagination = query.data?.pagination ?? { total: 0, limit: requestLimit, @@ -126,10 +141,6 @@ export function useKnowledgeBaseDocuments( hasMore: false, } - const hasProcessingDocs = documents.some( - (doc) => doc.processingStatus === 'pending' || doc.processingStatus === 'processing' - ) - const refreshDocuments = useCallback(async () => { await queryClient.invalidateQueries({ queryKey: knowledgeKeys.documents(knowledgeBaseId, paramsKey), @@ -158,7 +169,6 @@ export function useKnowledgeBaseDocuments( isFetching: query.isFetching, isPlaceholderData: query.isPlaceholderData, error: query.error ? getErrorMessage(query.error) : null, - hasProcessingDocuments: hasProcessingDocs, refreshDocuments, updateDocument, } diff --git a/apps/sim/hooks/queries/kb/connectors.test.ts b/apps/sim/hooks/queries/kb/connectors.test.ts index 14d7ead56ea..7fe550c006f 100644 --- a/apps/sim/hooks/queries/kb/connectors.test.ts +++ b/apps/sim/hooks/queries/kb/connectors.test.ts @@ -7,23 +7,240 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' const mocks = vi.hoisted(() => ({ requestJson: vi.fn(), useInfiniteQuery: vi.fn(), + useQuery: vi.fn(), + useMutation: vi.fn(), + cancelQueries: vi.fn(), + getQueryData: vi.fn(), + setQueryData: vi.fn(), + invalidateQueries: vi.fn(), })) vi.mock('@tanstack/react-query', () => ({ keepPreviousData: Symbol('keepPreviousData'), useInfiniteQuery: mocks.useInfiniteQuery, - useMutation: vi.fn(), - useQuery: vi.fn(), - useQueryClient: vi.fn(() => ({ invalidateQueries: vi.fn() })), + useMutation: mocks.useMutation, + useQuery: mocks.useQuery, + useQueryClient: vi.fn(() => ({ + cancelQueries: mocks.cancelQueries, + getQueryData: mocks.getQueryData, + setQueryData: mocks.setQueryData, + invalidateQueries: mocks.invalidateQueries, + })), })) vi.mock('@/lib/api/client/request', () => ({ requestJson: mocks.requestJson, })) -import { listKnowledgeConnectorDocumentsContract } from '@/lib/api/contracts/knowledge' +import { + type ConnectorData, + listKnowledgeConnectorDocumentsContract, +} from '@/lib/api/contracts/knowledge' import { MAX_KNOWLEDGE_CONNECTOR_DOCUMENT_PAGE_SIZE } from '@/lib/knowledge/constants' -import { useConnectorDocuments } from '@/hooks/queries/kb/connectors' +import { + CONNECTOR_SYNC_POLL_INTERVAL_MS, + connectorKeys, + isConnectorSyncingOrPending, + useConnectorDetail, + useConnectorDocuments, + useConnectorList, + useTriggerSync, +} from '@/hooks/queries/kb/connectors' + +const KB_ID = 'kb-1' + +function makeConnector(overrides: Partial = {}): ConnectorData { + return { + id: 'connector-1', + knowledgeBaseId: KB_ID, + connectorType: 'hubspot', + credentialId: 'credential-1', + sourceConfig: {}, + syncMode: 'full', + syncIntervalMinutes: 1440, + status: 'active', + lastSyncAt: null, + lastSyncError: null, + lastSyncDocCount: null, + nextSyncAt: null, + consecutiveFailures: 0, + createdAt: '2026-08-21T12:00:00.000Z', + updatedAt: '2026-08-21T12:00:00.000Z', + ...overrides, + } +} + +interface PollableQueryOptions { + refetchInterval: (query: { state: { data?: TData } }) => number | false +} + +function capturedQueryOptions(): PollableQueryOptions { + return mocks.useQuery.mock.calls.at(-1)?.[0] as PollableQueryOptions +} + +describe('isConnectorSyncingOrPending', () => { + it('treats a queued sync as in flight', () => { + expect(isConnectorSyncingOrPending(makeConnector({ status: 'pending' }))).toBe(true) + }) + + it('treats a running sync as in flight', () => { + expect(isConnectorSyncingOrPending(makeConnector({ status: 'syncing' }))).toBe(true) + }) + + /** + * The state this replaced: a just-created connector that had not synced yet + * was inferred to be pending from its `createdAt`. The server now says so + * itself, and an `active` row means idle no matter how recent it is. + */ + it('does not infer a queued sync from a freshly created unsynced connector', () => { + expect( + isConnectorSyncingOrPending( + makeConnector({ + status: 'active', + lastSyncAt: null, + createdAt: new Date().toISOString(), + }) + ) + ).toBe(false) + }) + + it.each(['active', 'paused', 'error', 'disabled'] as const)( + 'does not treat a %s connector as in flight', + (status) => { + expect(isConnectorSyncingOrPending(makeConnector({ status }))).toBe(false) + } + ) +}) + +describe('useConnectorList polling', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it.each(['pending', 'syncing'] as const)('polls while a connector is %s', (status) => { + useConnectorList(KB_ID) + const { refetchInterval } = capturedQueryOptions() + + expect(refetchInterval({ state: { data: [makeConnector({ status })] } })).toBe( + CONNECTOR_SYNC_POLL_INTERVAL_MS + ) + }) + + it('stops polling once every connector is idle', () => { + useConnectorList(KB_ID) + const { refetchInterval } = capturedQueryOptions() + + expect(refetchInterval({ state: { data: [makeConnector({ status: 'active' })] } })).toBe(false) + }) + + it('does not poll an empty list', () => { + useConnectorList(KB_ID) + const { refetchInterval } = capturedQueryOptions() + + expect(refetchInterval({ state: { data: [] } })).toBe(false) + }) +}) + +describe('useConnectorDetail polling', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it('polls the sync history while a sync is in flight', () => { + useConnectorDetail(KB_ID, 'connector-1') + const { refetchInterval } = capturedQueryOptions() + + expect(refetchInterval({ state: { data: makeConnector({ status: 'syncing' }) } })).toBe( + CONNECTOR_SYNC_POLL_INTERVAL_MS + ) + }) + + it('stops polling the sync history once the sync finishes', () => { + useConnectorDetail(KB_ID, 'connector-1') + const { refetchInterval } = capturedQueryOptions() + + expect(refetchInterval({ state: { data: makeConnector({ status: 'active' }) } })).toBe(false) + }) +}) + +describe('useTriggerSync optimistic state', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + function capturedMutationOptions() { + return mocks.useMutation.mock.calls.at(-1)?.[0] as { + onMutate: (vars: { knowledgeBaseId: string; connectorId: string }) => Promise + onError: (error: unknown, vars: unknown, context: unknown) => void + } + } + + it('marks the connector queued for the duration of the request', async () => { + const existing = [makeConnector({ status: 'active' })] + mocks.getQueryData.mockReturnValue(existing) + + useTriggerSync() + await capturedMutationOptions().onMutate({ knowledgeBaseId: KB_ID, connectorId: 'connector-1' }) + + /** `all`, not `lists`: the detail query polls the same status and must not land after the settle. */ + expect(mocks.cancelQueries).toHaveBeenCalledWith({ queryKey: connectorKeys.all(KB_ID) }) + const [, updater] = mocks.setQueryData.mock.calls.at(-1) as [ + unknown, + (connectors?: ConnectorData[]) => ConnectorData[] | undefined, + ] + expect(updater(existing)?.[0].status).toBe('pending') + }) + + it('restores the previous status when the request fails', async () => { + const existing = [makeConnector({ status: 'active' })] + mocks.getQueryData.mockReturnValue(existing) + + useTriggerSync() + const options = capturedMutationOptions() + const context = await options.onMutate({ knowledgeBaseId: KB_ID, connectorId: 'connector-1' }) + + mocks.setQueryData.mockClear() + options.onError(new Error('boom'), {}, context) + + const [, updater] = mocks.setQueryData.mock.calls.at(-1) as [ + unknown, + (connectors?: ConnectorData[]) => ConnectorData[] | undefined, + ] + expect(updater(existing)?.[0].status).toBe('active') + }) + + /** + * Two connectors can be in flight at once. A whole-list snapshot would make + * one connector's rollback discard the other's still-pending optimistic write. + */ + it('rolls back only the connector that failed', async () => { + const existing = [ + makeConnector({ id: 'connector-1', status: 'active' }), + makeConnector({ id: 'connector-2', status: 'active' }), + ] + mocks.getQueryData.mockReturnValue(existing) + + useTriggerSync() + const options = capturedMutationOptions() + const context = await options.onMutate({ knowledgeBaseId: KB_ID, connectorId: 'connector-1' }) + + /** connector-2 goes optimistically pending while connector-1 is still in flight. */ + const concurrent = existing.map((connector) => + connector.id === 'connector-2' ? { ...connector, status: 'pending' as const } : connector + ) + + mocks.setQueryData.mockClear() + options.onError(new Error('boom'), {}, context) + + const [, updater] = mocks.setQueryData.mock.calls.at(-1) as [ + unknown, + (connectors?: ConnectorData[]) => ConnectorData[] | undefined, + ] + const rolledBack = updater(concurrent) + expect(rolledBack?.find((c) => c.id === 'connector-1')?.status).toBe('active') + expect(rolledBack?.find((c) => c.id === 'connector-2')?.status).toBe('pending') + }) +}) interface ConnectorDocumentsPage { documents: Array<{ id: string }> diff --git a/apps/sim/hooks/queries/kb/connectors.ts b/apps/sim/hooks/queries/kb/connectors.ts index a00aad93346..e4fcc2aef40 100644 --- a/apps/sim/hooks/queries/kb/connectors.ts +++ b/apps/sim/hooks/queries/kb/connectors.ts @@ -1,5 +1,6 @@ import { keepPreviousData, + type QueryClient, useInfiniteQuery, useMutation, useQuery, @@ -29,11 +30,14 @@ export const CONNECTOR_LIST_STALE_TIME = 30 * 1000 export const CONNECTOR_DETAIL_STALE_TIME = 30 * 1000 export const CONNECTOR_DOCUMENT_LIST_STALE_TIME = 30 * 1000 +/** + * A knowledge base has exactly one connector list, so `lists` is both the + * prefix and the query's own key — there is no per-parameter leaf below it. + */ export const connectorKeys = { all: (knowledgeBaseId?: string) => [...knowledgeKeys.detail(knowledgeBaseId), 'connectors'] as const, lists: (knowledgeBaseId?: string) => [...connectorKeys.all(knowledgeBaseId), 'list'] as const, - list: (knowledgeBaseId?: string) => connectorKeys.lists(knowledgeBaseId), details: (knowledgeBaseId?: string) => [...connectorKeys.all(knowledgeBaseId), 'detail'] as const, detail: (knowledgeBaseId?: string, connectorId?: string) => [...connectorKeys.details(knowledgeBaseId), connectorId ?? ''] as const, @@ -64,24 +68,25 @@ async function fetchConnectorDetail( return result.data } -/** Stop polling for initial sync after 2 minutes */ -const PENDING_SYNC_WINDOW_MS = 2 * 60 * 1000 +export const CONNECTOR_SYNC_POLL_INTERVAL_MS = 3000 /** - * Checks if a connector is syncing or awaiting its first sync within the allowed window + * Whether a sync is queued or running for this connector. + * + * Reads server state only. The server writes `pending` the moment a sync is + * queued and `syncing` once a worker takes the lock, so there is no window to + * infer and no clock to compare against — an earlier version guessed from + * `createdAt`, which was wrong under queue backlog and under client clock skew. */ -export function isConnectorSyncingOrPending(connector: ConnectorData): boolean { - if (connector.status === 'syncing') return true - return ( - connector.status === 'active' && - !connector.lastSyncAt && - Date.now() - new Date(connector.createdAt).getTime() < PENDING_SYNC_WINDOW_MS - ) +export function isConnectorSyncingOrPending(connector: { + status: ConnectorData['status'] +}): boolean { + return connector.status === 'pending' || connector.status === 'syncing' } export function useConnectorList(knowledgeBaseId?: string) { return useQuery({ - queryKey: connectorKeys.list(knowledgeBaseId), + queryKey: connectorKeys.lists(knowledgeBaseId), queryFn: ({ signal }) => fetchConnectors(knowledgeBaseId as string, signal), enabled: Boolean(knowledgeBaseId), staleTime: CONNECTOR_LIST_STALE_TIME, @@ -89,7 +94,7 @@ export function useConnectorList(knowledgeBaseId?: string) { refetchInterval: (query) => { const connectors = query.state.data if (!connectors?.length) return false - return connectors.some(isConnectorSyncingOrPending) ? 3000 : false + return connectors.some(isConnectorSyncingOrPending) ? CONNECTOR_SYNC_POLL_INTERVAL_MS : false }, }) } @@ -102,9 +107,61 @@ export function useConnectorDetail(knowledgeBaseId?: string, connectorId?: strin enabled: Boolean(knowledgeBaseId && connectorId), staleTime: CONNECTOR_DETAIL_STALE_TIME, placeholderData: keepPreviousData, + /** + * The sync history this query carries is the thing a user watches during a + * sync, so it tracks the list's cadence instead of going stale behind an + * animating spinner. + */ + refetchInterval: (query) => { + const connector = query.state.data + if (!connector) return false + return isConnectorSyncingOrPending(connector) ? CONNECTOR_SYNC_POLL_INTERVAL_MS : false + }, }) } +function setCachedConnectorStatus( + queryClient: QueryClient, + knowledgeBaseId: string, + connectorId: string, + status: ConnectorData['status'] +) { + queryClient.setQueryData(connectorKeys.lists(knowledgeBaseId), (connectors) => + connectors?.map((connector) => + connector.id === connectorId ? { ...connector, status } : connector + ) + ) +} + +/** + * Applies an optimistic status to one connector and returns the status it had, + * which is all `onError` needs to undo it — the mutation variables already + * carry the ids. + * + * Both status-changing mutations resolve into the same list, so they share this + * write instead of each keeping a local `Set` of in-flight ids alongside it — + * that duplicated the server's own state and could not survive a remount. + * + * Deliberately not a snapshot of the whole array: two connectors can be in + * flight at once, and restoring a whole-list snapshot would roll the other + * one's optimistic write back along with this one's — or resurrect a status it + * had already moved past. + */ +function optimisticallySetConnectorStatus( + queryClient: QueryClient, + knowledgeBaseId: string, + connectorId: string, + status: ConnectorData['status'] +) { + const previousStatus = queryClient + .getQueryData(connectorKeys.lists(knowledgeBaseId)) + ?.find((connector) => connector.id === connectorId)?.status + + setCachedConnectorStatus(queryClient, knowledgeBaseId, connectorId, status) + + return previousStatus +} + interface CreateConnectorParams { knowledgeBaseId: string connectorType: string @@ -131,10 +188,13 @@ export function useCreateConnector() { return useMutation({ mutationFn: createConnector, + /** + * Only the connector list gains a row — a new connector has no documents + * yet, so the base's own totals do not move here. They move when the first + * sync lands, which `base.tsx` picks up on the syncing-to-idle transition. + */ onSettled: (_data, _error, { knowledgeBaseId }) => { - queryClient.invalidateQueries({ - queryKey: knowledgeKeys.detail(knowledgeBaseId), - }) + queryClient.invalidateQueries({ queryKey: connectorKeys.all(knowledgeBaseId) }) }, }) } @@ -167,10 +227,23 @@ export function useUpdateConnector() { return useMutation({ mutationFn: updateConnector, + onMutate: async ({ knowledgeBaseId, connectorId, updates }) => { + if (!updates.status) return undefined + await queryClient.cancelQueries({ queryKey: connectorKeys.all(knowledgeBaseId) }) + return optimisticallySetConnectorStatus( + queryClient, + knowledgeBaseId, + connectorId, + updates.status + ) + }, + onError: (_error, { knowledgeBaseId, connectorId }, previousStatus) => { + if (previousStatus) { + setCachedConnectorStatus(queryClient, knowledgeBaseId, connectorId, previousStatus) + } + }, onSettled: (_data, _error, { knowledgeBaseId }) => { - queryClient.invalidateQueries({ - queryKey: connectorKeys.all(knowledgeBaseId), - }) + queryClient.invalidateQueries({ queryKey: connectorKeys.all(knowledgeBaseId) }) }, }) } @@ -197,10 +270,28 @@ export function useDeleteConnector() { return useMutation({ mutationFn: deleteConnector, - onSettled: (_data, _error, { knowledgeBaseId }) => { + /** + * Removing a connector can take its documents with it, so the document + * lists and the base's own totals move — but nothing below them does. + * Invalidating `knowledgeKeys.detail` as a prefix would also refetch every + * cached document detail, chunk page, and chunk search in the base. + */ + onSettled: (_data, _error, { knowledgeBaseId, deleteDocuments }) => { + queryClient.invalidateQueries({ queryKey: connectorKeys.all(knowledgeBaseId) }) + queryClient.invalidateQueries({ queryKey: knowledgeKeys.documentLists(knowledgeBaseId) }) queryClient.invalidateQueries({ queryKey: knowledgeKeys.detail(knowledgeBaseId), + exact: true, }) + /** + * Only this branch takes documents with it, and any per-document detail, + * chunk page, or chunk search cached for one of them now points at a row + * that no longer exists. The ids are not in the response, so this is the + * narrowest prefix that reaches all of them. + */ + if (deleteDocuments) { + queryClient.invalidateQueries({ queryKey: knowledgeKeys.documentDetails(knowledgeBaseId) }) + } }, }) } @@ -229,14 +320,33 @@ export function useTriggerSync() { return useMutation({ mutationFn: triggerSync, /** - * The sync itself runs async — the connector list's own syncing poll surfaces its - * progress. Only the connector rows have anything to say yet. + * The server marks the connector `pending` as it queues the sync, so the + * optimistic write here only covers the request's own round trip — after + * which the refetch below carries the same status and the list's sync poll + * takes over through `pending` → `syncing` → `active`. */ - onSettled: (_data, _error, { knowledgeBaseId }) => { - queryClient.invalidateQueries({ - queryKey: connectorKeys.all(knowledgeBaseId), - }) + onMutate: async ({ knowledgeBaseId, connectorId }) => { + await queryClient.cancelQueries({ queryKey: connectorKeys.all(knowledgeBaseId) }) + return optimisticallySetConnectorStatus(queryClient, knowledgeBaseId, connectorId, 'pending') }, + /** + * Rolling back also stops the poll the optimistic `pending` started, so a + * refused sync does not leave the row spinning. + */ + onError: (_error, { knowledgeBaseId, connectorId }, previousStatus) => { + if (previousStatus) { + setCachedConnectorStatus(queryClient, knowledgeBaseId, connectorId, previousStatus) + } + queryClient.invalidateQueries({ queryKey: connectorKeys.all(knowledgeBaseId) }) + }, + /** + * Deliberately no invalidation on success. The route answers without + * awaiting the dispatch that writes `pending`, so an immediate refetch can + * still read `active`, discard the optimistic write, and stop the poll + * before it ever started — leaving the UI claiming idle for a sync that is + * running. The optimistic `pending` starts the poll instead, and the poll + * reconciles against whatever the server actually settles on. + */ }) } @@ -304,6 +414,39 @@ interface ConnectorDocumentMutationParams { documentIds: string[] } +/** + * Excluding or restoring moves the connector's own document list, the base's + * document lists that render those rows, and the base's totals. + * `knowledgeKeys.detail` is invalidated `exact` for that last one — as a prefix + * it would subsume every key here and refetch the base's chunk pages and chunk + * searches too. The affected rows are named in the request, so each one is + * invalidated directly rather than through a wider prefix. + */ +function invalidateConnectorDocumentChange( + queryClient: QueryClient, + { knowledgeBaseId, connectorId, documentIds }: ConnectorDocumentMutationParams +) { + queryClient.invalidateQueries({ + queryKey: connectorDocumentKeys.lists(knowledgeBaseId, connectorId), + }) + queryClient.invalidateQueries({ queryKey: knowledgeKeys.documentLists(knowledgeBaseId) }) + queryClient.invalidateQueries({ queryKey: knowledgeKeys.detail(knowledgeBaseId), exact: true }) + + /** + * One pass over the cache rather than one per id — `documentIds` reaches + * `MAX_KNOWLEDGE_CONNECTOR_DOCUMENT_MUTATION_ITEMS`, and every + * `invalidateQueries` call scans the whole cache. The prefix filter already + * restricts matches to `documentDetails`, under which `document()` puts the + * id at the position read here. + */ + const affected = new Set(documentIds) + const documentDetailsPrefix = knowledgeKeys.documentDetails(knowledgeBaseId) + queryClient.invalidateQueries({ + queryKey: documentDetailsPrefix, + predicate: (query) => affected.has(query.queryKey[documentDetailsPrefix.length] as string), + }) +} + async function excludeConnectorDocuments({ knowledgeBaseId, connectorId, @@ -322,14 +465,8 @@ export function useExcludeConnectorDocument() { return useMutation({ mutationFn: excludeConnectorDocuments, - onSettled: (_data, _error, { knowledgeBaseId, connectorId }) => { - queryClient.invalidateQueries({ - queryKey: connectorDocumentKeys.lists(knowledgeBaseId, connectorId), - }) - queryClient.invalidateQueries({ - queryKey: knowledgeKeys.detail(knowledgeBaseId), - }) - }, + onSettled: (_data, _error, variables) => + invalidateConnectorDocumentChange(queryClient, variables), }) } @@ -351,13 +488,7 @@ export function useRestoreConnectorDocument() { return useMutation({ mutationFn: restoreConnectorDocuments, - onSettled: (_data, _error, { knowledgeBaseId, connectorId }) => { - queryClient.invalidateQueries({ - queryKey: connectorDocumentKeys.lists(knowledgeBaseId, connectorId), - }) - queryClient.invalidateQueries({ - queryKey: knowledgeKeys.detail(knowledgeBaseId), - }) - }, + onSettled: (_data, _error, variables) => + invalidateConnectorDocumentChange(queryClient, variables), }) } diff --git a/apps/sim/hooks/queries/utils/knowledge-keys.ts b/apps/sim/hooks/queries/utils/knowledge-keys.ts index 743ffc81a81..e47704214d7 100644 --- a/apps/sim/hooks/queries/utils/knowledge-keys.ts +++ b/apps/sim/hooks/queries/utils/knowledge-keys.ts @@ -43,8 +43,16 @@ export const knowledgeKeys = { [...knowledgeKeys.detail(knowledgeBaseId), 'documents'] as const, documents: (knowledgeBaseId: string, paramsKey: string) => [...knowledgeKeys.documentLists(knowledgeBaseId), paramsKey] as const, + /** + * Prefix over every per-document cache in a base — each `document` entry and + * the `chunks` / `search` keys nested under it. Needed when a mutation + * invalidates documents it cannot name, so the alternative would be the + * `detail` prefix, which also drags in the connector and tag caches. + */ + documentDetails: (knowledgeBaseId: string) => + [...knowledgeKeys.detail(knowledgeBaseId), 'document'] as const, document: (knowledgeBaseId: string, documentId: string) => - [...knowledgeKeys.detail(knowledgeBaseId), 'document', documentId] as const, + [...knowledgeKeys.documentDetails(knowledgeBaseId), documentId] as const, documentTagDefinitions: (knowledgeBaseId: string, documentId: string) => [...knowledgeKeys.document(knowledgeBaseId, documentId), 'tagDefinitions'] as const, chunks: (knowledgeBaseId: string, documentId: string, paramsKey: string) => diff --git a/apps/sim/lib/api/contracts/knowledge/connectors.ts b/apps/sim/lib/api/contracts/knowledge/connectors.ts index bcb8d40d07e..d872cff4203 100644 --- a/apps/sim/lib/api/contracts/knowledge/connectors.ts +++ b/apps/sim/lib/api/contracts/knowledge/connectors.ts @@ -60,7 +60,8 @@ export const connectorDataSchema = z sourceConfig: z.record(z.string(), z.unknown()), syncMode: z.string().nullable(), syncIntervalMinutes: z.number(), - status: z.enum(['active', 'paused', 'syncing', 'error', 'disabled']), + /** `pending` means a sync is queued but no worker has taken the lock yet. */ + status: z.enum(['active', 'paused', 'pending', 'syncing', 'error', 'disabled']), lastSyncAt: z.string().nullable(), lastSyncError: z.string().nullable(), lastSyncDocCount: z.number().nullable(), diff --git a/apps/sim/lib/api/contracts/v2/knowledge.ts b/apps/sim/lib/api/contracts/v2/knowledge.ts index a45f021cb03..7c290fd1ca1 100644 --- a/apps/sim/lib/api/contracts/v2/knowledge.ts +++ b/apps/sim/lib/api/contracts/v2/knowledge.ts @@ -1491,8 +1491,8 @@ export const v2KnowledgeConnectorSchema = z .nonnegative() .describe('Scheduled synchronization interval in minutes; zero disables scheduled syncs.'), status: z - .enum(['active', 'paused', 'syncing', 'error', 'disabled']) - .describe('Current connector state.'), + .enum(['active', 'paused', 'pending', 'syncing', 'error', 'disabled']) + .describe('Current connector state. `pending` means a sync is queued but not yet running.'), lastSyncAt: v2TimestampSchema .nullable() .describe('Time of the most recent synchronization, or null before the first sync.'), diff --git a/apps/sim/lib/knowledge/connectors/queue.test.ts b/apps/sim/lib/knowledge/connectors/queue.test.ts index f867517f435..cd48bcc5b9d 100644 --- a/apps/sim/lib/knowledge/connectors/queue.test.ts +++ b/apps/sim/lib/knowledge/connectors/queue.test.ts @@ -1,7 +1,14 @@ /** * @vitest-environment node */ -import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing' +import { + dbChainMockFns, + hasMockCondition, + type MockCondition, + queueTableRows, + resetDbChainMock, + schemaMock, +} from '@sim/testing' import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest' const { mockExecuteSync, mockIsTriggerAvailable, mockResolveTriggerRegion, mockTrigger } = @@ -21,9 +28,14 @@ vi.mock('@/lib/knowledge/documents/service', () => ({ })) vi.mock('@/lib/knowledge/connectors/sync-engine', () => ({ executeSync: mockExecuteSync, + connectorIsLive: () => ({ type: 'connectorIsLive' }), })) -import { assertConnectorSyncPayload, dispatchSync } from '@/lib/knowledge/connectors/queue' +import { + assertConnectorSyncPayload, + dispatchSync, + SYNC_DISPATCH_FAILED_ERROR, +} from '@/lib/knowledge/connectors/queue' const BILLING_ATTRIBUTION = { actorUserId: 'external-admin', @@ -144,6 +156,140 @@ describe('connector sync queue', () => { expect(mockTrigger).not.toHaveBeenCalled() }) + it('marks the connector queued before handing the sync off', async () => { + await dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + requestId: 'request-1', + }) + + /** `pending` is the only thing distinguishing "a sync is coming" from "idle". */ + expect(dbChainMockFns.set).toHaveBeenCalledWith(expect.objectContaining({ status: 'pending' })) + expect(mockTrigger).toHaveBeenCalled() + }) + + it('opens a lease and takes a token when it queues', async () => { + await dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + requestId: 'request-1', + }) + + /** + * The lease is what the scheduler ages a stranded queue entry against — + * `updatedAt` cannot serve, because a pending connector is still editable + * and any unrelated write would renew the recovery it should trigger. + */ + const payload = dbChainMockFns.set.mock.calls[0][0] as Record + expect(payload.syncLockLeaseAt).toBeInstanceOf(Date) + expect(typeof payload.syncLockToken).toBe('string') + }) + + it('queues a connector that is already pending, so the create path gets a lease and token', async () => { + await dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + requestId: 'request-1', + }) + + /** + * A created connector is born `pending` in its INSERT but with no lease and + * no token. Skipping it here as a redundant write would leave it ageing + * against `updatedAt` — which any edit renews — and holding a token this + * dispatch cannot match, so a failed hand-off would never release it. + */ + const queueWhere = dbChainMockFns.where.mock.calls + .map((call) => call[0]) + .find((where) => + hasMockCondition( + where, + (node: MockCondition) => + node.type === 'ne' && node.left === schemaMock.knowledgeConnector.status + ) + ) + expect(queueWhere).toBeDefined() + + expect( + hasMockCondition( + queueWhere, + (node: MockCondition) => node.type === 'ne' && node.right === 'pending' + ) + ).toBe(false) + + /** A live run still owns its row: demoting it to `pending` would strand it. */ + expect( + hasMockCondition( + queueWhere, + (node: MockCondition) => node.type === 'ne' && node.right === 'syncing' + ) + ).toBe(true) + }) + + it('releases the queued connector when the hand-off throws', async () => { + mockTrigger.mockRejectedValueOnce(new Error('trigger unavailable')) + + await expect( + dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + requestId: 'request-1', + }) + ).rejects.toThrow('trigger unavailable') + + /** + * Left `pending`, the connector would sit with a sync that is never coming: + * the scheduler's due-sweep only looks at `active`/`error` rows. + */ + const released = dbChainMockFns.set.mock.calls.at(-1)?.[0] as Record + expect(released.status).toBe('error') + expect(released.lastSyncError).toBe(SYNC_DISPATCH_FAILED_ERROR) + expect(released.syncLockToken).toBeNull() + + /** + * The verdict is about the queue, not the connector, so it must not advance + * the auto-disable breaker — a queue outage would otherwise increment every + * connector in the fleet until they all disabled themselves. + */ + expect(released).not.toHaveProperty('consecutiveFailures') + expect(released.nextSyncAt).toBeInstanceOf(Date) + + /** + * Guarded on this dispatch's own token, not merely on `pending`. A hand-off + * can throw long after the scheduler reclaimed the queue entry and + * dispatched a replacement; without the token this dead dispatch would + * overwrite the live one. + */ + const queuedToken = (dbChainMockFns.set.mock.calls[0][0] as Record) + .syncLockToken + expect( + hasMockCondition( + dbChainMockFns.where.mock.calls.at(-1)?.[0], + (node: MockCondition) => + node.type === 'eq' && + node.left === schemaMock.knowledgeConnector.syncLockToken && + node.right === queuedToken + ) + ).toBe(true) + }) + + it('does not queue a connector whose knowledge base is gone', async () => { + resetDbChainMock() + queueTableRows(schemaMock.knowledgeConnector, [ + { + knowledgeBaseId: 'knowledge-base-1', + connectorArchivedAt: null, + connectorDeletedAt: null, + workspaceId: 'workspace-paid', + kbDeletedAt: new Date('2026-08-20T00:00:00.000Z'), + }, + ]) + + await dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + requestId: 'request-1', + }) + + expect(dbChainMockFns.set).not.toHaveBeenCalledWith( + expect.objectContaining({ status: 'pending' }) + ) + }) + it('rejects legacy payloads without billing attribution', () => { expect(() => assertConnectorSyncPayload({ diff --git a/apps/sim/lib/knowledge/connectors/queue.ts b/apps/sim/lib/knowledge/connectors/queue.ts index ad9163a5805..deafdf866d4 100644 --- a/apps/sim/lib/knowledge/connectors/queue.ts +++ b/apps/sim/lib/knowledge/connectors/queue.ts @@ -5,13 +5,13 @@ import { toError } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' import { isRecordLike } from '@sim/utils/object' import { tasks } from '@trigger.dev/sdk' -import { eq } from 'drizzle-orm' +import { and, eq, ne } from 'drizzle-orm' import { assertBillingAttributionSnapshot, type BillingAttributionSnapshot, } from '@/lib/billing/core/billing-attribution' import { resolveTriggerRegion } from '@/lib/core/async-jobs/region' -import { executeSync } from '@/lib/knowledge/connectors/sync-engine' +import { connectorIsLive, executeSync } from '@/lib/knowledge/connectors/sync-engine' import { isTriggerAvailable } from '@/lib/knowledge/documents/service' const logger = createLogger('ConnectorSyncQueue') @@ -71,6 +71,114 @@ export function assertConnectorSyncPayload(value: unknown): ConnectorSyncPayload } } +export const SYNC_DISPATCH_FAILED_ERROR = 'Sync could not be queued' + +/** + * Marks the connector as having a sync queued, and returns the token that owns + * that queued sync. + * + * Every dispatch path funnels through here, so `pending` is written in one + * place. It is what lets the UI show a queued sync from server state: until a + * worker takes the lock there is otherwise nothing on the row distinguishing + * "a sync is coming" from "idle", which is what previously forced the client to + * guess from `createdAt`. + * + * `pending` is a phase of the same lock `syncing` holds, not a state beside it, + * so it opens the lease and takes a token exactly as + * {@link buildSyncLockAcquisition} does. The lease is what the scheduler ages a + * stranded queue entry against — `updatedAt` cannot serve, because a pending + * connector is still editable and every unrelated write to the row would renew + * the recovery it is meant to trigger. The token is what makes the release + * below provably this dispatch's own. + * + * Deliberately still writes a row already `pending`. The create path is born + * `pending` in its INSERT but carries no lease and no token, so skipping it as + * a redundant write would leave every new connector ageing against `updatedAt` + * and holding a token this dispatch cannot match — defeating both guards above + * on exactly the path where a failed hand-off is most visible. The cost is one + * extra UPDATE per connector creation, which is rare; the scheduler's own + * dispatches only ever see `active`/`error` rows and are unaffected. + * + * Skips a `syncing` row for the reason the lock acquisition does: a run may + * have taken the lock between the caller's read and this write, and demoting a + * live run to `pending` would strand it, since the reaper only looks at + * `syncing` rows. + */ +async function markSyncPending(connectorId: string): Promise { + const dispatchToken = generateId() + const now = new Date() + + await db + .update(knowledgeConnector) + .set({ + status: 'pending', + syncLockToken: dispatchToken, + syncLockLeaseAt: now, + updatedAt: now, + }) + .where( + and( + eq(knowledgeConnector.id, connectorId), + ne(knowledgeConnector.status, 'syncing'), + connectorIsLive() + ) + ) + + return dispatchToken +} + +/** + * Releases a queued sync whose hand-off threw. + * + * Guarded on this dispatch's own token, not merely on `pending`: a hand-off can + * throw long after the scheduler reclaimed the queue entry and dispatched a + * replacement, and `status = 'pending'` alone would let this dead dispatch + * overwrite the live one — the same reason {@link holdsSyncLockToken} exists + * for `syncing`. + * + * Deliberately does NOT advance the failure ladder, unlike the scheduler's + * recovery of a stranded queue entry. The verdict here is observably about the + * queue, not the connector: the queue client itself threw. Laddering it would + * mean a Trigger.dev outage increments every connector in the fleet on every + * dispatch attempt until they auto-disable, each then needing a manual + * re-enable for a fault that was never theirs. `nextSyncAt` is pulled to now so + * the scheduler's due-sweep retries promptly once the queue recovers; a genuine + * per-connector problem still reaches the breaker through the run itself. + */ +async function releaseFailedDispatch( + connectorId: string, + dispatchToken: string, + error: unknown +): Promise { + const now = new Date() + try { + await db + .update(knowledgeConnector) + .set({ + status: 'error', + lastSyncError: SYNC_DISPATCH_FAILED_ERROR, + nextSyncAt: now, + syncLockToken: null, + syncLockLeaseAt: null, + updatedAt: now, + }) + .where( + and( + eq(knowledgeConnector.id, connectorId), + eq(knowledgeConnector.status, 'pending'), + eq(knowledgeConnector.syncLockToken, dispatchToken), + connectorIsLive() + ) + ) + } catch (releaseError) { + logger.error('Failed to release a connector whose sync dispatch failed', { + connectorId, + dispatchError: toError(error).message, + releaseError: toError(releaseError).message, + }) + } +} + /** * Dispatches a connector sync with billing attribution already fixed by the * authenticated or scheduled entry point. @@ -164,22 +272,44 @@ export async function dispatchSync( ] if (isTriggerAvailable()) { - await tasks.trigger('knowledge-connector-sync', payload, { - tags, - region: await resolveTriggerRegion(), - }) + const dispatchToken = await markSyncPending(connectorId) + + /** + * Everything between taking the queue entry and the hand-off landing has to + * sit inside this `try`. Resolving the region concurrently with + * `markSyncPending` looked free, but its rejection escaped before the token + * was ever bound, so the release below could not run and the row was left + * `pending` until the reaper's TTL. + */ + try { + await tasks.trigger('knowledge-connector-sync', payload, { + tags, + region: await resolveTriggerRegion(), + }) + } catch (error) { + await releaseFailedDispatch(connectorId, dispatchToken, error) + throw error + } logger.info('Dispatched connector sync to Trigger.dev', { connectorId, requestId }) return } + const dispatchToken = await markSyncPending(connectorId) + executeSync(connectorId, { fullSync: payload.fullSync, rehydrate: payload.rehydrate, billingAttribution: payload.billingAttribution, - }).catch((error) => { + }).catch(async (error) => { logger.error(`Sync failed for connector ${connectorId}`, { error: toError(error).message, requestId, }) + /** + * Only reaches a row still `pending` holding this dispatch's token: once + * `executeSync` takes the lock it overwrites the token and owns the terminal + * write. This covers the narrow case where it threw before acquiring it. + */ + await releaseFailedDispatch(connectorId, dispatchToken, error) }) } diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index bf5f42ed5b8..362c6ad18fd 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -656,7 +656,7 @@ export function stillHoldsSyncLock(connectorId: string, syncLockToken: string) { } /** The archived/deleted half of {@link stillHoldsSyncLock}. */ -function connectorIsLive() { +export function connectorIsLive() { return and(isNull(knowledgeConnector.archivedAt), isNull(knowledgeConnector.deletedAt)) } diff --git a/apps/sim/lib/knowledge/constants.ts b/apps/sim/lib/knowledge/constants.ts index 231833abde5..df4434758e7 100644 --- a/apps/sim/lib/knowledge/constants.ts +++ b/apps/sim/lib/knowledge/constants.ts @@ -177,3 +177,12 @@ export function getPlaceholderForFieldType(fieldType: string): string { return 'Enter value' } } + +/** + * How long a document may sit in `processing` before its run is treated as dead. + * + * Lives here rather than beside the server-side claim helpers so the client can + * read the same number without importing a module that pulls in the database + * client. `processing-claim.ts` re-exports it for its own callers. + */ +export const KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS = 10 * 60 * 1000 diff --git a/apps/sim/lib/knowledge/documents/processing-claim.ts b/apps/sim/lib/knowledge/documents/processing-claim.ts index 0d6290b680b..8d4389b54cb 100644 --- a/apps/sim/lib/knowledge/documents/processing-claim.ts +++ b/apps/sim/lib/knowledge/documents/processing-claim.ts @@ -4,10 +4,11 @@ import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { truncate } from '@sim/utils/string' import { and, eq, isNull } from 'drizzle-orm' +import { KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS } from '@/lib/knowledge/constants' const logger = createLogger('KnowledgeDocumentProcessingClaim') -export const KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS = 10 * 60 * 1000 +export { KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS } from '@/lib/knowledge/constants' interface ReclaimStaleDocumentProcessingClaimParams { knowledgeBaseId: string diff --git a/apps/sim/lib/knowledge/orchestration/connectors.test.ts b/apps/sim/lib/knowledge/orchestration/connectors.test.ts index 4f799a85b83..1a0f9382053 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.test.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.test.ts @@ -336,6 +336,33 @@ describe('performSyncKnowledgeConnector', () => { expect(mockDispatchSync).not.toHaveBeenCalled() }) + it.each(['pending', 'paused', 'disabled'] as const)( + 'refuses an on-demand sync on a %s connector', + async (status) => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status }, + ]) + + const outcome = await performSyncKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + resolveBillingAttribution, + }) + + /** + * `pending` already has a run queued. `paused`/`disabled` have no way + * back: queueing overwrites `status`, and every exit from the run writes + * its own verdict — success writes `active`, a lost queue entry writes + * `error`, which the due-sweep then keeps syncing. One "Sync now" would + * silently resume the connector for good. + */ + expect(outcome).toMatchObject({ success: false, errorCode: 'conflict' }) + expect(resolveBillingAttribution).not.toHaveBeenCalled() + expect(mockDispatchSync).not.toHaveBeenCalled() + } + ) + it('dispatches and records who asked for it', async () => { dbChainMockFns.limit.mockResolvedValueOnce([ { id: 'conn-1', connectorType: 'notion', status: 'active' }, diff --git a/apps/sim/lib/knowledge/orchestration/connectors.ts b/apps/sim/lib/knowledge/orchestration/connectors.ts index 1dc6dee58e9..9e9ee012776 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.ts @@ -295,7 +295,16 @@ export async function performCreateKnowledgeConnector( encryptedApiKey: resolvedEncryptedApiKey, sourceConfig: finalSourceConfig, syncIntervalMinutes, - status: 'active', + /** + * The initial sync is dispatched after this transaction commits, so + * the row is born with a sync already queued. `markSyncPending` writes + * this again moments later — this one exists so the create *response* + * is truthful, and the client renders the queued state immediately + * rather than an idle connector until its first refetch. The lease and + * ownership token that make the queue entry recoverable come from that + * later write, which is why it must not skip an already-`pending` row. + */ + status: 'pending', nextSyncAt, createdAt: now, updatedAt: now, @@ -447,6 +456,21 @@ export async function performUpdateKnowledgeConnector( if (existing.status === 'syncing') { return fail('Sync already in progress', 'conflict') } + /** + * A queued run has not read its config yet, so a status change is still safe + * and is deliberately allowed — refusing it would leave a connector stranded + * behind a lost queue entry unpausable until the reaper's TTL. The two config + * edits are refused for the same reasons the `syncing` guard above gives: + * `sourceConfig` would have the run list against one config and reconcile + * against another, and `syncIntervalMinutes` writes a `nextSyncAt` the run's + * terminal write overwrites moments later, silently discarding it. + */ + if ( + existing.status === 'pending' && + (updates.sourceConfig !== undefined || updates.syncIntervalMinutes !== undefined) + ) { + return fail('Sync already in progress', 'conflict') + } if (updates.syncIntervalMinutes !== undefined) { if (!kb.workspaceId && updates.syncIntervalMinutes > 0 && updates.syncIntervalMinutes < 60) { @@ -481,6 +505,15 @@ export async function performUpdateKnowledgeConnector( } if (updates.status !== undefined) { values.status = updates.status + /** + * Releases a queue entry this status change is walking away from, so no + * token survives on a row that is no longer `pending` and the reaper is not + * left with a lease it can never match. + */ + if (existing.status === 'pending') { + values.syncLockToken = null + values.syncLockLeaseAt = null + } if (updates.status === 'active') { values.consecutiveFailures = 0 values.lastSyncError = null @@ -723,9 +756,22 @@ export async function performSyncKnowledgeConnector( if (!connector) { return fail('Connector not found', 'not_found') } - if (connector.status === 'syncing') { + if (connector.status === 'syncing' || connector.status === 'pending') { return fail('Sync already in progress', 'conflict') } + /** + * A paused or disabled connector is not synced on demand. + * + * Nothing here can put the pause back: queueing overwrites `status`, and + * every exit from the run writes its own verdict — success writes `active`, + * and a lost queue entry writes `error`, which the scheduler's due-sweep then + * treats as a connector to keep syncing. So one "Sync now" on a paused + * connector silently resumes it for good. Resuming is a decision the caller + * has to make explicitly, through the status update that says so. + */ + if (connector.status === 'paused' || connector.status === 'disabled') { + return fail(`Connector is ${connector.status}. Resume it before triggering a sync.`, 'conflict') + } if (!kb.workspaceId) { return fail('Knowledge base is missing workspace billing context', 'conflict') } diff --git a/packages/db/schema.ts b/packages/db/schema.ts index 5af83230c34..a6f411227b5 100644 --- a/packages/db/schema.ts +++ b/packages/db/schema.ts @@ -4346,6 +4346,17 @@ export const knowledgeConnector = pgTable( sourceConfig: json('source_config').notNull(), syncMode: text('sync_mode').notNull().default('full'), syncIntervalMinutes: integer('sync_interval_minutes').notNull().default(1440), + /** + * One of `active`, `pending`, `syncing`, `error`, `paused`, `disabled`. + * + * `pending` and `syncing` are the two halves of a sync in flight: `pending` + * is written as the sync is handed to the queue, `syncing` when a worker + * takes the lock. The split exists because the queue depth between them is + * unbounded — without `pending` the row is indistinguishable from idle for + * as long as the hand-off takes, which is what forced readers to guess from + * `created_at`. A row left `pending` past the lock TTL is reclaimed by the + * scheduler, which is the only thing that ever observes a lost hand-off. + */ status: text('status').notNull().default('active'), lastSyncAt: timestamp('last_sync_at'), lastSyncError: text('last_sync_error'), diff --git a/packages/sim-cli/src/generated/v2-api.ts b/packages/sim-cli/src/generated/v2-api.ts index b6eac6251cd..b3dde8bd6bc 100644 --- a/packages/sim-cli/src/generated/v2-api.ts +++ b/packages/sim-cli/src/generated/v2-api.ts @@ -958,7 +958,7 @@ type CreateKnowledgeConnectorResponseRef0 = { sourceConfig: Record syncMode: string syncIntervalMinutes: number - status: 'active' | 'paused' | 'syncing' | 'error' | 'disabled' + status: 'active' | 'paused' | 'pending' | 'syncing' | 'error' | 'disabled' lastSyncAt: string | null lastSyncError: string | null lastSyncDocCount: number | null @@ -2741,7 +2741,7 @@ type GetKnowledgeConnectorResponseRef1 = { sourceConfig: Record syncMode: string syncIntervalMinutes: number - status: 'active' | 'paused' | 'syncing' | 'error' | 'disabled' + status: 'active' | 'paused' | 'pending' | 'syncing' | 'error' | 'disabled' lastSyncAt: string | null lastSyncError: string | null lastSyncDocCount: number | null @@ -3739,7 +3739,7 @@ type ListKnowledgeConnectorsResponseRef0 = { sourceConfig: Record syncMode: string syncIntervalMinutes: number - status: 'active' | 'paused' | 'syncing' | 'error' | 'disabled' + status: 'active' | 'paused' | 'pending' | 'syncing' | 'error' | 'disabled' lastSyncAt: string | null lastSyncError: string | null lastSyncDocCount: number | null @@ -5394,7 +5394,7 @@ type UpdateKnowledgeConnectorResponseRef0 = { sourceConfig: Record syncMode: string syncIntervalMinutes: number - status: 'active' | 'paused' | 'syncing' | 'error' | 'disabled' + status: 'active' | 'paused' | 'pending' | 'syncing' | 'error' | 'disabled' lastSyncAt: string | null lastSyncError: string | null lastSyncDocCount: number | null From d87064b18ff9f059ecc64e3e058b297af6d270b6 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 21 Aug 2026 20:03:54 -0700 Subject: [PATCH 2/5] fix(kb): refuse to start a queued run on a paused connector MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The queue outlives the decision to sync. Pausing a connector after its run was queued cleared the queue entry's token but left the task itself alive, and the lock CAS accepted any row that was not already `syncing` — so the worker took the paused row and wrote its own terminal `active` over the pause. Moves the rule to the two points that can enforce it: an explicit `LOCKABLE_CONNECTOR_STATUSES` allowlist on the lock acquisition, and the same allowlist on `markSyncPending`, which closes the mirror race where a dispatch already in flight rewrites a just-paused row back to `pending`. Queueing and starting now agree on one rule, and a skipped hand-off is reported as its own outcome rather than a concurrency conflict. Also patches the connector detail cache alongside the list on an optimistic status write, so an already-expanded card starts its own sync poll instead of showing stale history behind the list's spinner. --- apps/sim/hooks/queries/kb/connectors.test.ts | 46 +++++++++++-------- apps/sim/hooks/queries/kb/connectors.ts | 12 +++++ .../lib/knowledge/connectors/queue.test.ts | 33 +++++++------ apps/sim/lib/knowledge/connectors/queue.ts | 43 +++++++++++++---- .../knowledge/connectors/sync-engine.test.ts | 25 ++++++++++ .../lib/knowledge/connectors/sync-engine.ts | 34 +++++++++++++- 6 files changed, 150 insertions(+), 43 deletions(-) diff --git a/apps/sim/hooks/queries/kb/connectors.test.ts b/apps/sim/hooks/queries/kb/connectors.test.ts index 7fe550c006f..2a2dde7dee4 100644 --- a/apps/sim/hooks/queries/kb/connectors.test.ts +++ b/apps/sim/hooks/queries/kb/connectors.test.ts @@ -78,6 +78,16 @@ function capturedQueryOptions(): PollableQueryOptions { return mocks.useQuery.mock.calls.at(-1)?.[0] as PollableQueryOptions } +/** + * The status write patches the list and the detail cache, so pick the call for + * the list rather than whichever landed last. + */ +function lastListStatusUpdater() { + const listKey = JSON.stringify(connectorKeys.lists(KB_ID)) + const call = mocks.setQueryData.mock.calls.filter((c) => JSON.stringify(c[0]) === listKey).at(-1) + return call?.[1] as (connectors?: ConnectorData[]) => ConnectorData[] | undefined +} + describe('isConnectorSyncingOrPending', () => { it('treats a queued sync as in flight', () => { expect(isConnectorSyncingOrPending(makeConnector({ status: 'pending' }))).toBe(true) @@ -171,7 +181,11 @@ describe('useTriggerSync optimistic state', () => { function capturedMutationOptions() { return mocks.useMutation.mock.calls.at(-1)?.[0] as { onMutate: (vars: { knowledgeBaseId: string; connectorId: string }) => Promise - onError: (error: unknown, vars: unknown, context: unknown) => void + onError: ( + error: unknown, + vars: { knowledgeBaseId: string; connectorId: string }, + context: unknown + ) => void } } @@ -184,11 +198,7 @@ describe('useTriggerSync optimistic state', () => { /** `all`, not `lists`: the detail query polls the same status and must not land after the settle. */ expect(mocks.cancelQueries).toHaveBeenCalledWith({ queryKey: connectorKeys.all(KB_ID) }) - const [, updater] = mocks.setQueryData.mock.calls.at(-1) as [ - unknown, - (connectors?: ConnectorData[]) => ConnectorData[] | undefined, - ] - expect(updater(existing)?.[0].status).toBe('pending') + expect(lastListStatusUpdater()(existing)?.[0].status).toBe('pending') }) it('restores the previous status when the request fails', async () => { @@ -200,13 +210,13 @@ describe('useTriggerSync optimistic state', () => { const context = await options.onMutate({ knowledgeBaseId: KB_ID, connectorId: 'connector-1' }) mocks.setQueryData.mockClear() - options.onError(new Error('boom'), {}, context) + options.onError( + new Error('boom'), + { knowledgeBaseId: KB_ID, connectorId: 'connector-1' }, + context + ) - const [, updater] = mocks.setQueryData.mock.calls.at(-1) as [ - unknown, - (connectors?: ConnectorData[]) => ConnectorData[] | undefined, - ] - expect(updater(existing)?.[0].status).toBe('active') + expect(lastListStatusUpdater()(existing)?.[0].status).toBe('active') }) /** @@ -230,13 +240,13 @@ describe('useTriggerSync optimistic state', () => { ) mocks.setQueryData.mockClear() - options.onError(new Error('boom'), {}, context) + options.onError( + new Error('boom'), + { knowledgeBaseId: KB_ID, connectorId: 'connector-1' }, + context + ) - const [, updater] = mocks.setQueryData.mock.calls.at(-1) as [ - unknown, - (connectors?: ConnectorData[]) => ConnectorData[] | undefined, - ] - const rolledBack = updater(concurrent) + const rolledBack = lastListStatusUpdater()(concurrent) expect(rolledBack?.find((c) => c.id === 'connector-1')?.status).toBe('active') expect(rolledBack?.find((c) => c.id === 'connector-2')?.status).toBe('pending') }) diff --git a/apps/sim/hooks/queries/kb/connectors.ts b/apps/sim/hooks/queries/kb/connectors.ts index e4fcc2aef40..ba92aff2e9e 100644 --- a/apps/sim/hooks/queries/kb/connectors.ts +++ b/apps/sim/hooks/queries/kb/connectors.ts @@ -120,6 +120,14 @@ export function useConnectorDetail(knowledgeBaseId?: string, connectorId?: strin }) } +/** + * Writes the status into both caches that render it. + * + * The detail query drives its own sync poll off its own copy of `status`, so + * patching only the list would leave an already-expanded card reading `active`, + * never starting that poll, and showing stale sync history behind the list's + * spinner. + */ function setCachedConnectorStatus( queryClient: QueryClient, knowledgeBaseId: string, @@ -131,6 +139,10 @@ function setCachedConnectorStatus( connector.id === connectorId ? { ...connector, status } : connector ) ) + queryClient.setQueryData( + connectorKeys.detail(knowledgeBaseId, connectorId), + (detail) => (detail ? { ...detail, status } : detail) + ) } /** diff --git a/apps/sim/lib/knowledge/connectors/queue.test.ts b/apps/sim/lib/knowledge/connectors/queue.test.ts index cd48bcc5b9d..c18314cfff5 100644 --- a/apps/sim/lib/knowledge/connectors/queue.test.ts +++ b/apps/sim/lib/knowledge/connectors/queue.test.ts @@ -3,6 +3,7 @@ */ import { dbChainMockFns, + flattenMockConditions, hasMockCondition, type MockCondition, queueTableRows, @@ -29,6 +30,7 @@ vi.mock('@/lib/knowledge/documents/service', () => ({ vi.mock('@/lib/knowledge/connectors/sync-engine', () => ({ executeSync: mockExecuteSync, connectorIsLive: () => ({ type: 'connectorIsLive' }), + LOCKABLE_CONNECTOR_STATUSES: ['active', 'error', 'pending'], })) import { @@ -71,6 +73,8 @@ describe('connector sync queue', () => { kbDeletedAt: null, }, ]) + /** `markSyncPending` now reports whether it actually took the queue entry. */ + dbChainMockFns.returning.mockResolvedValue([{ id: 'connector-1' }]) mockIsTriggerAvailable.mockReturnValue(true) mockResolveTriggerRegion.mockResolvedValue('us-east-1') mockTrigger.mockResolvedValue({ id: 'run-1' }) @@ -201,25 +205,26 @@ describe('connector sync queue', () => { hasMockCondition( where, (node: MockCondition) => - node.type === 'ne' && node.left === schemaMock.knowledgeConnector.status + node.type === 'inArray' && node.column === schemaMock.knowledgeConnector.status ) ) expect(queueWhere).toBeDefined() - expect( - hasMockCondition( - queueWhere, - (node: MockCondition) => node.type === 'ne' && node.right === 'pending' - ) - ).toBe(false) + const lockable = flattenMockConditions(queueWhere).find( + (node: MockCondition) => + node.type === 'inArray' && node.column === schemaMock.knowledgeConnector.status + )?.values as string[] | undefined - /** A live run still owns its row: demoting it to `pending` would strand it. */ - expect( - hasMockCondition( - queueWhere, - (node: MockCondition) => node.type === 'ne' && node.right === 'syncing' - ) - ).toBe(true) + /** The create path is born `pending`, so queueing must still take that row. */ + expect(lockable).toContain('pending') + + /** + * A live run owns its row, and a paused or disabled connector must not be + * pulled back into a queued sync by a dispatch that raced the status change. + */ + expect(lockable).not.toContain('syncing') + expect(lockable).not.toContain('paused') + expect(lockable).not.toContain('disabled') }) it('releases the queued connector when the hand-off throws', async () => { diff --git a/apps/sim/lib/knowledge/connectors/queue.ts b/apps/sim/lib/knowledge/connectors/queue.ts index deafdf866d4..44e617c1b69 100644 --- a/apps/sim/lib/knowledge/connectors/queue.ts +++ b/apps/sim/lib/knowledge/connectors/queue.ts @@ -5,13 +5,17 @@ import { toError } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' import { isRecordLike } from '@sim/utils/object' import { tasks } from '@trigger.dev/sdk' -import { and, eq, ne } from 'drizzle-orm' +import { and, eq, inArray } from 'drizzle-orm' import { assertBillingAttributionSnapshot, type BillingAttributionSnapshot, } from '@/lib/billing/core/billing-attribution' import { resolveTriggerRegion } from '@/lib/core/async-jobs/region' -import { connectorIsLive, executeSync } from '@/lib/knowledge/connectors/sync-engine' +import { + connectorIsLive, + executeSync, + LOCKABLE_CONNECTOR_STATUSES, +} from '@/lib/knowledge/connectors/sync-engine' import { isTriggerAvailable } from '@/lib/knowledge/documents/service' const logger = createLogger('ConnectorSyncQueue') @@ -99,16 +103,20 @@ export const SYNC_DISPATCH_FAILED_ERROR = 'Sync could not be queued' * extra UPDATE per connector creation, which is rare; the scheduler's own * dispatches only ever see `active`/`error` rows and are unaffected. * - * Skips a `syncing` row for the reason the lock acquisition does: a run may - * have taken the lock between the caller's read and this write, and demoting a - * live run to `pending` would strand it, since the reaper only looks at - * `syncing` rows. + * Takes the entry only from a status a run may start from — the same + * {@link LOCKABLE_CONNECTOR_STATUSES} the lock acquisition uses, so queueing and + * starting agree on one rule. The dispatch-side guards run before this write and + * cannot see a status change that races it: without the allowlist, pausing a + * connector in the window between "Sync now" being accepted and this UPDATE + * landing would be silently overwritten back to `pending`. Returns `null` when + * it takes nothing, so the caller can skip a hand-off that would only be refused + * at the lock. */ -async function markSyncPending(connectorId: string): Promise { +async function markSyncPending(connectorId: string): Promise { const dispatchToken = generateId() const now = new Date() - await db + const taken = await db .update(knowledgeConnector) .set({ status: 'pending', @@ -119,12 +127,13 @@ async function markSyncPending(connectorId: string): Promise { .where( and( eq(knowledgeConnector.id, connectorId), - ne(knowledgeConnector.status, 'syncing'), + inArray(knowledgeConnector.status, LOCKABLE_CONNECTOR_STATUSES), connectorIsLive() ) ) + .returning({ id: knowledgeConnector.id }) - return dispatchToken + return taken.length > 0 ? dispatchToken : null } /** @@ -273,6 +282,13 @@ export async function dispatchSync( if (isTriggerAvailable()) { const dispatchToken = await markSyncPending(connectorId) + if (!dispatchToken) { + logger.info('Skipping sync dispatch: connector is not accepting a queued sync', { + connectorId, + requestId, + }) + return + } /** * Everything between taking the queue entry and the hand-off landing has to @@ -295,6 +311,13 @@ export async function dispatchSync( } const dispatchToken = await markSyncPending(connectorId) + if (!dispatchToken) { + logger.info('Skipping sync execution: connector is not accepting a queued sync', { + connectorId, + requestId, + }) + return + } executeSync(connectorId, { fullSync: payload.fullSync, diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts index f00e261c15d..f41f5027e4b 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts @@ -1747,6 +1747,31 @@ describe('buildSyncLockAcquisition', () => { }) }) +describe('LOCKABLE_CONNECTOR_STATUSES', () => { + it('refuses to start a run on a connector someone paused or disabled', async () => { + const { LOCKABLE_CONNECTOR_STATUSES } = await import('@/lib/knowledge/connectors/sync-engine') + + /** + * The queue outlives the decision to sync. A connector paused *after* its + * run was queued still has a task in flight, and a bare not-syncing test + * let that task take the lock and then write `active` over the pause — so + * one pause during the queue window was silently undone. The dispatch-side + * guards cannot see a status change that happens after they ran; this CAS + * is the only point that can. + */ + expect(LOCKABLE_CONNECTOR_STATUSES).not.toContain('paused') + expect(LOCKABLE_CONNECTOR_STATUSES).not.toContain('disabled') + + /** A queued run must still be lockable, or nothing would ever sync. */ + expect(LOCKABLE_CONNECTOR_STATUSES).toContain('pending') + expect(LOCKABLE_CONNECTOR_STATUSES).toContain('active') + expect(LOCKABLE_CONNECTOR_STATUSES).toContain('error') + + /** `syncing` is already locked; re-locking it would strand the live run. */ + expect(LOCKABLE_CONNECTOR_STATUSES).not.toContain('syncing') + }) +}) + describe('shouldHeartbeatSyncLock', () => { it('beats once the interval has elapsed', async () => { const { shouldHeartbeatSyncLock } = await import('@/lib/knowledge/connectors/sync-engine') diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index 362c6ad18fd..76402b2c285 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -691,6 +691,19 @@ export function holdsSyncLockToken(connectorId: string, syncLockToken: string) { * unrelated write to the row, so a config edit on a wedged connector used to * renew the lock it was meant to recover. */ +/** + * The statuses a run may take the lock from. + * + * An allowlist rather than `ne(status, 'syncing')`, because the queue outlives + * the decision to sync: a connector paused or disabled *after* its run was + * queued still had a task in flight, and a bare not-syncing test let that task + * lock the row and then write its own terminal status over the pause. This CAS + * is the single point where a run decides to start, so it is where the refusal + * belongs — the dispatch-side guards cannot see a status change that happens + * after they ran. + */ +export const LOCKABLE_CONNECTOR_STATUSES = ['active', 'error', 'pending'] as const + export function buildSyncLockAcquisition(syncLogId: string, now: Date) { return { status: 'syncing' as const, @@ -1567,7 +1580,7 @@ export async function executeSync( .where( and( eq(knowledgeConnector.id, connectorId), - ne(knowledgeConnector.status, 'syncing'), + inArray(knowledgeConnector.status, LOCKABLE_CONNECTOR_STATUSES), isNull(knowledgeConnector.archivedAt), isNull(knowledgeConnector.deletedAt) ) @@ -1575,6 +1588,25 @@ export async function executeSync( .returning({ id: knowledgeConnector.id }) if (lockResult.length === 0) { + /** + * Distinguishes the two ways the CAS can find no row. Costs one read on a + * path that already decided not to work, and the alternative is reporting a + * connector someone paused as a concurrency conflict. + */ + const [current] = await db + .select({ status: knowledgeConnector.status }) + .from(knowledgeConnector) + .where(eq(knowledgeConnector.id, connectorId)) + .limit(1) + + if (current?.status === 'paused' || current?.status === 'disabled') { + logger.info('Connector is not accepting syncs, skipping', { + connectorId, + status: current.status, + }) + return { ...result, error: 'connector_not_syncable' } + } + logger.info('Sync already in progress, skipping', { connectorId }) // Reported as an error so the task wrapper's `success: !result.error` does not // present a skipped run as a successful zero-document sync. From b2e20179be090b9ea4b0bd9a4e2039a2b90285c4 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 21 Aug 2026 20:12:33 -0700 Subject: [PATCH 3/5] fix(kb): make a queued sync prove it is the run that was queued MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `markSyncPending` minted an ownership token but only `releaseFailedDispatch` checked it, so the worker could consume a queue entry that was not its own. A task delayed past its lease is reclaimed and replaced; the status check alone let that stale task take the replacement's entry and run superseded options — a plain sync where the user had just asked for a full resync — while the replacement was turned away as `sync_in_progress`. Carries the token in the task payload and matches it at lock acquisition, the same discipline `holdsSyncLockToken` already applies to the `syncing` phase, extended to the phase before it. A superseded run is now reported as such rather than as a concurrency conflict. The payload field is optional for the rollout window only: tasks already in the queue carry no token, and stranding them would be worse than letting them fall back to the status check for one deploy. --- .../background/knowledge-connector-sync.ts | 9 +++-- .../lib/knowledge/connectors/queue.test.ts | 33 +++++++++++++++++++ apps/sim/lib/knowledge/connectors/queue.ts | 23 ++++++++++--- .../lib/knowledge/connectors/sync-engine.ts | 29 +++++++++++++++- 4 files changed, 87 insertions(+), 7 deletions(-) diff --git a/apps/sim/background/knowledge-connector-sync.ts b/apps/sim/background/knowledge-connector-sync.ts index 6efe6dbd26e..1076103324c 100644 --- a/apps/sim/background/knowledge-connector-sync.ts +++ b/apps/sim/background/knowledge-connector-sync.ts @@ -10,13 +10,18 @@ import { CONNECTOR_SYNC_MAX_DURATION_SECONDS } from '@/lib/knowledge/connectors/ const logger = createLogger('TriggerKnowledgeConnectorSync') export async function executeConnectorSyncJob(payload: unknown) { - const { connectorId, fullSync, rehydrate, requestId, billingAttribution } = + const { connectorId, fullSync, rehydrate, requestId, billingAttribution, dispatchToken } = assertConnectorSyncPayload(payload) logger.info(`[${requestId}] Starting connector sync: ${connectorId}`) try { - const result = await executeSync(connectorId, { billingAttribution, fullSync, rehydrate }) + const result = await executeSync(connectorId, { + billingAttribution, + fullSync, + rehydrate, + dispatchToken, + }) logger.info(`[${requestId}] Connector sync completed`, { connectorId, diff --git a/apps/sim/lib/knowledge/connectors/queue.test.ts b/apps/sim/lib/knowledge/connectors/queue.test.ts index c18314cfff5..3d1ae6f89fe 100644 --- a/apps/sim/lib/knowledge/connectors/queue.test.ts +++ b/apps/sim/lib/knowledge/connectors/queue.test.ts @@ -99,6 +99,8 @@ describe('connector sync queue', () => { rehydrate: undefined, requestId: 'request-1', billingAttribution: BILLING_ATTRIBUTION, + /** Minted per dispatch; its own test asserts it matches the queue entry. */ + dispatchToken: expect.any(String), }, { tags: [ @@ -295,6 +297,37 @@ describe('connector sync queue', () => { ) }) + it('stamps the queue entry token onto the task so the worker can prove ownership', async () => { + await dispatchSync('connector-1', { + billingAttribution: BILLING_ATTRIBUTION, + requestId: 'request-1', + }) + + /** + * Without this the status check alone lets a task delayed past its lease + * take the replacement entry the reaper's re-dispatch created, running + * superseded options while the replacement is turned away. + */ + const queuedToken = (dbChainMockFns.set.mock.calls[0][0] as Record) + .syncLockToken + expect(mockTrigger).toHaveBeenCalledWith( + 'knowledge-connector-sync', + expect.objectContaining({ dispatchToken: queuedToken }), + expect.anything() + ) + }) + + it('tolerates a payload queued before the token existed', () => { + /** In-flight tasks from before this field shipped must not be stranded. */ + expect( + assertConnectorSyncPayload({ + connectorId: 'connector-1', + requestId: 'request-1', + billingAttribution: BILLING_ATTRIBUTION, + }).dispatchToken + ).toBeUndefined() + }) + it('rejects legacy payloads without billing attribution', () => { expect(() => assertConnectorSyncPayload({ diff --git a/apps/sim/lib/knowledge/connectors/queue.ts b/apps/sim/lib/knowledge/connectors/queue.ts index 44e617c1b69..a1c408f26e9 100644 --- a/apps/sim/lib/knowledge/connectors/queue.ts +++ b/apps/sim/lib/knowledge/connectors/queue.ts @@ -33,6 +33,15 @@ export interface ConnectorSyncPayload { rehydrate?: boolean requestId: string billingAttribution: BillingAttributionSnapshot + /** + * The queue entry this task is allowed to consume, proving the run it starts + * is the one that was queued for it. + * + * Optional only for the rollout window: tasks queued before this field + * existed carry no token, and the lock falls back to the status check alone + * for them rather than stranding work already in the queue. + */ + dispatchToken?: string } export interface DispatchSyncOptions { @@ -62,6 +71,9 @@ export function assertConnectorSyncPayload(value: unknown): ConnectorSyncPayload if (value.rehydrate !== undefined && typeof value.rehydrate !== 'boolean') { throw new Error('Connector sync payload rehydrate must be a boolean when provided') } + if (value.dispatchToken !== undefined && !isNonEmptyString(value.dispatchToken)) { + throw new Error('Connector sync payload dispatchToken must be a string when provided') + } if (value.billingAttribution === undefined) { throw new Error('Connector sync payload requires billing attribution') } @@ -72,6 +84,7 @@ export function assertConnectorSyncPayload(value: unknown): ConnectorSyncPayload rehydrate: value.rehydrate as boolean | undefined, requestId: value.requestId, billingAttribution: assertBillingAttributionSnapshot(value.billingAttribution), + dispatchToken: value.dispatchToken as string | undefined, } } @@ -298,10 +311,11 @@ export async function dispatchSync( * `pending` until the reaper's TTL. */ try { - await tasks.trigger('knowledge-connector-sync', payload, { - tags, - region: await resolveTriggerRegion(), - }) + await tasks.trigger( + 'knowledge-connector-sync', + { ...payload, dispatchToken }, + { tags, region: await resolveTriggerRegion() } + ) } catch (error) { await releaseFailedDispatch(connectorId, dispatchToken, error) throw error @@ -323,6 +337,7 @@ export async function dispatchSync( fullSync: payload.fullSync, rehydrate: payload.rehydrate, billingAttribution: payload.billingAttribution, + dispatchToken, }).catch(async (error) => { logger.error(`Sync failed for connector ${connectorId}`, { error: toError(error).message, diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index 76402b2c285..ce0b0061dec 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -1480,6 +1480,11 @@ export async function executeSync( billingAttribution: BillingAttributionSnapshot fullSync?: boolean rehydrate?: boolean + /** + * The queue entry this run is allowed to consume. Absent only for tasks + * queued before the token existed; see {@link ConnectorSyncPayload}. + */ + dispatchToken?: string } ): Promise { const billingAttribution = assertBillingAttributionSnapshot(options?.billingAttribution) @@ -1581,6 +1586,20 @@ export async function executeSync( and( eq(knowledgeConnector.id, connectorId), inArray(knowledgeConnector.status, LOCKABLE_CONNECTOR_STATUSES), + /** + * Proves this run is consuming the queue entry that was made for it. + * + * A task delayed past the lease is reclaimed and replaced, and the + * status check alone would let that stale task take the replacement's + * entry — running superseded options (a plain sync where the user had + * just asked for a full resync) while the replacement is turned away as + * `sync_in_progress`. Matching the token is the same discipline + * {@link holdsSyncLockToken} already applies to the `syncing` phase, + * extended to the phase before it. + */ + ...(options.dispatchToken + ? [eq(knowledgeConnector.syncLockToken, options.dispatchToken)] + : []), isNull(knowledgeConnector.archivedAt), isNull(knowledgeConnector.deletedAt) ) @@ -1594,11 +1613,19 @@ export async function executeSync( * connector someone paused as a concurrency conflict. */ const [current] = await db - .select({ status: knowledgeConnector.status }) + .select({ + status: knowledgeConnector.status, + syncLockToken: knowledgeConnector.syncLockToken, + }) .from(knowledgeConnector) .where(eq(knowledgeConnector.id, connectorId)) .limit(1) + if (options.dispatchToken && current?.syncLockToken !== options.dispatchToken) { + logger.info('Sync superseded by a newer dispatch, skipping', { connectorId }) + return { ...result, error: 'dispatch_superseded' } + } + if (current?.status === 'paused' || current?.status === 'disabled') { logger.info('Connector is not accepting syncs, skipping', { connectorId, From b2aa5c06f2391f8b4b6ff01cb8229915baccc9dd Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 21 Aug 2026 20:18:49 -0700 Subject: [PATCH 4/5] fix(kb): report a paused connector as paused, not superseded MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pausing a queued connector releases its token, so testing ownership before status reported every pause-while-queued — the common case — as a superseded dispatch. The mismatch is the symptom there; the status is the reason. --- .../sim/lib/knowledge/connectors/sync-engine.ts | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index ce0b0061dec..3aafd238c3e 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -1621,11 +1621,13 @@ export async function executeSync( .where(eq(knowledgeConnector.id, connectorId)) .limit(1) - if (options.dispatchToken && current?.syncLockToken !== options.dispatchToken) { - logger.info('Sync superseded by a newer dispatch, skipping', { connectorId }) - return { ...result, error: 'dispatch_superseded' } - } - + /** + * Status is checked before ownership because pausing a queued connector + * releases its token, so a mismatch is the *symptom* there and the status is + * the actual reason. Testing ownership first would report every + * pause-while-queued — the common case — as a superseded dispatch, losing + * the distinction this branch exists to draw. + */ if (current?.status === 'paused' || current?.status === 'disabled') { logger.info('Connector is not accepting syncs, skipping', { connectorId, @@ -1634,6 +1636,11 @@ export async function executeSync( return { ...result, error: 'connector_not_syncable' } } + if (options.dispatchToken && current?.syncLockToken !== options.dispatchToken) { + logger.info('Sync superseded by a newer dispatch, skipping', { connectorId }) + return { ...result, error: 'dispatch_superseded' } + } + logger.info('Sync already in progress, skipping', { connectorId }) // Reported as an error so the task wrapper's `success: !result.error` does not // present a skipped run as a successful zero-document sync. From 70f25bbe5306e2d25abf2f65d316c105c4f40ae1 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 21 Aug 2026 20:42:46 -0700 Subject: [PATCH 5/5] fix(kb): stop a status update landing on a run that already started MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The update's guards ran against a row read moments earlier and the write carried no compare-and-set, so a worker taking the lock in between meant the write landed on a `syncing` row — overwriting the run's status and, because leaving `pending` also clears the lock columns, wiping the token its heartbeat and terminal write match on. That stranded a sync that had already begun. The write is now conditional on the status the request was authorized against, and a lost race is reported as a conflict rather than "not found". Also restores the in-flight guard on the pause control. The optimistic status flip relabels it Pause -> Resume immediately, so a second click could send `active` before the first pause settled and resume a connector the user meant to pause. Read from the mutation's own pending state rather than the local id set this PR removed — React Query already knows which row is in flight. --- .../connectors-section/connectors-section.tsx | 17 ++++++- .../orchestration/connectors.test.ts | 46 ++++++++++++++++++- .../lib/knowledge/orchestration/connectors.ts | 19 +++++++- 3 files changed, 79 insertions(+), 3 deletions(-) diff --git a/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/connectors-section/connectors-section.tsx b/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/connectors-section/connectors-section.tsx index 4a7fa356332..10b0a8b6e4a 100644 --- a/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/connectors-section/connectors-section.tsx +++ b/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/connectors-section/connectors-section.tsx @@ -88,7 +88,11 @@ export function ConnectorsSection({ className, }: ConnectorsSectionProps) { const { mutate: triggerSync } = useTriggerSync() - const { mutate: updateConnector } = useUpdateConnector() + const { + mutate: updateConnector, + isPending: isUpdatingConnector, + variables: updatingVariables, + } = useUpdateConnector() const { mutate: deleteConnector, isPending: isDeleting } = useDeleteConnector() const deleteDocumentsId = useId() const [deleteTarget, setDeleteTarget] = useState(null) @@ -178,6 +182,14 @@ export function ConnectorsSection({ workspaceId={workspaceId} knowledgeBaseId={knowledgeBaseId} canEdit={canEdit} + /** + * The optimistic status flip relabels this control Pause -> Resume + * immediately, so without a guard a second click would send + * `active` before the first pause settles and resume a connector + * the user meant to pause. Read from the mutation rather than a + * local id set: React Query already knows which row is in flight. + */ + isUpdating={isUpdatingConnector && updatingVariables?.connectorId === connector.id} onSync={(rehydrate) => handleSync(connector.id, rehydrate)} onTogglePause={() => handleTogglePause(connector)} onEdit={() => setEditingConnector(connector)} @@ -234,6 +246,7 @@ interface ConnectorCardProps { workspaceId: string knowledgeBaseId: string canEdit: boolean + isUpdating: boolean onSync: (rehydrate?: boolean) => void onEdit: () => void onTogglePause: () => void @@ -245,6 +258,7 @@ function ConnectorCard({ workspaceId, knowledgeBaseId, canEdit, + isUpdating, onSync, onEdit, onTogglePause, @@ -437,6 +451,7 @@ function ConnectorCard({ variant='ghost' className={CONNECTOR_ACTION_BUTTON_CLASSES} onClick={onTogglePause} + disabled={isUpdating} > {connector.status === 'paused' || connector.status === 'disabled' ? ( diff --git a/apps/sim/lib/knowledge/orchestration/connectors.test.ts b/apps/sim/lib/knowledge/orchestration/connectors.test.ts index 1a0f9382053..34a88f54a84 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.test.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.test.ts @@ -2,7 +2,14 @@ * @vitest-environment node */ import { document } from '@sim/db/schema' -import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing' +import { + dbChainMockFns, + hasMockCondition, + type MockCondition, + queueTableRows, + resetDbChainMock, + schemaMock, +} from '@sim/testing' import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest' const { @@ -289,6 +296,43 @@ describe('performUpdateKnowledgeConnector', () => { expect(outcome).toMatchObject({ success: true }) expect(mockRecordAudit).not.toHaveBeenCalled() }) + + it('refuses an update whose status moved after the guards ran', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'conn-1', connectorType: 'notion', status: 'pending' }, + ]) + /** The CAS matches nothing because a worker took the lock in between. */ + dbChainMockFns.returning.mockResolvedValueOnce([]) + + const outcome = await performUpdateKnowledgeConnector({ + ...ACTOR, + knowledgeBase: KB, + connectorId: 'conn-1', + updates: { status: 'paused' }, + }) + + /** + * Leaving `pending` clears the lock columns. Landing that on a row that has + * since gone `syncing` would wipe the token the run's heartbeat and terminal + * write match on, stranding a sync that had already started. + */ + expect(outcome).toMatchObject({ success: false, errorCode: 'conflict' }) + expect(mockRecordAudit).not.toHaveBeenCalled() + + /** + * The mock does not evaluate predicates, so assert the clause itself is + * present — an empty `returning()` alone would pass without it. + */ + expect( + hasMockCondition( + dbChainMockFns.where.mock.calls.at(-1)?.[0], + (node: MockCondition) => + node.type === 'eq' && + node.left === schemaMock.knowledgeConnector.status && + node.right === 'pending' + ) + ).toBe(true) + }) }) describe('performSyncKnowledgeConnector', () => { diff --git a/apps/sim/lib/knowledge/orchestration/connectors.ts b/apps/sim/lib/knowledge/orchestration/connectors.ts index 9e9ee012776..1e7c8a175db 100644 --- a/apps/sim/lib/knowledge/orchestration/connectors.ts +++ b/apps/sim/lib/knowledge/orchestration/connectors.ts @@ -534,6 +534,17 @@ export async function performUpdateKnowledgeConnector( and( eq(knowledgeConnector.id, connectorId), eq(knowledgeConnector.knowledgeBaseId, kb.id), + /** + * Compare-and-set on the status this request was authorized against. + * + * The guards above ran on a row read moments earlier, and a worker can + * take the lock in between. Without this the write lands on a row that + * is now `syncing`: it would overwrite the run's status and — because + * leaving `pending` also clears the lock columns — wipe the token its + * heartbeat and terminal write match on, stranding a sync that had + * already started. + */ + eq(knowledgeConnector.status, existing.status), isNull(knowledgeConnector.archivedAt), isNull(knowledgeConnector.deletedAt) ) @@ -541,7 +552,13 @@ export async function performUpdateKnowledgeConnector( .returning() if (!row) { - return fail('Connector not found', 'not_found') + /** + * Either the connector went away or its status moved under us. Both are + * conflicts rather than "not found": the caller's decision was made + * against a state that no longer holds, and re-reading to tell them apart + * would race the same way. + */ + return fail('Connector changed while the update was being applied', 'conflict') } updated = row } catch (error) {