diff --git a/apps/sim/lib/embeddings/client.test.ts b/apps/sim/lib/embeddings/client.test.ts index 78ce526bce9..4f7d27d9a1e 100644 --- a/apps/sim/lib/embeddings/client.test.ts +++ b/apps/sim/lib/embeddings/client.test.ts @@ -823,11 +823,8 @@ describe('knowledge embedding transport fallback', () => { }) /** - * OpenAI returns 429 for an exhausted balance as well as for a rate limit, but - * only one of them reopens. Retrying a spent account cannot succeed, and since - * the sweep re-queues failed documents every sync it turns into permanent load - * — this was observed burning every attempt on thousands of documents for - * weeks against an account with no credit. + * A spent account never reopens, and the sweep re-queues failed documents every + * sync — so retrying one burns the budget per document, indefinitely. */ it('does not retry a 429 that reports an exhausted balance', async () => { vi.useFakeTimers() diff --git a/apps/sim/lib/embeddings/client.ts b/apps/sim/lib/embeddings/client.ts index 9e39a691225..825f96d8b6c 100644 --- a/apps/sim/lib/embeddings/client.ts +++ b/apps/sim/lib/embeddings/client.ts @@ -79,10 +79,7 @@ const EMBEDDING_RETRY_BUDGET_MS = EMBEDDING_MAX_RETRIES * EMBEDDING_MAX_RETRY_DE export class EmbeddingAPIError extends Error { public status: number - /** - * The provider rejected this for an exhausted balance rather than a rate that - * will recover. Both arrive as 429. - */ + /** Rejected for an exhausted balance rather than a recoverable rate. Both are 429. */ public quotaExhausted?: boolean /** @@ -100,13 +97,8 @@ export class EmbeddingAPIError extends Error { /** * True when a rejection body reports an exhausted balance rather than a rate - * limit. - * - * OpenAI returns 429 for both, but only one of them reopens. `insufficient_quota` - * stands until somebody adds credit, so retrying it cannot succeed no matter how - * long the loop waits — and because a failed document is re-queued by the sweep - * on every sync, an account that has run out turns into a permanent load: the - * budget is spent per document, per attempt, forever. + * limit. OpenAI returns 429 for both, but only a rate limit reopens: a spent + * account stands until someone adds credit, so retrying it cannot succeed. */ function isQuotaExhaustionBody(errorText: string): boolean { try { @@ -148,13 +140,10 @@ function statedWaitOutlastsBudget(error: unknown): boolean { } /** - * Whether another attempt against the same provider could plausibly succeed. - * - * Deliberately narrower than {@link isTransientEmbeddingError}, which also - * decides whether the fallback chain should try a *different* provider. Those - * two questions differ: an exhausted balance rules out the key we just used, but - * says nothing about the next one in the chain, so a quota rejection stops the - * retries here while remaining eligible for failover. + * Whether another attempt against the *same* provider could succeed. Narrower + * than {@link isTransientEmbeddingError}, which decides whether to fail over to a + * different one: an exhausted balance rules out the key just used but says + * nothing about the next in the chain. */ function isWorthRetrying(error: unknown): boolean { if (!isTransientEmbeddingError(error)) return false diff --git a/apps/sim/lib/knowledge/documents/service.ts b/apps/sim/lib/knowledge/documents/service.ts index 2c5904773a4..0acef68e3ef 100644 --- a/apps/sim/lib/knowledge/documents/service.ts +++ b/apps/sim/lib/knowledge/documents/service.ts @@ -721,6 +721,7 @@ async function dispatchViaBatchTrigger( ): Promise { let dispatched = 0 const batchIds: string[] = [] + const undispatched: DocumentProcessingPayload[] = [] const region = await resolveTriggerRegion() for (let i = 0; i < jobPayloads.length; i += TRIGGER_BATCH_SIZE) { const chunk = jobPayloads.slice(i, i + TRIGGER_BATCH_SIZE) @@ -747,11 +748,25 @@ async function dispatchViaBatchTrigger( logger.error(`[${requestId}] Failed to batchTrigger ${chunk.length} document jobs`, { error: getErrorMessage(error), }) + undispatched.push(...chunk) } } if (batchIds.length > 0) { logger.info(`[${requestId}] Trigger.dev batches dispatched`, { batchIds }) } + + /** + * Only a total dispatch failure raises, so a chunk failing alone would leave its + * documents at `pending` with nothing recording why. Processing them here is + * slower than the queue but does not drop the work. + */ + if (undispatched.length > 0) { + logger.warn( + `[${requestId}] Processing ${undispatched.length} documents in-process after failed enqueue` + ) + dispatched += await dispatchInProcess(undispatched, requestId) + } + return dispatched } diff --git a/apps/sim/trigger.config.ts b/apps/sim/trigger.config.ts index 5f1a45cca2e..160c2808518 100644 --- a/apps/sim/trigger.config.ts +++ b/apps/sim/trigger.config.ts @@ -73,7 +73,17 @@ export default defineConfig({ '@daytona/sdk', ], extensions: [ - syncEnvVars(() => [{ name: 'DB_APP_NAME', value: 'sim-trigger' }]), + syncEnvVars(() => [ + { name: 'DB_APP_NAME', value: 'sim-trigger' }, + /** + * Workers run Trigger.dev by definition, but the flag saying so was only + * set on the app container, so `isTriggerAvailable()` was false in every + * task run and dispatched work silently took the in-process fallback. + * Ineffective where dispatching is impossible: the check also requires + * TRIGGER_SECRET_KEY, which only the Trigger.dev runtime provides. + */ + { name: 'TRIGGER_DEV_ENABLED', value: 'TRUE' }, + ]), additionalFiles({ files: [ './lib/execution/isolated-vm-worker.cjs',