From 25cae2fea9d62d1b8ae1103ed30743c82561a1fc Mon Sep 17 00:00:00 2001 From: RaymondVerhoef Date: Fri, 22 May 2026 16:35:56 +0200 Subject: [PATCH] fix: speed up handbook sync and stop llm_calls 404 noise Handbook sync ran files sequentially under a 5 req/min limiter with a hardcoded 60s LLM timeout, causing long syncs and AbortError timeouts on large files. Now: limiter at 20 req/min, files processed with concurrency 4, handbook extraction timeout raised to 180s, and near-empty files skip the LLM call. callLLM gains a timeoutMs option; passing a signal no longer silently disables the per-request timeout. llm_calls telemetry self-disables after the first 404 so deploys without the migration applied don't spam the console. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/components/admin/KnowledgeGraph.jsx | 59 +++++++++++++++++++------ src/lib/extractionPipeline.js | 1 + src/lib/llm.js | 17 +++++-- src/lib/llmRetry.js | 8 ++-- 4 files changed, 64 insertions(+), 21 deletions(-) diff --git a/src/components/admin/KnowledgeGraph.jsx b/src/components/admin/KnowledgeGraph.jsx index d363226..7426f65 100644 --- a/src/components/admin/KnowledgeGraph.jsx +++ b/src/components/admin/KnowledgeGraph.jsx @@ -261,25 +261,53 @@ const KnowledgeGraph = () => { } }); - setSyncResult({ added, modified, unchanged }); + setSyncResult({ added, modified, unchanged, failed: [] }); const filesToProcess = [...added, ...modified]; if (filesToProcess.length > 0) { - setSyncProgress(`Processing 0 of ${filesToProcess.length} files...`); - let count = 0; - for (const file of filesToProcess) { - count++; - setSyncProgress(`Processing ${count} of ${filesToProcess.length}: ${file.name}...`); - try { - const rawContent = await getFileContent('respellion', 'employee-handbook', file.path); - // Pacing is handled centrally by extractionLimiter inside analyzeHandbookDelta. - await analyzeHandbookDelta(rawContent, file.path); - await db.updateHandbookSyncState(file.path, file.sha); - } catch (err) { - console.error('Failed to process file:', file.path, err); + const total = filesToProcess.length; + let done = 0; + const failed = []; + setSyncProgress(`Processing 0 of ${total} files...`); + + // Run files in parallel with bounded concurrency. The extractionLimiter + // inside analyzeHandbookDelta still governs the actual API request rate; + // parallelism here just hides GitHub fetch latency and overlaps with the + // limiter's spacing instead of waiting serially. + const CONCURRENCY = 4; + const queue = filesToProcess.slice(); + + async function worker() { + while (queue.length > 0) { + const file = queue.shift(); + if (!file) return; + try { + const rawContent = await getFileContent('respellion', 'employee-handbook', file.path); + // Skip near-empty files — they only burn an LLM call to extract nothing. + if (rawContent.trim().length < 50) { + await db.updateHandbookSyncState(file.path, file.sha); + } else { + await analyzeHandbookDelta(rawContent, file.path); + await db.updateHandbookSyncState(file.path, file.sha); + } + } catch (err) { + console.error('Failed to process file:', file.path, err); + failed.push({ path: file.path, message: err?.message || String(err) }); + } finally { + done++; + setSyncProgress(`Processing ${done} of ${total} files (${failed.length} failed)...`); + } } } - setSyncProgress('Sync Complete! Click "Analyze & Optimize Graph" above to clean up and merge.'); + + await Promise.all(Array.from({ length: Math.min(CONCURRENCY, total) }, worker)); + + setSyncResult({ added, modified, unchanged, failed }); + setSyncProgress( + failed.length === 0 + ? 'Sync Complete! Click "Analyze & Optimize Graph" above to clean up and merge.' + : `Sync finished with ${failed.length} failure(s). See console for details.` + ); reloadKb(); } @@ -461,6 +489,9 @@ ${JSON.stringify({ topics: compactTopics, relations: compactRelations })}`;

Added files: {syncResult.added.length}

Modified files: {syncResult.modified.length}

Unchanged: {syncResult.unchanged.length}

+ {syncResult.failed?.length > 0 && ( +

Failed: {syncResult.failed.length}

+ )} )} {syncProgress && ( diff --git a/src/lib/extractionPipeline.js b/src/lib/extractionPipeline.js index 84bc259..0a4be2f 100644 --- a/src/lib/extractionPipeline.js +++ b/src/lib/extractionPipeline.js @@ -67,6 +67,7 @@ export async function analyzeHandbookDelta(fileContent, filePath, { signal } = { tools: [EMIT_HANDBOOK_DELTA_TOOL], toolChoice: { type: 'tool', name: EMIT_HANDBOOK_DELTA_TOOL.name }, maxTokens: 8192, + timeoutMs: 180_000, limiter: extractionLimiter, signal, }); diff --git a/src/lib/llm.js b/src/lib/llm.js index 59883ca..5ae41a0 100644 --- a/src/lib/llm.js +++ b/src/lib/llm.js @@ -112,9 +112,16 @@ function buildMessages({ messages, user }) { throw new Error('callLLM requires either `messages` or `user`.'); } +// Telemetry collection is optional. If the migration hasn't been applied on a +// given deploy, the first POST returns 404; we then disable further attempts +// to keep the console clean and avoid wasted round-trips. +let llmCallsDisabled = false; function logLlmCall(record) { + if (llmCallsDisabled) return; try { - pb.collection('llm_calls').create(record).catch(() => {}); + pb.collection('llm_calls').create(record).catch((err) => { + if (err?.status === 404) llmCallsDisabled = true; + }); } catch { /* collection may not exist yet — swallow */ } @@ -271,6 +278,7 @@ function validateToolInputs(toolUses, task, toolSchemas) { * @property {Record} [toolSchemas] Overrides for tool_use input validation. * @property {number} [maxTokens=4096] * @property {number} [temperature=0] + * @property {number} [timeoutMs=60000] Per-request timeout in ms. Increase for large structured extractions. * @property {AbortSignal} [signal] * @property {{ acquire: (opts?:{signal?:AbortSignal}) => Promise, pauseUntil: (untilMs:number) => void }} [limiter] */ @@ -291,6 +299,7 @@ export async function callLLM(options) { toolSchemas, maxTokens = 4096, temperature = 0, + timeoutMs = DEFAULT_TIMEOUT_MS, signal, limiter, } = options; @@ -321,9 +330,9 @@ export async function callLLM(options) { result = await withRetry( async () => { if (limiter) await limiter.acquire({ signal }); - const timeoutCtl = signal ? null : new AbortController(); - const timer = timeoutCtl ? setTimeout(() => timeoutCtl.abort(new DOMException('Timeout', 'AbortError')), DEFAULT_TIMEOUT_MS) : null; - const fetchSignal = linkSignals(signal, timeoutCtl?.signal); + const timeoutCtl = new AbortController(); + const timer = setTimeout(() => timeoutCtl.abort(new DOMException('Timeout', 'AbortError')), timeoutMs); + const fetchSignal = linkSignals(signal, timeoutCtl.signal); try { const response = await fetch(ANTHROPIC_URL, { diff --git a/src/lib/llmRetry.js b/src/lib/llmRetry.js index b2d2710..ae1e385 100644 --- a/src/lib/llmRetry.js +++ b/src/lib/llmRetry.js @@ -160,10 +160,12 @@ export function createLimiter({ rps = 1, burst = 1 } = {}) { /** * Shared limiter for the multi-call extraction loops (source chunks, - * handbook file sync). 5 requests/minute matches the lowest published - * Anthropic tier so we stay well clear of 429. + * handbook file sync). 20 req/min sits comfortably under Anthropic Tier 1 + * (50/min for Sonnet) and lets the handbook sync finish in a reasonable + * time without us redesigning the rate-limit story; if a 429 still slips + * through, `withRetry` honours Retry-After. */ -export const extractionLimiter = createLimiter({ rps: 5 / 60, burst: 1 }); +export const extractionLimiter = createLimiter({ rps: 20 / 60, burst: 2 }); /** * Run `fn(attempt)` with retry. `fn` may throw a `RetryableError` to request