From 711ff8721dbf13f9f28dcf5421189b7fb1bca196 Mon Sep 17 00:00:00 2001 From: Eva Date: Tue, 14 Jul 2026 02:15:57 +0700 Subject: [PATCH] fix(embeddings): make HTTP generation resumable --- .../src/integrations/resilient-fetch.ts | 8 +- gitnexus/.env.example | 3 + gitnexus/README.md | 5 +- gitnexus/src/core/embeddings/embedder.ts | 24 ++- .../src/core/embeddings/embedding-pipeline.ts | 81 +++++++- gitnexus/src/core/embeddings/http-client.ts | 153 ++++++++++++++- gitnexus/src/core/run-analyze.ts | 122 +++++++++++- gitnexus/src/server/analyze-job.ts | 17 ++ gitnexus/src/server/api.ts | 73 ++++++- gitnexus/src/storage/repo-manager.ts | 22 +++ gitnexus/test/unit/analyze-job.test.ts | 11 ++ .../test/unit/api-readonly-wiring.test.ts | 21 ++ gitnexus/test/unit/embedding-pipeline.test.ts | 183 ++++++++++++++++++ gitnexus/test/unit/http-embedder.test.ts | 128 ++++++++++++ .../unit/integrations/resilient-fetch.test.ts | 16 ++ gitnexus/test/unit/run-analyze.test.ts | 142 +++++++++++++- 16 files changed, 975 insertions(+), 34 deletions(-) diff --git a/gitnexus-shared/src/integrations/resilient-fetch.ts b/gitnexus-shared/src/integrations/resilient-fetch.ts index c91b9db3a..828dea731 100644 --- a/gitnexus-shared/src/integrations/resilient-fetch.ts +++ b/gitnexus-shared/src/integrations/resilient-fetch.ts @@ -33,6 +33,8 @@ export interface ResilientFetchOptions { breakerOptions?: CircuitBreakerOptions; /** Tuning knobs for the retry helper. */ retry?: Partial> & { + /** Upper bound on a single Retry-After wait. Defaults to RETRY_AFTER_CAP_MS. */ + retryAfterCapMs?: number; sleep?: RetryOptions['sleep']; random?: RetryOptions['random']; }; @@ -83,6 +85,7 @@ type Outcome = export function classifyOutcome( result: { kind: 'error'; err: unknown } | { kind: 'response'; resp: Response }, now: () => number, + retryAfterCapMs = RETRY_AFTER_CAP_MS, ): Outcome { if (result.kind === 'error') { // Both timer-fired aborts (`AbortSignal.timeout()` → `TimeoutError`) @@ -111,7 +114,7 @@ export function classifyOutcome( return { kind: 'retryable-status', resp, - afterMs: parsed !== null ? Math.min(parsed, RETRY_AFTER_CAP_MS) : undefined, + afterMs: parsed !== null ? Math.min(parsed, retryAfterCapMs) : undefined, }; } if (resp.status >= 500) return { kind: 'retryable-status', resp, afterMs: undefined }; @@ -176,6 +179,7 @@ export async function resilientFetch( maxAttempts: opts.retry?.maxAttempts ?? DEFAULT_RETRY.maxAttempts, baseDelayMs: opts.retry?.baseDelayMs ?? DEFAULT_RETRY.baseDelayMs, capDelayMs: opts.retry?.capDelayMs ?? DEFAULT_RETRY.capDelayMs, + retryAfterCapMs: opts.retry?.retryAfterCapMs ?? RETRY_AFTER_CAP_MS, }; const sleep = opts.retry?.sleep ?? defaultSleep; const random = opts.retry?.random ?? Math.random; @@ -202,7 +206,7 @@ export async function resilientFetch( result = { kind: 'error', err }; } - const outcome = classifyOutcome(result, now); + const outcome = classifyOutcome(result, now, retryConfig.retryAfterCapMs); switch (outcome.kind) { case 'success': diff --git a/gitnexus/.env.example b/gitnexus/.env.example index 8f2f83dc4..8b90c7ae7 100644 --- a/gitnexus/.env.example +++ b/gitnexus/.env.example @@ -7,6 +7,9 @@ # GITNEXUS_EMBEDDING_MODEL=BAAI/bge-large-en-v1.5 # GITNEXUS_EMBEDDING_DIMS=1024 # GITNEXUS_EMBEDDING_API_KEY=your-key +# GITNEXUS_EMBEDDING_MAX_ATTEMPTS=3 +# GITNEXUS_EMBEDDING_RETRY_CAP_MS=5000 +# GITNEXUS_EMBEDDING_MIN_INTERVAL_MS=0 # Works with Infinity, vLLM, TEI, llama.cpp, Ollama, LM Studio, or OpenAI. # See README for details. diff --git a/gitnexus/README.md b/gitnexus/README.md index 2dd0d8180..6ca03acfc 100644 --- a/gitnexus/README.md +++ b/gitnexus/README.md @@ -283,10 +283,13 @@ export GITNEXUS_EMBEDDING_URL=http://your-server:8080/v1 export GITNEXUS_EMBEDDING_MODEL=BAAI/bge-large-en-v1.5 export GITNEXUS_EMBEDDING_DIMS=1024 # optional, default 384 export GITNEXUS_EMBEDDING_API_KEY=your-key # optional, default: "unused" +export GITNEXUS_EMBEDDING_MAX_ATTEMPTS=3 # optional, total attempts (1-20) +export GITNEXUS_EMBEDDING_RETRY_CAP_MS=5000 # optional, maximum retry delay +export GITNEXUS_EMBEDDING_MIN_INTERVAL_MS=0 # optional, minimum request spacing gitnexus analyze . --embeddings ``` -Works with Infinity, vLLM, TEI, llama.cpp, Ollama, LM Studio, or OpenAI. When unset, local embeddings are used unchanged. +Works with Infinity, vLLM, TEI, llama.cpp, Ollama, LM Studio, or OpenAI. Retry and pacing settings are provider-neutral; provider-specific limits should be supplied through configuration. When unset, local embeddings are used unchanged. ## Multi-Repo Support diff --git a/gitnexus/src/core/embeddings/embedder.ts b/gitnexus/src/core/embeddings/embedder.ts index 9c4594a4d..e41c4f4c9 100644 --- a/gitnexus/src/core/embeddings/embedder.ts +++ b/gitnexus/src/core/embeddings/embedder.ts @@ -20,7 +20,12 @@ if (!process.env.ORT_LOG_LEVEL) { // initEmbedder, after the platform guard has passed (#1515). import type { FeatureExtractionPipeline, ProgressInfo } from '@huggingface/transformers'; import { DEFAULT_EMBEDDING_CONFIG, type EmbeddingConfig, type ModelProgress } from './types.js'; -import { isHttpMode, getHttpDimensions, httpEmbed } from './http-client.js'; +import { + isHttpMode, + getHttpDimensions, + httpEmbed, + type EmbeddingRequestOptions, +} from './http-client.js'; import { resolveEmbeddingConfig } from './config.js'; import { applyHfEnvOverrides, isHfDownloadFailure, withHfDownloadRetry } from './hf-env.js'; import { @@ -297,9 +302,13 @@ export const getEmbedder = (): FeatureExtractionPipeline => { * @param text - Text to embed * @returns Float32Array of embedding vector */ -export const embedText = async (text: string): Promise => { +export const embedText = async ( + text: string, + options: EmbeddingRequestOptions = {}, +): Promise => { + options.signal?.throwIfAborted(); if (isHttpMode()) { - const [vec] = await httpEmbed([text]); + const [vec] = await httpEmbed([text], options); return vec; } @@ -321,13 +330,17 @@ export const embedText = async (text: string): Promise => { * @param texts - Array of texts to embed * @returns Array of Float32Array embedding vectors */ -export const embedBatch = async (texts: string[]): Promise => { +export const embedBatch = async ( + texts: string[], + options: EmbeddingRequestOptions = {}, +): Promise => { + options.signal?.throwIfAborted(); if (texts.length === 0) { return []; } if (isHttpMode()) { - return httpEmbed(texts); + return httpEmbed(texts, options); } const embedder = getEmbedder(); @@ -337,6 +350,7 @@ export const embedBatch = async (texts: string[]): Promise => { pooling: 'mean', normalize: true, }); + options.signal?.throwIfAborted(); // Result shape is [batch_size, dimensions] // Need to split into individual vectors diff --git a/gitnexus/src/core/embeddings/embedding-pipeline.ts b/gitnexus/src/core/embeddings/embedding-pipeline.ts index 7b09e7314..37580b6f5 100644 --- a/gitnexus/src/core/embeddings/embedding-pipeline.ts +++ b/gitnexus/src/core/embeddings/embedding-pipeline.ts @@ -262,6 +262,24 @@ export interface EmbeddingPipelineResult { semanticMode: 'vector-index' | 'exact-scan'; } +export interface EmbeddingPipelineCheckpoint { + nodesProcessed: number; + totalNodes: number; + chunksProcessed: number; +} + +export interface EmbeddingPipelineCheckpointWindow extends EmbeddingPipelineCheckpoint { + nodeIds: string[]; +} + +export interface EmbeddingPipelineOptions { + signal?: AbortSignal; + checkpointEveryNodes?: number; + forceReembedNodeIds?: ReadonlySet; + onCheckpointWindowStart?: (window: EmbeddingPipelineCheckpointWindow) => Promise; + onCheckpoint?: (checkpoint: EmbeddingPipelineCheckpoint) => Promise; +} + /** * DELETE stale embedding rows for the given nodeIds so they can be re-inserted. * @@ -320,12 +338,20 @@ export const runEmbeddingPipeline = async ( config: Partial = {}, skipNodeIds?: Set, existingEmbeddings?: Map, + pipelineOptions: EmbeddingPipelineOptions = {}, ): Promise => { const finalConfig = resolveEmbeddingConfig(config); let totalChunks = 0; + const checkpointEveryNodes = pipelineOptions.checkpointEveryNodes ?? 5_000; + if (!Number.isSafeInteger(checkpointEveryNodes) || checkpointEveryNodes <= 0) { + throw new Error('checkpointEveryNodes must be a positive integer'); + } + const throwIfCancelled = (): void => pipelineOptions.signal?.throwIfAborted(); try { + throwIfCancelled(); const vectorAvailable = await ensureVectorExtensionAvailable(); + throwIfCancelled(); if (!vectorAvailable) { logger.warn(vectorUnavailableMessage); } @@ -346,6 +372,7 @@ export const runEmbeddingPipeline = async ( modelDownloadPercent: downloadPercent, }); }, finalConfig); + throwIfCancelled(); } onProgress({ @@ -360,6 +387,8 @@ export const runEmbeddingPipeline = async ( // Phase 2: Query embeddable nodes let nodes = await queryEmbeddableNodes(executeQuery); + throwIfCancelled(); + const embeddableNodeIds = new Set(nodes.map((node) => node.id)); // Incremental mode: compare content hashes, delete stale rows, skip fresh ones. // Computed hashes for stale nodes are cached so batchInsertEmbeddings can reuse them @@ -369,16 +398,20 @@ export const runEmbeddingPipeline = async ( // than all up front — see U6 / KTD7. `staleNodeIds` is consulted inside the // batch loop; it stays empty in full (non-incremental) mode so no deletes fire. const staleNodeIds = new Set(); - if (existingEmbeddings && existingEmbeddings.size > 0) { + const forceReembedNodeIds = pipelineOptions.forceReembedNodeIds; + if ( + (existingEmbeddings && existingEmbeddings.size > 0) || + (forceReembedNodeIds && forceReembedNodeIds.size > 0) + ) { const beforeCount = nodes.length; nodes = nodes.filter((n) => { - const existingHash = existingEmbeddings.get(n.id); + const existingHash = existingEmbeddings?.get(n.id); if (existingHash === undefined) { // New node — needs embedding return true; } const currentHash = contentHashForNode(n, finalConfig); - if (currentHash !== existingHash) { + if (currentHash !== existingHash || forceReembedNodeIds?.has(n.id)) { // Content changed — cache hash for reuse during insert, mark for DELETE + re-embed computedStaleHashes.set(n.id, currentHash); staleNodeIds.add(n.id); @@ -395,6 +428,14 @@ export const runEmbeddingPipeline = async ( } } + if (forceReembedNodeIds && forceReembedNodeIds.size > 0) { + const removedPendingNodeIds = [...forceReembedNodeIds].filter( + (nodeId) => !embeddableNodeIds.has(nodeId), + ); + await deleteStaleEmbeddingRows(executeWithReusedStatement, removedPendingNodeIds); + throwIfCancelled(); + } + const totalNodes = nodes.length; if (isDev) { @@ -402,6 +443,7 @@ export const runEmbeddingPipeline = async ( } if (totalNodes === 0) { + throwIfCancelled(); // Ensure the vector index exists even when no new nodes need embedding. // A prior crash or first-time incremental run may have left CodeEmbedding // rows without ever reaching index creation. @@ -425,6 +467,10 @@ export const runEmbeddingPipeline = async ( const batchSize = finalConfig.batchSize; const chunkSize = finalConfig.chunkSize; const overlap = finalConfig.overlap; + const checkpointWindowNodeCount = Math.max( + batchSize, + Math.ceil(checkpointEveryNodes / batchSize) * batchSize, + ); let processedNodes = 0; onProgress({ @@ -438,6 +484,18 @@ export const runEmbeddingPipeline = async ( // Process in batches of nodes for (let batchIndex = 0; batchIndex < totalNodes; batchIndex += batchSize) { + throwIfCancelled(); + if (pipelineOptions.onCheckpointWindowStart && batchIndex % checkpointWindowNodeCount === 0) { + await pipelineOptions.onCheckpointWindowStart({ + nodesProcessed: processedNodes, + totalNodes, + chunksProcessed: totalChunks, + nodeIds: nodes + .slice(batchIndex, batchIndex + checkpointWindowNodeCount) + .map((node) => node.id), + }); + throwIfCancelled(); + } const batch = nodes.slice(batchIndex, batchIndex + batchSize); // Chunk each node and generate text @@ -520,6 +578,7 @@ export const runEmbeddingPipeline = async ( // Preserves Kuzu's required DELETE-before-INSERT for vector-indexed rows. const batchStaleIds = batch.filter((n) => staleNodeIds.has(n.id)).map((n) => n.id); await deleteStaleEmbeddingRows(executeWithReusedStatement, batchStaleIds); + throwIfCancelled(); // Embed chunk texts in sub-batches to control memory const EMBED_SUB_BATCH = finalConfig.subBatchSize; @@ -529,7 +588,7 @@ export const runEmbeddingPipeline = async ( let embeddings: Float32Array[]; try { - embeddings = await embedBatch(subTexts); + embeddings = await embedBatch(subTexts, { signal: pipelineOptions.signal }); } catch (embedErr) { logger.error( { embedErr }, @@ -544,6 +603,7 @@ export const runEmbeddingPipeline = async ( })); await batchInsertEmbeddings(executeWithReusedStatement, dbUpdates); + throwIfCancelled(); } processedNodes += batch.length; @@ -558,9 +618,22 @@ export const runEmbeddingPipeline = async ( currentBatch: Math.floor(batchIndex / batchSize) + 1, totalBatches: Math.ceil(totalNodes / batchSize), }); + + if ( + pipelineOptions.onCheckpoint && + (processedNodes % checkpointWindowNodeCount === 0 || processedNodes === totalNodes) + ) { + await pipelineOptions.onCheckpoint({ + nodesProcessed: processedNodes, + totalNodes, + chunksProcessed: totalChunks, + }); + throwIfCancelled(); + } } // Phase 4: Create vector index + throwIfCancelled(); onProgress({ phase: 'indexing', percent: 90, diff --git a/gitnexus/src/core/embeddings/http-client.ts b/gitnexus/src/core/embeddings/http-client.ts index 121790251..c85d9ff02 100644 --- a/gitnexus/src/core/embeddings/http-client.ts +++ b/gitnexus/src/core/embeddings/http-client.ts @@ -16,6 +16,7 @@ import { CircuitOpenError, ResilientFetchExhaustedError, resilientFetch } from ' const HTTP_TIMEOUT_MS = 30_000; const HTTP_MAX_RETRIES = 2; const HTTP_RETRY_BACKOFF_MS = 1_000; +const HTTP_RETRY_CAP_MS = 5_000; const HTTP_BATCH_SIZE = 64; const DEFAULT_DIMS = 384; const HTTP_BREAKER_KEY = 'embeddings-http'; @@ -25,8 +26,85 @@ interface HttpConfig { model: string; apiKey: string; dimensions?: number; + maxAttempts: number; + retryCapMs: number; + minIntervalMs: number; } +export interface EmbeddingRequestOptions { + signal?: AbortSignal; +} + +let lastHttpRequestStartedAt: number | undefined; +let httpPaceQueue: Promise = Promise.resolve(); + +const parsePositiveIntegerEnv = (name: string, fallback: number, max: number): number => { + const raw = process.env[name]; + if (raw === undefined || raw === '') return fallback; + if (!/^\d+$/u.test(raw)) { + throw new Error(`${name} must be a positive integer, got "${raw}"`); + } + const parsed = Number(raw); + if (!Number.isSafeInteger(parsed) || parsed <= 0 || parsed > max) { + throw new Error(`${name} must be a positive integer <= ${max}, got "${raw}"`); + } + return parsed; +}; + +const parseNonNegativeIntegerEnv = (name: string, fallback: number, max: number): number => { + const raw = process.env[name]; + if (raw === undefined || raw === '') return fallback; + if (!/^\d+$/u.test(raw)) { + throw new Error(`${name} must be a non-negative integer, got "${raw}"`); + } + const parsed = Number(raw); + if (!Number.isSafeInteger(parsed) || parsed < 0 || parsed > max) { + throw new Error(`${name} must be a non-negative integer <= ${max}, got "${raw}"`); + } + return parsed; +}; + +const cancelledError = (): DOMException => + new DOMException('Embedding request cancelled', 'AbortError'); + +const throwIfAborted = (signal?: AbortSignal): void => { + if (signal?.aborted) throw cancelledError(); +}; + +const abortableSleep = (ms: number, signal?: AbortSignal): Promise => { + throwIfAborted(signal); + if (ms <= 0) return Promise.resolve(); + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + signal?.removeEventListener('abort', onAbort); + resolve(); + }, ms); + const onAbort = () => { + clearTimeout(timer); + signal?.removeEventListener('abort', onAbort); + reject(cancelledError()); + }; + signal?.addEventListener('abort', onAbort, { once: true }); + }); +}; + +const paceHttpRequest = async (minIntervalMs: number, signal?: AbortSignal): Promise => { + throwIfAborted(signal); + if (minIntervalMs <= 0) return; + const waitTurn = httpPaceQueue.then(async () => { + throwIfAborted(signal); + const waitMs = + lastHttpRequestStartedAt === undefined + ? 0 + : Math.max(0, lastHttpRequestStartedAt + minIntervalMs - Date.now()); + await abortableSleep(waitMs, signal); + throwIfAborted(signal); + lastHttpRequestStartedAt = Date.now(); + }); + httpPaceQueue = waitTurn.catch(() => undefined); + await waitTurn; +}; + /** * Stable lead of the {@link readConfig} malformed-`GITNEXUS_EMBEDDING_DIMS` * error. `readConfig` throws a plain `Error` (not an {@link HttpEmbeddingError}) @@ -74,6 +152,17 @@ const readConfig = (): HttpConfig | null => { model, apiKey: process.env.GITNEXUS_EMBEDDING_API_KEY ?? 'unused', dimensions, + maxAttempts: parsePositiveIntegerEnv( + 'GITNEXUS_EMBEDDING_MAX_ATTEMPTS', + HTTP_MAX_RETRIES + 1, + 20, + ), + retryCapMs: parsePositiveIntegerEnv( + 'GITNEXUS_EMBEDDING_RETRY_CAP_MS', + HTTP_RETRY_CAP_MS, + 300_000, + ), + minIntervalMs: parseNonNegativeIntegerEnv('GITNEXUS_EMBEDDING_MIN_INTERVAL_MS', 0, 300_000), }; }; @@ -120,11 +209,15 @@ export const safeUrl = (url: string): string => { * its masked form, then strip any residual `scheme://userinfo@` the transport may * have echoed in a normalized (non-exact) form. See #2385. */ -const sanitizeReason = (reason: string, url: string): string => - reason +const sanitizeReason = (reason: string, url: string, apiKey?: string): string => { + const withoutUrlCredentials = reason .split(url) .join(safeUrl(url)) .replace(/([a-z][a-z0-9+.-]*:\/\/)[^/@\s]*@/gi, '$1'); + return apiKey && apiKey !== 'unused' + ? withoutUrlCredentials.split(apiKey).join('[redacted]') + : withoutUrlCredentials; +}; /** * Error thrown by this module's HTTP embedding path (`httpEmbedBatch` / @@ -201,6 +294,10 @@ const httpEmbedBatch = async ( apiKey: string, batchIndex = 0, dimensions?: number, + requestOptions: EmbeddingRequestOptions = {}, + maxAttempts = HTTP_MAX_RETRIES + 1, + retryCapMs = HTTP_RETRY_CAP_MS, + minIntervalMs = 0, ): Promise => { const requestBody: { input: string[]; model: string; dimensions?: number } = { input: batch, @@ -212,11 +309,11 @@ const httpEmbedBatch = async ( let resp: Response; try { + throwIfAborted(requestOptions.signal); resp = await resilientFetch( url, { method: 'POST', - signal: AbortSignal.timeout(HTTP_TIMEOUT_MS), headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${apiKey}`, @@ -224,11 +321,35 @@ const httpEmbedBatch = async ( body: JSON.stringify(requestBody), }, { + fetchImpl: async (input, init) => { + await paceHttpRequest(minIntervalMs, requestOptions.signal); + throwIfAborted(requestOptions.signal); + const timeoutSignal = AbortSignal.timeout(HTTP_TIMEOUT_MS); + const signal = requestOptions.signal + ? AbortSignal.any([requestOptions.signal, timeoutSignal]) + : timeoutSignal; + return globalThis.fetch(input, { ...init, signal }); + }, breakerKey: HTTP_BREAKER_KEY, - retry: { maxAttempts: HTTP_MAX_RETRIES + 1, baseDelayMs: HTTP_RETRY_BACKOFF_MS }, + retry: { + maxAttempts, + baseDelayMs: HTTP_RETRY_BACKOFF_MS, + capDelayMs: retryCapMs, + retryAfterCapMs: retryCapMs, + sleep: (ms) => abortableSleep(ms, requestOptions.signal), + }, }, ); } catch (err) { + if ( + requestOptions.signal?.aborted || + (err instanceof DOMException && err.name === 'AbortError') + ) { + throw new HttpEmbeddingError( + `Embedding request cancelled (${safeUrl(url)}, batch ${batchIndex})`, + { cause: err }, + ); + } if (err instanceof CircuitOpenError) { throw new HttpEmbeddingError( `Embedding endpoint circuit open (${safeUrl(url)}, batch ${batchIndex}): retry in ${Math.ceil(err.retryAfterMs / 1000)}s`, @@ -247,10 +368,12 @@ const httpEmbedBatch = async ( { cause: err }, ); } - const reason = sanitizeReason(err instanceof Error ? err.message : String(err), url); + const reason = sanitizeReason(err instanceof Error ? err.message : String(err), url, apiKey); + const safeCause = new Error(reason); + safeCause.name = err instanceof Error ? err.name : 'EmbeddingTransportError'; throw new HttpEmbeddingError( `Embedding request failed (${safeUrl(url)}, batch ${batchIndex}): ${reason}`, - { cause: err }, + { cause: safeCause }, ); } @@ -290,7 +413,10 @@ const httpEmbedBatch = async ( * @param texts - Array of texts to embed * @returns Array of Float32Array embedding vectors */ -export const httpEmbed = async (texts: string[]): Promise => { +export const httpEmbed = async ( + texts: string[], + requestOptions: EmbeddingRequestOptions = {}, +): Promise => { if (texts.length === 0) return []; const config = readConfig(); @@ -309,6 +435,10 @@ export const httpEmbed = async (texts: string[]): Promise => { config.apiKey, batchIndex, config.dimensions, + requestOptions, + config.maxAttempts, + config.retryCapMs, + config.minIntervalMs, ); if (items.length !== batch.length) { @@ -347,7 +477,10 @@ export const httpEmbed = async (texts: string[]): Promise => { * @param text - Query text to embed * @returns Embedding vector as number array */ -export const httpEmbedQuery = async (text: string): Promise => { +export const httpEmbedQuery = async ( + text: string, + requestOptions: EmbeddingRequestOptions = {}, +): Promise => { const config = readConfig(); if (!config) throw new Error('HTTP embedding not configured'); @@ -359,6 +492,10 @@ export const httpEmbedQuery = async (text: string): Promise => { config.apiKey, 0, config.dimensions, + requestOptions, + config.maxAttempts, + config.retryCapMs, + config.minIntervalMs, ); if (!items.length) { throw new HttpEmbeddingError(`Embedding endpoint returned empty response (${safeUrl(url)})`); diff --git a/gitnexus/src/core/run-analyze.ts b/gitnexus/src/core/run-analyze.ts index c0ed5a06e..9f6143372 100644 --- a/gitnexus/src/core/run-analyze.ts +++ b/gitnexus/src/core/run-analyze.ts @@ -267,6 +267,22 @@ export interface AnalyzeOptions { skipNativeCloseOnExit?: boolean; } +interface EmbeddingIdentity { + model: string; + dimensions: number; +} + +const resolveEmbeddingIdentity = async (): Promise => { + const [{ getEmbeddingDimensions }, { resolveEmbeddingConfig }] = await Promise.all([ + import('./embeddings/embedder.js'), + import('./embeddings/config.js'), + ]); + return { + model: process.env.GITNEXUS_EMBEDDING_MODEL ?? resolveEmbeddingConfig().modelId, + dimensions: getEmbeddingDimensions(), + }; +}; + export interface AnalyzeResult { repoName: string; repoPath: string; @@ -760,6 +776,37 @@ export async function runFullAnalysis( } } + let resumeEmbeddingCheckpoint = false; + let pendingEmbeddingNodeIds = new Set(); + let embeddingIdentityForRun: EmbeddingIdentity | undefined; + if (existingMeta?.embeddingCheckpoint) { + if (options.dropEmbeddings) { + log('Discarding the interrupted embedding checkpoint (--drop-embeddings).'); + options = { ...options, force: true }; + } else { + embeddingIdentityForRun = await resolveEmbeddingIdentity(); + const checkpoint = existingMeta.embeddingCheckpoint; + if ( + checkpoint.model !== embeddingIdentityForRun.model || + checkpoint.dimensions !== embeddingIdentityForRun.dimensions + ) { + throw new Error( + `Cannot resume embedding checkpoint: it uses ${checkpoint.model} at ` + + `${checkpoint.dimensions} dimensions, but this run resolves ` + + `${embeddingIdentityForRun.model} at ${embeddingIdentityForRun.dimensions}. ` + + 'Restore the matching embedding configuration or pass --drop-embeddings to rebuild without it.', + ); + } + resumeEmbeddingCheckpoint = true; + pendingEmbeddingNodeIds = new Set(checkpoint.pendingNodeIds ?? []); + log( + `Previous analyze ended at an embedding checkpoint ` + + `(${checkpoint.nodesProcessed}/${checkpoint.totalNodes} nodes); resuming from persisted hashes` + + `${pendingEmbeddingNodeIds.size > 0 ? ` and regenerating ${pendingEmbeddingNodeIds.size} pending node(s)` : ''}.`, + ); + } + } + // ── Crash recovery: dirty flag forces full rebuild ──────────────── // If the previous incremental run set incrementalInProgress and didn't // clear it, the on-disk index may be in a half-state. Cheapest path @@ -898,7 +945,12 @@ export async function runFullAnalysis( } // ── Early-return: already up to date ────────────────────────────── - if (existingMeta && !options.force && existingMeta.lastCommit === currentCommit) { + if ( + existingMeta && + !existingMeta.embeddingCheckpoint && + !options.force && + existingMeta.lastCommit === currentCommit + ) { // Non-git folders have currentCommit = '' — always rebuild since we can't detect changes if (currentCommit !== '') { // For git repos, even if HEAD matches lastCommit, the working tree @@ -1031,9 +1083,11 @@ export async function runFullAnalysis( const { forceRegenerateEmbeddings, preserveExistingEmbeddings, - shouldGenerateEmbeddings, - shouldLoadCache, + shouldGenerateEmbeddings: derivedShouldGenerateEmbeddings, + shouldLoadCache: derivedShouldLoadCache, } = _deriveEmbeddingMode(options, existingEmbeddingCount); + const shouldGenerateEmbeddings = derivedShouldGenerateEmbeddings || resumeEmbeddingCheckpoint; + const shouldLoadCache = derivedShouldLoadCache || resumeEmbeddingCheckpoint; if (options.dropEmbeddings && existingEmbeddingCount > 0) { log( @@ -1735,7 +1789,7 @@ export async function runFullAnalysis( if (shouldGenerateEmbeddings) { const { skipForCap, capDisabled, nodeLimit } = deriveEmbeddingCap( stats.nodes, - options.embeddingsNodeLimit, + resumeEmbeddingCheckpoint ? 0 : options.embeddingsNodeLimit, ); if (!skipForCap) { embeddingSkipped = false; @@ -1801,6 +1855,8 @@ export async function runFullAnalysis( httpMode ? 'Connecting to embedding endpoint...' : 'Loading embedding model...', ); const { runEmbeddingPipeline } = await import('./embeddings/embedding-pipeline.js'); + embeddingIdentityForRun ??= await resolveEmbeddingIdentity(); + const embeddingIdentity = embeddingIdentityForRun; // Build a Map from cached embeddings for incremental mode let existingEmbeddings: Map | undefined; if (cachedEmbeddingNodeIds.size > 0) { @@ -1810,6 +1866,48 @@ export async function runFullAnalysis( } } + const saveEmbeddingCheckpoint = async ( + checkpoint: { + nodesProcessed: number; + totalNodes: number; + chunksProcessed: number; + }, + pendingNodeIds: string[], + embeddings: number | undefined, + ): Promise => { + const fileHashes: Record = {}; + for (const [key, value] of newFileHashes) fileHashes[key] = value; + await saveMeta(metaDir, { + ...(existingMeta ?? {}), + repoPath, + lastCommit: currentCommit, + indexedAt: new Date().toISOString(), + branch: branchLabel ?? existingMeta?.branch, + remoteUrl: hasGitDir(repoPath) ? getRemoteUrl(repoPath) : undefined, + stats: { + files: pipelineResult.totalFileCount, + nodes: stats.nodes, + edges: stats.edges, + communities: pipelineResult.communityResult?.stats.totalCommunities, + processes: pipelineResult.processResult?.stats.totalProcesses, + embeddings, + }, + schemaVersion: hasGitDir(repoPath) ? INCREMENTAL_SCHEMA_VERSION : undefined, + cjkSegmentation: getSearchFTSCjkSegmentation(), + fileHashes: hasGitDir(repoPath) ? fileHashes : undefined, + cacheKeys: [...parseCache.usedKeys], + incrementalInProgress: undefined, + embeddingCheckpoint: { + at: new Date().toISOString(), + ...checkpoint, + model: embeddingIdentity.model, + dimensions: embeddingIdentity.dimensions, + pendingNodeIds, + }, + pdg: resolvePdgConfig(options), + }); + }; + const embeddingResult = await runEmbeddingPipeline( executeQuery, executeWithReusedStatement, @@ -1826,6 +1924,21 @@ export async function runFullAnalysis( {}, cachedEmbeddingNodeIds.size > 0 ? cachedEmbeddingNodeIds : undefined, existingEmbeddings, + { + forceReembedNodeIds: pendingEmbeddingNodeIds, + onCheckpointWindowStart: async ({ nodeIds, ...checkpoint }) => { + await saveEmbeddingCheckpoint(checkpoint, nodeIds, existingMeta?.stats?.embeddings); + }, + onCheckpoint: async (checkpoint) => { + await checkpointOnce(); + const countResult = await executeQuery( + `MATCH (e:${EMBEDDING_TABLE_NAME}) RETURN count(e) AS cnt`, + ); + const countRow = countResult?.[0]; + const embeddings = Number(countRow?.cnt ?? countRow?.[0] ?? 0); + await saveEmbeddingCheckpoint(checkpoint, [], embeddings); + }, + }, ); if (embeddingResult.semanticMode === 'exact-scan') { semanticMode = 'exact-scan'; @@ -1952,6 +2065,7 @@ export async function runFullAnalysis( // so a sibling branch's prune can union it and not evict our shards. cacheKeys: [...parseCache.usedKeys], incrementalInProgress: undefined as RepoMeta['incrementalInProgress'], + embeddingCheckpoint: undefined, // The effective pdg config this run's DB rows were built under // (#2099 F1). `undefined` on pdg-off runs — this meta is a fresh // literal (no spread of existingMeta), so omission is what CLEARS the diff --git a/gitnexus/src/server/analyze-job.ts b/gitnexus/src/server/analyze-job.ts index d62912abd..5f67fcbfe 100644 --- a/gitnexus/src/server/analyze-job.ts +++ b/gitnexus/src/server/analyze-job.ts @@ -40,6 +40,7 @@ const JOB_TIMEOUT_MS = 30 * 60 * 1000; // 30 minutes export class JobManager { private jobs = new Map(); private children = new Map(); + private abortControllers = new Map(); private timeouts = new Map>(); private emitter = new EventEmitter(); private cleanupTimer: ReturnType; @@ -111,6 +112,7 @@ export class JobManager { if (this.isTerminal(job.status)) { job.completedAt = job.completedAt ?? Date.now(); + this.abortControllers.delete(id); } // Emit exactly one event per updateJob call to prevent SSE double-write @@ -150,6 +152,16 @@ export class JobManager { }); } + /** Register cancellable in-process work for a job. */ + registerAbortController(jobId: string, controller: AbortController): void { + const job = this.jobs.get(jobId); + if (!job || this.isTerminal(job.status)) { + controller.abort(); + return; + } + this.abortControllers.set(jobId, controller); + } + /** Cancel a running job — sends SIGTERM to child process. */ cancelJob(jobId: string, reason?: string): boolean { const job = this.jobs.get(jobId); @@ -159,6 +171,8 @@ export class JobManager { if (child) { child.kill('SIGTERM'); } + this.abortControllers.get(jobId)?.abort(); + this.abortControllers.delete(jobId); this.updateJob(jobId, { status: 'failed', @@ -181,6 +195,8 @@ export class JobManager { child.kill('SIGTERM'); } this.children.clear(); + for (const controller of this.abortControllers.values()) controller.abort(); + this.abortControllers.clear(); // Clear all timeouts for (const timer of this.timeouts.values()) { @@ -201,6 +217,7 @@ export class JobManager { for (const [id, job] of this.jobs) { if (this.isTerminal(job.status) && job.completedAt && now - job.completedAt > JOB_TTL_MS) { this.jobs.delete(id); + this.abortControllers.delete(id); } } } diff --git a/gitnexus/src/server/api.ts b/gitnexus/src/server/api.ts index 218c6e211..50b461f48 100644 --- a/gitnexus/src/server/api.ts +++ b/gitnexus/src/server/api.ts @@ -17,6 +17,7 @@ import { canonicalizePath, cloneDirBelongsToEntry, loadMeta, + saveMeta, listRegisteredRepos, getStoragePath, registryPathEquals, @@ -1776,17 +1777,15 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => status: 'analyzing' as any, progress: { phase: 'analyzing', percent: 0, message: 'Starting embedding generation...' }, }); + const embedController = new AbortController(); + embedJobManager.registerAbortController(job.id, embedController); // 30-minute timeout for embedding jobs (same as analyze jobs) const EMBED_TIMEOUT_MS = 30 * 60 * 1000; const embedTimeout = setTimeout(() => { const current = embedJobManager.getJob(job.id); if (current && current.status !== 'complete' && current.status !== 'failed') { - releaseRepoLock(repoLockPath); - embedJobManager.updateJob(job.id, { - status: 'failed', - error: 'Embedding timed out (30 minute limit)', - }); + embedJobManager.cancelJob(job.id, 'Embedding timed out (30 minute limit)'); } }, EMBED_TIMEOUT_MS); @@ -1797,6 +1796,50 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => await withLbugDb(lbugPath, async () => { const { runEmbeddingPipeline } = await import('../core/embeddings/embedding-pipeline.js'); + const [{ getEmbeddingDimensions }, { resolveEmbeddingConfig }] = await Promise.all([ + import('../core/embeddings/embedder.js'), + import('../core/embeddings/config.js'), + ]); + const embeddingIdentity = { + model: process.env.GITNEXUS_EMBEDDING_MODEL ?? resolveEmbeddingConfig().modelId, + dimensions: getEmbeddingDimensions(), + }; + let embeddingMeta = await loadMeta(entry.storagePath); + if (!embeddingMeta) { + throw new Error('Repository metadata is missing; run gitnexus analyze first'); + } + const priorCheckpoint = embeddingMeta.embeddingCheckpoint; + if ( + priorCheckpoint && + (priorCheckpoint.model !== embeddingIdentity.model || + priorCheckpoint.dimensions !== embeddingIdentity.dimensions) + ) { + throw new Error( + `Cannot resume embedding checkpoint: it uses ${priorCheckpoint.model} at ` + + `${priorCheckpoint.dimensions} dimensions, but this run resolves ` + + `${embeddingIdentity.model} at ${embeddingIdentity.dimensions}.`, + ); + } + const forceReembedNodeIds = new Set(priorCheckpoint?.pendingNodeIds ?? []); + const saveEmbeddingCheckpoint = async ( + checkpoint: { + nodesProcessed: number; + totalNodes: number; + chunksProcessed: number; + }, + pendingNodeIds: string[], + ): Promise => { + embeddingMeta = { + ...embeddingMeta, + embeddingCheckpoint: { + at: new Date().toISOString(), + ...checkpoint, + ...embeddingIdentity, + pendingNodeIds, + }, + }; + await saveMeta(entry.storagePath, embeddingMeta); + }; // Fetch existing content hashes for incremental embedding. // Delegated to lbug-adapter which owns the DB query logic and legacy-fallback handling. const { fetchExistingEmbeddingHashes } = await import('../core/lbug/lbug-adapter.js'); @@ -1831,6 +1874,17 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => {}, // config: use defaults undefined, // skipNodeIds existingEmbeddings, + { + signal: embedController.signal, + forceReembedNodeIds, + onCheckpointWindowStart: async ({ nodeIds, ...checkpoint }) => { + await saveEmbeddingCheckpoint(checkpoint, nodeIds); + }, + onCheckpoint: async (checkpoint) => { + await flushWAL(); + await saveEmbeddingCheckpoint(checkpoint, []); + }, + }, ); // Flush WAL so subsequent /api/search requests see the new @@ -1838,18 +1892,16 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => // handles this during process exit, but the server keeps the // connection open for other routes — a CHECKPOINT is enough. await flushWAL(); + embeddingMeta = { ...embeddingMeta, embeddingCheckpoint: undefined }; + await saveMeta(entry.storagePath, embeddingMeta); }); - clearTimeout(embedTimeout); - releaseRepoLock(repoLockPath); // Don't overwrite 'failed' if the job was cancelled while the pipeline was running const current = embedJobManager.getJob(job.id); if (!current || current.status !== 'failed') { embedJobManager.updateJob(job.id, { status: 'complete' }); } } catch (err: any) { - clearTimeout(embedTimeout); - releaseRepoLock(repoLockPath); const current = embedJobManager.getJob(job.id); if (!current || current.status !== 'failed') { embedJobManager.updateJob(job.id, { @@ -1857,6 +1909,9 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => error: err.message || 'Embedding generation failed', }); } + } finally { + clearTimeout(embedTimeout); + releaseRepoLock(repoLockPath); } })(); diff --git a/gitnexus/src/storage/repo-manager.ts b/gitnexus/src/storage/repo-manager.ts index 4c4e59f18..7755c8c2e 100644 --- a/gitnexus/src/storage/repo-manager.ts +++ b/gitnexus/src/storage/repo-manager.ts @@ -197,6 +197,28 @@ export interface RepoMeta { * under-expanded when the run died. */ droppedImporterChunks?: number; }; + /** + * Durable embedding-resume marker. Before a bounded write window begins, + * `pendingNodeIds` records every node that could become partially persisted; + * after the LadybugDB checkpoint it is cleared while progress is retained. + * A matching runtime resumes from persisted hashes and regenerates pending + * nodes; a model or dimension mismatch fails before mutation. + */ + embeddingCheckpoint?: { + at: string; + nodesProcessed: number; + totalNodes: number; + chunksProcessed: number; + model: string; + dimensions: number; + /** + * Nodes in the current checkpoint window. Any of these may have only a + * subset of their chunks persisted after an abrupt process termination, + * so resume must delete and regenerate them even when a persisted row has + * the current content hash. + */ + pendingNodeIds?: string[]; + }; /** * Name of the git branch this index represents (#2106). Absent for the * default/legacy single-branch case so the flat metadata file stays diff --git a/gitnexus/test/unit/analyze-job.test.ts b/gitnexus/test/unit/analyze-job.test.ts index 4f40d93b9..50b623364 100644 --- a/gitnexus/test/unit/analyze-job.test.ts +++ b/gitnexus/test/unit/analyze-job.test.ts @@ -119,6 +119,17 @@ describe('JobManager', () => { expect(manager.getJob(job.id)!.error).toBe('Cancelled by user'); }); + it('cancelJob aborts registered in-process work', () => { + const job = manager.createJob({ repoPath: '/tmp/repo' }); + manager.updateJob(job.id, { status: 'analyzing' }); + const controller = new AbortController(); + manager.registerAbortController(job.id, controller); + + manager.cancelJob(job.id, 'Cancelled by user'); + + expect(controller.signal.aborted).toBe(true); + }); + it('cancelJob returns false for terminal jobs', () => { const job = manager.createJob({ repoUrl: 'https://github.com/user/repo' }); manager.updateJob(job.id, { status: 'complete' }); diff --git a/gitnexus/test/unit/api-readonly-wiring.test.ts b/gitnexus/test/unit/api-readonly-wiring.test.ts index 7bb8681dc..8e86e7f31 100644 --- a/gitnexus/test/unit/api-readonly-wiring.test.ts +++ b/gitnexus/test/unit/api-readonly-wiring.test.ts @@ -57,4 +57,25 @@ describe('api read-only endpoint wiring', () => { expect(embedSection[0]).not.toMatch(/readOnly:\s*true/); } }); + + it('/api/embed keeps the repository lock until cancelled work actually stops', async () => { + const source = await readSource(); + const timeoutSection = source.match( + /const embedTimeout = setTimeout\([\s\S]*?\/\/ Run embedding pipeline asynchronously/, + ); + expect(timeoutSection).not.toBeNull(); + expect(timeoutSection?.[0]).not.toContain('releaseRepoLock(repoLockPath)'); + }); + + it('/api/embed persists and resumes bounded pending windows', async () => { + const source = await readSource(); + const embedSection = source.match( + /\/\/ Run embedding pipeline asynchronously[\s\S]*?res\.status\(202\)/, + ); + expect(embedSection).not.toBeNull(); + expect(embedSection?.[0]).toContain('forceReembedNodeIds'); + expect(embedSection?.[0]).toContain('onCheckpointWindowStart'); + expect(embedSection?.[0]).toContain('pendingNodeIds'); + expect(embedSection?.[0]).toContain('saveMeta'); + }); }); diff --git a/gitnexus/test/unit/embedding-pipeline.test.ts b/gitnexus/test/unit/embedding-pipeline.test.ts index 570593a2d..18afe0a33 100644 --- a/gitnexus/test/unit/embedding-pipeline.test.ts +++ b/gitnexus/test/unit/embedding-pipeline.test.ts @@ -549,6 +549,189 @@ describe('runEmbeddingPipeline incremental filter', () => { expect(insertN1).toBeLessThan(deleteN2); }); + it('stops at a batch boundary when cancellation is requested', async () => { + mockEmbedderSetup(); + const first = makeNode({ id: 'Function:first:src/first.ts', name: 'first' }); + const second = makeNode({ id: 'Function:second:src/second.ts', name: 'second' }); + const executeQuery = mockExecuteQuery([first, second]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + const controller = new AbortController(); + const checkpoints: number[] = []; + + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + const promise = runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + { batchSize: 1 }, + undefined, + new Map(), + { + signal: controller.signal, + checkpointEveryNodes: 1, + onCheckpoint: async ({ nodesProcessed }) => { + checkpoints.push(nodesProcessed); + controller.abort(); + }, + }, + ); + + await expect(promise).rejects.toThrow(/abort/i); + const insertedIds = stmtCalls + .filter((call) => call.cypher.includes('CREATE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + expect(insertedIds).toEqual([first.id]); + expect(checkpoints).toEqual([1]); + }); + + it('resumes idempotently from the hashes persisted before an interrupted checkpoint', async () => { + mockEmbedderSetup(); + const first = makeNode({ id: 'Function:first:src/first.ts', name: 'first' }); + const second = makeNode({ id: 'Function:second:src/second.ts', name: 'second' }); + const executeQuery = mockExecuteQuery([first, second]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await expect( + runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + { batchSize: 1 }, + undefined, + new Map(), + { + checkpointEveryNodes: 1, + onCheckpoint: async ({ nodesProcessed }) => { + if (nodesProcessed === 1) throw new Error('simulated interruption after checkpoint'); + }, + }, + ), + ).rejects.toThrow('simulated interruption'); + + const firstInsert = stmtCalls.find( + (call) => call.cypher.includes('CREATE') && call.params.some((p) => p.nodeId === first.id), + ); + expect(firstInsert).toBeDefined(); + const firstParam = firstInsert?.params.find((param) => param.nodeId === first.id); + if (!firstParam) throw new Error('expected first checkpoint insert'); + const firstHash = firstParam.contentHash; + + stmtCalls = []; + progressUpdates = []; + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + { batchSize: 1 }, + undefined, + new Map([[first.id, firstHash]]), + { checkpointEveryNodes: 1, onCheckpoint: async () => {} }, + ); + + const resumedIds = stmtCalls + .filter((call) => call.cypher.includes('CREATE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + expect(resumedIds).toEqual([second.id]); + }); + + it('re-embeds a pending-window node even when its persisted content hash matches', async () => { + mockEmbedderSetup(); + const node = makeNode({ + id: 'Function:pending:src/pending.ts', + name: 'pending', + filePath: 'src/pending.ts', + }); + const currentHash = contentHashForNode(node, DEFAULT_EMBEDDING_CONFIG); + const executeQuery = mockExecuteQuery([node]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + {}, + undefined, + new Map([[node.id, currentHash]]), + { forceReembedNodeIds: new Set([node.id]) }, + ); + + const deletedIds = stmtCalls + .filter((call) => call.cypher.includes('DELETE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + const insertedIds = stmtCalls + .filter((call) => call.cypher.includes('CREATE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + expect(deletedIds).toContain(node.id); + expect(insertedIds).toContain(node.id); + }); + + it('announces each checkpoint window before mutating any node in that window', async () => { + mockEmbedderSetup(); + const first = makeNode({ id: 'Function:first:src/first.ts', name: 'first' }); + const second = makeNode({ id: 'Function:second:src/second.ts', name: 'second' }); + const third = makeNode({ id: 'Function:third:src/third.ts', name: 'third' }); + const executeQuery = mockExecuteQuery([first, second, third]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + const windows: string[][] = []; + const mutationCountsAtWindowStart: number[] = []; + const checkpoints: number[] = []; + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + { batchSize: 1 }, + undefined, + new Map(), + { + checkpointEveryNodes: 2, + onCheckpointWindowStart: async ({ nodeIds }) => { + windows.push(nodeIds); + mutationCountsAtWindowStart.push(stmtCalls.length); + }, + onCheckpoint: async ({ nodesProcessed }) => { + checkpoints.push(nodesProcessed); + }, + }, + ); + + expect(windows).toEqual([[first.id, second.id], [third.id]]); + expect(mutationCountsAtWindowStart).toEqual([0, 2]); + expect(checkpoints).toEqual([2, 3]); + }); + + it('deletes pending-window rows whose node is no longer embeddable', async () => { + mockEmbedderSetup(); + const live = makeNode({ id: 'Function:live:src/live.ts', name: 'live' }); + const removedNodeId = 'Function:removed:src/removed.ts'; + const executeQuery = mockExecuteQuery([live]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + {}, + undefined, + new Map([[removedNodeId, 'persisted-partial-hash']]), + { forceReembedNodeIds: new Set([removedNodeId]) }, + ); + + const deletedIds = stmtCalls + .filter((call) => call.cypher.includes('DELETE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + expect(deletedIds).toContain(removedNodeId); + }); + it('deletes only stale nodes — new and unchanged nodes are never deleted (#2333 U6)', async () => { mockEmbedderSetup(); diff --git a/gitnexus/test/unit/http-embedder.test.ts b/gitnexus/test/unit/http-embedder.test.ts index b99cef182..9def24e3b 100644 --- a/gitnexus/test/unit/http-embedder.test.ts +++ b/gitnexus/test/unit/http-embedder.test.ts @@ -6,6 +6,9 @@ const ENV_KEYS = [ 'GITNEXUS_EMBEDDING_MODEL', 'GITNEXUS_EMBEDDING_API_KEY', 'GITNEXUS_EMBEDDING_DIMS', + 'GITNEXUS_EMBEDDING_MAX_ATTEMPTS', + 'GITNEXUS_EMBEDDING_RETRY_CAP_MS', + 'GITNEXUS_EMBEDDING_MIN_INTERVAL_MS', ] as const; /** 384d mock vector matching the default schema dimensions. */ @@ -16,6 +19,7 @@ describe('HTTP embedding backend', () => { const savedEnv = Object.fromEntries(ENV_KEYS.map((k) => [k, process.env[k]])); afterEach(() => { + vi.useRealTimers(); vi.unstubAllGlobals(); vi.resetModules(); // Restore env vars to pre-test state so a mid-test throw can't leak @@ -339,9 +343,25 @@ describe('HTTP embedding backend', () => { expect(isHttpEmbeddingError(err)).toBe(true); // The secret is gone; the masked host is retained so the message stays useful. expect(String(err)).not.toContain('secret'); + expect(String((err as Error & { cause?: unknown }).cause)).not.toContain('secret'); expect(String(err)).toContain('host.example'); }); + it('redacts the API key from both the message and diagnostic cause', async () => { + process.env.GITNEXUS_EMBEDDING_URL = 'https://host.example/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_API_KEY = 'super-secret-key'; + vi.stubGlobal( + 'fetch', + vi.fn().mockRejectedValue(new TypeError('transport rejected super-secret-key')), + ); + + const { embedText } = await import('../../src/core/embeddings/embedder.js'); + const err = await embedText('test').catch((error: unknown) => error); + expect(String(err)).not.toContain('super-secret-key'); + expect(String((err as Error & { cause?: unknown }).cause)).not.toContain('super-secret-key'); + }); + it('leaves a non-credential reason unchanged (no over-scrubbing)', async () => { process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; @@ -563,6 +583,114 @@ describe('HTTP embedding backend', () => { expect(fetch).toHaveBeenCalledTimes(2); expect(result).toBeInstanceOf(Float32Array); }); + + it('honors the configured total attempt bound', async () => { + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_MAX_ATTEMPTS = '1'; + vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: false, status: 503 })); + + const { embedText } = await import('../../src/core/embeddings/embedder.js'); + await expect(embedText('test')).rejects.toThrow('503'); + expect(fetch).toHaveBeenCalledTimes(1); + }); + + it('caps Retry-After with the configured retry cap', async () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_MAX_ATTEMPTS = '2'; + process.env.GITNEXUS_EMBEDDING_RETRY_CAP_MS = '2500'; + const ok = { ok: true, json: async () => ({ data: [{ embedding: mockVec }] }) }; + vi.stubGlobal( + 'fetch', + vi + .fn() + .mockResolvedValueOnce( + new Response('{}', { status: 429, headers: { 'Retry-After': '60' } }), + ) + .mockResolvedValueOnce(ok), + ); + + const { embedText } = await import('../../src/core/embeddings/embedder.js'); + const promise = embedText('test'); + await vi.advanceTimersByTimeAsync(2499); + expect(fetch).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1); + await expect(promise).resolves.toBeInstanceOf(Float32Array); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it('paces retries and successful batches through one minimum-interval queue', async () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_MAX_ATTEMPTS = '2'; + process.env.GITNEXUS_EMBEDDING_RETRY_CAP_MS = '1'; + process.env.GITNEXUS_EMBEDDING_MIN_INTERVAL_MS = '1000'; + const makeResp = (count: number) => ({ + ok: true, + json: async () => ({ data: Array.from({ length: count }, () => ({ embedding: mockVec })) }), + }); + vi.stubGlobal( + 'fetch', + vi + .fn() + .mockResolvedValueOnce({ ok: false, status: 503 }) + .mockResolvedValueOnce(makeResp(64)) + .mockResolvedValueOnce(makeResp(6)), + ); + + const { embedBatch } = await import('../../src/core/embeddings/embedder.js'); + const promise = embedBatch(Array.from({ length: 70 }, (_, i) => `text ${i}`)); + await vi.advanceTimersByTimeAsync(0); + expect(fetch).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(999); + expect(fetch).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1); + expect(fetch).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(999); + expect(fetch).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(1); + await expect(promise).resolves.toHaveLength(70); + expect(fetch).toHaveBeenCalledTimes(3); + }); + + it('cancels promptly while waiting for retry backoff', async () => { + vi.useFakeTimers(); + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_MAX_ATTEMPTS = '3'; + process.env.GITNEXUS_EMBEDDING_RETRY_CAP_MS = '60000'; + vi.stubGlobal( + 'fetch', + vi + .fn() + .mockResolvedValue(new Response('{}', { status: 429, headers: { 'Retry-After': '60' } })), + ); + const controller = new AbortController(); + + const { embedBatch } = await import('../../src/core/embeddings/embedder.js'); + const promise = embedBatch(['test'], { signal: controller.signal }); + await vi.advanceTimersByTimeAsync(1); + controller.abort(); + await expect(promise).rejects.toThrow(/cancelled/i); + expect(fetch).toHaveBeenCalledTimes(1); + }); + + it.each([ + ['GITNEXUS_EMBEDDING_MAX_ATTEMPTS', '0'], + ['GITNEXUS_EMBEDDING_RETRY_CAP_MS', '-1'], + ['GITNEXUS_EMBEDDING_MIN_INTERVAL_MS', 'nope'], + ])('rejects malformed resilience config %s=%s', async (key, value) => { + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env[key] = value; + const { embedText } = await import('../../src/core/embeddings/embedder.js'); + await expect(embedText('test')).rejects.toThrow(key); + }); }); describe('dimension mismatch on query path', () => { diff --git a/gitnexus/test/unit/integrations/resilient-fetch.test.ts b/gitnexus/test/unit/integrations/resilient-fetch.test.ts index 05299c0ba..fa7114e04 100644 --- a/gitnexus/test/unit/integrations/resilient-fetch.test.ts +++ b/gitnexus/test/unit/integrations/resilient-fetch.test.ts @@ -185,6 +185,22 @@ describe('resilientFetch', () => { expect(sleep).toHaveBeenCalledWith(50); }); + it('lets a caller set a stricter Retry-After cap', async () => { + let n = 0; + const fetchImpl = vi.fn(async () => { + n += 1; + return n === 1 ? jsonResp(429, { 'Retry-After': '60' }) : jsonResp(204); + }); + const sleep = vi.fn(async () => {}); + const { breaker } = makeBreaker(); + await resilientFetch(URL_STR, undefined, { + fetchImpl: fetchImpl as unknown as typeof fetch, + breaker, + retry: { sleep, capDelayMs: 2500, retryAfterCapMs: 2500 }, + }); + expect(sleep).toHaveBeenCalledWith(2500); + }); + it('401 returned as Response, no retry, breaker not incremented', async () => { const fetchImpl = vi.fn(async () => jsonResp(401)); const sleep = vi.fn(async () => {}); diff --git a/gitnexus/test/unit/run-analyze.test.ts b/gitnexus/test/unit/run-analyze.test.ts index 93a2f8915..af2c64ad8 100644 --- a/gitnexus/test/unit/run-analyze.test.ts +++ b/gitnexus/test/unit/run-analyze.test.ts @@ -1,7 +1,7 @@ import { execSync } from 'child_process'; import fs from 'fs/promises'; import path from 'path'; -import { describe, it, expect } from 'vitest'; +import { describe, it, expect, vi } from 'vitest'; import { deriveEmbeddingMode, deriveEmbeddingCap, @@ -17,6 +17,7 @@ import { } from '../../src/storage/repo-manager.js'; import { taintModelVersion } from '../../src/core/ingestion/taint/typescript-model.js'; import { createTempDir } from '../helpers/test-db.js'; +import { readEmbeddingNodeIds } from '../helpers/embedding-seed.js'; describe('run-analyze module', () => { it('exports runFullAnalysis as a function', async () => { @@ -74,6 +75,145 @@ describe('run-analyze module', () => { } }); + it('resumes a matching embedding checkpoint instead of taking the clean fast path', async () => { + const tmpRepo = await createTempDir('gitnexus-run-analyze-embedding-checkpoint-'); + const tmpHome = await createTempDir('gitnexus-run-analyze-embedding-checkpoint-home-'); + const saved = { + home: process.env.GITNEXUS_HOME, + url: process.env.GITNEXUS_EMBEDDING_URL, + model: process.env.GITNEXUS_EMBEDDING_MODEL, + dims: process.env.GITNEXUS_EMBEDDING_DIMS, + extension: process.env.GITNEXUS_LBUG_EXTENSION_INSTALL, + }; + try { + process.env.GITNEXUS_HOME = tmpHome.dbPath; + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_DIMS = '384'; + process.env.GITNEXUS_LBUG_EXTENSION_INSTALL = 'never'; + const vector = Array.from({ length: 384 }, (_, i) => i / 384); + const fetchMock = vi.fn().mockImplementation(async (_input, init?: RequestInit) => { + const body = JSON.parse(String(init?.body ?? '{}')) as { input?: unknown[] }; + const count = Array.isArray(body.input) ? body.input.length : 1; + return { + ok: true, + json: async () => ({ + data: Array.from({ length: count }, () => ({ embedding: vector })), + }), + }; + }); + vi.stubGlobal('fetch', fetchMock); + await fs.writeFile( + path.join(tmpRepo.dbPath, 'index.ts'), + 'export function checkpointResume() { return "ready"; }\n', + ); + execSync('git init', { cwd: tmpRepo.dbPath, stdio: 'pipe' }); + execSync('git add index.ts', { cwd: tmpRepo.dbPath, stdio: 'pipe' }); + execSync('git -c user.name=test -c user.email=test@test commit -m init', { + cwd: tmpRepo.dbPath, + stdio: 'pipe', + }); + + const { runFullAnalysis } = await import('../../src/core/run-analyze.js'); + await runFullAnalysis( + tmpRepo.dbPath, + { embeddings: true, skipAgentsMd: true, skipSkills: true }, + { onProgress: () => {} }, + ); + const { storagePath } = getStoragePaths(tmpRepo.dbPath); + const completed = await loadMeta(storagePath); + expect(completed).not.toBeNull(); + if (!completed) throw new Error('expected completed metadata'); + await saveMeta(storagePath, { + ...completed, + embeddingCheckpoint: { + at: new Date().toISOString(), + nodesProcessed: 1, + totalNodes: 1, + chunksProcessed: 1, + model: 'test-model', + dimensions: 384, + }, + } as RepoMeta); + fetchMock.mockClear(); + const logs: string[] = []; + + const resumed = await runFullAnalysis( + tmpRepo.dbPath, + { skipAgentsMd: true, skipSkills: true }, + { onProgress: () => {}, onLog: (message) => logs.push(message) }, + ); + + expect(resumed.alreadyUpToDate).not.toBe(true); + expect(fetchMock).not.toHaveBeenCalled(); + expect(logs.some((message) => message.includes('embedding checkpoint'))).toBe(true); + expect((await loadMeta(storagePath))?.embeddingCheckpoint).toBeUndefined(); + + const finalized = await loadMeta(storagePath); + if (!finalized) throw new Error('expected finalized metadata'); + const [pendingNodeId] = await readEmbeddingNodeIds(tmpRepo.dbPath); + if (!pendingNodeId) throw new Error('expected a persisted embedding node'); + await saveMeta(storagePath, { + ...finalized, + embeddingCheckpoint: { + at: new Date().toISOString(), + nodesProcessed: 0, + totalNodes: 1, + chunksProcessed: 0, + model: 'test-model', + dimensions: 384, + pendingNodeIds: [pendingNodeId], + }, + }); + fetchMock.mockClear(); + + await runFullAnalysis( + tmpRepo.dbPath, + { skipAgentsMd: true, skipSkills: true }, + { onProgress: () => {} }, + ); + + expect(fetchMock).toHaveBeenCalled(); + expect((await loadMeta(storagePath))?.embeddingCheckpoint).toBeUndefined(); + + const resumedPending = await loadMeta(storagePath); + if (!resumedPending) throw new Error('expected pending-window resume metadata'); + fetchMock.mockClear(); + await saveMeta(storagePath, { + ...resumedPending, + embeddingCheckpoint: { + at: new Date().toISOString(), + nodesProcessed: 1, + totalNodes: 2, + chunksProcessed: 1, + model: 'different-model', + dimensions: 384, + }, + }); + await expect( + runFullAnalysis( + tmpRepo.dbPath, + { skipAgentsMd: true, skipSkills: true }, + { onProgress: () => {} }, + ), + ).rejects.toThrow('Cannot resume embedding checkpoint'); + expect(fetchMock).not.toHaveBeenCalled(); + } finally { + vi.unstubAllGlobals(); + const restore = (key: string, value: string | undefined) => { + if (value === undefined) delete process.env[key]; + else process.env[key] = value; + }; + restore('GITNEXUS_HOME', saved.home); + restore('GITNEXUS_EMBEDDING_URL', saved.url); + restore('GITNEXUS_EMBEDDING_MODEL', saved.model); + restore('GITNEXUS_EMBEDDING_DIMS', saved.dims); + restore('GITNEXUS_LBUG_EXTENSION_INSTALL', saved.extension); + await tmpRepo.cleanup(); + await tmpHome.cleanup(); + } + }, 120_000); + it('plain analyze on another branch adopts the flat workspace slot (#2354)', async () => { const tmpRepo = await createTempDir('gitnexus-run-analyze-workspace-'); const tmpHome = await createTempDir('gitnexus-run-analyze-workspace-home-');