From 74140e07dd884b3018ca2a407b993cdd30033f77 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 18 Sep 2026 16:32:47 -0700 Subject: [PATCH 1/2] fix(knowledge): stop recovery from re-dispatching documents whose runs are still queued --- .../connector-lifecycle-locks.integration.ts | 33 +++++++++++ .../stored-document-recovery.integration.ts | 43 +++++++++++++++ .../connectors/sync-content-pass.test.ts | 24 ++++++++ .../knowledge/connectors/sync-content-pass.ts | 1 + .../knowledge/connectors/sync-primitives.ts | 6 +- .../processing-continuation-dispatch.test.ts | 17 ++++++ .../processing-continuation-dispatch.ts | 4 +- .../documents/processing-lane.test.ts | 55 ++++++++++++++++++- .../knowledge/documents/processing-lane.ts | 33 +++++++++++ .../documents/processing-queue.test.ts | 17 ++++++ .../documents/processing-recovery-policy.ts | 14 ++++- .../documents/processing-recovery.ts | 7 ++- .../documents/retry-processing-grace.test.ts | 5 +- apps/sim/lib/knowledge/documents/service.ts | 9 ++- apps/sim/lib/knowledge/documents/types.ts | 15 ++++- 15 files changed, 269 insertions(+), 14 deletions(-) diff --git a/apps/sim/lib/knowledge/__integration__/connector-lifecycle-locks.integration.ts b/apps/sim/lib/knowledge/__integration__/connector-lifecycle-locks.integration.ts index 8d59264625a..1b2114886e4 100644 --- a/apps/sim/lib/knowledge/__integration__/connector-lifecycle-locks.integration.ts +++ b/apps/sim/lib/knowledge/__integration__/connector-lifecycle-locks.integration.ts @@ -259,4 +259,37 @@ describe('source lifecycle KB guards', () => { .where(eq(document.id, retryDocumentId)) expect(unchanged.status).toBe('failed') }) + + it('moves an expired queued generation charge to its replacement instead of adding one', async () => { + const stampedAt = new Date(Date.now() - 24 * 60 * 60_000) + await db + .update(document) + .set({ + processingStatus: 'pending', + processingAttempts: 2, + processingQueuedAt: stampedAt, + processingQueueToken: 'expired-generation', + processingCompletedAt: null, + }) + .where(eq(document.id, retryDocumentId)) + await run('recover') + const [row] = await db + .select({ + status: document.processingStatus, + attempts: document.processingAttempts, + token: document.processingQueueToken, + }) + .from(document) + .where(eq(document.id, retryDocumentId)) + expect(row).toEqual({ status: 'pending', attempts: 1, token: null }) + }) + + it('keeps the charge of an attempt that reached a worker', async () => { + await run('recover') + const [row] = await db + .select({ attempts: document.processingAttempts }) + .from(document) + .where(eq(document.id, retryDocumentId)) + expect(row.attempts).toBe(1) + }) }) diff --git a/apps/sim/lib/knowledge/__integration__/stored-document-recovery.integration.ts b/apps/sim/lib/knowledge/__integration__/stored-document-recovery.integration.ts index 8d3a3f45a09..83eb212a2fa 100644 --- a/apps/sim/lib/knowledge/__integration__/stored-document-recovery.integration.ts +++ b/apps/sim/lib/knowledge/__integration__/stored-document-recovery.integration.ts @@ -244,6 +244,49 @@ describe('independent recovery of retained connector documents', () => { } }) + it('replaces expired queued generations without spending the budget, then indexes once', async () => { + const ids = await seed() + const file = await failedFile(ids) + const expire = (token: string) => + db + .update(document) + .set({ + processingStatus: 'pending', + processingCompletedAt: null, + processingError: null, + processingQueuedAt: old(), + processingQueueToken: token, + }) + .where(eq(document.id, file.documentId)) + let generation = 'old-fixture-generation' + await expire(generation) + for (let cycle = 0; cycle <= MAX_PROCESSING_ATTEMPTS; cycle++) { + expect(await recoverKnowledgeDocumentProcessing()).toBe(1) + const [row] = await db.select().from(document).where(eq(document.id, file.documentId)) + expect(row.processingAttempts).toBe(1) + expect(row.processingQueueToken).not.toBe(generation) + generation = row.processingQueueToken! + await expire(generation) + } + await db + .update(document) + .set({ processingQueuedAt: new Date() }) + .where(eq(document.id, file.documentId)) + expect(await recoverKnowledgeDocumentProcessing()).toBe(0) + + const events = await eventsFor(ids) + expect(events).toHaveLength(MAX_PROCESSING_ATTEMPTS + 1) + const before = fixture.embeddingCalls + for (const event of events) { + expect( + await outbox.processOutboxEventById(event.id, knowledgeDocumentProcessingOutboxHandlers) + ).toBe('completed') + } + expect(fixture.embeddingCalls).toBe(before + 1) + const [indexed] = await db.select().from(document).where(eq(document.id, file.documentId)) + expect(indexed.processingStatus, indexed.processingError ?? undefined).toBe('completed') + }) + it('keeps permanent, excluded, paused, fresh, deferred, and expired work out of automatic recovery', async () => { const ids = await seed() const file = await failedFile(ids) diff --git a/apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts b/apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts index 6539a0dfe88..ae813a75162 100644 --- a/apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts +++ b/apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts @@ -1,6 +1,8 @@ /** @vitest-environment node */ import { dbChainMockFns, + hasMockCondition, + type MockCondition, queueTableRows, resetDbChainMock as resetDatabaseMock, schemaMock, @@ -643,6 +645,28 @@ describe('content pass checkpoint intent', () => { ) }) +describe('resurrecting verified listed documents', () => { + it('only writes documents that are actually tombstoned', async () => { + sourceBody = { value: '

Current content

' } + await runPass({ access: 'admin' }) + const resurrect = dbChainMockFns.set.mock.invocationCallOrder.find((_order, index) => { + const [values] = dbChainMockFns.set.mock.calls[index] + return Object.keys(values).length === 1 && values.deletedAt === null + }) + expect(resurrect).toBeDefined() + const whereIndex = dbChainMockFns.where.mock.invocationCallOrder.findIndex( + (order) => order > resurrect! + ) + expect( + hasMockCondition( + dbChainMockFns.where.mock.calls[whereIndex][0], + (node: MockCondition) => + node.type === 'isNotNull' && node.column === schemaMock.document.deletedAt + ) + ).toBe(true) + }) +}) + describe('permission refresh through the shared content pass', () => { const current: StoredPage = { ...EXISTING, diff --git a/apps/sim/lib/knowledge/connectors/sync-content-pass.ts b/apps/sim/lib/knowledge/connectors/sync-content-pass.ts index 68b482446d1..8532a82cb8b 100644 --- a/apps/sim/lib/knowledge/connectors/sync-content-pass.ts +++ b/apps/sim/lib/knowledge/connectors/sync-content-pass.ts @@ -290,6 +290,7 @@ export async function runConnectorContentPass(input: ContentPassInput) { and( eq(document.connectorId, input.connectorId), inArray(document.externalId, verified.slice(offset, offset + 500)), + isNotNull(document.deletedAt), isNotNull(document.contentHash), isNull(document.archivedAt) ) diff --git a/apps/sim/lib/knowledge/connectors/sync-primitives.ts b/apps/sim/lib/knowledge/connectors/sync-primitives.ts index 562378d26b0..91e9cdae231 100644 --- a/apps/sim/lib/knowledge/connectors/sync-primitives.ts +++ b/apps/sim/lib/knowledge/connectors/sync-primitives.ts @@ -27,7 +27,10 @@ import { persistSkippedRetryHashes, updateDocument, } from '@/lib/knowledge/connectors/sync-persistence' -import { documentProcessingRecoveryCondition } from '@/lib/knowledge/documents/processing-recovery-policy' +import { + documentProcessingRecoveryCondition, + releaseUnclaimedDispatchAttempt, +} from '@/lib/knowledge/documents/processing-recovery-policy' import { DOCUMENT_PROCESSING_STALE_THRESHOLD_MS } from '@/lib/knowledge/documents/processing-timeouts.server' import type { DocumentData } from '@/lib/knowledge/documents/service' import { isTriggerAvailable, processDocumentsWithQueue } from '@/lib/knowledge/documents/service' @@ -1446,6 +1449,7 @@ export async function sweepStuckDocuments(input: SweepStuckDocumentsInput): Prom processingDeferredUntil: null, processingCompletedAt: null, processingError: null, + processingAttempts: releaseUnclaimedDispatchAttempt(sweepEvaluatedAt), chunkCount: 0, tokenCount: 0, characterCount: 0, diff --git a/apps/sim/lib/knowledge/documents/processing-continuation-dispatch.test.ts b/apps/sim/lib/knowledge/documents/processing-continuation-dispatch.test.ts index 83abbbad122..b5df897a695 100644 --- a/apps/sim/lib/knowledge/documents/processing-continuation-dispatch.test.ts +++ b/apps/sim/lib/knowledge/documents/processing-continuation-dispatch.test.ts @@ -19,6 +19,7 @@ import type { DocumentProcessingLane, DocumentProcessingPayload, } from '@/lib/knowledge/documents/processing-payload' +import { QUEUED_DISPATCH_GRACE_MS } from '@/lib/knowledge/documents/types' const BILLING_ATTRIBUTION = { actorUserId: 'user-1', @@ -73,4 +74,20 @@ describe('dispatchDocumentProcessingContinuation', () => { concurrencyKey: 'organization:org-1', }) }) + + /** Trigger.dev starts a delayed run's TTL when the delay ends; the deadline follows the stamp. */ + it('expires a continuation unstarted before recovery may replace it', async () => { + const deferredUntil = new Date(Date.now() + 15 * 60_000) + await dispatchDocumentProcessingContinuation( + { ...payload('backfill'), processingQueuedAt: deferredUntil.toISOString() }, + deferredUntil, + 'continuation-key', + true + ) + + const { ttl } = mockTrigger.mock.calls[0][2] + expect(mockTrigger.mock.calls[0][2]).toMatchObject({ delay: deferredUntil }) + expect(ttl).toBeGreaterThan(0) + expect(ttl * 1000).toBeLessThan(QUEUED_DISPATCH_GRACE_MS) + }) }) diff --git a/apps/sim/lib/knowledge/documents/processing-continuation-dispatch.ts b/apps/sim/lib/knowledge/documents/processing-continuation-dispatch.ts index 74335ad8d5c..21aa15e28a3 100644 --- a/apps/sim/lib/knowledge/documents/processing-continuation-dispatch.ts +++ b/apps/sim/lib/knowledge/documents/processing-continuation-dispatch.ts @@ -5,7 +5,7 @@ import { resolveTriggerRegion } from '@/lib/core/async-jobs/region' import { env } from '@/lib/core/config/env' import { isTriggerDevEnabled } from '@/lib/core/config/env-flags' import { isInsideTriggerRun } from '@/lib/core/config/trigger-runtime' -import { documentProcessingQueueOptions } from '@/lib/knowledge/documents/processing-lane' +import { documentProcessingRunOptions } from '@/lib/knowledge/documents/processing-lane' import type { DocumentProcessingPayload } from '@/lib/knowledge/documents/processing-payload' export interface DocumentProcessingContinuation { @@ -37,7 +37,7 @@ export async function dispatchDocumentProcessingContinuation( * deferred on quota resume as interactive work and escape the tenant's * bulk ceiling — the retry path would become the way around the limit. */ - ...documentProcessingQueueOptions(payload), + ...documentProcessingRunOptions(payload, deferredUntil), region, }) return diff --git a/apps/sim/lib/knowledge/documents/processing-lane.test.ts b/apps/sim/lib/knowledge/documents/processing-lane.test.ts index 24178163dd4..29319024f47 100644 --- a/apps/sim/lib/knowledge/documents/processing-lane.test.ts +++ b/apps/sim/lib/knowledge/documents/processing-lane.test.ts @@ -1,14 +1,19 @@ /** * @vitest-environment node */ -import { describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { BACKFILL_PROCESSING_QUEUE_NAME, documentProcessingQueueOptions, + documentProcessingRunExpiry, INTERACTIVE_PROCESSING_QUEUE_NAME, } from '@/lib/knowledge/documents/processing-lane' import type { DocumentProcessingPayload } from '@/lib/knowledge/documents/processing-payload' import { resolveDocumentProcessingLane } from '@/lib/knowledge/documents/processing-payload' +import { + QUEUED_DISPATCH_GRACE_MS, + QUEUED_DISPATCH_START_DEADLINE_MS, +} from '@/lib/knowledge/documents/types' const BILLING_ATTRIBUTION = { actorUserId: 'user-1', @@ -130,3 +135,51 @@ describe('documentProcessingQueueOptions', () => { ) }) }) + +describe('documentProcessingRunExpiry', () => { + const NOW = new Date('2026-09-18T12:00:00.000Z') + + beforeEach(() => { + vi.useFakeTimers() + vi.setSystemTime(NOW) + }) + afterEach(() => { + vi.useRealTimers() + }) + + it('expires an undelayed run at the start deadline, before recovery may replace it', () => { + expect(QUEUED_DISPATCH_START_DEADLINE_MS).toBeLessThan(QUEUED_DISPATCH_GRACE_MS) + expect(documentProcessingRunExpiry({ processingQueuedAt: NOW.toISOString() })).toEqual({ + ttl: QUEUED_DISPATCH_START_DEADLINE_MS / 1000, + }) + }) + + it('gives a late relay only the time its generation has left', () => { + const stampedAt = new Date(NOW.getTime() - 60 * 60_000) + expect(documentProcessingRunExpiry({ processingQueuedAt: stampedAt.toISOString() })).toEqual({ + ttl: (QUEUED_DISPATCH_START_DEADLINE_MS - 60 * 60_000) / 1000, + }) + }) + + it('expires a generation already past its deadline at the minimum', () => { + const stampedAt = new Date(NOW.getTime() - QUEUED_DISPATCH_GRACE_MS) + expect(documentProcessingRunExpiry({ processingQueuedAt: stampedAt.toISOString() })).toEqual({ + ttl: 1, + }) + }) + + /** Trigger.dev starts a delayed run's TTL when its delay ends, not when it is triggered. */ + it('counts a delayed run from the end of its delay', () => { + const deferredUntil = new Date(NOW.getTime() + 30 * 60_000) + expect( + documentProcessingRunExpiry( + { processingQueuedAt: deferredUntil.toISOString() }, + deferredUntil + ) + ).toEqual({ ttl: QUEUED_DISPATCH_START_DEADLINE_MS / 1000 }) + }) + + it('leaves a payload without a queue stamp on the queue default', () => { + expect(documentProcessingRunExpiry({})).toEqual({}) + }) +}) diff --git a/apps/sim/lib/knowledge/documents/processing-lane.ts b/apps/sim/lib/knowledge/documents/processing-lane.ts index 9a64f51113a..10323c514a9 100644 --- a/apps/sim/lib/knowledge/documents/processing-lane.ts +++ b/apps/sim/lib/knowledge/documents/processing-lane.ts @@ -3,6 +3,7 @@ import type { DocumentProcessingBillingContext, DocumentProcessingPayload, } from '@/lib/knowledge/documents/processing-payload' +import { QUEUED_DISPATCH_START_DEADLINE_MS } from '@/lib/knowledge/documents/types' /** * Queue backing the interactive lane. @@ -55,3 +56,35 @@ export function documentProcessingQueueOptions(payload: DocumentProcessingPayloa concurrencyKey: documentProcessingTenantKey(payload), } } + +/** + * Trigger.dev `ttl` expiring a run unstarted at its generation's start deadline + * (queue stamp + {@link QUEUED_DISPATCH_START_DEADLINE_MS}). Trigger.dev counts `ttl` + * from enqueue, which for a delayed run is `notBefore`. A generation already past its + * deadline gets the one-second minimum; one without a stamp keeps the queue default. + */ +export function documentProcessingRunExpiry( + payload: Pick, + notBefore?: Date +): { ttl?: number } { + if (!payload.processingQueuedAt) return {} + const deadline = + new Date(payload.processingQueuedAt).getTime() + QUEUED_DISPATCH_START_DEADLINE_MS + const enqueuedAt = Math.max(Date.now(), notBefore?.getTime() ?? 0) + return { ttl: Math.max(1, Math.floor((deadline - enqueuedAt) / 1000)) } +} + +/** + * Every Trigger.dev option a `knowledge-process-document` dispatch needs: its lane's + * per-tenant queue and its start deadline. Dispatch sites use this rather than the + * parts so none can enqueue a run that outlives recovery's grace. + */ +export function documentProcessingRunOptions( + payload: DocumentProcessingPayload, + notBefore?: Date +): { queue: string; concurrencyKey: string; ttl?: number } { + return { + ...documentProcessingQueueOptions(payload), + ...documentProcessingRunExpiry(payload, notBefore), + } +} diff --git a/apps/sim/lib/knowledge/documents/processing-queue.test.ts b/apps/sim/lib/knowledge/documents/processing-queue.test.ts index 64debf56f43..e8fe245d813 100644 --- a/apps/sim/lib/knowledge/documents/processing-queue.test.ts +++ b/apps/sim/lib/knowledge/documents/processing-queue.test.ts @@ -368,6 +368,23 @@ describe('processDocumentsWithQueue dispatch backend', () => { }) }) + it('expires each queued run before recovery may replace its generation', async () => { + markInsideTriggerRun() + + await processDocumentsWithQueue( + [DOCUMENT], + 'knowledge-base-1', + {}, + 'request-1', + BILLING_ATTRIBUTION, + 'backfill' + ) + + const [item] = mockBatchTrigger.mock.calls[0][1] + expect(item.options.ttl).toBeGreaterThan(0) + expect(item.options.ttl * 1000).toBeLessThan(QUEUED_DISPATCH_GRACE_MS) + }) + /** * The starvation this split exists to prevent: connector backfill must not be * able to occupy the queue a person's upload is admitted through. diff --git a/apps/sim/lib/knowledge/documents/processing-recovery-policy.ts b/apps/sim/lib/knowledge/documents/processing-recovery-policy.ts index 4c2bb0e9053..60331c64bb3 100644 --- a/apps/sim/lib/knowledge/documents/processing-recovery-policy.ts +++ b/apps/sim/lib/knowledge/documents/processing-recovery-policy.ts @@ -1,5 +1,5 @@ import { document } from '@sim/db/schema' -import { and, eq, gt, isNotNull, isNull, lt, lte, or, sql } from 'drizzle-orm' +import { and, eq, gt, isNotNull, isNull, lt, lte, or, type SQL, sql } from 'drizzle-orm' import { DOCUMENT_PROCESSING_STALE_THRESHOLD_MS } from '@/lib/knowledge/documents/processing-timeouts.server' import { MAX_PROCESSING_ATTEMPTS, QUEUED_DISPATCH_GRACE_MS } from '@/lib/knowledge/documents/types' @@ -39,3 +39,15 @@ export function documentProcessingRecoveryCondition( ) ) } + +/** + * `processingAttempts` with the charge of a replaced, never-claimed queued generation given + * back, so the replacement's own charge does not spend the budget on queue wait. Only a + * `pending`, stamped, non-deferred row past the queue grace qualifies: no worker claimed it + * (a claim sets `processing`, a deferral sets `processingDeferredUntil`), and replacing its + * token fences it from ever claiming. Every other row keeps its count. + */ +export function releaseUnclaimedDispatchAttempt(now: Date): SQL { + const graceCutoff = new Date(now.getTime() - QUEUED_DISPATCH_GRACE_MS) + return sql`CASE WHEN ${document.processingStatus} = 'pending' AND ${document.processingDeferredUntil} IS NULL AND ${document.processingQueuedAt} < ${sql.param(graceCutoff, document.processingQueuedAt)} THEN GREATEST(${document.processingAttempts} - 1, 0) ELSE ${document.processingAttempts} END` +} diff --git a/apps/sim/lib/knowledge/documents/processing-recovery.ts b/apps/sim/lib/knowledge/documents/processing-recovery.ts index 4a0a457d2ac..e02cb5fab96 100644 --- a/apps/sim/lib/knowledge/documents/processing-recovery.ts +++ b/apps/sim/lib/knowledge/documents/processing-recovery.ts @@ -16,7 +16,10 @@ import { createOrganizationDocumentProcessingBillingContext, createWorkspaceDocumentProcessingBillingContext, } from '@/lib/knowledge/documents/processing-payload' -import { documentProcessingRecoveryCondition } from '@/lib/knowledge/documents/processing-recovery-policy' +import { + documentProcessingRecoveryCondition, + releaseUnclaimedDispatchAttempt, +} from '@/lib/knowledge/documents/processing-recovery-policy' const logger = createLogger('KnowledgeDocumentRecovery') @@ -230,7 +233,7 @@ async function recoverStoredDocumentBatch( processingCompletedAt: null, processingError: null, processingRecoveryAfter: null, - processingAttempts: sql`${document.processingAttempts} + 1`, + processingAttempts: sql`${releaseUnclaimedDispatchAttempt(now)} + 1`, }) .where(eq(document.id, doc.id)) await enqueueOutboxEvent(tx, KNOWLEDGE_DOCUMENT_RECOVERY_OUTBOX_EVENT, payload, { diff --git a/apps/sim/lib/knowledge/documents/retry-processing-grace.test.ts b/apps/sim/lib/knowledge/documents/retry-processing-grace.test.ts index 4c9db45ba91..7da1075f29f 100644 --- a/apps/sim/lib/knowledge/documents/retry-processing-grace.test.ts +++ b/apps/sim/lib/knowledge/documents/retry-processing-grace.test.ts @@ -240,7 +240,10 @@ describe('retryDocumentProcessing requeue guard', () => { processingQueuedAt: null, processingQueueToken: null, }) - expect(reset?.[0]).not.toHaveProperty('processingAttempts') + /** A retry never resets the budget; it only releases a replaced unclaimed generation's charge. */ + expect( + (reset?.[0].processingAttempts as { toSQL: () => { sql: string } }).toSQL().sql + ).toContain('GREATEST') }) it('also requeues a pending document whose dispatch is certainly lost', async () => { diff --git a/apps/sim/lib/knowledge/documents/service.ts b/apps/sim/lib/knowledge/documents/service.ts index 7b1bbef023b..9446abcbb74 100644 --- a/apps/sim/lib/knowledge/documents/service.ts +++ b/apps/sim/lib/knowledge/documents/service.ts @@ -103,7 +103,7 @@ import { recordUndispatchedDocumentFailure, } from '@/lib/knowledge/documents/processing-claim' import type { DocumentProcessingContinuation } from '@/lib/knowledge/documents/processing-continuation-dispatch' -import { documentProcessingQueueOptions } from '@/lib/knowledge/documents/processing-lane' +import { documentProcessingRunOptions } from '@/lib/knowledge/documents/processing-lane' import { enqueueKnowledgeDocumentProcessing } from '@/lib/knowledge/documents/processing-outbox-event' import { assertDocumentProcessingBillingContext, @@ -121,6 +121,7 @@ import { ProviderCapacityContinuationExhaustedError, } from '@/lib/knowledge/documents/processing-provider-deferral' import { scheduleDocumentProcessingQuotaContinuation } from '@/lib/knowledge/documents/processing-quota-continuation' +import { releaseUnclaimedDispatchAttempt } from '@/lib/knowledge/documents/processing-recovery-policy' import { documentProcessingOutcomeSelection, getDocumentProcessingOutcome, @@ -1250,7 +1251,7 @@ async function dispatchViaBatchTrigger( `knowledgeBaseId:${payload.knowledgeBaseId}`, `documentId:${payload.documentId}`, ], - ...documentProcessingQueueOptions(payload), + ...documentProcessingRunOptions(payload), region, }, })) @@ -3459,7 +3460,8 @@ export async function retryDocumentProcessing( * window closed for a document created moments ago whose first dispatch is * still in flight. */ - const queuedGraceCutoff = new Date(Date.now() - QUEUED_DISPATCH_GRACE_MS) + const requestedAt = new Date() + const queuedGraceCutoff = new Date(requestedAt.getTime() - QUEUED_DISPATCH_GRACE_MS) const requeued = await db.transaction(async (tx) => { const reset = await tx .update(document) @@ -3475,6 +3477,7 @@ export async function retryDocumentProcessing( processingDeferredUntil: null, processingCompletedAt: null, processingError: null, + processingAttempts: releaseUnclaimedDispatchAttempt(requestedAt), chunkCount: 0, tokenCount: 0, characterCount: 0, diff --git a/apps/sim/lib/knowledge/documents/types.ts b/apps/sim/lib/knowledge/documents/types.ts index 4bf7435f690..79e4b946fd2 100644 --- a/apps/sim/lib/knowledge/documents/types.ts +++ b/apps/sim/lib/knowledge/documents/types.ts @@ -10,9 +10,9 @@ * a short-interval connector can still burn several inside one transient * outage. A dispatch that provably reached nothing is refunded — see * `clearDocumentsQueued` — which refunds each newly claimed dispatch that - * provably failed before processing began. An accepted dispatch whose remote - * run never starts still stays charged. Three left too little room for those; - * five still bounds the spend + * provably failed before processing began, and recovery gives back the charge of + * a queued generation it replaces unclaimed (`releaseUnclaimedDispatchAttempt`). + * Three left too little room for those; five still bounds the spend * well inside `RETRY_WINDOW_DAYS`. * * Reaching it is a dead letter, not a deletion: the document keeps its `failed` @@ -47,6 +47,15 @@ export const MAX_PROCESSING_ATTEMPTS = 5 */ export const QUEUED_DISPATCH_GRACE_MS = 240 * 60 * 1000 +/** + * How long after its queue stamp a dispatched processing run may still start; the queue + * expires it after that. Kept below {@link QUEUED_DISPATCH_GRACE_MS}, with half an hour + * for clock skew, so recovery never replaces a generation whose run could still start. + * Without it Trigger.dev holds an unstarted run for fourteen days, and a backlog longer + * than the grace had every sweep add a run per waiting document. + */ +export const QUEUED_DISPATCH_START_DEADLINE_MS = QUEUED_DISPATCH_GRACE_MS - 30 * 60 * 1000 + /** Worst-case wall clock for one processing run across its retry budget. */ export function worstCaseProcessingMinutes( maxDurationSeconds: number, From 8f362e9416e1689ea542bc0ec8b46cdeab6c9902 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 18 Sep 2026 16:40:37 -0700 Subject: [PATCH 2/2] test(knowledge): prove sweep dispatches the replacement and keeps worker-reached charges --- .../connector-lifecycle-locks.integration.ts | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/apps/sim/lib/knowledge/__integration__/connector-lifecycle-locks.integration.ts b/apps/sim/lib/knowledge/__integration__/connector-lifecycle-locks.integration.ts index 1b2114886e4..1b8825d8a08 100644 --- a/apps/sim/lib/knowledge/__integration__/connector-lifecycle-locks.integration.ts +++ b/apps/sim/lib/knowledge/__integration__/connector-lifecycle-locks.integration.ts @@ -32,6 +32,7 @@ import { deferConnectorSync } from '@/lib/knowledge/connectors/sync-deferral' import { completeSuccessfulSync } from '@/lib/knowledge/connectors/sync-engine' import { createContentSyncLease } from '@/lib/knowledge/connectors/sync-lock' import { sweepStuckDocuments } from '@/lib/knowledge/connectors/sync-primitives' +import { processDocumentsWithQueue } from '@/lib/knowledge/documents/service' import { deleteKnowledgeBase } from '@/lib/knowledge/service' import { GitHubRequestDeferredError } from '@/connectors/github/request' import type { SyncResult } from '@/connectors/types' @@ -282,14 +283,24 @@ describe('source lifecycle KB guards', () => { .from(document) .where(eq(document.id, retryDocumentId)) expect(row).toEqual({ status: 'pending', attempts: 1, token: null }) + expect(vi.mocked(processDocumentsWithQueue).mock.lastCall?.[0]).toEqual([ + expect.objectContaining({ documentId: retryDocumentId }), + ]) }) - it('keeps the charge of an attempt that reached a worker', async () => { + it.each([ + ['failed', {}], + ['stale processing', { processingStatus: 'processing', processingStartedAt: new Date(0) }], + ] as const)('keeps the charge of a %s attempt that reached a worker', async (_state, row) => { + await db + .update(document) + .set({ ...row, processingAttempts: 2, processingQueuedAt: new Date(0) }) + .where(eq(document.id, retryDocumentId)) await run('recover') - const [row] = await db - .select({ attempts: document.processingAttempts }) + const [after] = await db + .select({ status: document.processingStatus, attempts: document.processingAttempts }) .from(document) .where(eq(document.id, retryDocumentId)) - expect(row.attempts).toBe(1) + expect(after).toEqual({ status: 'pending', attempts: 2 }) }) })