From e5be52a511d9cccabb94ea4119e684d849317ddc Mon Sep 17 00:00:00 2001 From: Brendan McMullen Date: Wed, 19 Aug 2026 10:57:46 -0700 Subject: [PATCH 1/3] Sidecar builds at maintenance: compaction finalizes a file, the index follows (LLP 0265 T6) Compaction is the moment a data file stops changing, so it is the one point where a hypgrep index can be built once and stay valid against its rows. maintainCache now follows every committed rewrite of the grep dataset with a sidecar-build pass over the new generation's files, in a worker thread (createIndex is seconds of straight-line CPU and the daemon is single-threaded), one file at a time. The worker handle and thread are ports of the server's index-worker pair, with one behavioral fix: the worker holds an event-loop ref exactly while a build is in flight, because an always-unref'd worker deadlocks any process whose loop would otherwise drain while awaiting the build. Sidecar existence is the completion marker, no ledger: the publish is a write-then-rename, a killed daemon leaves nothing half-claimed, and the next pass rebuilds whatever is missing. A file whose build keeps failing is quarantined after three attempts (in-memory, process-lifetime; a restart is the retry) and the scan tier serves it forever after: index presence is purely a performance property. The build pass can never fail the partition's own maintenance verdict. Two hazards found and closed on the way: - countDataFiles and measureDataDir counted sidecars (*.parquet in data/), which would have made every just-indexed partition read as "grew since compaction" and rewrite itself every tick through the LLP 0199 baseline gate. Both now exclude .index.parquet; a test pins that a second unforced tick stays converged. - A corrupt sidecar used to fail the whole search; the indexed tier now runs into local buffers and commits only on success, so an unreadable sidecar degrades that one file to the brute scan with no double count. GREP_DATASET joins the shared searchable-columns module so the search service and the build pass cannot disagree about which dataset carries indexes. Tests: per-file build and existence-marker idempotency, the quarantine budget with the scan tier still serving, the corrupt-sidecar fallback, maintenance building indexes for exactly the grep dataset, sidecars not re-triggering compaction, and a retired generation dying whole with its sidecars inside (the no-GC-code guarantee). The compaction-effectiveness tests' liveDataFiles helper learns the same sidecar exclusion the production counters did. Co-Authored-By: Claude Fable 5 --- src/core/cache/maintenance.js | 58 ++++- src/core/cache/types.d.ts | 6 + src/core/search/grep_service.js | 65 ++++-- src/core/search/index_worker.js | 161 ++++++++++++++ src/core/search/index_worker_thread.js | 106 +++++++++ src/core/search/searchable_columns.js | 9 + src/core/search/sidecar_build.js | 142 ++++++++++++ .../cache-compaction-effectiveness.test.js | 4 +- test/core/search-sidecar-build.test.js | 203 ++++++++++++++++++ 9 files changed, 727 insertions(+), 27 deletions(-) create mode 100644 src/core/search/index_worker.js create mode 100644 src/core/search/index_worker_thread.js create mode 100644 src/core/search/sidecar_build.js create mode 100644 test/core/search-sidecar-build.test.js diff --git a/src/core/cache/maintenance.js b/src/core/cache/maintenance.js index b8cfe0d3..4d7cde5c 100644 --- a/src/core/cache/maintenance.js +++ b/src/core/cache/maintenance.js @@ -18,6 +18,8 @@ import { createLocalIcebergIO, tableUrlForDir } from './iceberg/resolver.js' import { columnsFromIcebergSchema } from './iceberg/schema.js' import { appendRowsToTable, currentPartitionSpec, currentSchema, scanRowsFromTable, sortColumnsFromMetadata, tableExists } from './iceberg/store.js' import { openStreamingAppend } from './iceberg/stream_append.js' +import { buildSidecarsForTable } from '../search/sidecar_build.js' +import { GREP_DATASET } from '../search/searchable_columns.js' import { isPlainObject } from '../util/json_util.js' /** @@ -245,6 +247,45 @@ export async function maintainCache(opts) { report.errorMessage = err instanceof Error ? err.message : String(err) totalFailed++ } + // The grep sidecar build, on the files the rewrite just finalized: + // compaction is the moment a file stops changing, so this is the one + // point in a file's life where an index can be built once and stay + // valid (LLP 0264 #lifecycle). Only the grep dataset carries indexes, + // and only a rewrite that committed has new files to index. Isolated + // from the partition's own verdict: an index that cannot be built + // costs speed, never the tick, and never correctness (the scan tier + // serves whatever has no sidecar). + // @ref LLP 0264#lifecycle [implements]: sidecars are built at maintenance right after compaction finalizes the generation's files + if (!opts.dryRun && report.compacted && !report.failed && part.dataset === GREP_DATASET) { + try { + const cursorAfter = readCursorSync(part.path) + const liveDir = path.join(part.path, generationLayout(cursorAfter).liveDir) + await withSpan( + 'maintenance.grep_index', + { + [Attr.COMPONENT]: 'cache', + [Attr.OPERATION]: 'maintenance.grep_index', + [Attr.DATASET]: part.dataset, + status: 'ok', + }, + async (span) => { + const built = await buildSidecarsForTable({ tableDir: liveDir }) + report.sidecarsBuilt = built.built + report.sidecarsFailed = built.failed + built.quarantined + span.setAttribute('sidecars_built', built.built) + span.setAttribute('sidecars_present', built.present) + span.setAttribute('sidecars_failed', built.failed) + span.setAttribute('sidecars_quarantined', built.quarantined) + }, + { component: 'cache' } + ) + } catch (err) { + // Index absence is served by the scan tier, so a build-pass throw + // is a warning on the report, never a failed partition. + report.sidecarsFailed = (report.sidecarsFailed ?? 0) + 1 + report.sidecarError = err instanceof Error ? err.message : String(err) + } + } reports.push(report) if (!report.failed) maintained++ totalSnapshotsExpired += report.snapshotsExpired @@ -1468,8 +1509,13 @@ function liveDataFileCount(partitionDir) { function countDataFiles(tableDir) { const dataDir = path.join(tableDir, 'data') try { + // Grep sidecars live beside their data files as `.index.parquet` + // and MUST stay out of this count: it feeds the compaction heuristics + // and the LLP 0199 baseline gate, so counting sidecars would read a + // just-compacted-and-indexed partition as "grew since compaction" and + // rewrite it every tick forever. return fs.readdirSync(dataDir, { withFileTypes: true }) - .filter((e) => e.isFile() && e.name.endsWith('.parquet')) + .filter((e) => e.isFile() && e.name.endsWith('.parquet') && !e.name.endsWith('.index.parquet')) .length } catch { return 0 @@ -1516,22 +1562,28 @@ function measureMetadataDir(tableDir) { } /** + * Data bytes only: grep sidecars are excluded for the same reason + * `countDataFiles` excludes them; the avg-file-size heuristic divides + * these bytes by that count, so the two must see the same file set. + * * @param {string} tableDir * @returns {number} */ function measureDataDir(tableDir) { - return measureDir(path.join(tableDir, 'data')) + return measureDir(path.join(tableDir, 'data'), (name) => !name.endsWith('.index.parquet')) } /** * @param {string} dir + * @param {(name: string) => boolean} [include] * @returns {number} */ -function measureDir(dir) { +function measureDir(dir, include) { let total = 0 try { for (const entry of fs.readdirSync(dir, { withFileTypes: true })) { if (!entry.isFile()) continue + if (include && !include(entry.name)) continue try { total += fs.statSync(path.join(dir, entry.name)).size } catch { /* skip */ } diff --git a/src/core/cache/types.d.ts b/src/core/cache/types.d.ts index 22842b0b..c90bd66f 100644 --- a/src/core/cache/types.d.ts +++ b/src/core/cache/types.d.ts @@ -295,6 +295,12 @@ export interface MaintenancePartitionReport { dataFilesAfter: number /** Bytes the compaction rewrite actually wrote; absent when it did not run. */ compactedBytesWritten?: number + /** Grep sidecars built for the just-compacted generation; absent when the build did not run. */ + sidecarsBuilt?: number + /** Files whose sidecar build failed or is quarantined; the scan tier serves them. */ + sidecarsFailed?: number + /** The build pass's own error, when the pass itself threw (never fails the partition). */ + sidecarError?: string // Compaction of this partition is known not to reduce its data-file // count under the writer running now: either this run's rewrite // reproduced the count it started from, or a previous one did and the diff --git a/src/core/search/grep_service.js b/src/core/search/grep_service.js index 890a4785..81d2d7ef 100644 --- a/src/core/search/grep_service.js +++ b/src/core/search/grep_service.js @@ -19,7 +19,7 @@ import { resolveCallerClass, } from '../query/visibility.js' import { cellText, compileMatcher, makeSnippet, MAX_MATCH_COLUMNS } from './matcher.js' -import { SCAN_COLUMNS, SEARCHABLE_COLUMNS } from './searchable_columns.js' +import { GREP_DATASET, SCAN_COLUMNS, SEARCHABLE_COLUMNS } from './searchable_columns.js' /** * The local grep-search service: the client half of LLP 0264, mirroring the @@ -58,7 +58,7 @@ import { SCAN_COLUMNS, SEARCHABLE_COLUMNS } from './searchable_columns.js' * @import { UsagePolicyResolver } from '../../../src/core/usage-policy/types.js' */ -const DATASET = 'ai_gateway_messages' +const DATASET = GREP_DATASET /** * Rows between abort checks inside one brute-scanned file. The deadline has @@ -221,29 +221,48 @@ export async function executeGrepSearch(args) { } } if (indexFile) { - indexedFiles += 1 - // No `limit` is passed down: a purged or withheld row is filtered - // AFTER parquetFind accepts it, so a passed-down limit would count - // rows this walk then discards and under-return. The generator is - // simply not pulled past the budget instead. - const rows = parquetFind({ - query: matcher.hypQuery, - url: file.filePath, - indexFile, - asyncBufferFactory: async ({ url }) => await io.reader(url), - rowFilter: accept, - signal, - }) - for await (const row of rows) { - if (file.deletedPositions?.has(BigInt(/** @type {number} */ (row.__index__)))) continue - if (withheld?.(row)) { - localOnly.withheldRows += 1 - continue + // The attempt runs into local buffers and commits only when the + // index tier finished (or the budget stopped it): a sidecar that + // turns out to be unreadable mid-read (torn by an external + // writer; the build's own publish is atomic) must degrade this + // one file to the scan tier below without double-counting the + // rows the broken attempt already saw. + /** @type {GrepSearchHit[]} */ + const found = [] + let withheldHere = 0 + try { + // No `limit` is passed down: a purged or withheld row is + // filtered AFTER parquetFind accepts it, so a passed-down limit + // would count rows this walk then discards and under-return. + // The generator is simply not pulled past the budget instead. + const rows = parquetFind({ + query: matcher.hypQuery, + url: file.filePath, + indexFile, + asyncBufferFactory: async ({ url }) => await io.reader(url), + rowFilter: accept, + signal, + }) + for await (const row of rows) { + if (file.deletedPositions?.has(BigInt(/** @type {number} */ (row.__index__)))) continue + if (withheld?.(row)) { + withheldHere += 1 + continue + } + found.push(toHit(row, matcher)) + if (hits.length + found.length >= budget) break } - hits.push(toHit(row, matcher)) - if (hits.length >= budget) return + indexedFiles += 1 + localOnly.withheldRows += withheldHere + hits.push(...found) + return + } catch (err) { + if (isAbort(err)) throw err + getLogger('query').warn('grep_search.sidecar_unreadable', { + [Attr.COMPONENT]: 'query', + error_message: err instanceof Error ? err.message : String(err), + }) } - return } scannedFiles += 1 const sourceFile = await io.reader(file.filePath) diff --git a/src/core/search/index_worker.js b/src/core/search/index_worker.js new file mode 100644 index 00000000..605e4f74 --- /dev/null +++ b/src/core/search/index_worker.js @@ -0,0 +1,161 @@ +// @ts-check + +import { Worker } from 'node:worker_threads' + +/** + * The main-thread handle on the grep sidecar build worker, ported from the + * server's `index-worker.js` shape by decision. + * + * `createIndex` is seconds of straight-line CPU per sidecar and the daemon + * is single-threaded, so running it inline stops ingest, queries, and the + * health probe for as long as it runs. Yielding between files does not + * help: one file is already longer than any request budget. So the build + * moves to a worker thread and the main loop keeps only the scheduling, + * the file reads, and the publish. + * + * One worker serves one sidecar-build pass and is closed with it: within a + * pass the loop builds one file at a time, so a pool would buy nothing, + * while a worker that outlived the pass would hold an index-sized heap + * between maintenance ticks for no one. + * + * @ref LLP 0264#lifecycle [implements]: the build runs in a worker thread so maintenance never blocks the daemon loop + */ + +/** + * @param {{ + * log?: { info(msg: string, fields?: object): void, warn(msg: string, fields?: object): void }, + * threadUrl?: URL, + * }} [args] threadUrl swaps the worker module so tests can drive + * uncommanded death and protocol breaches; production callers omit it + */ +export function createIndexWorker({ log, threadUrl } = {}) { + /** @type {Worker | null} */ + let worker = null + /** + * The live worker's in-flight builds. Reassigned per spawn: each + * worker's handlers capture their own map, so a crashed worker's late + * error or exit event can only reject its own builds, never a + * replacement worker's. + * @type {Map void, reject: (err: Error) => void }>} + */ + let pending = new Map() + let nextId = 1 + let closed = false + /** Re-evaluate the live worker's ref state; rebound per spawn. */ + let syncRef = () => {} + + function ensureWorker() { + if (worker) return worker + const started = new Worker(threadUrl ?? new URL('./index_worker_thread.js', import.meta.url)) + /** @type {typeof pending} */ + const owned = new Map() + pending = owned + + /** + * Fail this worker's in-flight builds. A worker that dies (OOM, + * terminate, an unparseable module) must surface as a rejected build + * so the caller records the failure and moves on; a silently hung + * promise would wedge the maintenance pass forever. + * + * @param {Error} err + */ + function failAll(err) { + const inflight = [...owned.values()] + owned.clear() + for (const entry of inflight) entry.reject(err) + } + + /** + * Hold the event loop open exactly while a build is in flight. An + * always-unref'd worker deadlocks any process whose loop would + * otherwise drain (the awaiting caller's promise is resolved only by + * a worker message that an empty loop never waits for), while an + * always-ref'd one would hold a shutting-down daemon for the seconds + * a build takes. Ref-while-pending gives both callers what they mean: + * an awaited build completes, an idle worker never keeps the process + * up. + */ + function updateRef() { + if (owned.size > 0) started.ref() + else started.unref() + } + + started.on('message', (/** @type {{ id: number, index?: ArrayBuffer, error?: string }} */ message) => { + const entry = owned.get(message.id) + if (!entry) return + owned.delete(message.id) + updateRef() + // A message with no index bytes and no error is a protocol breach, + // and resolving it would publish an empty sidecar that lists as + // done. The length test is the whole guard: `new ArrayBuffer(0)` is + // truthy, so a presence-only check would let zero bytes through as a + // successful build. + if (message.index && message.index.byteLength > 0) entry.resolve(new Uint8Array(message.index)) + else entry.reject(new Error(message.error ?? 'index worker answered with no index bytes and no error')) + }) + started.on('error', (err) => { + if (worker === started) worker = null + failAll(err instanceof Error ? err : new Error(String(err))) + }) + started.on('exit', (code) => { + if (worker === started) worker = null + failAll(new Error(`grep index worker exited before answering (code ${code})`)) + }) + started.unref() + syncRef = updateRef + worker = started + log?.info('grep_index.worker_started', {}) + return started + } + + return { + /** + * Build one sidecar's bytes off the event loop. A whole, exclusively + * owned source buffer is transferred (detached here on return); + * anything else, such as a view into Node's Buffer pool, is copied + * first. Either way callers must not touch `sourceBytes` afterwards. + * + * @param {Uint8Array} sourceBytes + * @returns {Promise} + */ + build(sourceBytes) { + if (closed) return Promise.reject(new Error('grep index worker is closed')) + const active = ensureWorker() + const id = nextId + nextId += 1 + const source = transferable(sourceBytes) + return new Promise((resolve, reject) => { + pending.set(id, { resolve, reject }) + syncRef() + active.postMessage({ id, source }, [source]) + }) + }, + /** Terminate the worker; in-flight builds reject through the exit hook. */ + async close() { + closed = true + const active = worker + worker = null + if (active) await active.terminate() + }, + } +} + +/** + * An ArrayBuffer that is safe to hand to `postMessage`'s transfer list. + * Node's Buffer allocator hands out views into a shared pool for small + * reads, and transferring a pooled buffer would detach every unrelated + * Buffer sharing it, so anything that is not a whole, exclusively owned + * buffer is copied first. + * + * @param {Uint8Array} bytes + * @returns {ArrayBuffer} + */ +function transferable(bytes) { + const buffer = bytes.buffer + if (buffer instanceof ArrayBuffer && bytes.byteOffset === 0 && bytes.byteLength === buffer.byteLength) { + return buffer + } + const copy = new ArrayBuffer(bytes.byteLength) + new Uint8Array(copy).set(bytes) + return copy +} diff --git a/src/core/search/index_worker_thread.js b/src/core/search/index_worker_thread.js new file mode 100644 index 00000000..79c4315e --- /dev/null +++ b/src/core/search/index_worker_thread.js @@ -0,0 +1,106 @@ +// @ts-check + +import { parentPort } from 'node:worker_threads' + +import { parquetMetadataAsync, parquetSchema } from 'hyparquet' +import { ByteWriter } from 'hyparquet-writer' +import { createIndex } from 'hypgrep' + +import { SEARCHABLE_COLUMNS } from './searchable_columns.js' + +/** + * The worker end of the grep sidecar build, ported from the server's + * `index-worker-thread.js`. One message is one sidecar: source bytes in, + * index bytes out, both moved by transfer, so neither multi-megabyte + * buffer is duplicated to cross the thread boundary. Transfer is not the + * same as never copying: `getBuffer()` slices the writer's backing store, + * so the index is copied once here before it is handed to the transfer + * list, and a source the caller could not transfer outright was copied + * once on the way in. + * + * This thread does no IO. It never opens the cache and never writes a + * sidecar: the caller reads the source and performs the single + * write-then-rename that publishes the result. That is what keeps sidecar + * existence honest as the completion marker (LLP 0264 #lifecycle) across + * a thread that can be killed at any instant. + * + * @import { FileMetaData, SchemaTree } from 'hyparquet' + */ + +if (!parentPort) throw new Error('index_worker_thread must be started as a worker thread') +const port = parentPort + +port.on('message', (/** @type {{ id: number, source: ArrayBuffer }} */ message) => { + void handle(message) +}) + +/** + * Build one index and post it back. Errors travel as a message, not as an + * uncaught rejection: a source file the builder cannot parse must fail + * that one file, not tear down the thread mid-pass. + * + * @param {{ id: number, source: ArrayBuffer }} message + */ +async function handle({ id, source }) { + try { + const bytes = new Uint8Array(source) + // hypgrep reads through an AsyncBuffer; the whole file is already + // resident, so slicing is a copy out of memory. + const sourceFile = { + byteLength: bytes.byteLength, + /** + * @param {number} [start] + * @param {number} [end] + * @returns {ArrayBuffer} + */ + slice(start, end) { + const view = bytes.subarray(start ?? 0, end ?? bytes.byteLength) + const out = new ArrayBuffer(view.byteLength) + new Uint8Array(out).set(view) + return out + }, + } + const indexFile = new ByteWriter() + const metadata = await parquetMetadataAsync(sourceFile) + // @ref LLP 0264#shared [implements]: only the searchable columns are indexed; hypgrep's default would n-gram every string column, and the server measured system_text alone at 90.8% of decoded index text + const textColumns = searchableStringColumns(metadata) + await createIndex({ sourceFile, sourceMetadata: metadata, indexFile, textColumns }) + const index = indexFile.getBuffer() + port.postMessage({ id, index }, [index]) + } catch (err) { + const error = err instanceof Error ? err.message : String(err) + port.postMessage({ id, error }) + } +} + +/** + * The source's string leaf columns narrowed to the searchable set, in + * schema order. hypgrep's root export offers no schema walk, so the + * string-leaf test (UTF8 converted type or STRING logical type on a + * childless node) is restated here against the same hyparquet schema tree + * hypgrep itself reads. A source none of whose string columns are + * searchable fails loudly: building an index that can never match is a + * misconfiguration, not a degenerate success. The caller isolates the + * throw, counts it against this file, and quarantines it after the + * attempt budget; search still serves the file by scanning. + * + * @param {FileMetaData} metadata + * @returns {string[]} + */ +function searchableStringColumns(metadata) { + /** @type {string[]} */ + const stringColumns = [] + /** @param {SchemaTree} node */ + function walk(node) { + const { element, children } = node + const isString = element.converted_type === 'UTF8' || element.logical_type?.type === 'STRING' + if (isString && children.length === 0) stringColumns.push(element.name) + for (const child of children) walk(child) + } + walk(parquetSchema(metadata)) + const textColumns = stringColumns.filter((name) => SEARCHABLE_COLUMNS.has(name)) + if (textColumns.length === 0) { + throw new Error(`source has no searchable string columns (string columns: ${stringColumns.join(', ') || 'none'})`) + } + return textColumns +} diff --git a/src/core/search/searchable_columns.js b/src/core/search/searchable_columns.js index d7739f3f..3d15720c 100644 --- a/src/core/search/searchable_columns.js +++ b/src/core/search/searchable_columns.js @@ -38,6 +38,15 @@ export const SEARCHABLE_COLUMNS = constantSet([ 'git_remote', ]) +/** + * The one dataset grep search covers, on both repositories: the client + * greps its own `ai_gateway_messages` cache, the server the same dataset's + * cache and archive. Named here beside the columns it scopes so the search + * service and the sidecar-build pass cannot disagree about which tables + * carry indexes. + */ +export const GREP_DATASET = 'ai_gateway_messages' + /** * A Set that cannot be added to, deleted from, or cleared. `SCAN_COLUMNS` * below is a load-time snapshot of the allowlist, so a caller mutating the diff --git a/src/core/search/sidecar_build.js b/src/core/search/sidecar_build.js new file mode 100644 index 00000000..ecd6fa4f --- /dev/null +++ b/src/core/search/sidecar_build.js @@ -0,0 +1,142 @@ +// @ts-check + +import fs from 'node:fs' +import fsPromises from 'node:fs/promises' + +import { urlToPath } from '../cache/iceberg/resolver.js' +import { listLiveDataFiles } from '../cache/iceberg/store.js' +import { getLogger } from '../observability/index.js' +import { createIndexWorker } from './index_worker.js' + +/** + * The sidecar build pass: give every live data file of a just-compacted + * table a hypgrep `.index.parquet` beside it, so the grep service's + * indexed tier serves the partition's history instead of brute-scanning + * it. Runs at maintenance, after the compaction that finalized the files + * (LLP 0264 #lifecycle): compaction is the moment a file stops changing, + * so its index can never go stale against its own rows (purge is handled + * at read time by position, not by rebuild). + * + * Sidecar existence IS the completion marker: there is no ledger to + * drift, a killed daemon leaves nothing half-claimed (the publish is a + * write-then-rename, all or nothing), and the next pass simply rebuilds + * whatever is missing. A file that cannot be indexed is quarantined after + * a bounded number of attempts and served by the scan tier forever after: + * index presence is purely a performance property, never a correctness + * one. + */ + +/** + * How many failed builds one file may cost before the pass stops + * attempting it. Three, because the two failure families this counter + * separates are cheap to tell apart: a transient one (a worker killed by + * shutdown) clears well inside three passes, while a deterministic one + * (the file hypgrep cannot index) costs three builds to prove and then + * costs nothing. The ledger is in-memory and process-lifetime on purpose: + * a daemon restart is the retry, and a persisted poison list would + * outlive the bug it recorded. + */ +const MAX_INDEX_ATTEMPTS = 3 + +/** + * @param {{ maxAttempts?: number }} [args] + */ +export function createIndexQuarantine({ maxAttempts = MAX_INDEX_ATTEMPTS } = {}) { + /** @type {Map} */ + const failures = new Map() + return { + /** @param {string} key */ + isQuarantined(key) { + return (failures.get(key) ?? 0) >= maxAttempts + }, + /** @param {string} key */ + recordFailure(key) { + const attempts = (failures.get(key) ?? 0) + 1 + failures.set(key, attempts) + return { attempts, quarantined: attempts >= maxAttempts } + }, + /** @param {string} key */ + clear(key) { + failures.delete(key) + }, + } +} + +/** The ledger every pass in this process shares unless a caller injects its own (tests do). */ +const processQuarantine = createIndexQuarantine() + +/** + * The sidecar path beside a data file: hypgrep's own default, which is a + * contract: any reader with byte access to the cache can search it with + * the stock hypgrep CLI, no daemon involved. The grep service probes + * exactly this path. + * + * @param {string} dataFilePath + * @returns {string} + */ +export function sidecarPathFor(dataFilePath) { + return dataFilePath.replace(/\.parquet$/i, '.index.parquet') +} + +/** + * Build the missing sidecars for one Iceberg table directory, one file at + * a time (the pass's memory bound is one data file plus its index). Files + * whose sidecar already exists are skipped by the existence marker; + * quarantined files are skipped without spending a build. A failure is + * recorded, logged, and isolated to its file: the rest of the pass runs. + * + * @param {{ + * tableDir: string, + * quarantine?: ReturnType, + * worker?: ReturnType, + * log?: { info(msg: string, fields?: object): void, warn(msg: string, fields?: object): void }, + * }} args + * @returns {Promise<{ built: number, present: number, failed: number, quarantined: number }>} + */ +export async function buildSidecarsForTable({ tableDir, quarantine = processQuarantine, worker, log }) { + const logger = log ?? getLogger('cache') + const report = { built: 0, present: 0, failed: 0, quarantined: 0 } + const files = await listLiveDataFiles(tableDir) + if (files.length === 0) return report + const ownWorker = worker ?? createIndexWorker({ log: logger }) + try { + for (const file of files) { + const sourcePath = urlToPath(file.filePath) + const sidecarPath = sidecarPathFor(sourcePath) + if (fs.existsSync(sidecarPath)) { + report.present += 1 + continue + } + if (quarantine.isQuarantined(sourcePath)) { + report.quarantined += 1 + continue + } + try { + const bytes = await fsPromises.readFile(sourcePath) + const index = await ownWorker.build(bytes) + // Publish atomically: rename is the only step that makes the + // sidecar exist, so a crash mid-write can never leave a partial + // file that lists as a finished index. The `.tmp` suffix also + // keeps the in-flight file out of every `*.parquet` count. + const tmpPath = `${sidecarPath}.tmp` + await fsPromises.writeFile(tmpPath, index) + await fsPromises.rename(tmpPath, sidecarPath) + quarantine.clear(sourcePath) + report.built += 1 + } catch (err) { + const { attempts, quarantined } = quarantine.recordFailure(sourcePath) + report.failed += 1 + const message = err instanceof Error ? err.message : String(err) + logger.warn('grep_index.build_failed', { attempts, quarantined, error_message: message }) + if (quarantined) { + logger.warn('grep_index.file_quarantined', { attempts }) + } + } + } + } finally { + // A caller-provided worker outlives the pass (the caller owns its + // lifecycle); one created here is closed with it. + if (!worker) await ownWorker.close() + } + return report +} diff --git a/test/core/cache-compaction-effectiveness.test.js b/test/core/cache-compaction-effectiveness.test.js index 8ac07ea6..ae0b6541 100644 --- a/test/core/cache-compaction-effectiveness.test.js +++ b/test/core/cache-compaction-effectiveness.test.js @@ -118,7 +118,9 @@ async function liveDataFiles(dir) { const dataDir = path.join(dir, cursor.tableDir ?? 'table', 'data') const entries = await fs.readdir(dataDir, { withFileTypes: true }) return entries - .filter((e) => e.isFile() && e.name.endsWith('.parquet')) + // Data files only: compaction now leaves a grep sidecar beside each + // one, and truncating a sidecar would not make the rewrite fail. + .filter((e) => e.isFile() && e.name.endsWith('.parquet') && !e.name.endsWith('.index.parquet')) .map((e) => path.join(dataDir, e.name)) } diff --git a/test/core/search-sidecar-build.test.js b/test/core/search-sidecar-build.test.js new file mode 100644 index 00000000..fc9d6836 --- /dev/null +++ b/test/core/search-sidecar-build.test.js @@ -0,0 +1,203 @@ +// @ts-check + +import test from 'node:test' +import assert from 'node:assert/strict' +import fsSync from 'node:fs' +import fs from 'node:fs/promises' +import os from 'node:os' +import path from 'node:path' + +import { urlToPath } from '../../src/core/cache/iceberg/resolver.js' +import { listLiveDataFiles } from '../../src/core/cache/iceberg/store.js' +import { maintainCache } from '../../src/core/cache/maintenance.js' +import { appendRowsToSourceTable } from '../../src/core/cache/partition.js' +import { createQueryStorageService, resolveIcebergDir } from '../../src/core/cache/storage.js' +import { executeGrepSearch } from '../../src/core/search/grep_service.js' +import { buildSidecarsForTable, createIndexQuarantine, sidecarPathFor } from '../../src/core/search/sidecar_build.js' +import { aiGatewayDatasetRegistration } from '../../hypaware-core/plugins-workspace/ai-gateway/src/dataset.js' + +/** + * @import { ColumnSpec } from '../../hypaware-plugin-kernel-types.js' + */ + +const DATASET = 'ai_gateway_messages' + +/** @type {ColumnSpec[]} */ +const COLUMNS = [ + { name: 'session_id', type: 'STRING', nullable: false }, + { name: 'conversation_id', type: 'STRING', nullable: true }, + { name: 'agent_id', type: 'STRING', nullable: true }, + { name: 'cwd', type: 'STRING', nullable: true }, + { name: 'content_text', type: 'STRING', nullable: true }, + { name: 'date', type: 'STRING', nullable: false }, + { name: 'part_id', type: 'STRING', nullable: false }, + { name: 'message_id', type: 'STRING', nullable: false }, + { name: 'message_created_at', type: 'TIMESTAMP', nullable: false }, + { name: 'client_name', type: 'STRING', nullable: true }, +] + +let rowSeq = 0 + +/** @param {Record} [over] */ +function mkRow(over = {}) { + rowSeq += 1 + const date = typeof over.date === 'string' ? over.date : '2026-08-10' + return { + session_id: 's1', + conversation_id: null, + agent_id: null, + cwd: '/home/open-proj', + content_text: null, + date, + part_id: `m${rowSeq}#0`, + message_id: `m${rowSeq}`, + message_created_at: new Date(`${date}T00:00:00Z`).getTime() + rowSeq * 1000, + client_name: 'test', + ...over, + } +} + +/** + * @param {Record[][]} batches + * @param {string} [dataset] + */ +async function makeCache(batches, dataset = DATASET) { + const cacheRoot = await fs.mkdtemp(path.join(os.tmpdir(), 'hyp-sidecar-')) + const declaration = aiGatewayDatasetRegistration().cachePartitioning + for (const batch of batches) { + await appendRowsToSourceTable(cacheRoot, dataset, ['source=test'], COLUMNS, batch, { declaration }) + } + const storage = createQueryStorageService({ cacheRoot }) + const partitionDir = path.join(cacheRoot, 'datasets', dataset, 'source=test') + return { cacheRoot, storage, partitionDir, tableDir: () => resolveIcebergDir(partitionDir) } +} + +/** A worker stand-in whose every build fails. */ +function failingWorker() { + return { + build: () => Promise.reject(new Error('synthetic build failure')), + close: async () => {}, + } +} + +const quietLog = { info() {}, warn() {} } + +const OLD = mkRow({ date: '2026-08-10', session_id: 's1', content_text: 'alpha needle one' }) +const NEW = mkRow({ date: '2026-08-12', session_id: 's2', content_text: 'the needle two' }) + +test('buildSidecarsForTable builds one sidecar per file, idempotently, and grep serves them', async () => { + const { storage, tableDir } = await makeCache([[OLD], [NEW]]) + const first = await buildSidecarsForTable({ tableDir: tableDir(), log: quietLog }) + assert.equal(first.built >= 2, true) + assert.equal(first.present, 0) + assert.equal(first.failed, 0) + for (const file of await listLiveDataFiles(tableDir())) { + assert.ok(fsSync.existsSync(sidecarPathFor(urlToPath(file.filePath))), 'every live file has a sidecar') + } + const second = await buildSidecarsForTable({ tableDir: tableDir(), log: quietLog }) + assert.equal(second.built, 0) + assert.equal(second.present, first.built, 'existence is the completion marker; nothing rebuilds') + + const res = await executeGrepSearch({ storage, query: 'needle', limit: 10, includeLocalOnly: true }) + assert.equal(res.hits.length, 2) + assert.equal(res.indexedFiles, first.built, 'the search runs on the indexed tier') + assert.equal(res.scannedFiles, 0) + assert.deepEqual(res.hits.map((h) => h.sessionId), ['s2', 's1']) +}) + +test('a failing build quarantines after three attempts and the scan tier still serves the file', async () => { + const { storage, tableDir } = await makeCache([[OLD]]) + const quarantine = createIndexQuarantine() + for (let attempt = 1; attempt <= 3; attempt++) { + const report = await buildSidecarsForTable({ + tableDir: tableDir(), quarantine, worker: failingWorker(), log: quietLog, + }) + assert.equal(report.failed, 1, `attempt ${attempt} spends a build and fails`) + assert.equal(report.built, 0) + } + const afterQuarantine = await buildSidecarsForTable({ + tableDir: tableDir(), quarantine, worker: failingWorker(), log: quietLog, + }) + assert.equal(afterQuarantine.failed, 0, 'a quarantined file costs no further builds') + assert.equal(afterQuarantine.quarantined, 1) + + const res = await executeGrepSearch({ storage, query: 'needle', limit: 10, includeLocalOnly: true }) + assert.equal(res.hits.length, 1, 'the unindexed file is served by the scan tier') + assert.equal(res.scannedFiles, 1) + assert.equal(res.indexedFiles, 0) +}) + +test('a corrupt sidecar degrades that one file to the scan tier instead of failing the search', async () => { + const { storage, tableDir } = await makeCache([[OLD]]) + const [file] = await listLiveDataFiles(tableDir()) + await fs.writeFile(sidecarPathFor(urlToPath(file.filePath)), 'not a parquet file') + const res = await executeGrepSearch({ storage, query: 'needle', limit: 10, includeLocalOnly: true }) + assert.equal(res.hits.length, 1) + assert.equal(res.indexedFiles, 0) + assert.equal(res.scannedFiles, 1, 'the unreadable sidecar fell back to the brute scan') +}) + +test('maintenance compaction finalizes files and builds their sidecars', async () => { + const { cacheRoot, storage, tableDir } = await makeCache([[OLD], [NEW]]) + const result = await maintainCache({ cacheRoot, force: true }) + const report = result.partitions.find((p) => p.dataset === DATASET) + assert.ok(report) + assert.equal(report.compacted, true) + assert.ok((report.sidecarsBuilt ?? 0) >= 1, 'the rewrite queued index builds for its files') + assert.equal(report.sidecarsFailed ?? 0, 0) + + const files = await listLiveDataFiles(tableDir()) + assert.ok(files.length >= 1) + for (const file of files) { + assert.ok(fsSync.existsSync(sidecarPathFor(urlToPath(file.filePath))), 'every finalized file is indexed') + } + const res = await executeGrepSearch({ storage, query: 'needle', limit: 10, includeLocalOnly: true }) + assert.equal(res.hits.length, 2) + assert.equal(res.scannedFiles, 0) +}) + +test('sidecars do not re-trigger compaction: the data-file counters exclude them', async () => { + const { cacheRoot } = await makeCache([[OLD], [NEW]]) + await maintainCache({ cacheRoot, force: true }) + // No new data flushed since the rewrite; a second unforced tick must see + // a converged partition, not one that "grew" by its own index files. + const second = await maintainCache({ cacheRoot }) + const report = second.partitions.find((p) => p.dataset === DATASET) + assert.ok(report) + assert.equal(report.compacted, false, 'the sidecars did not read as growth') +}) + +test('a non-grep dataset is compacted without sidecars', async () => { + const { cacheRoot, tableDir } = await makeCache([[mkRow({ content_text: 'needle' })]], 'other_dataset') + const result = await maintainCache({ cacheRoot, force: true }) + const report = result.partitions.find((p) => p.dataset === 'other_dataset') + assert.ok(report) + assert.equal(report.compacted, true) + assert.equal(report.sidecarsBuilt, undefined, 'the build pass never ran') + for (const file of await listLiveDataFiles(tableDir())) { + assert.equal(fsSync.existsSync(sidecarPathFor(urlToPath(file.filePath))), false) + } +}) + +test('a retired generation dies whole, sidecars included', async () => { + const { cacheRoot, partitionDir } = await makeCache([[OLD], [NEW]]) + await maintainCache({ cacheRoot, force: true }) + const compactedDir = resolveIcebergDir(partitionDir) + assert.ok((await listLiveDataFiles(compactedDir)).length >= 1) + + // New data, then a second rewrite: the first compacted generation (with + // its sidecars inside) is retired. + const declaration = aiGatewayDatasetRegistration().cachePartitioning + await appendRowsToSourceTable(cacheRoot, DATASET, ['source=test'], COLUMNS, + [mkRow({ date: '2026-08-14', session_id: 's3', content_text: 'needle three' })], { declaration }) + await maintainCache({ cacheRoot, force: true }) + assert.notEqual(resolveIcebergDir(partitionDir), compactedDir, 'a fresh generation is live') + assert.ok(fsSync.existsSync(path.join(compactedDir, '.retired')), 'the old generation is marked retired') + + // Backdate the marker past the grace period; the next tick's sweep + // reclaims the directory, and the sidecars go with it because they live + // inside it: the no-GC-code guarantee this test exists to pin. + await fs.writeFile(path.join(compactedDir, '.retired'), new Date(0).toISOString()) + await maintainCache({ cacheRoot }) + assert.equal(fsSync.existsSync(compactedDir), false, 'the retired generation and its sidecars are gone') +}) From 07cd46617e2c8e8bae4a7ba296fa71d887c8bb36 Mon Sep 17 00:00:00 2001 From: test Date: Thu, 20 Aug 2026 15:18:06 +0000 Subject: [PATCH 2/3] Review fixes on the sidecar build: one sidecar-path contract, a collision-safe publish, and failures that name their file - Hoist `sidecarPathFor` beside `GREP_DATASET` in searchable_columns.js. The build pass and the search service each carried their own copy of the `.index.parquet` rule; two copies of a path contract drift into a build that writes an index nobody probes for. - Give the publish scratch file a random token. A fixed `.tmp` is only atomic for a single writer: the daemon tick and a hand-run `hyp` over the same cache would interleave into one scratch file and rename the mixture into place as a finished sidecar. The scratch file is now also removed on the failure path. - Name the data file on `grep_index.build_failed` / `grep_index.file_quarantined` / `grep_search.sidecar_unreadable`, and add the component/operation attributes, so three warnings can be told apart as one poisoned file or three. - Append rather than spread the indexed tier's buffered hits: `limit` reaches the service unvalidated and one file can fill the budget. - Correct the module docs: the pass runs only behind a committed compaction, which always publishes a fresh generation, so it never re-attempts a file it skipped or failed on, and a daemon restart is not a retry. Add the LLP 0264#lifecycle ref the module realizes. Co-Authored-By: Claude Opus 5 (1M context) --- src/core/search/grep_service.js | 14 ++++- src/core/search/searchable_columns.js | 17 ++++++ src/core/search/sidecar_build.js | 79 +++++++++++++++++--------- test/core/search-sidecar-build.test.js | 3 +- 4 files changed, 81 insertions(+), 32 deletions(-) diff --git a/src/core/search/grep_service.js b/src/core/search/grep_service.js index 81d2d7ef..52464ace 100644 --- a/src/core/search/grep_service.js +++ b/src/core/search/grep_service.js @@ -19,7 +19,7 @@ import { resolveCallerClass, } from '../query/visibility.js' import { cellText, compileMatcher, makeSnippet, MAX_MATCH_COLUMNS } from './matcher.js' -import { GREP_DATASET, SCAN_COLUMNS, SEARCHABLE_COLUMNS } from './searchable_columns.js' +import { GREP_DATASET, SCAN_COLUMNS, SEARCHABLE_COLUMNS, sidecarPathFor } from './searchable_columns.js' /** * The local grep-search service: the client half of LLP 0264, mirroring the @@ -210,7 +210,7 @@ export async function executeGrepSearch(args) { // #lifecycle): probe the filesystem, then degrade this one file to // the scan tier if the read races a delete. Results stay exact // either way; only the wall clock changes. - const sidecarUrl = file.filePath.replace(/\.parquet$/i, '.index.parquet') + const sidecarUrl = sidecarPathFor(file.filePath) /** @type {Awaited> | null} */ let indexFile = null if (fs.existsSync(urlToPath(sidecarUrl))) { @@ -254,12 +254,20 @@ export async function executeGrepSearch(args) { } indexedFiles += 1 localOnly.withheldRows += withheldHere - hits.push(...found) + // Appended, not spread: `limit` reaches this service unvalidated + // and one file may fill the whole budget, and a spread of that + // many arguments is an argument-count overflow, not a push. + for (const hit of found) hits.push(hit) return } catch (err) { if (isAbort(err)) throw err getLogger('query').warn('grep_search.sidecar_unreadable', { [Attr.COMPONENT]: 'query', + [Attr.OPERATION]: 'query.grep_search', + // Named, because this warning is the only notice that a + // sidecar needs deleting: nothing rebuilds one in place, so + // the file it points at is the actionable part of the line. + sidecar_file: urlToPath(sidecarUrl), error_message: err instanceof Error ? err.message : String(err), }) } diff --git a/src/core/search/searchable_columns.js b/src/core/search/searchable_columns.js index 3d15720c..07b6257d 100644 --- a/src/core/search/searchable_columns.js +++ b/src/core/search/searchable_columns.js @@ -47,6 +47,23 @@ export const SEARCHABLE_COLUMNS = constantSet([ */ export const GREP_DATASET = 'ai_gateway_messages' +/** + * The sidecar path beside a data file: hypgrep's own default, which is a + * contract. Any reader with byte access to the cache can search a file + * with the stock hypgrep CLI, no daemon involved. It lives beside the + * allowlist for the same reason `GREP_DATASET` does: the build pass that + * publishes a sidecar and the search service that probes for one must + * spell this path identically, or the build writes an index nobody looks + * for and every file silently falls back to the scan tier. Takes a + * filesystem path or a `file://` URL; only the extension is rewritten. + * + * @param {string} dataFile + * @returns {string} + */ +export function sidecarPathFor(dataFile) { + return dataFile.replace(/\.parquet$/i, '.index.parquet') +} + /** * A Set that cannot be added to, deleted from, or cleared. `SCAN_COLUMNS` * below is a load-time snapshot of the allowlist, so a caller mutating the diff --git a/src/core/search/sidecar_build.js b/src/core/search/sidecar_build.js index ecd6fa4f..fa5604ed 100644 --- a/src/core/search/sidecar_build.js +++ b/src/core/search/sidecar_build.js @@ -1,12 +1,14 @@ // @ts-check +import { randomUUID } from 'node:crypto' import fs from 'node:fs' import fsPromises from 'node:fs/promises' import { urlToPath } from '../cache/iceberg/resolver.js' import { listLiveDataFiles } from '../cache/iceberg/store.js' -import { getLogger } from '../observability/index.js' +import { Attr, getLogger } from '../observability/index.js' import { createIndexWorker } from './index_worker.js' +import { sidecarPathFor } from './searchable_columns.js' /** * The sidecar build pass: give every live data file of a just-compacted @@ -19,11 +21,20 @@ import { createIndexWorker } from './index_worker.js' * * Sidecar existence IS the completion marker: there is no ledger to * drift, a killed daemon leaves nothing half-claimed (the publish is a - * write-then-rename, all or nothing), and the next pass simply rebuilds - * whatever is missing. A file that cannot be indexed is quarantined after - * a bounded number of attempts and served by the scan tier forever after: - * index presence is purely a performance property, never a correctness - * one. + * write-then-rename, all or nothing), and a pass over a generation + * rebuilds whatever that generation is missing. A file that cannot be + * indexed is quarantined after a bounded number of attempts and served by + * the scan tier: index presence is purely a performance property, never a + * correctness one. + * + * What that does NOT buy, and callers must not assume it does: a retry. + * The pass runs only behind a committed compaction, and a compaction + * always publishes a fresh generation directory, so the files this pass + * skipped or failed on are gone by the time another pass runs. A missing + * sidecar is repaired by the next compaction rewriting the rows into a + * new file, not by re-attempting the old one. + * + * @ref LLP 0264#lifecycle [implements]: sidecar existence is the idempotency marker, no ledger; an unindexed or poisoned file is brute-scanned, so index state is never a correctness input */ /** @@ -33,8 +44,10 @@ import { createIndexWorker } from './index_worker.js' * shutdown) clears well inside three passes, while a deterministic one * (the file hypgrep cannot index) costs three builds to prove and then * costs nothing. The ledger is in-memory and process-lifetime on purpose: - * a daemon restart is the retry, and a persisted poison list would - * outlive the bug it recorded. + * a persisted poison list would outlive the bug it recorded. Note it only + * bites where one path is offered to more than one pass, which under the + * compaction gate above means a caller driving this module directly; the + * maintenance pass sees fresh paths every time. */ const MAX_INDEX_ATTEMPTS = 3 @@ -65,19 +78,6 @@ export function createIndexQuarantine({ maxAttempts = MAX_INDEX_ATTEMPTS } = {}) /** The ledger every pass in this process shares unless a caller injects its own (tests do). */ const processQuarantine = createIndexQuarantine() -/** - * The sidecar path beside a data file: hypgrep's own default, which is a - * contract: any reader with byte access to the cache can search it with - * the stock hypgrep CLI, no daemon involved. The grep service probes - * exactly this path. - * - * @param {string} dataFilePath - * @returns {string} - */ -export function sidecarPathFor(dataFilePath) { - return dataFilePath.replace(/\.parquet$/i, '.index.parquet') -} - /** * Build the missing sidecars for one Iceberg table directory, one file at * a time (the pass's memory bound is one data file plus its index). Files @@ -111,25 +111,48 @@ export async function buildSidecarsForTable({ tableDir, quarantine = processQuar report.quarantined += 1 continue } + // Publish atomically: rename is the only step that makes the + // sidecar exist, so a crash mid-write can never leave a partial file + // that lists as a finished index. The scratch name carries a random + // token because a fixed `.tmp` is only safe for one writer: + // the daemon's tick and a hand-run `hyp` sharing a cache would + // interleave their writes into the same scratch file and then rename + // the mixture into place as a finished sidecar. It also ends outside + // `.parquet`, so an in-flight or abandoned file joins no data-file + // count, and it is removed on the failure path rather than left to + // wait for the generation's retirement. + const tmpPath = `${sidecarPath}.${randomUUID()}.tmp` try { const bytes = await fsPromises.readFile(sourcePath) const index = await ownWorker.build(bytes) - // Publish atomically: rename is the only step that makes the - // sidecar exist, so a crash mid-write can never leave a partial - // file that lists as a finished index. The `.tmp` suffix also - // keeps the in-flight file out of every `*.parquet` count. - const tmpPath = `${sidecarPath}.tmp` await fsPromises.writeFile(tmpPath, index) await fsPromises.rename(tmpPath, sidecarPath) quarantine.clear(sourcePath) report.built += 1 } catch (err) { + await fsPromises.rm(tmpPath, { force: true }).catch(() => {}) const { attempts, quarantined } = quarantine.recordFailure(sourcePath) report.failed += 1 const message = err instanceof Error ? err.message : String(err) - logger.warn('grep_index.build_failed', { attempts, quarantined, error_message: message }) + // The data file is named on every line: with one warning per failed + // build and a per-file attempt budget, an operator who cannot tell + // three retries of one poisoned file from three distinct failures + // cannot act on either. + logger.warn('grep_index.build_failed', { + [Attr.COMPONENT]: 'cache', + [Attr.OPERATION]: 'maintenance.grep_index', + data_file: sourcePath, + attempts, + quarantined, + error_message: message, + }) if (quarantined) { - logger.warn('grep_index.file_quarantined', { attempts }) + logger.warn('grep_index.file_quarantined', { + [Attr.COMPONENT]: 'cache', + [Attr.OPERATION]: 'maintenance.grep_index', + data_file: sourcePath, + attempts, + }) } } } diff --git a/test/core/search-sidecar-build.test.js b/test/core/search-sidecar-build.test.js index fc9d6836..7cd57f99 100644 --- a/test/core/search-sidecar-build.test.js +++ b/test/core/search-sidecar-build.test.js @@ -13,7 +13,8 @@ import { maintainCache } from '../../src/core/cache/maintenance.js' import { appendRowsToSourceTable } from '../../src/core/cache/partition.js' import { createQueryStorageService, resolveIcebergDir } from '../../src/core/cache/storage.js' import { executeGrepSearch } from '../../src/core/search/grep_service.js' -import { buildSidecarsForTable, createIndexQuarantine, sidecarPathFor } from '../../src/core/search/sidecar_build.js' +import { sidecarPathFor } from '../../src/core/search/searchable_columns.js' +import { buildSidecarsForTable, createIndexQuarantine } from '../../src/core/search/sidecar_build.js' import { aiGatewayDatasetRegistration } from '../../hypaware-core/plugins-workspace/ai-gateway/src/dataset.js' /** From bf79490e527ab0dc9ddeb1adb12dd2d092af8e2d Mon Sep 17 00:00:00 2001 From: philcunliffe Date: Fri, 21 Aug 2026 01:13:25 +0000 Subject: [PATCH 3/3] Review round 2: an aborted indexed read keeps its rows, and the build scratch is not data bytes The merge resolution's per-file buffer degraded one file instead of failing the query, but it also made a deadline throw away everything the index had already produced for the file it landed in. hypgrep checks the signal at every coalesced range boundary, so a deadline lands inside a file, and on a newest-first walk that is the newest file the caller most wants. Committed before the abort propagates: safe because an abort ends the walk, so the file is never rescanned and no row can be counted twice. grep_search.sidecar_unreadable named only the sidecar, but parquetFind opens the source data file through the same factory and runs the row filter per row, so a torn source parquet lands in that catch too and points the operator at a healthy index. Renamed to grep_search.indexed_read_failed and both files are named. measureDataDir excluded `*.index.parquet` but not the build's publish scratch, `.index.parquet..tmp`, which survives a kill between write and rename with no reaper until the generation retires. countDataFiles already skips it, so counting its bytes broke the shared-file-set invariant in the dangerous direction: needsCompaction compacts on a LOW average, so a large orphan makes a fragmented partition read as healthy. Test pins it. Also recorded, not fixed: a sidecar freezes the allowlist it was built over (hypgrep stores hypgrep.text_columns in the index and prunes to it, and nothing compares that stamp to today's SEARCHABLE_COLUMNS), so #977 has to invalidate existing sidecars rather than only build new ones. Co-Authored-By: Claude Opus 5 (1M context) --- src/core/cache/maintenance.js | 12 +++++++- src/core/search/grep_service.js | 40 ++++++++++++++++++++------ src/core/search/sidecar_build.js | 13 +++++++++ test/core/search-sidecar-build.test.js | 28 ++++++++++++++++++ 4 files changed, 84 insertions(+), 9 deletions(-) diff --git a/src/core/cache/maintenance.js b/src/core/cache/maintenance.js index 4d7cde5c..351255cb 100644 --- a/src/core/cache/maintenance.js +++ b/src/core/cache/maintenance.js @@ -1566,11 +1566,21 @@ function measureMetadataDir(tableDir) { * `countDataFiles` excludes them; the avg-file-size heuristic divides * these bytes by that count, so the two must see the same file set. * + * The test is `includes`, not `endsWith`, because the build's publish + * scratch (`.index.parquet..tmp`) is index bytes too, and it + * is the half of the pair that survives a crash: `countDataFiles` already + * skips it for want of a `.parquet` suffix, so counting its bytes would + * break the shared-file-set invariant above in the dangerous direction. + * The average would read HIGHER than the partition really is, and + * `needsCompaction` compacts when the average is LOW, so a genuinely + * fragmented partition would look healthy and go unrewritten until its + * generation retires. + * * @param {string} tableDir * @returns {number} */ function measureDataDir(tableDir) { - return measureDir(path.join(tableDir, 'data'), (name) => !name.endsWith('.index.parquet')) + return measureDir(path.join(tableDir, 'data'), (name) => !name.includes('.index.parquet')) } /** diff --git a/src/core/search/grep_service.js b/src/core/search/grep_service.js index 45944a74..3d5b2ce6 100644 --- a/src/core/search/grep_service.js +++ b/src/core/search/grep_service.js @@ -275,6 +275,11 @@ export async function executeGrepSearch(args) { * degrading the file a decision this function can still take at any * point in the read. * + * An abort is the one failure that commits the buffer instead of + * discarding it: it ends the walk rather than degrading the file, so + * there is no rescan to double-count against and the rows the index + * already produced belong in the partial answer. + * * @param {{ filePath: string, deletedPositions: Set | undefined }} file * @param {Awaited>} indexFile * @param {string} sidecarUrl @@ -307,23 +312,42 @@ export async function executeGrepSearch(args) { if (found.length >= budget * 2) trimBuffer(found) } } catch (err) { - if (isAbort(err, signal)) throw err - getLogger('query').warn('grep_search.sidecar_unreadable', { + if (isAbort(err, signal)) { + // Commit before the abort propagates. A deadline lands INSIDE a + // file, not between files (hypgrep checks the signal at every + // coalesced range boundary), and a newest-first walk makes the + // interrupted file the newest one the caller most wants, so + // discarding the buffer would answer zero for exactly that file + // and lose its withheld-row count out of the report. Safe + // precisely because an abort ends the walk: this file is never + // rescanned, so no row can be counted twice. It still does not + // count as indexed, because it was not served whole. + for (const hit of found) hits.push(hit) + localOnly.withheldRows += withheldHere + throw err + } + // Both files are named, because the read that failed spans both: + // `parquetFind` opens the source data file through the same + // factory as the sidecar and runs the row filter per row, so a + // torn source parquet reaches this line too and then fails the + // rescan below. Deleting the sidecar is the usual remedy and this + // warning is its only notice (nothing rebuilds one in place), but + // the line must not claim to have proved which file is at fault. + getLogger('query').warn('grep_search.indexed_read_failed', { [Attr.COMPONENT]: 'query', [Attr.OPERATION]: 'query.grep_search', - // Named, because this warning is the only notice that a sidecar - // needs deleting: nothing rebuilds one in place, so the file it - // points at is the actionable part of the line. sidecar_file: urlToPath(sidecarUrl), + data_file: urlToPath(file.filePath), error_message: err instanceof Error ? err.message : String(err), }) return false } indexedFiles += 1 localOnly.withheldRows += withheldHere - // Appended, not spread: `limit` reaches this service unvalidated and - // one file may fill the whole buffer, and a spread of that many - // arguments is an argument-count overflow, not a push. + // Appended, not spread: `limit` is validated as a positive safe + // integer but is not bounded above, so one file may fill a buffer of + // millions, and a spread of that many arguments is an argument-count + // overflow, not a push. for (const hit of found) hits.push(hit) if (hits.length >= budget * 2) trimHits() return true diff --git a/src/core/search/sidecar_build.js b/src/core/search/sidecar_build.js index fa5604ed..7ea002fe 100644 --- a/src/core/search/sidecar_build.js +++ b/src/core/search/sidecar_build.js @@ -34,6 +34,19 @@ import { sidecarPathFor } from './searchable_columns.js' * sidecar is repaired by the next compaction rewriting the rows into a * new file, not by re-attempting the old one. * + * A sidecar also freezes the allowlist it was built over. hypgrep records + * the indexed columns in the index itself (`hypgrep.text_columns`) and + * prunes candidate blocks to them, and nothing on the read side compares + * that stamp against today's `SEARCHABLE_COLUMNS`. So ADDING a column to + * the allowlist does not reach a file that is already indexed, and under + * the no-retry lifecycle above it never will while that generation lives: + * the new column answers on the scan tier and zero on the indexed one, + * which is the tier disagreement the shared allowlist exists to prevent. + * Any change to the set (hyparam/hypaware#977 is the first one queued) has + * to invalidate the existing sidecars, not merely start building new ones. + * Recorded here rather than left to be rediscovered, in the same spirit as + * the column note in `searchable_columns.js`. + * * @ref LLP 0264#lifecycle [implements]: sidecar existence is the idempotency marker, no ledger; an unindexed or poisoned file is brute-scanned, so index state is never a correctness input */ diff --git a/test/core/search-sidecar-build.test.js b/test/core/search-sidecar-build.test.js index 7cd57f99..3d9047c4 100644 --- a/test/core/search-sidecar-build.test.js +++ b/test/core/search-sidecar-build.test.js @@ -168,6 +168,34 @@ test('sidecars do not re-trigger compaction: the data-file counters exclude them assert.equal(report.compacted, false, 'the sidecars did not read as growth') }) +test('an orphaned publish scratch counts as index bytes, not data bytes', async () => { + // A build killed between the write and the rename leaves + // `.index.parquet..tmp` in the live data dir, and nothing + // reaps it before the generation retires. `countDataFiles` already skips + // it (no `.parquet` suffix), so the byte measure has to skip it too: the + // avg-file-size heuristic compacts when the average is LOW, so counting a + // large orphan makes a fragmented partition read as healthy and go + // unrewritten. + const { cacheRoot, tableDir } = await makeCache([[OLD], [NEW]]) + const files = await listLiveDataFiles(tableDir()) + assert.ok(files.length >= 2) + let dataBytes = 0 + for (const file of files) dataBytes += (await fs.stat(urlToPath(file.filePath))).size + const avgBytes = dataBytes / files.length + const orphan = `${sidecarPathFor(urlToPath(files[0].filePath))}.orphaned-build.tmp` + await fs.writeFile(orphan, Buffer.alloc(dataBytes * 4)) + + // Due by a hair on the real data bytes; not due at all if the orphan's + // bytes join the average. + const result = await maintainCache({ + cacheRoot, + config: { compact_file_count: 1000, compact_avg_file_bytes: Math.ceil(avgBytes) + 1 }, + }) + const report = result.partitions.find((p) => p.dataset === DATASET) + assert.ok(report) + assert.equal(report.compacted, true, 'the orphaned scratch did not inflate the average file size') +}) + test('a non-grep dataset is compacted without sidecars', async () => { const { cacheRoot, tableDir } = await makeCache([[mkRow({ content_text: 'needle' })]], 'other_dataset') const result = await maintainCache({ cacheRoot, force: true })