perf(kernel): direct-to-store decode — buffers flow to the store worker, main thread never materializes nodes
Kernel-routed files ship their flat tables from the parse worker to the store worker as buffers (tryKernelExtractRaw → kernelBuffers on the result → KernelStoreBundle); the store worker decodes and finalizes (finalizeStoreBundle shared with the object path so filter semantics can never drift). Files with applicable framework extract() hooks keep the decoded path; the no-writer fallback materializes via materializeKernelResult. Byte-identical dumps re-verified on dubbo, excalidraw, express, gson; full suite green (2,467). Measurement (plan §4d): dubbo's parse-loop wall is 94% store-writer busy time — the many-core fresh-index wall is single-writer SQLite ingest, not extraction or main-thread work. d2s improves the writer lane ~11% (structured-clone deserialization avoided on the writer) and frees the main thread; the remaining many-core gap is a store-architecture arc (deferred index builds, multi-file transactions, buffer→bind), out of the kernel project's scope. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5
parent
03d54e47a1
commit
28068fa0f1
@@ -264,6 +264,31 @@ Default routing: `DEFAULT_ROUTED = {typescript, tsx, javascript, jsx}` in
|
|||||||
Windows VM still deferred (same fallback rationale as §4b).
|
Windows VM still deferred (same fallback rationale as §4b).
|
||||||
- Default routing now includes `java`.
|
- Default routing now includes `java`.
|
||||||
|
|
||||||
|
### 4d. Direct-to-store decode (2026-07-16) — and where the wall ACTUALLY is
|
||||||
|
|
||||||
|
Kernel-routed files now ship their flat buffers from the parse worker all the way
|
||||||
|
to the STORE WORKER, which decodes + finalizes them there (`tryKernelExtractRaw` →
|
||||||
|
`ExtractionResult.kernelBuffers` → `KernelStoreBundle` → `decodeKernelBundle`;
|
||||||
|
filter semantics shared via `finalizeStoreBundle`). The main thread's per-file work
|
||||||
|
drops to O(1) + the content hash — it never materializes per-node objects, and both
|
||||||
|
postMessage hops move flat bytes instead of object graphs. Files whose applicable
|
||||||
|
frameworks carry an `extract()` hook keep the decoded path (hooks merge into decoded
|
||||||
|
results); non-writer paths (main-thread store, tests) materialize via
|
||||||
|
`materializeKernelResult`. Byte-identical dumps re-verified on dubbo, excalidraw,
|
||||||
|
express, gson.
|
||||||
|
|
||||||
|
**Measurement that closes the §4c question:** with the store worker instrumented,
|
||||||
|
dubbo's parse-loop wall is **94% store-writer busy time** (4,202ms of 4,493ms on the
|
||||||
|
kernel arm). The many-core fresh-index wall is the single-writer SQLite ingest —
|
||||||
|
not extraction, not main-thread work. d2s still improves the writer lane ~11%
|
||||||
|
(4,726→4,202ms: buffers skip structured-clone deserialization ON the writer) and
|
||||||
|
frees the main thread, but the remaining cbm gap on many-core medium repos is a
|
||||||
|
STORE-ARCHITECTURE question (their RAM-first design defers all durability). Next
|
||||||
|
levers there (a separate perf arc, not this project): deferred/bulk index builds
|
||||||
|
during the parse phase, multi-file write transactions, buffer→bind without object
|
||||||
|
materialization. Note the #1320-arc post-mortem already measured statement batching
|
||||||
|
and sorted inserts as ~zero on this path — B-tree maintenance is the floor.
|
||||||
|
|
||||||
## 4. Per-language tracker
|
## 4. Per-language tracker
|
||||||
|
|
||||||
Tiers: **T1** = mostly `.scm` + mapping config. **T2** = needs bespoke pre/post passes kept
|
Tiers: **T1** = mostly `.scm` + mapping config. **T2** = needs bespoke pre/post passes kept
|
||||||
|
|||||||
+51
-32
@@ -23,7 +23,8 @@ import {
|
|||||||
import { QueryBuilder } from '../db/queries';
|
import { QueryBuilder } from '../db/queries';
|
||||||
import { extractFromSource } from './tree-sitter';
|
import { extractFromSource } from './tree-sitter';
|
||||||
import { ParseWorkerPool, resolveParsePoolSize, resolveParseTimeoutMs } from './parse-pool';
|
import { ParseWorkerPool, resolveParsePoolSize, resolveParseTimeoutMs } from './parse-pool';
|
||||||
import { StoreWriter, StoreBundle } from './store-writer';
|
import { StoreWriter, StoreBundle, finalizeStoreBundle } from './store-writer';
|
||||||
|
import { materializeKernelResult } from './kernel';
|
||||||
import { detectLanguage, isSourceFile, isLanguageSupported, isFileLevelOnlyLanguage, initGrammars, loadGrammarsForLanguages, readGrammarWasmBytes } from './grammars';
|
import { detectLanguage, isSourceFile, isLanguageSupported, isFileLevelOnlyLanguage, initGrammars, loadGrammarsForLanguages, readGrammarWasmBytes } from './grammars';
|
||||||
import { loadExtensionOverrides, loadIncludeIgnoredPatterns, loadExcludePatterns, loadIncludePatterns } from '../project-config';
|
import { loadExtensionOverrides, loadIncludeIgnoredPatterns, loadExcludePatterns, loadIncludePatterns } from '../project-config';
|
||||||
import { isCodeGraphDataDir } from '../directory';
|
import { isCodeGraphDataDir } from '../directory';
|
||||||
@@ -1706,16 +1707,34 @@ export class ExtractionOrchestrator {
|
|||||||
const bp = walBackpressure?.();
|
const bp = walBackpressure?.();
|
||||||
if (bp) await bp;
|
if (bp) await bp;
|
||||||
|
|
||||||
|
// Kernel deferred-decode results carry table sizes in kernelCounts
|
||||||
|
// (their object arrays are empty — decode happens at the store).
|
||||||
|
const nodeCount = result.kernelCounts?.nodes ?? result.nodes.length;
|
||||||
|
const edgeCount = result.kernelCounts?.edges ?? result.edges.length;
|
||||||
|
|
||||||
// Store: on the writer thread when active (fresh DB — bundles applied
|
// Store: on the writer thread when active (fresh DB — bundles applied
|
||||||
// in the same file order this chain dispatches them), else on the main
|
// in the same file order this chain dispatches them), else on the main
|
||||||
// thread (SQLite connections are per-thread).
|
// thread (SQLite connections are per-thread).
|
||||||
if (result.nodes.length > 0 || result.errors.length === 0) {
|
if (nodeCount > 0 || result.errors.length === 0) {
|
||||||
const language = detectLanguage(filePath, content, overrides);
|
const language = detectLanguage(filePath, content, overrides);
|
||||||
if (storeWriter) {
|
if (storeWriter) {
|
||||||
|
if (result.kernelBuffers) {
|
||||||
|
// Buffers go to the writer as-is; the worker decodes + finalizes.
|
||||||
|
// The main thread's only per-file work stays O(1) + the content hash.
|
||||||
|
storeWriter.send({
|
||||||
|
kernel: true,
|
||||||
|
filePath,
|
||||||
|
language,
|
||||||
|
buffers: result.kernelBuffers,
|
||||||
|
file: this.buildFileRecord(filePath, content, language, stats, nodeCount, result.errors),
|
||||||
|
});
|
||||||
|
} else {
|
||||||
storeWriter.send(this.buildFreshStoreBundle(filePath, content, language, stats, result));
|
storeWriter.send(this.buildFreshStoreBundle(filePath, content, language, stats, result));
|
||||||
|
}
|
||||||
await storeWriter.waitBelow(STORE_WRITER_WINDOW);
|
await storeWriter.waitBelow(STORE_WRITER_WINDOW);
|
||||||
} else {
|
} else {
|
||||||
await this.storeExtractionResult(filePath, content, language, stats, result, commitYield);
|
const materialized = materializeKernelResult(result, filePath, language);
|
||||||
|
await this.storeExtractionResult(filePath, content, language, stats, materialized, commitYield);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1726,10 +1745,10 @@ export class ExtractionOrchestrator {
|
|||||||
errors.push(...result.errors);
|
errors.push(...result.errors);
|
||||||
}
|
}
|
||||||
|
|
||||||
if (result.nodes.length > 0) {
|
if (nodeCount > 0) {
|
||||||
filesIndexed++;
|
filesIndexed++;
|
||||||
totalNodes += result.nodes.length;
|
totalNodes += nodeCount;
|
||||||
totalEdges += result.edges.length;
|
totalEdges += edgeCount;
|
||||||
} else if (result.errors.some((e) => e.severity === 'error')) {
|
} else if (result.errors.some((e) => e.severity === 'error')) {
|
||||||
filesErrored++;
|
filesErrored++;
|
||||||
} else {
|
} else {
|
||||||
@@ -2383,6 +2402,27 @@ export class ExtractionOrchestrator {
|
|||||||
* check, no cross-file edge snapshot (both are re-index concerns — a fresh
|
* check, no cross-file edge snapshot (both are re-index concerns — a fresh
|
||||||
* database has neither). Filters mirror storeExtractionResult exactly.
|
* database has neither). Filters mirror storeExtractionResult exactly.
|
||||||
*/
|
*/
|
||||||
|
/** The FileRecord for a fresh-index store (nodeCount is the PRE-filter count). */
|
||||||
|
private buildFileRecord(
|
||||||
|
filePath: string,
|
||||||
|
content: string,
|
||||||
|
language: Language,
|
||||||
|
stats: fs.Stats,
|
||||||
|
nodeCount: number,
|
||||||
|
resultErrors: ExtractionResult['errors']
|
||||||
|
): FileRecord {
|
||||||
|
return {
|
||||||
|
path: filePath,
|
||||||
|
contentHash: hashContent(content),
|
||||||
|
language,
|
||||||
|
size: stats.size,
|
||||||
|
modifiedAt: stats.mtimeMs,
|
||||||
|
indexedAt: Date.now(),
|
||||||
|
nodeCount,
|
||||||
|
errors: resultErrors.length > 0 ? resultErrors : undefined,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
private buildFreshStoreBundle(
|
private buildFreshStoreBundle(
|
||||||
filePath: string,
|
filePath: string,
|
||||||
content: string,
|
content: string,
|
||||||
@@ -2390,33 +2430,12 @@ export class ExtractionOrchestrator {
|
|||||||
stats: fs.Stats,
|
stats: fs.Stats,
|
||||||
result: ExtractionResult
|
result: ExtractionResult
|
||||||
): StoreBundle {
|
): StoreBundle {
|
||||||
const validNodes = result.nodes.filter((n) => n.id && n.kind && n.name && n.filePath && n.language);
|
return finalizeStoreBundle(
|
||||||
const insertedIds = new Set(validNodes.map((n) => n.id));
|
result,
|
||||||
const validEdges = result.edges.filter(
|
filePath,
|
||||||
(e) => insertedIds.has(e.source) && insertedIds.has(e.target)
|
|
||||||
);
|
|
||||||
const validRefs = result.unresolvedReferences
|
|
||||||
.filter((ref) => insertedIds.has(ref.fromNodeId))
|
|
||||||
.map((ref) => ({
|
|
||||||
...ref,
|
|
||||||
filePath: ref.filePath ?? filePath,
|
|
||||||
language: ref.language ?? language,
|
|
||||||
}));
|
|
||||||
return {
|
|
||||||
nodes: validNodes,
|
|
||||||
edges: validEdges,
|
|
||||||
refs: validRefs,
|
|
||||||
file: {
|
|
||||||
path: filePath,
|
|
||||||
contentHash: hashContent(content),
|
|
||||||
language,
|
language,
|
||||||
size: stats.size,
|
this.buildFileRecord(filePath, content, language, stats, result.nodes.length, result.errors)
|
||||||
modifiedAt: stats.mtimeMs,
|
);
|
||||||
indexedAt: Date.now(),
|
|
||||||
nodeCount: result.nodes.length,
|
|
||||||
errors: result.errors.length > 0 ? result.errors : undefined,
|
|
||||||
},
|
|
||||||
};
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
@@ -17,6 +17,11 @@
|
|||||||
import type { ExtractionResult, Language } from '../../types';
|
import type { ExtractionResult, Language } from '../../types';
|
||||||
import { getKernel, kernelSupports } from './loader';
|
import { getKernel, kernelSupports } from './loader';
|
||||||
import { decodeExtractBuffers } from './decode';
|
import { decodeExtractBuffers } from './decode';
|
||||||
|
import {
|
||||||
|
KERNEL_ABI_VERSION as LAYOUT_ABI,
|
||||||
|
META as LAYOUT_META,
|
||||||
|
NONE as LAYOUT_NONE,
|
||||||
|
} from './layout';
|
||||||
|
|
||||||
export { getKernel, kernelSupports, resetKernelForTests } from './loader';
|
export { getKernel, kernelSupports, resetKernelForTests } from './loader';
|
||||||
export { decodeExtractBuffers } from './decode';
|
export { decodeExtractBuffers } from './decode';
|
||||||
@@ -66,6 +71,84 @@ export function kernelRoutes(language: Language): boolean {
|
|||||||
/** Warned-once registry so a broken language logs a single line, not one per file. */
|
/** Warned-once registry so a broken language logs a single line, not one per file. */
|
||||||
const warned = new Set<string>();
|
const warned = new Set<string>();
|
||||||
|
|
||||||
|
/** The raw table buffers + the cheap facts the orchestrator needs pre-decode. */
|
||||||
|
export interface KernelRawResult {
|
||||||
|
buffers: NonNullable<ExtractionResult['kernelBuffers']>;
|
||||||
|
counts: { nodes: number; edges: number; refs: number };
|
||||||
|
errors: ExtractionResult['errors'];
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Extract via the kernel WITHOUT decoding — the bulk-index fast path. The
|
||||||
|
* tables ride to the store boundary as buffers (decoded on the store worker),
|
||||||
|
* so the main thread never materializes per-node objects. Returns null under
|
||||||
|
* exactly the conditions tryKernelExtract does, PLUS when the language has a
|
||||||
|
* registered post() pass (post passes operate on decoded results, so those
|
||||||
|
* languages keep the decoded path).
|
||||||
|
*/
|
||||||
|
export function tryKernelExtractRaw(
|
||||||
|
filePath: string,
|
||||||
|
source: string,
|
||||||
|
language: Language
|
||||||
|
): KernelRawResult | null {
|
||||||
|
if (!kernelRoutes(language) || POST_PASSES[language]) return null;
|
||||||
|
const kernel = getKernel();
|
||||||
|
if (!kernel) return null;
|
||||||
|
try {
|
||||||
|
const buffers = kernel.extractFile(filePath, source, language);
|
||||||
|
const meta = buffers.meta;
|
||||||
|
if (meta.readUInt8(LAYOUT_META.version) !== LAYOUT_ABI) {
|
||||||
|
throw new Error(`kernel buffer ABI ${meta.readUInt8(0)} != expected ${LAYOUT_ABI}`);
|
||||||
|
}
|
||||||
|
const counts = {
|
||||||
|
nodes: meta.readUInt32LE(LAYOUT_META.nodeCount),
|
||||||
|
edges: meta.readUInt32LE(LAYOUT_META.edgeCount),
|
||||||
|
refs: meta.readUInt32LE(LAYOUT_META.refCount),
|
||||||
|
};
|
||||||
|
let errors: ExtractionResult['errors'] = [];
|
||||||
|
const errorsOff = meta.readUInt32LE(LAYOUT_META.errorsOff);
|
||||||
|
if (errorsOff !== LAYOUT_NONE) {
|
||||||
|
const errorsLen = meta.readUInt32LE(LAYOUT_META.errorsLen);
|
||||||
|
errors = JSON.parse(
|
||||||
|
buffers.arena.toString('utf8', errorsOff, errorsOff + errorsLen)
|
||||||
|
) as ExtractionResult['errors'];
|
||||||
|
}
|
||||||
|
return { buffers, counts, errors };
|
||||||
|
} catch (err) {
|
||||||
|
const message = err instanceof Error ? err.message : String(err);
|
||||||
|
if (message.includes('defer:')) return null;
|
||||||
|
if (!warned.has(language)) {
|
||||||
|
warned.add(language);
|
||||||
|
process.stderr.write(
|
||||||
|
`[codegraph-kernel] ${language} extraction failed (${message}) — falling back to the wasm path\n`
|
||||||
|
);
|
||||||
|
}
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Decode a buffer-carrying result (see ExtractionResult.kernelBuffers) into a
|
||||||
|
* plain, fully-materialized ExtractionResult — the fallback for store paths
|
||||||
|
* that need objects (main-thread store, tests).
|
||||||
|
*/
|
||||||
|
export function materializeKernelResult(
|
||||||
|
result: ExtractionResult,
|
||||||
|
filePath: string,
|
||||||
|
language: Language
|
||||||
|
): ExtractionResult {
|
||||||
|
if (!result.kernelBuffers) return result;
|
||||||
|
const b = result.kernelBuffers;
|
||||||
|
const asBuf = (u: Uint8Array) => Buffer.from(u.buffer, u.byteOffset, u.byteLength);
|
||||||
|
const decoded = decodeExtractBuffers(
|
||||||
|
{ meta: asBuf(b.meta), nodes: asBuf(b.nodes), edges: asBuf(b.edges), refs: asBuf(b.refs), arena: asBuf(b.arena) },
|
||||||
|
filePath,
|
||||||
|
language
|
||||||
|
);
|
||||||
|
decoded.durationMs = result.durationMs;
|
||||||
|
return decoded;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Extract via the native kernel. Returns null when the kernel doesn't apply
|
* Extract via the native kernel. Returns null when the kernel doesn't apply
|
||||||
* (not routed / not available / kill switch) — the caller falls back to the
|
* (not routed / not available / kill switch) — the caller falls back to the
|
||||||
|
|||||||
@@ -16,6 +16,8 @@ try {
|
|||||||
import { parentPort } from 'worker_threads';
|
import { parentPort } from 'worker_threads';
|
||||||
import { extractFromSource } from './tree-sitter';
|
import { extractFromSource } from './tree-sitter';
|
||||||
import { detectLanguage, loadGrammarsForLanguages, resetParser } from './grammars';
|
import { detectLanguage, loadGrammarsForLanguages, resetParser } from './grammars';
|
||||||
|
import { tryKernelExtractRaw } from './kernel';
|
||||||
|
import { getAllFrameworkResolvers, getApplicableFrameworks } from '../resolution/frameworks';
|
||||||
import type { Language, ExtractionResult } from '../types';
|
import type { Language, ExtractionResult } from '../types';
|
||||||
|
|
||||||
// Emscripten prints `Aborted()` (and a follow-up RuntimeError diag
|
// Emscripten prints `Aborted()` (and a follow-up RuntimeError diag
|
||||||
@@ -80,7 +82,36 @@ parentPort!.on('message', async (msg: { type: string; id?: number; filePath?: st
|
|||||||
// codegraph.json extension overrides) and sends it; fall back to detection
|
// codegraph.json extension overrides) and sends it; fall back to detection
|
||||||
// for older callers / safety.
|
// for older callers / safety.
|
||||||
const language = msg.language ?? detectLanguage(filePath!, content);
|
const language = msg.language ?? detectLanguage(filePath!, content);
|
||||||
const result: ExtractionResult = extractFromSource(filePath!, content!, language, frameworkNames);
|
|
||||||
|
// Kernel deferred-decode fast path: ship the file's tables as flat
|
||||||
|
// buffers and decode at the STORE boundary, so the main thread never
|
||||||
|
// materializes per-node objects (nor pays their structured-clone cost —
|
||||||
|
// buffer clone is a flat memcpy). Only when no applicable framework has
|
||||||
|
// an extract() hook: those merge extra nodes/refs into the DECODED
|
||||||
|
// result inside extractFromSource, so such files keep the decoded path.
|
||||||
|
let result: ExtractionResult | undefined;
|
||||||
|
const frameworksNeedDecode =
|
||||||
|
frameworkNames && frameworkNames.length > 0
|
||||||
|
? getApplicableFrameworks(
|
||||||
|
getAllFrameworkResolvers().filter((r) => frameworkNames.includes(r.name)),
|
||||||
|
language
|
||||||
|
).some((fw) => !!fw.extract)
|
||||||
|
: false;
|
||||||
|
if (!frameworksNeedDecode) {
|
||||||
|
const raw = tryKernelExtractRaw(filePath!, content!, language);
|
||||||
|
if (raw) {
|
||||||
|
result = {
|
||||||
|
nodes: [],
|
||||||
|
edges: [],
|
||||||
|
unresolvedReferences: [],
|
||||||
|
errors: raw.errors,
|
||||||
|
durationMs: 0,
|
||||||
|
kernelBuffers: raw.buffers,
|
||||||
|
kernelCounts: raw.counts,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
}
|
||||||
|
result ??= extractFromSource(filePath!, content!, language, frameworkNames);
|
||||||
|
|
||||||
// Periodic parser reset to reclaim WASM heap memory
|
// Periodic parser reset to reclaim WASM heap memory
|
||||||
const count = (parseCounts.get(language) ?? 0) + 1;
|
const count = (parseCounts.get(language) ?? 0) + 1;
|
||||||
|
|||||||
@@ -30,7 +30,8 @@ try {
|
|||||||
import { parentPort } from 'worker_threads';
|
import { parentPort } from 'worker_threads';
|
||||||
import { QueryBuilder } from '../db/queries';
|
import { QueryBuilder } from '../db/queries';
|
||||||
import { createDatabase, SqliteDatabase } from '../db/sqlite-adapter';
|
import { createDatabase, SqliteDatabase } from '../db/sqlite-adapter';
|
||||||
import type { StoreBundle } from './store-writer';
|
import { finalizeStoreBundle, type KernelStoreBundle, type StoreBundle } from './store-writer';
|
||||||
|
import { decodeExtractBuffers } from './kernel/decode';
|
||||||
|
|
||||||
if (!parentPort) {
|
if (!parentPort) {
|
||||||
throw new Error('store-worker must be run as a worker thread');
|
throw new Error('store-worker must be run as a worker thread');
|
||||||
@@ -42,10 +43,27 @@ let queries: QueryBuilder | null = null;
|
|||||||
|
|
||||||
type InMessage =
|
type InMessage =
|
||||||
| { type: 'open'; dbPath: string; fastInit: boolean }
|
| { type: 'open'; dbPath: string; fastInit: boolean }
|
||||||
| { type: 'bundle'; bundle: StoreBundle }
|
| { type: 'bundle'; bundle: StoreBundle | KernelStoreBundle }
|
||||||
| { type: 'drain'; id: number }
|
| { type: 'drain'; id: number }
|
||||||
| { type: 'close' };
|
| { type: 'close' };
|
||||||
|
|
||||||
|
/** Decode a kernel bundle's buffers into the standard pre-filtered StoreBundle. */
|
||||||
|
function decodeKernelBundle(bundle: KernelStoreBundle): StoreBundle {
|
||||||
|
const asBuf = (u: Uint8Array) => Buffer.from(u.buffer, u.byteOffset, u.byteLength);
|
||||||
|
const decoded = decodeExtractBuffers(
|
||||||
|
{
|
||||||
|
meta: asBuf(bundle.buffers.meta),
|
||||||
|
nodes: asBuf(bundle.buffers.nodes),
|
||||||
|
edges: asBuf(bundle.buffers.edges),
|
||||||
|
refs: asBuf(bundle.buffers.refs),
|
||||||
|
arena: asBuf(bundle.buffers.arena),
|
||||||
|
},
|
||||||
|
bundle.filePath,
|
||||||
|
bundle.language
|
||||||
|
);
|
||||||
|
return finalizeStoreBundle(decoded, bundle.filePath, bundle.language, bundle.file);
|
||||||
|
}
|
||||||
|
|
||||||
port.on('message', (msg: InMessage) => {
|
port.on('message', (msg: InMessage) => {
|
||||||
try {
|
try {
|
||||||
switch (msg.type) {
|
switch (msg.type) {
|
||||||
@@ -70,7 +88,8 @@ port.on('message', (msg: InMessage) => {
|
|||||||
}
|
}
|
||||||
case 'bundle': {
|
case 'bundle': {
|
||||||
if (!queries) throw new Error('store-worker: bundle before open');
|
if (!queries) throw new Error('store-worker: bundle before open');
|
||||||
queries.storeFileBundle(msg.bundle);
|
const bundle = 'kernel' in msg.bundle ? decodeKernelBundle(msg.bundle) : msg.bundle;
|
||||||
|
queries.storeFileBundle(bundle);
|
||||||
port.postMessage({ type: 'ack' });
|
port.postMessage({ type: 'ack' });
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -8,7 +8,7 @@
|
|||||||
*/
|
*/
|
||||||
|
|
||||||
import { Worker } from 'worker_threads';
|
import { Worker } from 'worker_threads';
|
||||||
import { Node, Edge, UnresolvedReference, FileRecord } from '../types';
|
import { ExtractionResult, Language, Node, Edge, UnresolvedReference, FileRecord } from '../types';
|
||||||
|
|
||||||
/** One file's complete store payload (pre-filtered — see storeFileBundle). */
|
/** One file's complete store payload (pre-filtered — see storeFileBundle). */
|
||||||
export interface StoreBundle {
|
export interface StoreBundle {
|
||||||
@@ -18,6 +18,50 @@ export interface StoreBundle {
|
|||||||
file: FileRecord;
|
file: FileRecord;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A kernel deferred-decode payload: the file's raw table buffers plus the
|
||||||
|
* FileRecord the main thread built from meta counts. The store WORKER decodes
|
||||||
|
* and finalizes (same filters as the object path), so per-node objects never
|
||||||
|
* exist on the main thread.
|
||||||
|
*/
|
||||||
|
export interface KernelStoreBundle {
|
||||||
|
kernel: true;
|
||||||
|
filePath: string;
|
||||||
|
language: Language;
|
||||||
|
buffers: NonNullable<ExtractionResult['kernelBuffers']>;
|
||||||
|
file: FileRecord;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The validation/denormalization every bundle gets before storeFileBundle —
|
||||||
|
* shared by the orchestrator's object path and the store worker's kernel
|
||||||
|
* decode path so the two can never drift:
|
||||||
|
* - nodes missing identity fields are dropped (#42-class safety),
|
||||||
|
* - edges must connect inserted nodes (FK integrity),
|
||||||
|
* - refs must originate from inserted nodes and carry the denormalized
|
||||||
|
* filePath/language the resolver reads.
|
||||||
|
*/
|
||||||
|
export function finalizeStoreBundle(
|
||||||
|
result: Pick<ExtractionResult, 'nodes' | 'edges' | 'unresolvedReferences'>,
|
||||||
|
filePath: string,
|
||||||
|
language: Language,
|
||||||
|
file: FileRecord
|
||||||
|
): StoreBundle {
|
||||||
|
const validNodes = result.nodes.filter((n) => n.id && n.kind && n.name && n.filePath && n.language);
|
||||||
|
const insertedIds = new Set(validNodes.map((n) => n.id));
|
||||||
|
const validEdges = result.edges.filter(
|
||||||
|
(e) => insertedIds.has(e.source) && insertedIds.has(e.target)
|
||||||
|
);
|
||||||
|
const validRefs = result.unresolvedReferences
|
||||||
|
.filter((ref) => insertedIds.has(ref.fromNodeId))
|
||||||
|
.map((ref) => ({
|
||||||
|
...ref,
|
||||||
|
filePath: ref.filePath ?? filePath,
|
||||||
|
language: ref.language ?? language,
|
||||||
|
}));
|
||||||
|
return { nodes: validNodes, edges: validEdges, refs: validRefs, file };
|
||||||
|
}
|
||||||
|
|
||||||
export class StoreWriter {
|
export class StoreWriter {
|
||||||
private worker: Worker;
|
private worker: Worker;
|
||||||
private readyPromise: Promise<void>;
|
private readyPromise: Promise<void>;
|
||||||
@@ -102,7 +146,7 @@ export class StoreWriter {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/** Post one file's bundle. Throws immediately if the writer already failed. */
|
/** Post one file's bundle. Throws immediately if the writer already failed. */
|
||||||
send(bundle: StoreBundle): void {
|
send(bundle: StoreBundle | KernelStoreBundle): void {
|
||||||
if (this.firstError) throw this.firstError;
|
if (this.firstError) throw this.firstError;
|
||||||
if (this.exited) throw new Error('store worker already exited');
|
if (this.exited) throw new Error('store worker already exited');
|
||||||
this.outstanding++;
|
this.outstanding++;
|
||||||
|
|||||||
@@ -276,6 +276,23 @@ export interface ExtractionResult {
|
|||||||
|
|
||||||
/** Extraction duration in milliseconds */
|
/** Extraction duration in milliseconds */
|
||||||
durationMs: number;
|
durationMs: number;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Deferred-decode transport (native kernel, bulk-index path): when present,
|
||||||
|
* `nodes`/`edges`/`unresolvedReferences` are EMPTY and the file's tables
|
||||||
|
* ride as flat buffers to be decoded at the store boundary (the store
|
||||||
|
* worker), so the MAIN thread never materializes per-node objects.
|
||||||
|
* `kernelCounts` carries the table sizes for bookkeeping. Decode into a
|
||||||
|
* plain result with `materializeKernelResult` (src/extraction/kernel).
|
||||||
|
*/
|
||||||
|
kernelBuffers?: {
|
||||||
|
meta: Uint8Array;
|
||||||
|
nodes: Uint8Array;
|
||||||
|
edges: Uint8Array;
|
||||||
|
refs: Uint8Array;
|
||||||
|
arena: Uint8Array;
|
||||||
|
};
|
||||||
|
kernelCounts?: { nodes: number; edges: number; refs: number };
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
Reference in New Issue
Block a user