From 60cb166889b69d027d7e91d90df5ebd4554884fd Mon Sep 17 00:00:00 2001 From: Gergo Magyar Date: Thu, 17 Sep 2026 13:29:20 +0000 Subject: [PATCH] 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. --- gitnexus/src/core/run-analyze.ts | 17 +-- .../test/unit/embedding-restore-spill.test.ts | 13 +- .../test/unit/run-analyze-fts-repair.test.ts | 112 +++++++++++++++++- 3 files changed, 132 insertions(+), 10 deletions(-) diff --git a/gitnexus/src/core/run-analyze.ts b/gitnexus/src/core/run-analyze.ts index 6e8a6c4e6..8ea0b0f66 100644 --- a/gitnexus/src/core/run-analyze.ts +++ b/gitnexus/src/core/run-analyze.ts @@ -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(); - const restoreAttemptedNodeIds = new Set(); + const restoreFailedNodeIds = new Set(); 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(); 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); diff --git a/gitnexus/test/unit/embedding-restore-spill.test.ts b/gitnexus/test/unit/embedding-restore-spill.test.ts index 04693ed16..b11695110 100644 --- a/gitnexus/test/unit/embedding-restore-spill.test.ts +++ b/gitnexus/test/unit/embedding-restore-spill.test.ts @@ -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/, diff --git a/gitnexus/test/unit/run-analyze-fts-repair.test.ts b/gitnexus/test/unit/run-analyze-fts-repair.test.ts index 4f641f8a7..2f0304ef4 100644 --- a/gitnexus/test/unit/run-analyze-fts-repair.test.ts +++ b/gitnexus/test/unit/run-analyze-fts-repair.test.ts @@ -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 | 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()), + 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 | 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(); + } + }); }); /**