From ba6bfc0156fc306250a1c2a44bfb44060c55b0d5 Mon Sep 17 00:00:00 2001 From: panbergco Date: Fri, 28 Aug 2026 02:09:50 +0300 Subject: [PATCH] Make long HTTP embedding jobs resumable Retry endpoint timeouts when explicitly configured, creating a fresh abort signal for every attempt. Add gitnexus embeddings to fill an existing index in place: every successful batch is durable, and rerunning skips vectors whose content hash still matches instead of rebuilding the structural graph. --- gitnexus/src/cli/embeddings.ts | 120 ++++++++++++++++++++ gitnexus/src/cli/i18n/en.ts | 2 + gitnexus/src/cli/i18n/zh-CN.ts | 1 + gitnexus/src/cli/index.ts | 18 +-- gitnexus/src/core/embeddings/http-client.ts | 45 +++++++- gitnexus/test/unit/http-embedder.test.ts | 22 ++++ 6 files changed, 200 insertions(+), 8 deletions(-) diff --git a/gitnexus/src/cli/embeddings.ts b/gitnexus/src/cli/embeddings.ts index a63b342c4..5137254fa 100644 --- a/gitnexus/src/cli/embeddings.ts +++ b/gitnexus/src/cli/embeddings.ts @@ -1,4 +1,24 @@ +import path from 'node:path'; import { cliError, cliInfo, cliWarn } from './cli-message.js'; +import { getGitRoot } from '../storage/git.js'; +import { getStoragePaths, loadMeta, saveMeta } from '../storage/repo-manager.js'; +import { + closeLbug, + executeQuery, + executeWithReusedStatement, + initLbug, + loadCachedEmbeddings, +} from '../core/lbug/lbug-adapter.js'; +import { EMBEDDING_TABLE_NAME } from '../core/lbug/schema.js'; +import { runEmbeddingPipeline } from '../core/embeddings/embedding-pipeline.js'; +import { resolveEmbeddingIdentity } from '../core/embeddings/embedding-identity.js'; +import { + decideEmbeddingResume, + mintInterruptedCheckpoint, + mintPartialCheckpoint, + type EmbeddingCheckpoint, + type EmbeddingCheckpointProgress, +} from '../core/embedding-checkpoint.js'; import { getEmbeddingRuntimeDir, getEmbeddingStackSpecs, @@ -67,3 +87,103 @@ export const embeddingsInstallCommand = async ( } cliInfo('✓ Embedding runtime installed. `gitnexus analyze --embeddings` is ready.'); }; + +/** Add missing embeddings directly to a healthy index, checkpointing every batch. */ +export const embeddingsSyncCommand = async (inputPath?: string): Promise => { + const repoPath = inputPath ? path.resolve(inputPath) : getGitRoot(process.cwd()); + if (!repoPath) throw new Error('Not inside a git repository. Pass a repository path.'); + + const { lbugPath, metaPath } = getStoragePaths(repoPath); + const metaDir = path.dirname(metaPath); + const meta = await loadMeta(metaDir); + if (!meta) throw new Error(`No GitNexus index found for ${repoPath}. Run gitnexus analyze first.`); + if (meta.incrementalInProgress) { + throw new Error('The structural index is incomplete. Run gitnexus analyze --force first.'); + } + + const identity = resolveEmbeddingIdentity(); + let forceReembedNodeIds: ReadonlySet | undefined; + let resumedFrom: EmbeddingCheckpoint | undefined; + if (meta.embeddingCheckpoint) { + const decision = decideEmbeddingResume(meta.embeddingCheckpoint, identity); + if (decision.action === 'abort') throw new Error(decision.error); + cliInfo(decision.log); + if (decision.action === 'resume') { + forceReembedNodeIds = decision.pendingNodeIds; + resumedFrom = decision.resumedFrom; + } + } + + await initLbug(lbugPath); + try { + const cached = await loadCachedEmbeddings(); + const existing = new Map( + cached.embeddings.map((row) => [row.nodeId, row.contentHash ?? '']), + ); + let lastPercent = -1; + + const countEmbeddings = async (): Promise => { + try { + const rows = await executeQuery( + `MATCH (e:${EMBEDDING_TABLE_NAME}) RETURN count(e) AS cnt`, + ); + return Number(rows?.[0]?.cnt ?? rows?.[0]?.[0] ?? 0); + } catch { + return undefined; + } + }; + const saveCheckpoint = async ( + checkpoint: EmbeddingCheckpointProgress, + pendingNodeIds: string[], + embeddings?: number, + ): Promise => { + const latest = (await loadMeta(metaDir)) ?? meta; + await saveMeta(metaDir, { + ...latest, + ...(embeddings === undefined ? {} : { stats: { ...latest.stats, embeddings } }), + embeddingCheckpoint: mintInterruptedCheckpoint(identity, checkpoint, pendingNodeIds), + }); + }; + + cliInfo(`Embedding ${repoPath}`); + cliInfo(`Checkpointed vectors already present: ${cached.embeddings.length}`); + + const result = await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + (progress) => { + const percent = Math.floor(progress.percent); + if (percent !== lastPercent && (percent % 5 === 0 || percent === 100)) { + lastPercent = percent; + cliInfo(` ${percent}% — ${progress.nodesProcessed ?? 0}/${progress.totalNodes ?? '?'} nodes`); + } + }, + {}, + undefined, + existing.size ? existing : undefined, + { + forceReembedNodeIds, + onCheckpointWindowStart: async ({ nodeIds, ...checkpoint }) => { + await saveCheckpoint(checkpoint, nodeIds); + }, + onCheckpoint: async (checkpoint) => { + await saveCheckpoint(checkpoint, [], await countEmbeddings()); + }, + }, + ); + + const embeddings = await countEmbeddings(); + if (embeddings === undefined) throw new Error('Could not verify persisted embedding count.'); + const latest = (await loadMeta(metaDir)) ?? meta; + await saveMeta(metaDir, { + ...latest, + stats: { ...latest.stats, embeddings }, + embeddingCheckpoint: result.failedNodeIds.length + ? mintPartialCheckpoint(identity, result, resumedFrom) + : undefined, + }); + cliInfo(`Embeddings ready: ${embeddings}`); + } finally { + await closeLbug(); + } +}; diff --git a/gitnexus/src/cli/i18n/en.ts b/gitnexus/src/cli/i18n/en.ts index e893e61ac..257c94e69 100644 --- a/gitnexus/src/cli/i18n/en.ts +++ b/gitnexus/src/cli/i18n/en.ts @@ -133,6 +133,8 @@ export const en = { 'help.command.uninstall.description': 'Reverse `setup`: remove GitNexus MCP entries, skills, and hooks from all detected editors', 'help.command.analyze.description': 'Index a repository (full analysis)', + 'help.command.embeddings.sync.description': + 'Add missing embeddings to an existing index, checkpointing each batch for safe resume', 'help.command.index.description': 'Register an existing .gitnexus/ folder into the global registry (no re-analysis needed)', 'help.command.serve.description': 'Start local HTTP server for web UI connection', diff --git a/gitnexus/src/cli/i18n/zh-CN.ts b/gitnexus/src/cli/i18n/zh-CN.ts index 2c066f010..d27b5d56a 100644 --- a/gitnexus/src/cli/i18n/zh-CN.ts +++ b/gitnexus/src/cli/i18n/zh-CN.ts @@ -134,6 +134,7 @@ export const zhCN = { 'help.command.uninstall.description': '撤销 `setup`：从所有检测到的编辑器中移除 GitNexus 的 MCP 配置、技能和钩子', 'help.command.analyze.description': '索引仓库（完整分析）', + 'help.command.embeddings.sync.description': '向现有索引添加缺失的嵌入，并逐批次保存检查点以安全续跑', 'help.command.index.description': '将现有 .gitnexus/ 文件夹注册到全局注册表（无需重新分析）', 'help.command.serve.description': '启动供 Web UI 连接的本地 HTTP 服务器', 'help.command.mcp.description': diff --git a/gitnexus/src/cli/index.ts b/gitnexus/src/cli/index.ts index 1ccf75c2f..12667a846 100644 --- a/gitnexus/src/cli/index.ts +++ b/gitnexus/src/cli/index.ts @@ -259,15 +259,13 @@ program .description('Show runtime platform capabilities and embedding configuration') .action(createLazyAction(() => import('./doctor.js'), 'doctorCommand')); -program +const embeddings = program .command('embeddings') - .description('Manage the on-demand local embedding runtime') + .description(t('help.command.embeddings.description')); + +embeddings .command('install') - .description( - 'Install the local embedding stack (@huggingface/transformers + onnxruntime-node) on demand. ' + - 'Heals installs where npm skipped the optional packages (e.g. behind an HTTP proxy, #2370). ' + - 'Downloads only from your configured npm registry — mirrors and proxies apply.', - ) + .description(t('help.command.embeddings.install.description')) .option( '--cuda', "Also download the CUDA GPU binaries (runs onnxruntime-node's NuGet postinstall; " + @@ -276,6 +274,12 @@ program .option('--force', 'Install into the runtime prefix even when the stack already resolves') .action(createLazyAction(() => import('./embeddings.js'), 'embeddingsInstallCommand')); +embeddings + .command('sync [path]') + .description(t('help.command.embeddings.sync.description')) + .addHelpText('after', () => t('help.analyze.environment')) + .action(createLbugLazyAction(() => import('./embeddings.js'), 'embeddingsSyncCommand')); + program .command('clean') .description('Delete GitNexus index for current repo') diff --git a/gitnexus/src/core/embeddings/http-client.ts b/gitnexus/src/core/embeddings/http-client.ts index 7919b2376..565473b78 100644 --- a/gitnexus/src/core/embeddings/http-client.ts +++ b/gitnexus/src/core/embeddings/http-client.ts @@ -39,6 +39,7 @@ interface HttpConfig { retryCapMs: number; minIntervalMs: number; timeoutMs: number; + retryTimeouts: boolean; requestDimensions?: number; } @@ -202,6 +203,8 @@ const readConfig = (): HttpConfig | null => { DEFAULT_HTTP_TIMEOUT_MS, MAX_HTTP_TIMEOUT_MS, ), + retryTimeouts: + parseNonNegativeIntegerEnv('GITNEXUS_EMBEDDING_RETRY_TIMEOUTS', 0, 1) === 1, requestDimensions, }; }; @@ -338,6 +341,16 @@ class RetryableEmbeddingBodyError extends Error { } } +class RetryableEmbeddingTimeoutError extends Error { + constructor(readonly timeoutMs: number, options?: { cause?: unknown }) { + super( + `Embedding request timed out after ${timeoutMs}ms`, + options?.cause !== undefined ? { cause: options.cause } : undefined, + ); + this.name = 'RetryableEmbeddingTimeoutError'; + } +} + /** * Build the message for a 2xx body carrying the wrong number of vectors. * @@ -384,6 +397,7 @@ const httpEmbedBatch = async ( retryCapMs = HTTP_RETRY_CAP_MS, minIntervalMs = 0, timeoutMs = DEFAULT_HTTP_TIMEOUT_MS, + retryTimeouts = false, ): Promise => { const requestBody: { input: string[]; model: string; dimensions?: number } = { input: batch, @@ -428,7 +442,20 @@ const httpEmbedBatch = async ( const signal = requestOptions.signal ? AbortSignal.any([requestOptions.signal, timeoutSignal]) : timeoutSignal; - const attemptResp = await globalThis.fetch(input, { ...init, signal }); + let attemptResp: Response; + try { + attemptResp = await globalThis.fetch(input, { ...init, signal }); + } catch (err) { + if ( + retryTimeouts && + !requestOptions.signal?.aborted && + err instanceof DOMException && + err.name === 'TimeoutError' + ) { + throw new RetryableEmbeddingTimeoutError(timeoutMs, { cause: err }); + } + throw err; + } // Non-OK bodies are none of our business: hand the response straight // back so `resilientFetch` keeps classifying 4xx/5xx/429 unchanged. if (!attemptResp.ok) return attemptResp; @@ -456,6 +483,14 @@ const httpEmbedBatch = async ( // same timeout would take 3 attempts instead of 1, count toward the // process-global `embeddings-http` breaker, and reach the operator as // "unparseable response" so they never reach for the timeout knob. + if ( + retryTimeouts && + !requestOptions.signal?.aborted && + err instanceof DOMException && + err.name === 'TimeoutError' + ) { + throw new RetryableEmbeddingTimeoutError(timeoutMs, { cause: err }); + } if (isTerminalNetworkError(err)) throw err; throw new RetryableEmbeddingBodyError(unparseableMessage(), { cause: err }); } @@ -503,6 +538,12 @@ const httpEmbedBatch = async ( if (err instanceof RetryableEmbeddingBodyError) { throw new HttpEmbeddingError(err.terminalMessage, { cause: err.cause }); } + if (err instanceof RetryableEmbeddingTimeoutError) { + throw new HttpEmbeddingError( + `${err.message} after ${maxAttempts} attempt(s) (${safeUrl(url)}, batch ${batchIndex})`, + { cause: err.cause }, + ); + } if (err instanceof CircuitOpenError) { throw new HttpEmbeddingError( `Embedding endpoint circuit open (${safeUrl(url)}, batch ${batchIndex}): retry in ${Math.ceil(err.retryAfterMs / 1000)}s`, @@ -580,6 +621,7 @@ export const httpEmbed = async ( config.retryCapMs, config.minIntervalMs, config.timeoutMs, + config.retryTimeouts, ); // Defensive backstop, deliberately kept: `httpEmbedBatch` now rejects a @@ -655,6 +697,7 @@ export const httpEmbedQuery = async ( config.retryCapMs, config.minIntervalMs, config.timeoutMs, + config.retryTimeouts, ); // Defensive backstop like the `httpEmbed` one above: an empty `data` array is // now a cardinality mismatch (0 vectors for 1 text) rejected and retried diff --git a/gitnexus/test/unit/http-embedder.test.ts b/gitnexus/test/unit/http-embedder.test.ts index 397481569..5daf565f6 100644 --- a/gitnexus/test/unit/http-embedder.test.ts +++ b/gitnexus/test/unit/http-embedder.test.ts @@ -11,6 +11,7 @@ const ENV_KEYS = [ 'GITNEXUS_EMBEDDING_RETRY_CAP_MS', 'GITNEXUS_EMBEDDING_MIN_INTERVAL_MS', 'GITNEXUS_EMBEDDING_REQUEST_DIMS', + 'GITNEXUS_EMBEDDING_RETRY_TIMEOUTS', ] as const; /** 384d mock vector matching the default schema dimensions. */ @@ -733,6 +734,27 @@ describe('HTTP embedding backend', () => { expect(fetch).toHaveBeenCalledTimes(1); }); + it('retries a timeout when explicitly configured', async () => { + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_RETRY_TIMEOUTS = '1'; + process.env.GITNEXUS_EMBEDDING_MAX_ATTEMPTS = '2'; + process.env.GITNEXUS_EMBEDDING_RETRY_CAP_MS = '1'; + + const timeoutErr = new DOMException( + 'The operation was aborted due to timeout', + 'TimeoutError', + ); + const ok = { ok: true, json: async () => ({ data: [{ embedding: mockVec }] }) }; + vi.stubGlobal('fetch', vi.fn().mockRejectedValueOnce(timeoutErr).mockResolvedValueOnce(ok)); + + const { embedText } = await import('../../src/core/embeddings/embedder.js'); + const result = await embedText('test'); + + expect(fetch).toHaveBeenCalledTimes(2); + expect(result).toBeInstanceOf(Float32Array); + }); + it('retries on network error then succeeds', async () => { process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model';