mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-10-14 04:08:05 +00:00
Address PR review feedback (#3310)
Mark a node stale when any restore batch fails so leftover chunks are deleted and rembedded, and exercise a full-length bad-magic spill header.
This commit is contained in:
Gergo Magyar
parent
0e9116c0b7
commit
60cb166889
3 changed files with 132 additions and 10 deletions
|
|
@ -3662,11 +3662,10 @@ async function runFullAnalysisInner(
|
|||
// delete outcome) and this process holds the exclusive DB lock (no
|
||||
// concurrent writer).
|
||||
// Materialize runs outside the insert catch so a spill I/O failure is not
|
||||
// treated as a benign PK conflict. Failed restore batches are omitted from
|
||||
// the Phase 4 skip map so those nodes can be re-embedded.
|
||||
// treated as a benign PK conflict. Any node with a failed restore batch is
|
||||
// marked stale in the Phase 4 map so leftover chunks are deleted and rembedded.
|
||||
let restoredEmbeddingCount = 0;
|
||||
const restoredEmbeddingNodeIds = new Set<string>();
|
||||
const restoreAttemptedNodeIds = new Set<string>();
|
||||
const restoreFailedNodeIds = new Set<string>();
|
||||
if (cacheRowCount(cachedSnapshot) > 0) {
|
||||
const cachedDims = snapshotEmbeddingDims(cachedSnapshot);
|
||||
const { EMBEDDING_DIMS } = await import('./lbug/schema.js');
|
||||
|
|
@ -3698,7 +3697,6 @@ async function runFullAnalysisInner(
|
|||
let spillReader: EmbeddingSpillReader | undefined;
|
||||
try {
|
||||
for (const batch of chunk(rowsToRestore, EMBED_BATCH)) {
|
||||
for (const row of batch) restoreAttemptedNodeIds.add(row.nodeId);
|
||||
let materialized;
|
||||
try {
|
||||
if (!spillReader && cachedSnapshot.spill && cachedSnapshot.embeddings.length === 0) {
|
||||
|
|
@ -3706,6 +3704,7 @@ async function runFullAnalysisInner(
|
|||
}
|
||||
materialized = materializeCachedEmbeddings(cachedSnapshot, batch, spillReader);
|
||||
} catch (err) {
|
||||
for (const row of batch) restoreFailedNodeIds.add(row.nodeId);
|
||||
log(
|
||||
`Warning: could not materialize ${batch.length} cached embedding(s) for restore ` +
|
||||
`(${(err as Error).message}); those nodes will be re-embedded if this run generates embeddings.`,
|
||||
|
|
@ -3715,8 +3714,8 @@ async function runFullAnalysisInner(
|
|||
try {
|
||||
await batchInsert(executeWithReusedStatement, materialized);
|
||||
restoredEmbeddingCount += batch.length;
|
||||
for (const row of batch) restoredEmbeddingNodeIds.add(row.nodeId);
|
||||
} catch (err) {
|
||||
for (const row of batch) restoreFailedNodeIds.add(row.nodeId);
|
||||
log(
|
||||
`Warning: could not restore ${batch.length} cached embedding(s) ` +
|
||||
`(${(err as Error).message}); those nodes will be re-embedded if this run generates embeddings.`,
|
||||
|
|
@ -4025,7 +4024,11 @@ async function runFullAnalysisInner(
|
|||
if (cachedSnapshot.embeddingNodeIds.size > 0) {
|
||||
existingEmbeddings = new Map<string, string>();
|
||||
for (const e of cachedSnapshot.rows) {
|
||||
if (restoreAttemptedNodeIds.has(e.nodeId) && !restoredEmbeddingNodeIds.has(e.nodeId)) {
|
||||
if (restoreFailedNodeIds.has(e.nodeId)) {
|
||||
// Any failed batch for this node: mark stale so Phase 4 DELETEs
|
||||
// leftover chunks and re-embeds. Omitting the id would treat the
|
||||
// node as new and PK-conflict on rows that already restored.
|
||||
existingEmbeddings.set(e.nodeId, STALE_HASH_SENTINEL);
|
||||
continue;
|
||||
}
|
||||
existingEmbeddings.set(e.nodeId, e.contentHash ?? STALE_HASH_SENTINEL);
|
||||
|
|
|
|||
|
|
@ -168,8 +168,19 @@ describe('embedding-restore-spill (#3306)', () => {
|
|||
).toThrow(/dim mismatch/);
|
||||
abortCachedEmbeddingsBuilder(builder);
|
||||
|
||||
const shortPath = path.join(os.tmpdir(), `gitnexus-embed-restore-short-${process.pid}.bin`);
|
||||
writeFileSync(shortPath, Buffer.from('NOPE'));
|
||||
spills.push({ path: shortPath });
|
||||
expect(() => readSpillVectors({ path: shortPath, dims: DIMS, rowCount: 1 }, [0])).toThrow(
|
||||
/invalid embedding spill header/,
|
||||
);
|
||||
|
||||
const badMagic = Buffer.alloc(12);
|
||||
badMagic.write('NOPE', 0, 4, 'ascii');
|
||||
badMagic.writeUInt8(1, 4);
|
||||
badMagic.writeUInt32LE(DIMS, 5);
|
||||
const badPath = path.join(os.tmpdir(), `gitnexus-embed-restore-bad-${process.pid}.bin`);
|
||||
writeFileSync(badPath, Buffer.from('NOPE'));
|
||||
writeFileSync(badPath, badMagic);
|
||||
spills.push({ path: badPath });
|
||||
expect(() => readSpillVectors({ path: badPath, dims: DIMS, rowCount: 1 }, [0])).toThrow(
|
||||
/invalid embedding spill header/,
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ import {
|
|||
saveMeta,
|
||||
type RepoMeta,
|
||||
} from '../../src/storage/repo-manager.js';
|
||||
import { EMBEDDING_DIMS } from '../../src/core/lbug/schema.js';
|
||||
import { EMBEDDING_DIMS, STALE_HASH_SENTINEL } from '../../src/core/lbug/schema.js';
|
||||
import { getIndexIncompleteReasons } from '../../src/core/index-freshness.js';
|
||||
import type {
|
||||
EmbeddingPipelineOptions,
|
||||
|
|
@ -1418,12 +1418,120 @@ describe('runFullAnalysis wipe-and-restore vector-index stamp (tri-review 466951
|
|||
);
|
||||
|
||||
expect(runEmbeddingPipeline).toHaveBeenCalled();
|
||||
expect(existingEmbeddings?.has(RESTORED_NODE_ID)).toBe(false);
|
||||
expect(existingEmbeddings?.get(RESTORED_NODE_ID)).toBe(STALE_HASH_SENTINEL);
|
||||
expect(logs.some((m) => m.includes('Warning: could not restore'))).toBe(true);
|
||||
} finally {
|
||||
await tmpRepo.cleanup();
|
||||
}
|
||||
});
|
||||
|
||||
it('marks a node stale when only some of its restore batches succeed', async () => {
|
||||
const RESTORED_NODE_ID = 'Function:src/app.ts:handler:1';
|
||||
const CACHED_HASH = 'cached-stable-hash';
|
||||
const stubNode = {
|
||||
id: RESTORED_NODE_ID,
|
||||
label: 'Function',
|
||||
name: 'handler',
|
||||
properties: { filePath: 'src/app.ts' },
|
||||
};
|
||||
const embeddings = Array.from({ length: 201 }, (_, chunkIndex) => ({
|
||||
nodeId: RESTORED_NODE_ID,
|
||||
chunkIndex,
|
||||
startLine: chunkIndex,
|
||||
endLine: chunkIndex + 1,
|
||||
embedding: new Array(EMBEDDING_DIMS).fill(0),
|
||||
contentHash: CACHED_HASH,
|
||||
}));
|
||||
let existingEmbeddings: Map<string, string> | undefined;
|
||||
vi.doMock('../../src/core/lbug/lbug-adapter.js', () => ({
|
||||
initLbug: vi.fn(async () => undefined),
|
||||
loadGraphToLbug: vi.fn(async () => undefined),
|
||||
getLbugStats: vi.fn(async () => ({ nodes: 2, edges: 0, communities: 0, processes: 0 })),
|
||||
executeQuery: vi.fn(async () => []),
|
||||
executeWithReusedStatement: vi.fn(async () => []),
|
||||
closeLbug: vi.fn(async () => undefined),
|
||||
wipeLbugDbFiles: vi.fn(async () => undefined),
|
||||
tryFlushWAL: vi.fn(async () => true),
|
||||
loadCachedEmbeddings: vi.fn(async () => ({
|
||||
embeddingNodeIds: new Set([RESTORED_NODE_ID]),
|
||||
embeddings,
|
||||
})),
|
||||
deleteNodesForFile: vi.fn(async () => undefined),
|
||||
deleteNodesForFiles: vi.fn(async () => undefined),
|
||||
deleteAllCommunitiesAndProcesses: vi.fn(async () => undefined),
|
||||
queryImporters: vi.fn(async () => []),
|
||||
queryImportersBatch: vi.fn(async () => []),
|
||||
loadFTSExtension: vi.fn(async () => false),
|
||||
}));
|
||||
vi.doMock('../../src/core/search/fts-indexes.js', () => ({
|
||||
initialiseSearchFTSStemmer: vi.fn(() => 'porter'),
|
||||
createSearchFTSIndexes: vi.fn(async () => []),
|
||||
verifySearchFTSIndexes: vi.fn(async () => []),
|
||||
}));
|
||||
vi.doMock('../../src/core/ingestion/pipeline.js', () => ({
|
||||
runPipelineFromRepo: vi.fn(async (repoPath: string) => ({
|
||||
repoPath,
|
||||
totalFileCount: 1,
|
||||
graph: {
|
||||
forEachNode: (fn: (node: typeof stubNode) => void) => fn(stubNode),
|
||||
getNode: (id: string) => (id === RESTORED_NODE_ID ? stubNode : undefined),
|
||||
},
|
||||
})),
|
||||
}));
|
||||
vi.doMock('../../src/storage/repo-manager.js', async (importActual) => ({
|
||||
...(await importActual<typeof import('../../src/storage/repo-manager.js')>()),
|
||||
registerRepo: vi.fn(async () => 'partial-restore-skip-set-repo'),
|
||||
ensureGitNexusIgnored: vi.fn(async () => undefined),
|
||||
}));
|
||||
const runEmbeddingPipeline = vi.fn(
|
||||
async (
|
||||
_executeQuery: unknown,
|
||||
_executeWithReusedStatement: unknown,
|
||||
_onProgress: unknown,
|
||||
_config: unknown,
|
||||
_cachedNodeIds: unknown,
|
||||
embeddings: Map<string, string> | undefined,
|
||||
) => {
|
||||
existingEmbeddings = embeddings;
|
||||
return {
|
||||
nodesProcessed: 0,
|
||||
chunksProcessed: 0,
|
||||
vectorIndexReady: false,
|
||||
semanticMode: 'exact-scan' as const,
|
||||
failedNodeIds: [],
|
||||
};
|
||||
},
|
||||
);
|
||||
let inserts = 0;
|
||||
vi.doMock('../../src/core/embeddings/embedding-pipeline.js', () => ({
|
||||
runEmbeddingPipeline,
|
||||
buildVectorIndex: vi.fn(async () => false),
|
||||
batchInsertEmbeddings: vi.fn(async () => {
|
||||
inserts += 1;
|
||||
if (inserts > 1) throw new Error('second batch insert failed');
|
||||
}),
|
||||
}));
|
||||
|
||||
const tmpRepo = await createTempDir('gitnexus-run-analyze-partial-restore-');
|
||||
try {
|
||||
const { storagePath } = getStoragePaths(tmpRepo.dbPath);
|
||||
await fs.mkdir(storagePath, { recursive: true });
|
||||
await saveMeta(storagePath, {
|
||||
repoPath: tmpRepo.dbPath,
|
||||
lastCommit: '',
|
||||
indexedAt: new Date().toISOString(),
|
||||
stats: { embeddings: 201 },
|
||||
});
|
||||
|
||||
const { runFullAnalysis } = await import('../../src/core/run-analyze.js');
|
||||
await runFullAnalysis(tmpRepo.dbPath, { force: true }, { onProgress: () => {} });
|
||||
|
||||
expect(inserts).toBe(2);
|
||||
expect(existingEmbeddings?.get(RESTORED_NODE_ID)).toBe(STALE_HASH_SENTINEL);
|
||||
} finally {
|
||||
await tmpRepo.cleanup();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue