mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-08-28 05:25:25 +00:00
fix(embeddings): make HTTP generation resumable
This commit is contained in:
Eva
parent
c6445096eb
commit
711ff8721d
16 changed files with 975 additions and 34 deletions
|
|
@ -33,6 +33,8 @@ export interface ResilientFetchOptions {
|
|||
breakerOptions?: CircuitBreakerOptions;
|
||||
/** Tuning knobs for the retry helper. */
|
||||
retry?: Partial<Pick<RetryOptions, 'maxAttempts' | 'baseDelayMs' | 'capDelayMs'>> & {
|
||||
/** 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':
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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<Float32Array> => {
|
||||
export const embedText = async (
|
||||
text: string,
|
||||
options: EmbeddingRequestOptions = {},
|
||||
): Promise<Float32Array> => {
|
||||
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<Float32Array> => {
|
|||
* @param texts - Array of texts to embed
|
||||
* @returns Array of Float32Array embedding vectors
|
||||
*/
|
||||
export const embedBatch = async (texts: string[]): Promise<Float32Array[]> => {
|
||||
export const embedBatch = async (
|
||||
texts: string[],
|
||||
options: EmbeddingRequestOptions = {},
|
||||
): Promise<Float32Array[]> => {
|
||||
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<Float32Array[]> => {
|
|||
pooling: 'mean',
|
||||
normalize: true,
|
||||
});
|
||||
options.signal?.throwIfAborted();
|
||||
|
||||
// Result shape is [batch_size, dimensions]
|
||||
// Need to split into individual vectors
|
||||
|
|
|
|||
|
|
@ -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<string>;
|
||||
onCheckpointWindowStart?: (window: EmbeddingPipelineCheckpointWindow) => Promise<void>;
|
||||
onCheckpoint?: (checkpoint: EmbeddingPipelineCheckpoint) => Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* DELETE stale embedding rows for the given nodeIds so they can be re-inserted.
|
||||
*
|
||||
|
|
@ -320,12 +338,20 @@ export const runEmbeddingPipeline = async (
|
|||
config: Partial<EmbeddingConfig> = {},
|
||||
skipNodeIds?: Set<string>,
|
||||
existingEmbeddings?: Map<string, string>,
|
||||
pipelineOptions: EmbeddingPipelineOptions = {},
|
||||
): Promise<EmbeddingPipelineResult> => {
|
||||
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<string>();
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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<void> = 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<void> => {
|
||||
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<void> => {
|
||||
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<EmbeddingItem[]> => {
|
||||
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<Float32Array[]> => {
|
||||
export const httpEmbed = async (
|
||||
texts: string[],
|
||||
requestOptions: EmbeddingRequestOptions = {},
|
||||
): Promise<Float32Array[]> => {
|
||||
if (texts.length === 0) return [];
|
||||
|
||||
const config = readConfig();
|
||||
|
|
@ -309,6 +435,10 @@ export const httpEmbed = async (texts: string[]): Promise<Float32Array[]> => {
|
|||
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<Float32Array[]> => {
|
|||
* @param text - Query text to embed
|
||||
* @returns Embedding vector as number array
|
||||
*/
|
||||
export const httpEmbedQuery = async (text: string): Promise<number[]> => {
|
||||
export const httpEmbedQuery = async (
|
||||
text: string,
|
||||
requestOptions: EmbeddingRequestOptions = {},
|
||||
): Promise<number[]> => {
|
||||
const config = readConfig();
|
||||
if (!config) throw new Error('HTTP embedding not configured');
|
||||
|
||||
|
|
@ -359,6 +492,10 @@ export const httpEmbedQuery = async (text: string): Promise<number[]> => {
|
|||
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)})`);
|
||||
|
|
|
|||
|
|
@ -267,6 +267,22 @@ export interface AnalyzeOptions {
|
|||
skipNativeCloseOnExit?: boolean;
|
||||
}
|
||||
|
||||
interface EmbeddingIdentity {
|
||||
model: string;
|
||||
dimensions: number;
|
||||
}
|
||||
|
||||
const resolveEmbeddingIdentity = async (): Promise<EmbeddingIdentity> => {
|
||||
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<string>();
|
||||
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<nodeId, contentHash> from cached embeddings for incremental mode
|
||||
let existingEmbeddings: Map<string, string> | 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<void> => {
|
||||
const fileHashes: Record<string, string> = {};
|
||||
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
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ const JOB_TIMEOUT_MS = 30 * 60 * 1000; // 30 minutes
|
|||
export class JobManager {
|
||||
private jobs = new Map<string, AnalyzeJob>();
|
||||
private children = new Map<string, ChildProcess>();
|
||||
private abortControllers = new Map<string, AbortController>();
|
||||
private timeouts = new Map<string, ReturnType<typeof setTimeout>>();
|
||||
private emitter = new EventEmitter();
|
||||
private cleanupTimer: ReturnType<typeof setInterval>;
|
||||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<void> => {
|
||||
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);
|
||||
}
|
||||
})();
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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' });
|
||||
|
|
|
|||
|
|
@ -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');
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
||||
|
|
|
|||
|
|
@ -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', () => {
|
||||
|
|
|
|||
|
|
@ -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 () => {});
|
||||
|
|
|
|||
|
|
@ -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-');
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue