diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a26d39a..ef2a59b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -72,3 +72,20 @@ jobs: if grep -qi rocksdb "$RUNNER_TEMP/fulltext-linkage.txt"; then exit 1; fi - name: Verify unwind profile run: cargo rustc --locked --release --features node-api -- --print cfg | grep 'panic="unwind"' + + benchmark-smoke: + name: Native benchmark smoke + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: '24' + cache: npm + - uses: dtolnay/rust-toolchain@1.90.0 + - uses: Swatinem/rust-cache@v2 + - run: npm ci --ignore-scripts + - run: npm run build:native + - run: npm run build:typescript + - run: npm run benchmark:smoke diff --git a/README.md b/README.md index 7b2f2c3..3d2a0a6 100644 --- a/README.md +++ b/README.md @@ -3,9 +3,8 @@ Native Tantivy full-text indexing for Node.js, with a native filesystem backend and a planned caller-owned rocksdb-js backend for Harper. -This repository is under active development. The initial scaffold exposes runtime capability -information through the native entry point; indexing and search APIs are tracked separately and -are not yet available. +This repository is under active development. The native entry point provides a standalone Tantivy +index backed by `MmapDirectory`. Harper releases will use only the planned RocksDB entry point. ## Requirements @@ -21,15 +20,50 @@ targets are added only after their artifacts are loaded and tested on the target ## Native usage ```js -import { runtimeInfo } from '@harperfast/fulltext/native'; - -const info = await runtimeInfo(); -console.log(info.tantivyVersion); +import { encodeMutationBatch, openNativeFullTextIndex } from '@harperfast/fulltext/native'; + +const index = await openNativeFullTextIndex({ + path: './search/products', + indexId: 'products', + generation: 'v1', + fields: [{ name: 'title', weight: 3 }, { name: 'description' }], + analyzer: 'english@1', + limits: { + indexingThreads: 2, + searchThreads: 4, + writerMemoryBytes: 60_000_000, + maxQueuedCommands: 128, + maxQueuedBytes: 64 * 1024 * 1024, + maxBatchBytes: 8 * 1024 * 1024, + }, +}); + +await index.apply( + encodeMutationBatch({ + upserts: [{ id: 'shoe-1', fields: { title: 'Trail running shoe', description: 'Waterproof' } }], + }), +); +await index.commit(); +await index.reload(); +console.log(await index.search({ text: 'waterproof running shoes', limit: 10 })); +await index.close(); ``` -`runtimeInfo()` is asynchronous so later search and indexing operations can remain off the Node.js -event loop without changing the public calling convention. Its first call may synchronously load -the native artifact; search, indexing, commit, and storage work will use the package executor. +Mutation batches are versioned packed values, so indexing crosses Node-API once per batch rather +than once per document. One dedicated actor owns Tantivy's single writer for each index. A bounded +search pool shares immutable searchers and can execute reads while indexing or commit work is in +progress. Queue limits reject overload with `E_QUEUE_FULL` rather than blocking the JavaScript +thread. + +Search uses BM25. `total` is a bounded result by default so Tantivy can retain block-max WAND +pruning. Set `exactTotal: true` only when an exact match count is worth a second full-match +traversal. `positions` defaults on for future phrase queries; `surfaceTerms` defaults off because it +stores source text for future highlighting and suggestions. Both settings are persisted and must +match when the index is reopened. + +`close()` rejects uncommitted data by default. Use `close({ mode: 'rollback' })` to discard it +explicitly. `commit()` publishes mutations, and `reload()` makes the latest commit visible to this +handle's searches. ## Storage boundaries @@ -51,8 +85,17 @@ npm run build:debug npm test npm run lint npm run format:check +npm run benchmark:native -- --documents 100000 --concurrency 4 --commit-every 25000 ``` +The benchmark generates a deterministic, high-cardinality product catalog and emits one versioned +JSON record. It reports packing, apply, durable end-to-end ingestion, actor queue and execution +time, commit distributions, reload cost, warm and cold BM25 p50/p95/p99, exact-total overhead, +index bytes, and periodically sampled process RSS. `--commit-every` sets the target number of mutations between +durability points; it materially affects throughput and peak memory because replacement-safe +upserts include delete terms. CI runs only the correctness smoke profile; timing comparisons +require controlled hardware. + Generated Node-API declarations in `ts/addon.d.ts` are private implementation types. Consumers use only the types exported from a package entry point. diff --git a/benchmarks/native.mjs b/benchmarks/native.mjs new file mode 100644 index 0000000..d83f60e --- /dev/null +++ b/benchmarks/native.mjs @@ -0,0 +1,224 @@ +import assert from 'node:assert'; +import { mkdtemp, readdir, rm, stat } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { performance } from 'node:perf_hooks'; + +import { encodeMutationBatch, openNativeFullTextIndex, runtimeInfo } from '../dist/native.js'; + +const smoke = process.argv.includes('--smoke'); +const documents = integerArgument('--documents', smoke ? 1_000 : 100_000); +const batchSize = integerArgument('--batch-size', smoke ? 250 : 1_000); +const queryCount = integerArgument('--queries', smoke ? 20 : 500); +const concurrency = integerArgument('--concurrency', 4); +const commitEvery = integerArgument('--commit-every', documents); +const indexPath = await mkdtemp(path.join(tmpdir(), 'harper-fulltext-benchmark-')); +const config = { + path: indexPath, + indexId: 'products-benchmark', + generation: 'benchmark-v1', + fields: [{ name: 'title', weight: 3 }, { name: 'description' }, { name: 'category', weight: 1.5 }], + analyzer: 'english@1', + positions: true, + surfaceTerms: false, + limits: { + indexingThreads: Math.min(4, Math.max(1, concurrency)), + searchThreads: Math.min(8, Math.max(1, concurrency)), + writerMemoryBytes: 60_000_000, + maxQueuedCommands: 128, + maxQueuedBytes: 64 * 1024 * 1024, + maxBatchBytes: 16 * 1024 * 1024, + }, +}; + +try { + const info = await runtimeInfo(); + let index = await openNativeFullTextIndex(config); + let packingMilliseconds = 0; + let applyMilliseconds = 0; + let packedBytes = 0; + let uncommittedDocuments = 0; + let peakRssBytes = process.memoryUsage.rss(); + const rssSampler = setInterval(() => { + peakRssBytes = Math.max(peakRssBytes, process.memoryUsage.rss()); + }, 50); + rssSampler.unref(); + const commitLatencies = []; + const indexingStarted = performance.now(); + for (let start = 0; start < documents; start += batchSize) { + const end = Math.min(start + batchSize, documents); + const batch = Array.from({ length: end - start }, (_, offset) => product(start + offset)); + const packingStarted = performance.now(); + const packed = encodeMutationBatch({ upserts: batch }, config.limits.maxBatchBytes); + packingMilliseconds += performance.now() - packingStarted; + packedBytes += packed.byteLength; + const applyStarted = performance.now(); + assert.strictEqual(await index.apply(packed), batch.length); + applyMilliseconds += performance.now() - applyStarted; + peakRssBytes = Math.max(peakRssBytes, process.memoryUsage.rss()); + uncommittedDocuments += batch.length; + if (end === documents || uncommittedDocuments >= commitEvery) { + const commitStarted = performance.now(); + await index.commit(); + commitLatencies.push(performance.now() - commitStarted); + peakRssBytes = Math.max(peakRssBytes, process.memoryUsage.rss()); + uncommittedDocuments = 0; + } + } + const durableIndexingMilliseconds = performance.now() - indexingStarted; + const afterIndexing = index.status(); + const reloadStarted = performance.now(); + await index.reload(); + const reloadMilliseconds = performance.now() - reloadStarted; + + const correctness = await index.search({ text: 'waterproof trail shoes', exactTotal: true, limit: 10 }); + assert(correctness.total > 0); + assert(correctness.hits.some((hit) => hit.id.startsWith('product-'))); + const queryMix = ['waterproof trail shoes', 'wireless headphones', 'cotton blue shirt', 'outdoor product']; + for (const query of queryMix) { + await index.search({ text: query, limit: 10 }); + } + const warm = await measureSearch(index, queryMix, queryCount, concurrency, false); + const exactComparisonCount = Math.max(4, Math.floor(queryCount / 10)); + const approximateSingle = await measureSearch(index, queryMix, exactComparisonCount, 1, false); + const exact = await measureSearch(index, queryMix, exactComparisonCount, 1, true); + const status = index.status(); + peakRssBytes = Math.max(peakRssBytes, process.memoryUsage.rss()); + clearInterval(rssSampler); + await index.close(); + + const reopenStarted = performance.now(); + index = await openNativeFullTextIndex(config); + const reopenMilliseconds = performance.now() - reopenStarted; + const cold = await measureSearch(index, queryMix, Math.min(20, queryCount), 1, false); + await index.close(); + const sortedCommitLatencies = [...commitLatencies].sort((left, right) => left - right); + const output = { + formatVersion: 1, + backend: 'tantivy-mmap', + runtime: info, + host: { + platform: process.platform, + arch: process.arch, + node: process.version, + cpus: navigator.hardwareConcurrency, + }, + workload: { + documents, + batchSize, + packedBytes, + averagePackedBytesPerDocument: packedBytes / documents, + queryCount, + concurrency, + fields: config.fields.length, + indexingThreads: config.limits.indexingThreads, + searchThreads: config.limits.searchThreads, + writerMemoryBytes: config.limits.writerMemoryBytes, + }, + indexing: { + packingMilliseconds, + applyMilliseconds, + durableEndToEndMilliseconds: durableIndexingMilliseconds, + durableDocumentsPerSecond: (documents * 1_000) / durableIndexingMilliseconds, + durablePackedMiBPerSecond: (packedBytes / 1024 / 1024) * (1_000 / durableIndexingMilliseconds), + writerQueueMilliseconds: Number(afterIndexing.metrics.writerQueueNanoseconds) / 1e6, + writerExecutionMilliseconds: Number(afterIndexing.metrics.writerExecutionNanoseconds) / 1e6, + commitEveryDocuments: commitEvery, + commitCount: commitLatencies.length, + commitMillisecondsTotal: commitLatencies.reduce((total, latency) => total + latency, 0), + commitP50Milliseconds: percentile(sortedCommitLatencies, 0.5), + commitP95Milliseconds: percentile(sortedCommitLatencies, 0.95), + commitP99Milliseconds: percentile(sortedCommitLatencies, 0.99), + reloadMilliseconds, + }, + search: { + warmApproximate: warm, + warmApproximateSingle: approximateSingle, + warmExactTotal: exact, + coldAfterReopen: cold, + reopenMilliseconds, + }, + resources: { + indexBytes: await directoryBytes(indexPath), + peakRssBytes, + postCloseRssBytes: process.memoryUsage.rss(), + }, + metrics: { + writerQueueNanoseconds: status.metrics.writerQueueNanoseconds.toString(), + writerExecutionNanoseconds: status.metrics.writerExecutionNanoseconds.toString(), + searchQueueNanoseconds: status.metrics.searchQueueNanoseconds.toString(), + searchExecutionNanoseconds: status.metrics.searchExecutionNanoseconds.toString(), + }, + }; + assert(output.indexing.durableDocumentsPerSecond > 0); + assert(output.search.warmApproximate.p99Milliseconds > 0); + assert(output.resources.indexBytes > 0); + console.log(JSON.stringify(output, null, 2)); +} finally { + await rm(indexPath, { recursive: true, force: true }); +} + +function product(id) { + const variants = [ + ['Waterproof Trail Running Shoes', 'Lightweight outdoor footwear with durable grip', 'shoes'], + ['Wireless Noise Cancelling Headphones', 'Portable audio product with long battery life', 'electronics'], + ['Organic Cotton Blue Shirt', 'Comfortable everyday apparel in multiple sizes', 'clothing'], + ['Stainless Steel Water Bottle', 'Insulated outdoor product for hiking and travel', 'outdoors'], + ]; + const [title, description, category] = variants[id % variants.length]; + return { id: `product-${id}`, fields: { title: `${title} ${id}`, description, category } }; +} + +async function measureSearch(index, queries, count, parallelism, exactTotal) { + const latencies = []; + const started = performance.now(); + for (let offset = 0; offset < count; offset += parallelism) { + await Promise.all( + Array.from({ length: Math.min(parallelism, count - offset) }, async (_, lane) => { + const queryStarted = performance.now(); + const result = await index.search({ + text: queries[(offset + lane) % queries.length], + limit: 10, + exactTotal, + }); + assert(result.hits.length > 0); + latencies.push(performance.now() - queryStarted); + }), + ); + } + const milliseconds = performance.now() - started; + latencies.sort((left, right) => left - right); + return { + queries: count, + concurrency: parallelism, + queriesPerSecond: (count * 1_000) / milliseconds, + p50Milliseconds: percentile(latencies, 0.5), + p95Milliseconds: percentile(latencies, 0.95), + p99Milliseconds: percentile(latencies, 0.99), + }; +} + +function percentile(values, fraction) { + return values[Math.min(values.length - 1, Math.ceil(values.length * fraction) - 1)]; +} + +function integerArgument(name, fallback) { + const index = process.argv.indexOf(name); + if (index === -1) { + return fallback; + } + const value = Number(process.argv[index + 1]); + if (!Number.isSafeInteger(value) || value <= 0) { + throw new Error(`${name} must be a positive integer`); + } + return value; +} + +async function directoryBytes(directory) { + let bytes = 0; + for (const entry of await readdir(directory, { withFileTypes: true })) { + const entryPath = path.join(directory, entry.name); + bytes += entry.isDirectory() ? await directoryBytes(entryPath) : (await stat(entryPath)).size; + } + return bytes; +} diff --git a/docs/native-backend-implementation.md b/docs/native-backend-implementation.md new file mode 100644 index 0000000..bcd25d8 --- /dev/null +++ b/docs/native-backend-implementation.md @@ -0,0 +1,387 @@ +# Native Tantivy backend implementation + +## Intent + +Implement `@harperfast/fulltext/native` as a usable standalone full-text index backed directly by +Tantivy's `MmapDirectory`. This backend is the behavioral and performance reference for the future +RocksDB-backed directory. Harper will never select it in a product release. + +This change covers the smallest end-to-end slice needed to measure real behavior: native index +creation and reopen, batched document upsert/delete, explicit commit and reload, BM25 search, +status, and deterministic close. Phrase, fuzzy, prefix, autocomplete, +suggestions, highlighting, derived-index watermarks, and RocksDB storage remain separate work. + +It deliberately implements narrow prerequisites from issues #14 and #17 without closing either +issue. From #14 it uses one versioned packed mutation request, one packed search result, stable error +codes, and the lifecycle shape required by this backend. From #17 it uses one bounded dedicated +writer actor and one bounded search executor per index so sustained work stays off JavaScript and +libuv while search remains independent of write/commit latency. It does not implement the final +process-wide governor, shared cross-index search pool, multi-environment handle registry, +cancellation, or derived nonblocking admission. + +## Invariant + +Every storage backend uses the same engine-facing schema, mutation, commit, search, and lifecycle +contracts; the native backend contributes only canonical path handling and Tantivy +`MmapDirectory` construction. + +## Verified constraints + +- The package currently exports only `runtimeInfo()` from the hand-written native facade. + `verify: ts/native.ts:1-23, package.json:6-11` +- The addon already contains a panic boundary and per-handle poison primitive. + `verify: src/boundary.rs:1-38, src/lib.rs:31-53` +- The committed Phase 0 work already exercises a complete Tantivy create, write, commit, query, and + reopen lifecycle through the same `Directory` contract intended for RocksDB. + `verify: src/directory_harness.rs:108-153` +- Tantivy 0.26.1 is pinned and compiled into the addon. Its `Index::open` wraps a supplied directory + in `ManagedDirectory`, while `Index::writer_with_num_threads` acquires the writer lock and divides + the supplied memory budget across the requested indexing threads. + `verify: Cargo.toml:20-25; tantivy 0.26.1 src/index/index.rs:509-590` +- `MmapDirectory::open` requires an existing directory, canonicalizes it, and owns its mmap cache, + watcher, filesystem access, and lock behavior. + `verify: tantivy 0.26.1 src/directory/mmap_directory/mod.rs:166-175,232-295` +- The reader uses `ReloadPolicy::Manual`, which does not call `Directory::watch`; Tantivy's mmap + watcher starts its polling thread only when `watch()` is called. The native backend therefore + does not add a metadata-watcher thread per open index. + `verify: src/engine.rs:136-141; tantivy 0.26.1 src/reader/mod.rs:80-98; src/directory/mmap_directory/file_watcher.rs:35-71` +- napi-rs `AsyncTask` executes on the shared libuv pool, so it is not the execution primitive for + sustained indexing or search. + `verify: napi 2.16.17 src/task.rs:6-14` + +## Public slice + +The hand-written TypeScript facade remains authoritative. Generated addon declarations remain +private. + +```ts +interface NativeFullTextIndexOptions { + path: string; + indexId: string; + generation: string; + fields: Array<{ name: string; weight?: number }>; + analyzer: 'english@1'; + stopWords?: boolean; + positions?: boolean; + surfaceTerms?: boolean; + limits: { + indexingThreads: number; + writerMemoryBytes: number; + maxQueuedCommands: number; + maxQueuedBytes: number; + maxBatchBytes: number; + searchThreads: number; + }; +} + +interface FullTextMutationBatch { + upserts?: Array<{ id: string; fields: Record }>; + deletes?: string[]; +} + +interface SearchRequest { + text: string; + operator?: 'any' | 'all'; + fields?: string[]; + offset?: number; + limit?: number; + exactTotal?: boolean; +} + +interface SearchResult { + total: number; + totalRelation: 'exact' | 'lower-bound'; + hits: Array<{ id: string; score: number }>; +} + +interface FullTextStatus { + state: 'open' | 'closing' | 'closed' | 'poisoned'; + uncommittedMutations: bigint; + writerQueuedCommands: bigint; + writerQueuedBytes: bigint; + searchQueuedCommands: bigint; + searchQueuedBytes: bigint; + commitOpstamp: bigint; + metrics: { + writerQueueNanoseconds: bigint; + writerExecutionNanoseconds: bigint; + searchQueueNanoseconds: bigint; + searchExecutionNanoseconds: bigint; + }; +} +``` + +These native-only limits are required in the pre-1.0 standalone factory so measurements are +reproducible and memory is bounded without prematurely implementing #17's final process-wide budget +allocator. The shared engine accepts resolved limits; the future runtime governor will supply them +for both storage backends, and the Harper schema will never expose them. + +The exported flow is: + +1. `openNativeFullTextIndex(options)` creates the directory when absent, canonicalizes it, reserves + the canonical path, and asynchronously creates or reopens Tantivy state. +2. `encodeMutationBatch(batch)` creates the versioned packed request. `apply(packedBatch)` copies + it once into Rust-owned memory, validates it once in Rust, and enqueues one command. Upsert is + delete-by-ID followed by add, so a committed ID has at most one live document. +3. `commit()` serializes behind earlier writer commands and publishes through Tantivy's ordinary + commit path. It does not imply reader reload. +4. `reload()` crosses the writer barrier and then refreshes the reader on the search executor; + `search()` uses one captured immutable searcher without waiting behind indexing or commit work. +5. `close()` rejects new work, settles admitted commands, shuts down the writer, releases the path + reservation, and is idempotent. +6. `status()` reads bounded counters and state without entering either sustained-work queue. + +Only English analysis is accepted. `positions` defaults to true and selects frequencies with or +without positions. Changing that default in a future release is an index-format change, not a +silent reinterpretation. `generation` is an opaque caller-owned identity for this physical index +generation; it is persisted in the engine fingerprint and must match on reopen. It is not a Tantivy +opstamp or a Harper transaction-log position. +`surfaceTerms` stores original field values in the same Tantivy field for later highlighting and +suggestion work; it defaults off because of its storage cost. Field weights are applied as query +boosts and therefore may change without rebuilding the physical schema. + +The public search method accepts a small typed request and decodes a versioned native result buffer +into bounded result objects. The N-API boundary receives one operation per batch or search; +per-document native calls are not exposed. The benchmark reports engine execution separately from +N-API plus result-decoding time so storage and boundary costs cannot be confused. + +`encodeMutationBatch()` enforces the configured byte limit incrementally before allocation growth, +but it is a convenience encoder, not claimed to be an event-loop-free ingestion path. It performs +UTF-8 encoding on its caller's JavaScript thread. Production callers may build the documented +packed format in their own worker; Harper's projection path produces it directly. `apply()` then +makes one required copy into Rust-owned memory before asynchronous admission. Both costs are +measured, and the benchmark pre-encodes its engine-only corpus so directory results exclude them. + +## Native architecture + +```text +Node worker + │ one packed mutation or small search request + ▼ +N-API handle registry ── canonical path reservation + │ + ├─ write/commit/reload barrier ─► bounded writer queue ─► dedicated writer actor + │ └─ Tantivy IndexWriter + │ └─ indexing/merge workers + └─ search/reload ─► bounded search queue ─► small search worker pool + └─ shared IndexReader/Searcher + +writer actor ─► shared engine ─► MmapDirectory (native) + later └─ RocksDbDirectory (same engine) +``` + +The native addon owns these threads and queues; no sustained operation runs on the JavaScript event +loop or libuv pool. The writer actor is the sole owner of `IndexWriter`, making mutation, commit, +rollback, and shutdown ordering explicit. A small configurable search pool shares the +`IndexReader`; each request captures its immutable `Searcher`, so searches overlap each other as +well as indexing and commit. `reload()` first crosses the writer queue as a barrier, reloads the +shared reader under a short coordination lock, and publishes the new searcher to subsequent +requests. Tantivy remains free to use its +configured indexing and merge workers behind the writer actor. Independent indexes and their +searches may run concurrently. #17 later replaces the per-index search pools with the bounded +process pool without changing engine behavior. + +Both queues are bounded by command count and retained bytes. A JS-owned buffer is copied once into a +Rust-owned `Vec` before admission; no native thread borrows memory owned by a Node environment. +Queue wait and engine execution time are reported separately in status/benchmark output. Effective +writer arena, indexing-thread, and search-thread values are printed in every benchmark record. + +The registry rejects a second live open of the same canonical path. This is narrower than the +eventual shared multi-environment registry, but it preserves the one-writer invariant without +pretending two JavaScript handles have coordinated close ownership. The later registry issue can +replace rejection with reference-counted shared handles without changing the index contract. + +## Schema and query behavior + +Each Tantivy schema contains an internal indexed string fast field for the raw ID using Tantivy's raw tokenizer +and one declared text field per configured source field. At creation, the engine atomically writes a +small backend-neutral identity sidecar through `Directory::atomic_write()` and calls +`Directory::sync_directory()`. It contains a versioned fingerprint of the logical index ID, bounded +generation, analyzer identity, stop-word policy, positions, surface-term storage, and structural +schema. It deliberately excludes the Node wire ABI and Tantivy package version: wire changes do not +change durable semantics, and Tantivy performs its own index-format compatibility check. Reopen +compares both the generated Tantivy schema and this fingerprint before creating a writer. The +immutable sidecar is separate from Tantivy's per-commit payload, which remains available for +standalone checkpoints and derived watermarks. +Unknown mutation fields, missing IDs, duplicate schema +field names, empty queries, unknown search fields, oversized batches, and excessive result windows +fail before search/index work. + +Create treats `{sidecar, meta.json}` as a pair. If neither exists, it writes and syncs the sidecar +first and then creates the Tantivy index. If both exist, it reopens and verifies them. A sidecar-only +state is an interrupted empty create and is completed automatically after verifying the sidecar; +`meta.json` without the sidecar is rejected as incomplete because it may contain durable data whose +identity cannot be proven. Absent, unparseable, or mismatched identity is never accepted as +legacy-compatible. Tantivy's persisted schema and versioned tokenizer names independently verify +the structural and analyzer portions of the identity. Every declared count and length in packed input is checked +against the remaining bytes before arithmetic or allocation, the version tag is checked first, and +invalid UTF-8 is rejected on the writer actor rather than the JavaScript thread. Admission checks +the fixed header and structural bounds only. `indexId` and `generation` have fixed encoded-length +limits because they are persisted. + +The English analyzer is versioned by name and composed from Tantivy's tokenizer primitives: +`SimpleTokenizer`, `RemoveLongFilter`, `LowerCaser`, optional built-in English +`StopWordFilter`, and English `Stemmer`. The same registered analyzer tokenizes indexed text and +search text. + +Search builds a typed Boolean query rather than exposing Tantivy's query-string syntax. Each +analyzed term is searched across the selected fields, applying configured field boosts. `any` +scores documents matching at least one term; `all` requires every analyzed term to match at least +one selected field. Tantivy's normal scorer supplies BM25. The default result reports a bounded +lower total (`offset + returned hits`, with `totalRelation: 'lower-bound'` when the page is full) and +runs `TopDocs::order_by_score()` alone. In pinned Tantivy 0.26.1 that collector invokes +`Weight::for_each_pruning`, and Boolean term unions select the block-WAND implementation. Exact +total is explicit per query, runs a separate `Count`, and is benchmarked separately at the same +concurrency because it must visit all matches. The initial schema resolves hit IDs through that fast +field once per result segment, avoiding stored-document decompression on every result. The ID is not +duplicated in Tantivy's document store. + +## Failure and lifecycle behavior + +All N-API exports retain `catch_unwind`, and both native thread entries catch unwind around +initialization and every command. A panic poisons only the affected handle, drains and rejects every +queued promise, and leaves no admitted promise unsettled. Filesystem, incomplete-create, identity, +schema, query, resource, queue, closed, and native failures map to stable package error codes present +in the TypeScript allowlist while preserving the cause message. A competing process holding +Tantivy's filesystem writer lock maps to a distinct retryable lock-busy code. A source-parity test +compares the Rust error table with the TypeScript allowlist, while integration tests assert codes on +representative synchronous and asynchronous failures. No Rust type or Tantivy object crosses the +public API or Node worker. + +Successful `commit()` delegates to Tantivy 0.26.1's ordinary commit path and resolves only after it +returns. The process-kill test verifies publication and process-crash recovery. The existing +`KvDirectory` contract suite supplies the durable file/publication ordering checks; injecting a +commit failure through the complete shared engine remains part of Rocks-backend hardening. Engine +payload publication is intentionally deferred to the derived-index checkpoint work. A commit or +post-validation mutation failure poisons the writer generation; callers close and reopen from the +last durable commit rather than guessing which uncommitted opstamps survived. + +`close()` defaults to require-clean: uncommitted mutations fail close rather than being silently +committed or discarded. That failure restores the open state so the caller can commit or call +`close({ mode: 'rollback' })`; it does not strand the path reservation in `closing`. A poisoned +handle always permits rollback close, and a default close after a terminal commit failure performs +the same forced teardown because there is no valid writer state left to preserve. A successful close +joins Tantivy's merge threads, stops the search executor, and only then releases the canonical path. +Closing transitions through open, closing, and closed states; it settles admitted commands, and +repeated successful close calls resolve. Process exit does not promise an implicit final commit. + +Each Node environment registers one N-API asynchronous cleanup hook, shared by every index it opens. +Environment teardown stops accepting work, detaches JavaScript completions, schedules rollback close +for every tracked runtime before waiting, and keeps the native environment alive until writer +shutdown, search-thread joins, and path release complete. The hook uses napi-rs's cleanup primitive +and returns only after the native actors stop, so hook completion and environment destruction stay +on Node's environment thread. All runtimes share one 30-second deadline rather than consuming that +budget serially per index, and a caught panic still returns control to napi-rs so worker/process +shutdown cannot hang indefinitely. A worker terminated while an index is still opening waits on the +same bounded completion signal. Completions detached during teardown leave their native thread-safe +function to Node's environment finalization instead of accessing that environment after the hook +returns. Handles and completions are never reused across workers. + +## Performance experiment + +Add a release-build benchmark that generates a deterministic high-cardinality product-style corpus +and drives the public native API. It emits one versioned JSON record containing environment +metadata and: + +- durable documents and packed MiB indexed per second by batch size and commit cadence; +- separately reported packing, apply, writer queue, and writer execution time; +- commit latency distribution and reload time; +- warm BM25 search p50/p95/p99 and throughput at configurable concurrency, using approximate totals + by default and same-concurrency single-worker approximate/exact profiles; +- cold-after-reopen search p50/p95/p99; +- index bytes, periodically sampled peak RSS, and post-close RSS; and +- document count, field count, average packed bytes, thread/memory budgets, Tantivy version, and + host metadata needed to interpret the numbers. + +The default local profile is short enough for engineering iteration. A larger profile is selected +explicitly. Correctness assertions run before timing results are accepted: expected IDs must rank, +reopen must preserve results, and every operation count must match. The benchmark's +replacement-safe upserts emit delete terms, so `--commit-every` is an explicit workload dimension +rather than allowing an unbounded final commit to masquerade as a production ingestion profile. +The benchmark does not claim the 100-million-document or Harper p99-under-50-ms release gate; those +remain the paired fixed-host work in issue #15. This slice establishes the native engine and N-API +baseline that issue #15 will compare against RocksDB. + +CI runs correctness tests and an explicit small benchmark-smoke command that performs ranking, +commit, close, and reopen assertions before validating nonzero measurements. Shared runners enforce +no timing threshold. Performance thresholds require controlled hardware and release-over-release +history. + +## Verification + +- Rust unit tests: schema equality, analyzer behavior, batch decode bounds, upsert/delete ordering, + query construction, close state, duplicate path rejection, and queue saturation. +- Node tests through `@harperfast/fulltext/native`: create, apply, commit, reload, BM25 ranking, + reopen, mutation validation, schema mismatch, close modes, and event-loop responsiveness. +- Process tests: kill the indexer immediately after a successful commit and verify the committed + corpus after reopen; kill before commit and verify it is absent. A worker-thread test verifies + promises settle only into their originating Node environment. +- Concurrency tests: duplicate canonical opens admit one writer, overload rejects without blocking + JavaScript, indexing leaves the event loop responsive, and worker termination detaches + completions and releases the writer. +- Decoder tests cover invalid counts, lengths, and UTF-8; randomized decoder fuzzing remains part of + the hardening work. +- Existing Directory contract tests remain unchanged. +- `npm run check` and package artifact verification run before review. +- The release benchmark runs locally at two dataset sizes and commit cadences; raw JSON is retained + with the PR verification notes. It is built explicitly in release mode before measurements are + taken. + +End-to-end route: a Node integration test loads the built addon through the published native +subpath, creates an on-disk index, mutates and commits it, searches it, closes it, reopens it, and +repeats the search. + +## Approaches considered + +### Different layer: implement native storage only in Harper + +The candidate layer is Harper's `DerivedIndexBackend`, which could own the engine and call a thin +native filesystem binding. Rejected because the invariant is one engine for standalone RocksDB, +native filesystem users, and Harper-derived delivery; putting the engine in Harper would force the +two standalone modes either to depend on Harper or to fork search/index behavior. + +### Deeper cause: implement the complete storage-neutral runtime before either backend + +The bad state to prevent is a backend opening durable data whose analyzer or engine semantics it +cannot interpret. The candidate is a backend-neutral persisted fingerprint and commit durability +contract before either backend ships. This is adopted in the chosen approach. The complete #14/#17 +surface remains rejected for this issue because derived watermarks, multi-environment sharing, and a +global scheduler are not needed to enforce that durable compatibility invariant. + +### Do less: expose Tantivy's existing filesystem API or query parser directly + +Rejected because it would expose a third-party API, permit unbounded query syntax, and create a +public contract the Rocks and Harper integrations could not safely govern. A directory-only smoke +also cannot provide the performance baseline Kyle requested. + +### Do less on runtime: engine benchmark plus separate per-index writer and search executors + +Adopted. The engine is generic over `Directory`, and the Node slice uses byte-bounded per-index +execution rather than implementing #17's process governor and shared cross-index search pool. +Keeping search separate from the writer is the minimum needed for a baseline that measures Tantivy +instead of temporary writer-queue head-of-line blocking. + +### Identity sidecar versus repeating identity in every commit payload + +The sidecar is chosen. Repeating identity in every commit payload couples immutable engine identity +to future checkpoint/watermark publication and lets any omitted `set_payload()` erase it. An +immutable sidecar written through the same `Directory` contract prevents that failure and works for +both `MmapDirectory` and `RocksDbDirectory` without custom filesystem code. + +### Chosen: one shared engine slice with a thin MmapDirectory constructor + +This is the only option that simultaneously produces a usable standalone backend, keeps native +storage out of Harper, avoids reimplementing Tantivy filesystem primitives, establishes bounded +off-event-loop execution, and yields an apples-to-apples reference for the Rocks directory. + +## Explicit deferrals + +- Derived-index delivery, checkpoints/watermarks, replay, and Harper lifecycle hooks. +- RocksDbDirectory and rocksdb-js lease use. +- Phrase, fuzzy, prefix, autocomplete, suggestions, highlighting, snippets, and filters. +- Shared handles across multiple Node worker environments. +- A handle-lifetime response dispatcher that replaces the initial per-operation thread-safe + callback; this is part of the shared multi-environment runtime in issue #17. +- Durable benchmark publication, fixed-host regression thresholds, and Rocks/native comparison. +- Process-wide runtime budgets, cancellation, and cursor-based + deep pagination. diff --git a/package.json b/package.json index 9ff9e4a..48d20b3 100644 --- a/package.json +++ b/package.json @@ -33,6 +33,8 @@ "build:native": "napi build --platform --js false --dts ts/addon.d.ts --release --features node-api", "build:test-native": "napi build --platform --js false --dts ts/addon.d.ts --features test-panic", "build:typescript": "tsc -p tsconfig.json", + "benchmark:native": "npm run build && node benchmarks/native.mjs", + "benchmark:smoke": "node benchmarks/native.mjs --smoke", "check": "npm run format:check && npm run lint && npm run test", "format": "prettier --write . && cargo fmt", "format:check": "prettier --check . && cargo fmt --check", diff --git a/src/engine.rs b/src/engine.rs new file mode 100644 index 0000000..92c0a99 --- /dev/null +++ b/src/engine.rs @@ -0,0 +1,576 @@ +use std::collections::{HashMap, HashSet}; +use std::path::Path; + +use tantivy::collector::{Count, TopDocs}; +use tantivy::directory::Directory; +use tantivy::query::{BooleanQuery, BoostQuery, Occur, Query, TermQuery}; +use tantivy::schema::{Field, IndexRecordOption, Schema, TantivyDocument, TextFieldIndexing, TextOptions}; +use tantivy::tokenizer::{ + Language, LowerCaser, RemoveLongFilter, SimpleTokenizer, Stemmer, StopWordFilter, TextAnalyzer, +}; +use tantivy::{Index, IndexReader, IndexSettings, IndexWriter, ReloadPolicy, Searcher, Term}; + +use crate::error::{FulltextError, Result}; +use crate::protocol::{EngineConfig, MutationBatch, SearchOperator, SearchRequest}; + +const ID_FIELD_NAME: &str = "__fulltext_id"; +const IDENTITY_PATH: &str = ".harper-fulltext-identity"; +const META_PATH: &str = "meta.json"; +const ANALYZER_NAME: &str = "english@1"; + +#[derive(Clone)] +pub struct Engine { + index: Index, + id_field: Field, + fields: Vec, + field_lookup: HashMap, + analyzer: TextAnalyzer, +} + +#[derive(Clone)] +struct EngineField { + name: String, + field: Field, + weight: f32, +} + +pub struct Writer { + inner: IndexWriter, + id_field: Field, + fields: Vec, + field_lookup: HashMap, +} + +pub(crate) struct PreparedBatch { + mutation_count: u64, + deletes: Vec, + documents: Vec<(String, TantivyDocument)>, +} + +#[derive(Clone, Debug, PartialEq)] +pub struct SearchHit { + pub id: String, + pub score: f32, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum TotalRelation { + Exact, + LowerBound, +} + +#[derive(Clone, Debug, PartialEq)] +pub struct SearchResult { + pub total: u64, + pub total_relation: TotalRelation, + pub hits: Vec, +} + +impl Engine { + pub fn open(directory: D, config: &EngineConfig) -> Result { + let (schema, id_field, fields) = build_schema(config)?; + let expected_identity = identity_bytes(config); + let sidecar_exists = directory.exists(Path::new(IDENTITY_PATH)).map_err(storage_error)?; + let meta_exists = directory.exists(Path::new(META_PATH)).map_err(storage_error)?; + + if sidecar_exists { + let actual = directory.atomic_read(Path::new(IDENTITY_PATH)).map_err(storage_error)?; + if actual != expected_identity { + return Err(FulltextError::new( + "E_IDENTITY_MISMATCH", + "the persisted index identity does not match the requested configuration", + )); + } + } else if meta_exists { + return Err(FulltextError::new( + "E_INCOMPLETE_CREATE", + "meta.json exists without a fulltext identity sidecar", + )); + } else { + directory + .atomic_write(Path::new(IDENTITY_PATH), &expected_identity) + .map_err(storage_error)?; + directory.sync_directory().map_err(storage_error)?; + } + + let index = if meta_exists { + Index::open(directory).map_err(index_error)? + } else { + Index::create(directory, schema.clone(), IndexSettings::default()).map_err(index_error)? + }; + if index.schema() != schema { + return Err(FulltextError::new( + "E_SCHEMA_MISMATCH", + "the persisted Tantivy schema does not match the requested configuration", + )); + } + let analyzer = build_analyzer(config.stop_words)?; + index.tokenizers().register(ANALYZER_NAME, analyzer.clone()); + let field_lookup = fields + .iter() + .enumerate() + .map(|(index, field)| (field.name.clone(), index)) + .collect(); + Ok(Self { + index, + id_field, + fields, + field_lookup, + analyzer, + }) + } + + pub fn writer(&self, config: &EngineConfig) -> Result { + let inner = self + .index + .writer_with_num_threads(config.limits.indexing_threads, config.limits.writer_memory_bytes) + .map_err(index_error)?; + Ok(Writer { + inner, + id_field: self.id_field, + fields: self.fields.clone(), + field_lookup: self.field_lookup.clone(), + }) + } + + pub fn reader(&self) -> Result { + self.index + .reader_builder() + .reload_policy(ReloadPolicy::Manual) + .try_into() + .map_err(index_error) + } + + pub fn search(&self, searcher: &Searcher, request: &SearchRequest) -> Result { + let selected = self.selected_fields(&request.fields)?; + let query = self.query(&request.text, request.operator, &selected)?; + let top_docs = searcher + .search( + query.as_ref(), + &TopDocs::with_limit(request.limit) + .and_offset(request.offset) + .order_by_score(), + ) + .map_err(index_error)?; + let mut hits_by_segment = HashMap::new(); + for (index, (_, address)) in top_docs.iter().enumerate() { + hits_by_segment + .entry(address.segment_ord) + .or_insert_with(Vec::new) + .push((index, address.doc_id)); + } + let mut ids = vec![String::new(); top_docs.len()]; + for (segment_ord, segment_hits) in hits_by_segment { + let segment = &searcher.segment_readers()[segment_ord as usize]; + let column = segment + .fast_fields() + .str(ID_FIELD_NAME) + .map_err(index_error)? + .ok_or_else(|| FulltextError::new("E_NATIVE_FAILURE", "search segment has no ID fast field"))?; + let mut id = Vec::new(); + for (index, doc_id) in segment_hits { + let ordinal = column + .term_ords(doc_id) + .next() + .ok_or_else(|| FulltextError::new("E_NATIVE_FAILURE", "search hit has no ID ordinal"))?; + id.clear(); + if !column + .dictionary() + .ord_to_term(ordinal, &mut id) + .map_err(storage_error)? + { + return Err(FulltextError::new( + "E_NATIVE_FAILURE", + "search hit ID ordinal is missing", + )); + } + ids[index] = std::str::from_utf8(&id) + .map_err(|_| FulltextError::new("E_NATIVE_FAILURE", "search hit ID is not UTF-8"))? + .to_owned(); + } + } + let hits = top_docs + .into_iter() + .zip(ids) + .map(|((score, _), id)| SearchHit { id, score }) + .collect::>(); + let (total, total_relation) = if request.exact_total { + ( + searcher.search(query.as_ref(), &Count).map_err(index_error)? as u64, + TotalRelation::Exact, + ) + } else if request.offset == 0 && hits.len() < request.limit { + (hits.len() as u64, TotalRelation::Exact) + } else if !hits.is_empty() && hits.len() < request.limit { + ((request.offset + hits.len()) as u64, TotalRelation::Exact) + } else { + ( + if hits.is_empty() { + 0 + } else { + (request.offset + hits.len()) as u64 + }, + TotalRelation::LowerBound, + ) + }; + Ok(SearchResult { + total, + total_relation, + hits, + }) + } + + fn selected_fields(&self, requested: &[String]) -> Result> { + if requested.is_empty() { + return Ok(self.fields.iter().collect()); + } + let mut seen = HashSet::with_capacity(requested.len()); + let mut fields = Vec::with_capacity(requested.len()); + for name in requested { + if !seen.insert(name) { + return Err(FulltextError::invalid(format!("duplicate search field {name}"))); + } + let index = self + .field_lookup + .get(name) + .ok_or_else(|| FulltextError::invalid(format!("unknown search field {name}")))?; + fields.push(&self.fields[*index]); + } + Ok(fields) + } + + fn query(&self, text: &str, operator: SearchOperator, fields: &[&EngineField]) -> Result> { + let mut analyzer = self.analyzer.clone(); + let mut stream = analyzer.token_stream(text); + let mut tokens = Vec::new(); + stream.process(&mut |token| tokens.push(token.text.clone())); + if tokens.is_empty() { + return Err(FulltextError::invalid("search text produced no searchable terms")); + } + let outer_occur = match operator { + SearchOperator::Any => Occur::Should, + SearchOperator::All => Occur::Must, + }; + let mut terms = Vec::with_capacity(tokens.len()); + for token in tokens { + let mut alternatives: Vec<(Occur, Box)> = Vec::with_capacity(fields.len()); + for field in fields { + let query: Box = Box::new(TermQuery::new( + Term::from_field_text(field.field, &token), + IndexRecordOption::WithFreqs, + )); + let query = if field.weight == 1.0 { + query + } else { + Box::new(BoostQuery::new(query, field.weight)) + }; + alternatives.push((Occur::Should, query)); + } + terms.push((outer_occur, Box::new(BooleanQuery::new(alternatives)) as Box)); + } + Ok(Box::new(BooleanQuery::new(terms))) + } +} + +impl Writer { + pub fn apply(&self, batch: MutationBatch) -> Result { + let prepared = self.prepare(batch)?; + self.apply_prepared(prepared) + } + + pub(crate) fn prepare(&self, batch: MutationBatch) -> Result { + let mutation_count = batch.upserts.len() + batch.deletes.len(); + for id in &batch.deletes { + if id.is_empty() { + return Err(FulltextError::invalid("delete ID must not be empty")); + } + } + let mut documents = Vec::with_capacity(batch.upserts.len()); + for upsert in batch.upserts { + if upsert.id.is_empty() { + return Err(FulltextError::invalid("upsert ID must not be empty")); + } + let mut seen = HashSet::with_capacity(upsert.fields.len()); + let mut document = TantivyDocument::default(); + document.add_text(self.id_field, &upsert.id); + for (name, values) in upsert.fields { + if !seen.insert(name.clone()) { + return Err(FulltextError::invalid(format!("duplicate mutation field {name}"))); + } + let index = self + .field_lookup + .get(&name) + .ok_or_else(|| FulltextError::invalid(format!("unknown mutation field {name}")))?; + for value in values { + document.add_text(self.fields[*index].field, &value); + } + } + documents.push((upsert.id, document)); + } + Ok(PreparedBatch { + mutation_count: mutation_count as u64, + deletes: batch.deletes, + documents, + }) + } + + pub(crate) fn apply_prepared(&self, prepared: PreparedBatch) -> Result { + for id in prepared.deletes { + self.inner.delete_term(Term::from_field_text(self.id_field, &id)); + } + for (id, document) in prepared.documents { + self.inner.delete_term(Term::from_field_text(self.id_field, &id)); + self.inner.add_document(document).map_err(index_error)?; + } + Ok(prepared.mutation_count) + } + + pub fn commit(&mut self) -> Result { + self.inner.commit().map_err(index_error) + } + + pub fn rollback(&mut self) -> Result { + self.inner.rollback().map_err(index_error) + } + + pub fn close(self) -> Result<()> { + self.inner.wait_merging_threads().map_err(index_error) + } +} + +fn build_schema(config: &EngineConfig) -> Result<(Schema, Field, Vec)> { + let mut builder = Schema::builder(); + let id_indexing = TextFieldIndexing::default() + .set_tokenizer("raw") + .set_index_option(IndexRecordOption::Basic) + .set_fieldnorms(false); + let id_options = TextOptions::default().set_indexing_options(id_indexing).set_fast(None); + let id_field = builder.add_text_field(ID_FIELD_NAME, id_options); + let record = if config.positions { + IndexRecordOption::WithFreqsAndPositions + } else { + IndexRecordOption::WithFreqs + }; + let mut fields = Vec::with_capacity(config.fields.len()); + for field in &config.fields { + let indexing = TextFieldIndexing::default() + .set_tokenizer(ANALYZER_NAME) + .set_index_option(record); + let mut options = TextOptions::default().set_indexing_options(indexing); + if config.surface_terms { + options = options.set_stored(); + } + let schema_field = builder.add_text_field(&field.name, options); + fields.push(EngineField { + name: field.name.clone(), + field: schema_field, + weight: field.weight, + }); + } + Ok((builder.build(), id_field, fields)) +} + +fn build_analyzer(stop_words: bool) -> Result { + let mut builder = TextAnalyzer::builder(SimpleTokenizer::default()) + .filter_dynamic(RemoveLongFilter::limit(40)) + .filter_dynamic(LowerCaser); + if stop_words { + let stop_filter = StopWordFilter::new(Language::English) + .ok_or_else(|| FulltextError::new("E_NATIVE_FAILURE", "English stop words are unavailable"))?; + builder = builder.filter_dynamic(stop_filter); + } + Ok(builder.filter_dynamic(Stemmer::new(Language::English)).build()) +} + +fn identity_bytes(config: &EngineConfig) -> Vec { + let mut bytes = b"HTFI\x01\x00".to_vec(); + push_string(&mut bytes, &config.index_id); + push_string(&mut bytes, &config.generation); + push_string(&mut bytes, &config.analyzer); + bytes.extend_from_slice(&[ + config.stop_words as u8, + config.positions as u8, + config.surface_terms as u8, + ]); + bytes.extend_from_slice(&(config.fields.len() as u16).to_le_bytes()); + for field in &config.fields { + push_string(&mut bytes, &field.name); + } + bytes +} + +fn push_string(bytes: &mut Vec, value: &str) { + bytes.extend_from_slice(&(value.len() as u32).to_le_bytes()); + bytes.extend_from_slice(value.as_bytes()); +} + +fn storage_error(error: impl std::fmt::Display) -> FulltextError { + FulltextError::new("E_STORAGE", error.to_string()) +} + +fn index_error(error: tantivy::TantivyError) -> FulltextError { + match error { + tantivy::TantivyError::LockFailure(tantivy::directory::error::LockError::LockBusy, _) => { + FulltextError::new("E_LOCK_BUSY", "another writer owns the Tantivy index lock") + } + tantivy::TantivyError::OpenDirectoryError(_) + | tantivy::TantivyError::OpenReadError(_) + | tantivy::TantivyError::OpenWriteError(_) + | tantivy::TantivyError::IoError(_) => storage_error(error), + other => FulltextError::native(other), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::protocol::{FieldConfig, Limits}; + use tantivy::directory::RamDirectory; + + fn config() -> EngineConfig { + EngineConfig { + path: "unused".to_owned(), + index_id: "products".to_owned(), + generation: "one".to_owned(), + fields: vec![ + FieldConfig { + name: "title".to_owned(), + weight: 3.0, + }, + FieldConfig { + name: "description".to_owned(), + weight: 1.0, + }, + ], + analyzer: ANALYZER_NAME.to_owned(), + stop_words: true, + positions: true, + surface_terms: false, + limits: Limits { + indexing_threads: 1, + search_threads: 2, + writer_memory_bytes: 15_000_000, + max_queued_commands: 8, + max_queued_bytes: 1 << 20, + max_batch_bytes: 1 << 20, + }, + } + } + + fn batch() -> MutationBatch { + MutationBatch { + upserts: vec![ + crate::protocol::Upsert { + id: "one".to_owned(), + fields: vec![("title".to_owned(), vec!["Running Shoes".to_owned()])], + }, + crate::protocol::Upsert { + id: "two".to_owned(), + fields: vec![("description".to_owned(), vec!["shoe rack".to_owned()])], + }, + ], + deletes: Vec::new(), + } + } + + #[test] + fn indexes_searches_updates_and_reopens() { + let directory = RamDirectory::create(); + let config = config(); + let engine = Engine::open(directory.clone(), &config).unwrap(); + let mut writer = engine.writer(&config).unwrap(); + assert_eq!(writer.apply(batch()).unwrap(), 2); + writer.commit().unwrap(); + let reader = engine.reader().unwrap(); + reader.reload().unwrap(); + let request = SearchRequest { + text: "shoes".to_owned(), + operator: SearchOperator::Any, + fields: Vec::new(), + offset: 0, + limit: 10, + exact_total: true, + }; + let result = engine.search(&reader.searcher(), &request).unwrap(); + assert_eq!(result.total, 2); + assert_eq!(result.hits[0].id, "one"); + writer + .apply(MutationBatch { + upserts: Vec::new(), + deletes: vec!["one".to_owned()], + }) + .unwrap(); + writer.commit().unwrap(); + reader.reload().unwrap(); + assert_eq!(engine.search(&reader.searcher(), &request).unwrap().total, 1); + writer.close().unwrap(); + let reopened = Engine::open(directory, &config).unwrap(); + let reopened_reader = reopened.reader().unwrap(); + assert_eq!(reopened.search(&reopened_reader.searcher(), &request).unwrap().total, 1); + } + + #[test] + fn rejects_identity_mismatch() { + let directory = RamDirectory::create(); + let config = config(); + Engine::open(directory.clone(), &config).unwrap(); + let mut different = config; + different.generation = "two".to_owned(); + let error = match Engine::open(directory, &different) { + Ok(_) => panic!("identity mismatch was accepted"), + Err(error) => error, + }; + assert_eq!(error.code, "E_IDENTITY_MISMATCH"); + } + + #[test] + fn same_engine_runs_on_the_kv_directory() { + let directory = crate::phase0::FaultingDirectory::new(crate::phase0::FaultingKv::default()); + let config = config(); + let engine = Engine::open(directory.clone(), &config).unwrap(); + let mut writer = engine.writer(&config).unwrap(); + writer.apply(batch()).unwrap(); + writer.commit().unwrap(); + let reader = engine.reader().unwrap(); + let request = SearchRequest { + text: "running shoes".to_owned(), + operator: SearchOperator::All, + fields: Vec::new(), + offset: 0, + limit: 10, + exact_total: true, + }; + assert_eq!(engine.search(&reader.searcher(), &request).unwrap().hits[0].id, "one"); + writer.close().unwrap(); + let reopened = Engine::open(directory, &config).unwrap(); + let reopened_reader = reopened.reader().unwrap(); + assert_eq!( + reopened.search(&reopened_reader.searcher(), &request).unwrap().hits[0].id, + "one" + ); + } + + #[test] + fn completes_an_interrupted_sidecar_first_create() { + let directory = RamDirectory::create(); + let config = config(); + directory + .atomic_write(Path::new(IDENTITY_PATH), &identity_bytes(&config)) + .unwrap(); + Engine::open(directory.clone(), &config).unwrap(); + assert!(directory.exists(Path::new(META_PATH)).unwrap()); + } + + #[test] + fn rejects_meta_without_an_identity_sidecar() { + let directory = RamDirectory::create(); + let config = config(); + let (schema, _, _) = build_schema(&config).unwrap(); + Index::create(directory.clone(), schema, IndexSettings::default()).unwrap(); + let error = match Engine::open(directory, &config) { + Ok(_) => panic!("meta without an identity sidecar was accepted"), + Err(error) => error, + }; + assert_eq!(error.code, "E_INCOMPLETE_CREATE"); + } +} diff --git a/src/error.rs b/src/error.rs new file mode 100644 index 0000000..d819668 --- /dev/null +++ b/src/error.rs @@ -0,0 +1,53 @@ +use std::fmt; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct FulltextError { + pub code: &'static str, + pub message: String, +} + +impl FulltextError { + pub fn new(code: &'static str, message: impl Into) -> Self { + Self { + code, + message: message.into(), + } + } + + pub fn invalid(message: impl Into) -> Self { + Self::new("E_INVALID_ARGUMENT", message) + } + + pub fn native(error: impl fmt::Display) -> Self { + Self::new("E_NATIVE_FAILURE", error.to_string()) + } +} + +impl fmt::Display for FulltextError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(formatter, "{}: {}", self.code, self.message) + } +} + +impl std::error::Error for FulltextError {} + +pub type Result = std::result::Result; + +pub const ERROR_CODES: &[&str] = &[ + "E_CLOSED", + "E_DIRTY_CLOSE", + "E_DUPLICATE_OPEN", + "E_IDENTITY_MISMATCH", + "E_INCOMPLETE_CREATE", + "E_INVALID_ARGUMENT", + "E_LOCK_BUSY", + "E_NATIVE_ABI_MISMATCH", + "E_NATIVE_ADDON_NOT_FOUND", + "E_NATIVE_CAPABILITY_MISMATCH", + "E_NATIVE_FAILURE", + "E_NATIVE_PANIC", + "E_POISONED", + "E_QUEUE_FULL", + "E_SCHEMA_MISMATCH", + "E_STORAGE", +]; diff --git a/src/lib.rs b/src/lib.rs index 7c2f315..86ed122 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2,6 +2,9 @@ #![deny(unsafe_op_in_unsafe_fn)] pub mod directory_harness; +pub mod engine; +pub mod error; +pub mod protocol; #[cfg(any(test, feature = "phase0"))] pub mod phase0; @@ -12,6 +15,9 @@ pub mod rocks_lease; #[cfg(feature = "node-api")] mod boundary; +#[cfg(feature = "node-api")] +pub mod native; + #[cfg(feature = "node-api")] use napi_derive::napi; diff --git a/src/native.rs b/src/native.rs new file mode 100644 index 0000000..ebbf069 --- /dev/null +++ b/src/native.rs @@ -0,0 +1,1057 @@ +use std::collections::{HashMap, HashSet, VecDeque}; +use std::fs; +use std::mem; +use std::panic::{catch_unwind, AssertUnwindSafe}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicU8, Ordering}; +use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak}; +use std::thread; +use std::time::{Duration, Instant}; + +use napi::bindgen_prelude::Buffer; +use napi::threadsafe_function::{ErrorStrategy, ThreadSafeCallContext, ThreadsafeFunction, ThreadsafeFunctionCallMode}; +use napi::{Env, JsFunction}; +use napi_derive::napi; +use tantivy::directory::MmapDirectory; +use tantivy::IndexReader; + +use crate::boundary; +use crate::engine::{Engine, SearchResult, TotalRelation, Writer}; +use crate::error::{FulltextError, Result}; +use crate::protocol::{ + decode_batch, decode_open, decode_search, validate_batch_header, validate_search_header, EngineConfig, +}; + +const STATE_OPEN: u8 = 0; +const STATE_CLOSING: u8 = 1; +const STATE_CLOSED: u8 = 2; +const STATE_POISONED: u8 = 3; +const CLEANUP_TIMEOUT: Duration = Duration::from_secs(30); + +static NEXT_HANDLE: AtomicU32 = AtomicU32::new(1); +static REGISTRY: OnceLock> = OnceLock::new(); + +#[derive(Default)] +struct Registry { + handles: HashMap>, + paths: HashMap, + opening: HashSet, + cancelled: HashSet, + environments: HashMap>, +} + +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +enum PathIdentity { + #[cfg(unix)] + Unix(u64, u64), + #[cfg(not(unix))] + Path(PathBuf), +} + +struct Runtime { + handle: u32, + path_identity: PathIdentity, + config: EngineConfig, + engine: Arc, + reader: Arc, + writer_queue: Arc>, + search_queue: Arc>, + state: AtomicU8, + environment: Arc, + uncommitted_mutations: AtomicU64, + commit_opstamp: AtomicU64, + writer_queue_nanoseconds: AtomicU64, + writer_execution_nanoseconds: AtomicU64, + search_queue_nanoseconds: AtomicU64, + search_execution_nanoseconds: AtomicU64, + search_threads: Mutex>>, + closed: Arc, +} + +struct CompletionSignal { + done: Mutex, + ready: Condvar, +} + +struct EnvironmentState { + alive: Arc, + handles: Mutex>>, +} + +struct QueueState { + items: VecDeque>, + bytes: usize, + closed: bool, +} + +struct Queued { + value: T, + bytes: usize, + enqueued: Instant, +} + +struct BoundedQueue { + state: Mutex>, + ready: Condvar, + max_commands: usize, + max_bytes: usize, + queued_commands: AtomicU64, + queued_bytes: AtomicU64, +} + +type Callback = ThreadsafeFunction, ErrorStrategy::Fatal>; + +struct Completion { + callback: Option, + env_alive: Arc, +} + +struct WriterCommand { + operation: WriterOperation, + completion: Completion, +} + +enum WriterOperation { + Apply(Vec), + Commit, + Reload, + Close { rollback: bool }, +} + +enum WriterOutcome { + Continue(Result>), + Stop(Result>), + Poison(Result>, FulltextError), +} + +struct SearchCommand { + request: Vec, + completion: Completion, +} + +#[napi(catch_unwind, skip_typescript, js_name = "__nativeOpen")] +pub fn native_open(env: Env, packed_config: Buffer, callback: JsFunction) -> boundary::Result<()> { + boundary::run_stateless(|| { + let environment = environment_state(&env)?; + let opening_done = Arc::new(CompletionSignal::new()); + let completion = completion(callback, environment.alive.clone())?; + let handle = next_handle().map_err(fulltext_napi_error)?; + registry().opening.insert(handle); + environment.track(handle, opening_done.clone()); + let bytes = packed_config.to_vec(); + let thread_opening_done = opening_done.clone(); + let thread_environment = environment.clone(); + if let Err(error) = thread::Builder::new() + .name(format!("fulltext-open-{handle}")) + .spawn(move || open_on_thread(handle, bytes, completion, thread_opening_done, thread_environment)) + { + registry().opening.remove(&handle); + environment.release(handle); + opening_done.signal(); + return Err(napi_error("E_NATIVE_FAILURE", error)); + } + Ok(()) + })? +} + +#[napi(catch_unwind, skip_typescript, js_name = "__nativeApply")] +pub fn native_apply(handle: u32, packed_batch: Buffer, callback: JsFunction) -> boundary::Result<()> { + boundary::run_stateless(|| { + let runtime = runtime(handle)?; + validate_batch_header(&packed_batch, runtime.config.limits.max_batch_bytes).map_err(fulltext_napi_error)?; + runtime + .writer_queue + .check_capacity(packed_batch.len()) + .map_err(fulltext_napi_error)?; + let completion = completion(callback, runtime.environment.alive.clone())?; + let bytes = packed_batch.to_vec(); + runtime.enqueue_writer( + WriterCommand { + operation: WriterOperation::Apply(bytes), + completion, + }, + packed_batch.len(), + ) + })? +} + +#[napi(catch_unwind, skip_typescript, js_name = "__nativeCommit")] +pub fn native_commit(handle: u32, callback: JsFunction) -> boundary::Result<()> { + boundary::run_stateless(|| { + let runtime = runtime(handle)?; + let completion = completion(callback, runtime.environment.alive.clone())?; + runtime.enqueue_writer( + WriterCommand { + operation: WriterOperation::Commit, + completion, + }, + 0, + ) + })? +} + +#[napi(catch_unwind, skip_typescript, js_name = "__nativeReload")] +pub fn native_reload(handle: u32, callback: JsFunction) -> boundary::Result<()> { + boundary::run_stateless(|| { + let runtime = runtime(handle)?; + let completion = completion(callback, runtime.environment.alive.clone())?; + runtime.enqueue_writer( + WriterCommand { + operation: WriterOperation::Reload, + completion, + }, + 0, + ) + })? +} + +#[napi(catch_unwind, skip_typescript, js_name = "__nativeSearch")] +pub fn native_search(handle: u32, packed_request: Buffer, callback: JsFunction) -> boundary::Result<()> { + boundary::run_stateless(|| { + let runtime = runtime(handle)?; + runtime.require_open()?; + validate_search_header(&packed_request).map_err(fulltext_napi_error)?; + runtime + .search_queue + .check_capacity(packed_request.len()) + .map_err(fulltext_napi_error)?; + let completion = completion(callback, runtime.environment.alive.clone())?; + let request = packed_request.to_vec(); + runtime + .search_queue + .try_push(SearchCommand { request, completion }, packed_request.len()) + .map_err(fulltext_napi_error) + })? +} + +#[napi(catch_unwind, skip_typescript, js_name = "__nativeClose")] +pub fn native_close(handle: u32, rollback: bool, callback: JsFunction) -> boundary::Result<()> { + boundary::run_stateless(|| { + let runtime = runtime(handle)?; + let completion = completion(callback, runtime.environment.alive.clone())?; + match runtime + .state + .compare_exchange(STATE_OPEN, STATE_CLOSING, Ordering::AcqRel, Ordering::Acquire) + { + Ok(_) => runtime + .writer_queue + .push_force( + WriterCommand { + operation: WriterOperation::Close { rollback }, + completion, + }, + 0, + ) + .map_err(fulltext_napi_error), + Err(STATE_CLOSED) => { + completion.success(Vec::new()); + Ok(()) + } + Err(STATE_POISONED) => runtime + .writer_queue + .push_force( + WriterCommand { + operation: WriterOperation::Close { rollback: true }, + completion, + }, + 0, + ) + .map_err(fulltext_napi_error), + Err(_) => Err(fulltext_napi_error(FulltextError::new( + "E_CLOSED", + "index is closing or poisoned", + ))), + } + })? +} + +#[cfg(feature = "test-panic")] +#[napi(catch_unwind, skip_typescript, js_name = "__testPoisonNativeHandle")] +pub fn test_poison_native_handle(handle: u32) -> boundary::Result<()> { + boundary::run_stateless(|| { + let runtime = runtime(handle)?; + runtime.poison(FulltextError::new("E_POISONED", "test poison")); + Ok(()) + })? +} + +#[napi(catch_unwind, skip_typescript, js_name = "__nativeStatus")] +pub fn native_status(handle: u32) -> boundary::Result { + boundary::run_stateless(|| { + let runtime = runtime(handle)?; + Ok(Buffer::from(success_envelope(runtime.status_bytes()))) + })? +} + +impl Runtime { + fn start( + handle: u32, + path_identity: PathIdentity, + config: EngineConfig, + engine: Engine, + writer: Writer, + reader: IndexReader, + environment: Arc, + ) -> Result> { + let search_thread_count = config.limits.search_threads; + let writer_queue = Arc::new(BoundedQueue::new( + config.limits.max_queued_commands, + config.limits.max_queued_bytes, + )); + let search_queue = Arc::new(BoundedQueue::new( + config.limits.max_queued_commands, + config.limits.max_queued_bytes, + )); + let runtime = Arc::new(Self { + handle, + path_identity, + config, + engine: Arc::new(engine), + reader: Arc::new(reader), + writer_queue, + search_queue, + state: AtomicU8::new(STATE_OPEN), + environment, + uncommitted_mutations: AtomicU64::new(0), + commit_opstamp: AtomicU64::new(0), + writer_queue_nanoseconds: AtomicU64::new(0), + writer_execution_nanoseconds: AtomicU64::new(0), + search_queue_nanoseconds: AtomicU64::new(0), + search_execution_nanoseconds: AtomicU64::new(0), + search_threads: Mutex::new(Vec::with_capacity(search_thread_count)), + closed: Arc::new(CompletionSignal::new()), + }); + let writer_runtime = runtime.clone(); + thread::Builder::new() + .name(format!("fulltext-writer-{handle}")) + .spawn(move || writer_loop(writer_runtime, writer)) + .map_err(FulltextError::native)?; + for worker in 0..search_thread_count { + let search_runtime = runtime.clone(); + let join = match thread::Builder::new() + .name(format!("fulltext-search-{handle}-{worker}")) + .spawn(move || search_loop(search_runtime)) + { + Ok(join) => join, + Err(error) => { + runtime.writer_queue.close(); + runtime.search_queue.close(); + for join in std::mem::take(&mut *lock(&runtime.search_threads)) { + let _ = join.join(); + } + return Err(FulltextError::native(error)); + } + }; + lock(&runtime.search_threads).push(join); + } + Ok(runtime) + } + + fn require_open(&self) -> boundary::Result<()> { + match self.state.load(Ordering::Acquire) { + STATE_OPEN => Ok(()), + STATE_POISONED => Err(fulltext_napi_error(FulltextError::new( + "E_POISONED", + "index is poisoned", + ))), + _ => Err(fulltext_napi_error(FulltextError::new( + "E_CLOSED", + "index is closing or closed", + ))), + } + } + + fn enqueue_writer(&self, command: WriterCommand, bytes: usize) -> boundary::Result<()> { + self.require_open()?; + self.writer_queue.try_push(command, bytes).map_err(fulltext_napi_error) + } + + fn force_close(&self) { + let previous = self.state.swap(STATE_CLOSING, Ordering::AcqRel); + if previous == STATE_CLOSED || previous == STATE_CLOSING { + return; + } + let _ = self.writer_queue.push_force( + WriterCommand { + operation: WriterOperation::Close { rollback: true }, + completion: Completion { + callback: None, + env_alive: self.environment.alive.clone(), + }, + }, + 0, + ); + } + + fn poison(&self, error: FulltextError) { + self.state.store(STATE_POISONED, Ordering::Release); + for command in self.writer_queue.drain() { + command.value.fail(error.clone()); + } + for command in self.search_queue.close() { + command.value.completion.failure(error.clone()); + } + } + + fn status_bytes(&self) -> Vec { + let mut bytes = Vec::with_capacity(80); + bytes.push(self.state.load(Ordering::Acquire)); + push_u64(&mut bytes, self.uncommitted_mutations.load(Ordering::Acquire)); + push_u64(&mut bytes, self.writer_queue.queued_commands.load(Ordering::Relaxed)); + push_u64(&mut bytes, self.writer_queue.queued_bytes.load(Ordering::Relaxed)); + push_u64(&mut bytes, self.search_queue.queued_commands.load(Ordering::Relaxed)); + push_u64(&mut bytes, self.search_queue.queued_bytes.load(Ordering::Relaxed)); + push_u64(&mut bytes, self.commit_opstamp.load(Ordering::Acquire)); + push_u64(&mut bytes, self.writer_queue_nanoseconds.load(Ordering::Relaxed)); + push_u64(&mut bytes, self.writer_execution_nanoseconds.load(Ordering::Relaxed)); + push_u64(&mut bytes, self.search_queue_nanoseconds.load(Ordering::Relaxed)); + push_u64(&mut bytes, self.search_execution_nanoseconds.load(Ordering::Relaxed)); + bytes + } + + fn signal_closed(&self) { + self.closed.signal(); + } + + fn wait_closed(&self, timeout: Duration) -> bool { + self.closed.wait(timeout) + } +} + +impl CompletionSignal { + fn new() -> Self { + Self { + done: Mutex::new(false), + ready: Condvar::new(), + } + } + + fn signal(&self) { + *lock(&self.done) = true; + self.ready.notify_all(); + } + + fn wait(&self, timeout: Duration) -> bool { + let done = lock(&self.done); + if *done { + return true; + } + let (done, _) = self + .ready + .wait_timeout_while(done, timeout, |done| !*done) + .unwrap_or_else(|error| error.into_inner()); + *done + } +} + +impl BoundedQueue { + fn new(max_commands: usize, max_bytes: usize) -> Self { + Self { + state: Mutex::new(QueueState { + items: VecDeque::new(), + bytes: 0, + closed: false, + }), + ready: Condvar::new(), + max_commands, + max_bytes, + queued_commands: AtomicU64::new(0), + queued_bytes: AtomicU64::new(0), + } + } + + fn try_push(&self, value: T, bytes: usize) -> Result<()> { + let mut state = lock(&self.state); + self.validate_capacity(&state, bytes)?; + state.bytes += bytes; + state.items.push_back(Queued { + value, + bytes, + enqueued: Instant::now(), + }); + self.queued_commands.fetch_add(1, Ordering::Relaxed); + self.queued_bytes.fetch_add(bytes as u64, Ordering::Relaxed); + self.ready.notify_one(); + Ok(()) + } + + fn check_capacity(&self, bytes: usize) -> Result<()> { + self.validate_capacity(&lock(&self.state), bytes) + } + + fn validate_capacity(&self, state: &QueueState, bytes: usize) -> Result<()> { + if state.closed { + return Err(FulltextError::new("E_CLOSED", "operation queue is closed")); + } + if state.items.len() >= self.max_commands || state.bytes.saturating_add(bytes) > self.max_bytes { + return Err(FulltextError::new( + "E_QUEUE_FULL", + "operation queue limits are exhausted", + )); + } + Ok(()) + } + + fn push_force(&self, value: T, bytes: usize) -> Result<()> { + let mut state = lock(&self.state); + if state.closed { + return Err(FulltextError::new("E_CLOSED", "operation queue is closed")); + } + state.bytes += bytes; + state.items.push_back(Queued { + value, + bytes, + enqueued: Instant::now(), + }); + self.queued_commands.fetch_add(1, Ordering::Relaxed); + self.queued_bytes.fetch_add(bytes as u64, Ordering::Relaxed); + self.ready.notify_one(); + Ok(()) + } + + fn pop(&self) -> Option> { + let mut state = lock(&self.state); + loop { + if let Some(item) = state.items.pop_front() { + state.bytes -= item.bytes; + self.queued_commands.fetch_sub(1, Ordering::Relaxed); + self.queued_bytes.fetch_sub(item.bytes as u64, Ordering::Relaxed); + return Some(item); + } + if state.closed { + return None; + } + state = self.ready.wait(state).unwrap_or_else(|error| error.into_inner()); + } + } + + fn close(&self) -> Vec> { + let mut state = lock(&self.state); + state.closed = true; + let items = drain_queue(&mut state, &self.queued_commands, &self.queued_bytes); + self.ready.notify_all(); + items + } + + fn drain(&self) -> Vec> { + let mut state = lock(&self.state); + drain_queue(&mut state, &self.queued_commands, &self.queued_bytes) + } + + fn shutdown_after_drain(&self) { + let mut state = lock(&self.state); + state.closed = true; + self.ready.notify_all(); + } +} + +fn drain_queue(state: &mut QueueState, commands: &AtomicU64, bytes: &AtomicU64) -> Vec> { + let items = state.items.drain(..).collect::>(); + state.bytes = 0; + commands.store(0, Ordering::Relaxed); + bytes.store(0, Ordering::Relaxed); + items +} + +impl Completion { + fn success(mut self, body: Vec) { + self.send(success_envelope(body)); + } + + fn failure(mut self, error: FulltextError) { + self.send(error_envelope(error)); + } + + fn send(&mut self, bytes: Vec) { + if self.env_alive.load(Ordering::Acquire) { + if let Some(callback) = self.callback.take() { + let _ = callback.call(bytes, ThreadsafeFunctionCallMode::NonBlocking); + } + } else if let Some(callback) = self.callback.take() { + mem::forget(callback); + } + } +} + +impl Drop for Completion { + fn drop(&mut self) { + if self.callback.is_some() { + self.send(error_envelope(FulltextError::new( + "E_CLOSED", + "native operation ended before completion", + ))); + } + } +} + +impl WriterCommand { + fn fail(self, error: FulltextError) { + self.completion.failure(error); + } +} + +fn writer_loop(runtime: Arc, writer: Writer) { + let mut writer = Some(writer); + while let Some(queued) = runtime.writer_queue.pop() { + runtime + .writer_queue_nanoseconds + .fetch_add(duration_ns(queued.enqueued.elapsed()), Ordering::Relaxed); + let WriterCommand { operation, completion } = queued.value; + let started = Instant::now(); + let outcome = catch_unwind(AssertUnwindSafe(|| match operation { + WriterOperation::Apply(bytes) => { + match decode_batch(&bytes).and_then(|batch| active_writer(&writer)?.prepare(batch)) { + Ok(prepared) => match active_writer(&writer).and_then(|writer| writer.apply_prepared(prepared)) { + Ok(count) => { + runtime.uncommitted_mutations.fetch_add(count, Ordering::AcqRel); + WriterOutcome::Continue(Ok(u64_body(count))) + } + Err(error) => WriterOutcome::Poison( + Err(error), + FulltextError::new( + "E_POISONED", + "a mutation failed after writer state changed and the index generation is terminal", + ), + ), + }, + Err(error) => WriterOutcome::Continue(Err(error)), + } + } + WriterOperation::Commit => match active_writer_mut(&mut writer).and_then(Writer::commit) { + Ok(opstamp) => { + runtime.uncommitted_mutations.store(0, Ordering::Release); + runtime.commit_opstamp.store(opstamp, Ordering::Release); + WriterOutcome::Continue(Ok(u64_body(opstamp))) + } + Err(error) => WriterOutcome::Poison( + Err(error), + FulltextError::new( + "E_POISONED", + "a prior commit failed and the index generation is terminal", + ), + ), + }, + WriterOperation::Reload => WriterOutcome::Continue( + runtime + .reader + .reload() + .map(|()| Vec::new()) + .map_err(FulltextError::native), + ), + WriterOperation::Close { rollback } => { + if !rollback && runtime.uncommitted_mutations.load(Ordering::Acquire) > 0 { + runtime.state.store(STATE_OPEN, Ordering::Release); + WriterOutcome::Continue(Err(FulltextError::new( + "E_DIRTY_CLOSE", + "index has uncommitted mutations; commit or close with rollback", + ))) + } else { + let close_result = writer + .take() + .ok_or_else(|| FulltextError::new("E_POISONED", "writer is unavailable")) + .and_then(|mut owned_writer| { + if rollback { + owned_writer.rollback()?; + } + owned_writer.close() + }); + runtime.search_queue.shutdown_after_drain(); + for join in std::mem::take(&mut *lock(&runtime.search_threads)) { + let _ = join.join(); + } + runtime.writer_queue.close(); + runtime.state.store(STATE_CLOSED, Ordering::Release); + release_runtime(runtime.handle, &runtime.path_identity, &runtime.environment); + runtime.signal_closed(); + WriterOutcome::Stop(close_result.map(|()| Vec::new())) + } + } + })); + runtime + .writer_execution_nanoseconds + .fetch_add(duration_ns(started.elapsed()), Ordering::Relaxed); + match outcome { + Ok(WriterOutcome::Continue(result)) => settle(completion, result), + Ok(WriterOutcome::Stop(result)) => { + settle(completion, result); + return; + } + Ok(WriterOutcome::Poison(result, poison)) => { + settle(completion, result); + runtime.poison(poison); + } + Err(_) => { + completion.failure(FulltextError::new("E_NATIVE_PANIC", "native writer actor panicked")); + runtime.poison(FulltextError::new("E_NATIVE_PANIC", "native writer actor panicked")); + runtime.writer_queue.close(); + release_runtime(runtime.handle, &runtime.path_identity, &runtime.environment); + runtime.signal_closed(); + return; + } + } + } + runtime.state.store(STATE_CLOSED, Ordering::Release); + release_runtime(runtime.handle, &runtime.path_identity, &runtime.environment); + runtime.signal_closed(); +} + +fn settle(completion: Completion, result: Result>) { + match result { + Ok(body) => completion.success(body), + Err(error) => completion.failure(error), + } +} + +fn active_writer(writer: &Option) -> Result<&Writer> { + writer + .as_ref() + .ok_or_else(|| FulltextError::new("E_POISONED", "writer is unavailable")) +} + +fn active_writer_mut(writer: &mut Option) -> Result<&mut Writer> { + writer + .as_mut() + .ok_or_else(|| FulltextError::new("E_POISONED", "writer is unavailable")) +} + +fn search_loop(runtime: Arc) { + while let Some(queued) = runtime.search_queue.pop() { + runtime + .search_queue_nanoseconds + .fetch_add(duration_ns(queued.enqueued.elapsed()), Ordering::Relaxed); + let started = Instant::now(); + let result = catch_unwind(AssertUnwindSafe(|| { + decode_search(&queued.value.request) + .and_then(|request| runtime.engine.search(&runtime.reader.searcher(), &request)) + })); + runtime + .search_execution_nanoseconds + .fetch_add(duration_ns(started.elapsed()), Ordering::Relaxed); + match result { + Ok(Ok(result)) => queued.value.completion.success(search_body(result)), + Ok(Err(error)) => queued.value.completion.failure(error), + Err(_) => { + queued + .value + .completion + .failure(FulltextError::new("E_NATIVE_PANIC", "native search actor panicked")); + runtime.poison(FulltextError::new("E_NATIVE_PANIC", "native search actor panicked")); + return; + } + } + } +} + +fn open_on_thread( + handle: u32, + bytes: Vec, + completion: Completion, + opening_done: Arc, + environment: Arc, +) { + let result = catch_unwind(AssertUnwindSafe(|| open_runtime(handle, bytes, environment.clone()))); + let opened = matches!(result, Ok(Ok(()))); + match result { + Ok(Ok(())) => completion.success(u32_body(handle)), + Ok(Err(error)) => completion.failure(error), + Err(_) => completion.failure(FulltextError::new("E_NATIVE_PANIC", "native index open panicked")), + } + registry().opening.remove(&handle); + if !opened { + environment.release(handle); + } + opening_done.signal(); +} + +fn open_runtime(handle: u32, bytes: Vec, environment: Arc) -> Result<()> { + let config = decode_open(&bytes)?; + let canonical = create_and_canonicalize(Path::new(&config.path))?; + let path_identity = path_identity(&canonical)?; + { + let mut registry = registry(); + if registry.cancelled.remove(&handle) || !environment.alive.load(Ordering::Acquire) { + registry.opening.remove(&handle); + return Err(FulltextError::new( + "E_CLOSED", + "Node environment closed during index open", + )); + } + if registry.paths.contains_key(&path_identity) { + return Err(FulltextError::new( + "E_DUPLICATE_OPEN", + "the physical index is already open", + )); + } + registry.paths.insert(path_identity.clone(), handle); + } + let result = (|| { + let directory = MmapDirectory::open(&canonical).map_err(storage_error)?; + let engine = Engine::open(directory, &config)?; + let writer = engine.writer(&config)?; + let reader = engine.reader()?; + let runtime = Runtime::start( + handle, + path_identity.clone(), + config, + engine, + writer, + reader, + environment.clone(), + )?; + let mut registry = registry(); + if registry.cancelled.remove(&handle) || !environment.alive.load(Ordering::Acquire) { + drop(registry); + runtime.force_close(); + let _ = runtime.wait_closed(CLEANUP_TIMEOUT); + return Err(FulltextError::new( + "E_CLOSED", + "Node environment closed during index open", + )); + } + registry.handles.insert(handle, runtime); + registry.opening.remove(&handle); + Ok(()) + })(); + if result.is_err() { + release_runtime(handle, &path_identity, &environment); + } + result +} + +fn completion(callback: JsFunction, env_alive: Arc) -> boundary::Result { + let callback = callback + .create_threadsafe_function::, Buffer, _, ErrorStrategy::Fatal>( + 0, + |context: ThreadSafeCallContext>| Ok(vec![Buffer::from(context.value)]), + ) + .map_err(|error| napi_error("E_NATIVE_FAILURE", error))?; + Ok(Completion { + callback: Some(callback), + env_alive, + }) +} + +fn runtime(handle: u32) -> boundary::Result> { + registry() + .handles + .get(&handle) + .cloned() + .ok_or_else(|| napi_error("E_CLOSED", "unknown or closed fulltext index handle")) +} + +fn cleanup_handle(handle: u32) -> Option> { + let runtime = { + let mut registry = registry(); + match registry.handles.get(&handle).cloned() { + Some(runtime) => Some(runtime), + None if registry.opening.remove(&handle) => { + registry.cancelled.insert(handle); + None + } + None => None, + } + }; + if let Some(runtime) = runtime { + runtime.force_close(); + Some(runtime) + } else { + None + } +} + +struct EnvironmentHookData { + key: usize, + environment: Arc, +} + +enum CleanupWait { + Runtime(Arc), + Opening(Arc), +} + +fn environment_state(env: &Env) -> boundary::Result> { + let key = env.raw() as usize; + if let Some(environment) = registry().environments.get(&key).and_then(Weak::upgrade) { + return Ok(environment); + } + let environment = Arc::new(EnvironmentState { + alive: Arc::new(AtomicBool::new(true)), + handles: Mutex::new(HashMap::new()), + }); + env.add_async_cleanup_hook( + EnvironmentHookData { + key, + environment: environment.clone(), + }, + |data| { + let _ = catch_unwind(AssertUnwindSafe(|| finish_environment_cleanup(data))); + }, + ) + .map_err(|error| napi_error("E_NATIVE_FAILURE", error))?; + registry().environments.insert(key, Arc::downgrade(&environment)); + Ok(environment) +} + +fn finish_environment_cleanup(data: EnvironmentHookData) { + data.environment.alive.store(false, Ordering::Release); + let tracked = data.environment.take_handles(); + let waits = tracked + .into_iter() + .map(|(handle, opening_done)| match cleanup_handle(handle) { + Some(runtime) => CleanupWait::Runtime(runtime), + None => CleanupWait::Opening(opening_done), + }) + .collect::>(); + let remove_environment = registry() + .environments + .get(&data.key) + .and_then(Weak::upgrade) + .is_some_and(|environment| Arc::ptr_eq(&environment, &data.environment)); + if remove_environment { + registry().environments.remove(&data.key); + } + let deadline = Instant::now() + CLEANUP_TIMEOUT; + let finished = waits.into_iter().all(|wait| { + let remaining = deadline.saturating_duration_since(Instant::now()); + !remaining.is_zero() && wait.wait(remaining) + }); + if !finished { + eprintln!("fulltext native cleanup exceeded {} seconds", CLEANUP_TIMEOUT.as_secs()); + } +} + +impl EnvironmentState { + fn track(&self, handle: u32, opening_done: Arc) { + lock(&self.handles).insert(handle, opening_done); + } + + fn release(&self, handle: u32) { + lock(&self.handles).remove(&handle); + } + + fn take_handles(&self) -> HashMap> { + mem::take(&mut *lock(&self.handles)) + } +} + +impl CleanupWait { + fn wait(&self, timeout: Duration) -> bool { + match self { + Self::Runtime(runtime) => runtime.wait_closed(timeout), + Self::Opening(signal) => signal.wait(timeout), + } + } +} + +fn release_runtime(handle: u32, identity: &PathIdentity, environment: &EnvironmentState) { + let mut registry = registry(); + registry.handles.remove(&handle); + if registry.paths.get(identity) == Some(&handle) { + registry.paths.remove(identity); + } + drop(registry); + environment.release(handle); +} + +fn registry() -> std::sync::MutexGuard<'static, Registry> { + REGISTRY + .get_or_init(Default::default) + .lock() + .unwrap_or_else(|error| error.into_inner()) +} + +fn create_and_canonicalize(path: &Path) -> Result { + fs::create_dir_all(path).map_err(storage_error)?; + fs::canonicalize(path).map_err(storage_error) +} + +#[cfg(unix)] +fn path_identity(path: &Path) -> Result { + use std::os::unix::fs::MetadataExt; + let metadata = fs::metadata(path).map_err(storage_error)?; + Ok(PathIdentity::Unix(metadata.dev(), metadata.ino())) +} + +#[cfg(windows)] +fn path_identity(path: &Path) -> Result { + Ok(PathIdentity::Path(PathBuf::from(path.to_string_lossy().to_lowercase()))) +} + +#[cfg(all(not(unix), not(windows)))] +fn path_identity(path: &Path) -> Result { + Ok(PathIdentity::Path(path.to_path_buf())) +} + +fn next_handle() -> Result { + let handle = NEXT_HANDLE.fetch_add(1, Ordering::Relaxed); + if handle == 0 { + Err(FulltextError::new("E_NATIVE_FAILURE", "native handle space exhausted")) + } else { + Ok(handle) + } +} + +fn success_envelope(body: Vec) -> Vec { + let mut bytes = b"FTRP\x01\x00\x00".to_vec(); + bytes.extend_from_slice(&body); + bytes +} + +fn error_envelope(error: FulltextError) -> Vec { + let mut bytes = b"FTRP\x01\x00\x01".to_vec(); + push_string(&mut bytes, error.code); + push_string(&mut bytes, &error.message); + bytes +} + +fn u32_body(value: u32) -> Vec { + value.to_le_bytes().to_vec() +} + +fn u64_body(value: u64) -> Vec { + value.to_le_bytes().to_vec() +} + +fn search_body(result: SearchResult) -> Vec { + let mut bytes = Vec::new(); + push_u64(&mut bytes, result.total); + bytes.push(match result.total_relation { + TotalRelation::Exact => 0, + TotalRelation::LowerBound => 1, + }); + bytes.extend_from_slice(&(result.hits.len() as u32).to_le_bytes()); + for hit in result.hits { + bytes.extend_from_slice(&hit.score.to_le_bytes()); + push_string(&mut bytes, &hit.id); + } + bytes +} + +fn push_string(bytes: &mut Vec, value: &str) { + bytes.extend_from_slice(&(value.len() as u32).to_le_bytes()); + bytes.extend_from_slice(value.as_bytes()); +} + +fn push_u64(bytes: &mut Vec, value: u64) { + bytes.extend_from_slice(&value.to_le_bytes()); +} + +fn duration_ns(duration: std::time::Duration) -> u64 { + u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX) +} + +fn lock(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(|error| error.into_inner()) +} + +fn fulltext_napi_error(error: FulltextError) -> napi::Error<&'static str> { + napi::Error::new(error.code, error.message) +} + +fn napi_error(code: &'static str, error: impl std::fmt::Display) -> napi::Error<&'static str> { + napi::Error::new(code, error.to_string()) +} + +fn storage_error(error: impl std::fmt::Display) -> FulltextError { + FulltextError::new("E_STORAGE", error.to_string()) +} diff --git a/src/protocol.rs b/src/protocol.rs new file mode 100644 index 0000000..38f0efe --- /dev/null +++ b/src/protocol.rs @@ -0,0 +1,390 @@ +use crate::error::{FulltextError, Result}; + +pub const PROTOCOL_VERSION: u16 = 1; +const MAX_STRING_BYTES: usize = 1 << 20; +const MAX_FIELDS: usize = 1_024; + +#[derive(Clone, Debug, PartialEq)] +pub struct FieldConfig { + pub name: String, + pub weight: f32, +} + +#[derive(Clone, Debug, PartialEq)] +pub struct Limits { + pub indexing_threads: usize, + pub search_threads: usize, + pub writer_memory_bytes: usize, + pub max_queued_commands: usize, + pub max_queued_bytes: usize, + pub max_batch_bytes: usize, +} + +#[derive(Clone, Debug, PartialEq)] +pub struct EngineConfig { + pub path: String, + pub index_id: String, + pub generation: String, + pub fields: Vec, + pub analyzer: String, + pub stop_words: bool, + pub positions: bool, + pub surface_terms: bool, + pub limits: Limits, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Upsert { + pub id: String, + pub fields: Vec<(String, Vec)>, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct MutationBatch { + pub upserts: Vec, + pub deletes: Vec, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum SearchOperator { + Any, + All, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SearchRequest { + pub text: String, + pub operator: SearchOperator, + pub fields: Vec, + pub offset: usize, + pub limit: usize, + pub exact_total: bool, +} + +pub fn decode_open(bytes: &[u8]) -> Result { + let mut cursor = Cursor::new(bytes, *b"FTOP")?; + let path = cursor.string()?; + let index_id = cursor.string()?; + let generation = cursor.string()?; + let analyzer = cursor.string()?; + let stop_words = cursor.boolean()?; + let positions = cursor.boolean()?; + let surface_terms = cursor.boolean()?; + let field_count = cursor.u16()? as usize; + if field_count == 0 || field_count > MAX_FIELDS { + return Err(FulltextError::invalid("fields must contain between 1 and 1024 entries")); + } + let mut fields = Vec::with_capacity(field_count); + for _ in 0..field_count { + let name = cursor.string()?; + let weight = cursor.f32()?; + if !weight.is_finite() || weight <= 0.0 { + return Err(FulltextError::invalid( + "field weights must be finite and greater than zero", + )); + } + fields.push(FieldConfig { name, weight }); + } + let limits = Limits { + indexing_threads: cursor.u16()? as usize, + search_threads: cursor.u16()? as usize, + writer_memory_bytes: cursor.u64_usize()?, + max_queued_commands: cursor.u32()? as usize, + max_queued_bytes: cursor.u64_usize()?, + max_batch_bytes: cursor.u64_usize()?, + }; + cursor.finish()?; + validate_config(EngineConfig { + path, + index_id, + generation, + fields, + analyzer, + stop_words, + positions, + surface_terms, + limits, + }) +} + +pub fn validate_batch_header(bytes: &[u8], max_batch_bytes: usize) -> Result<()> { + if bytes.len() > max_batch_bytes { + return Err(FulltextError::new( + "E_INVALID_ARGUMENT", + format!("mutation batch is {} bytes; maximum is {max_batch_bytes}", bytes.len()), + )); + } + let mut cursor = Cursor::new(bytes, *b"FTMB")?; + let _ = cursor.u32()?; + let _ = cursor.u32()?; + Ok(()) +} + +pub fn validate_search_header(bytes: &[u8]) -> Result<()> { + let _ = Cursor::new(bytes, *b"FTSQ")?; + Ok(()) +} + +pub fn decode_batch(bytes: &[u8]) -> Result { + let mut cursor = Cursor::new(bytes, *b"FTMB")?; + let upsert_count = cursor.u32()? as usize; + let delete_count = cursor.u32()? as usize; + let minimum_bytes = upsert_count + .checked_mul(6) + .and_then(|bytes| { + delete_count + .checked_mul(4) + .and_then(|deletes| bytes.checked_add(deletes)) + }) + .ok_or_else(|| FulltextError::invalid("mutation count overflow"))?; + if minimum_bytes > cursor.remaining() { + return Err(FulltextError::invalid("mutation counts exceed the packed batch length")); + } + let mut upserts = Vec::with_capacity(upsert_count); + for _ in 0..upsert_count { + let id = cursor.string()?; + let field_count = cursor.u16()? as usize; + if field_count > MAX_FIELDS { + return Err(FulltextError::invalid("upsert field count exceeds 1024")); + } + if field_count > cursor.remaining() / 6 { + return Err(FulltextError::invalid( + "upsert field count exceeds the packed batch length", + )); + } + let mut fields = Vec::with_capacity(field_count); + for _ in 0..field_count { + let name = cursor.string()?; + let value_count = cursor.u16()? as usize; + if value_count > cursor.remaining() / 4 { + return Err(FulltextError::invalid("value count exceeds the packed batch length")); + } + let mut values = Vec::with_capacity(value_count); + for _ in 0..value_count { + values.push(cursor.string()?); + } + fields.push((name, values)); + } + upserts.push(Upsert { id, fields }); + } + if delete_count > cursor.remaining() / 4 { + return Err(FulltextError::invalid("delete count exceeds the packed batch length")); + } + let mut deletes = Vec::with_capacity(delete_count); + for _ in 0..delete_count { + deletes.push(cursor.string()?); + } + cursor.finish()?; + Ok(MutationBatch { upserts, deletes }) +} + +pub fn decode_search(bytes: &[u8]) -> Result { + let mut cursor = Cursor::new(bytes, *b"FTSQ")?; + let text = cursor.string()?; + let operator = match cursor.u8()? { + 0 => SearchOperator::Any, + 1 => SearchOperator::All, + _ => return Err(FulltextError::invalid("unknown search operator")), + }; + let field_count = cursor.u16()? as usize; + if field_count > MAX_FIELDS { + return Err(FulltextError::invalid("search field count exceeds 1024")); + } + let mut fields = Vec::with_capacity(field_count); + for _ in 0..field_count { + fields.push(cursor.string()?); + } + let offset = cursor.u32()? as usize; + let limit = cursor.u32()? as usize; + let exact_total = cursor.boolean()?; + cursor.finish()?; + if text.trim().is_empty() { + return Err(FulltextError::invalid("search text must not be empty")); + } + if limit == 0 || offset.saturating_add(limit) > 10_000 { + return Err(FulltextError::invalid("search window must be between 1 and 10000")); + } + Ok(SearchRequest { + text, + operator, + fields, + offset, + limit, + exact_total, + }) +} + +fn validate_config(config: EngineConfig) -> Result { + if config.path.is_empty() || config.index_id.is_empty() || config.generation.is_empty() { + return Err(FulltextError::invalid( + "path, indexId, and generation must not be empty", + )); + } + if config.index_id.len() > 4_096 || config.generation.len() > 4_096 { + return Err(FulltextError::invalid( + "indexId and generation must not exceed 4096 UTF-8 bytes", + )); + } + if config.analyzer != "english@1" { + return Err(FulltextError::invalid("only analyzer english@1 is supported")); + } + let mut names = std::collections::HashSet::with_capacity(config.fields.len()); + for field in &config.fields { + if field.name.is_empty() || field.name == "__fulltext_id" || !names.insert(field.name.as_str()) { + return Err(FulltextError::invalid( + "field names must be non-empty, unique, and not reserved", + )); + } + } + let limits = &config.limits; + if limits.indexing_threads == 0 || limits.search_threads == 0 || limits.max_queued_commands == 0 { + return Err(FulltextError::invalid( + "thread and queue command limits must be greater than zero", + )); + } + if limits.indexing_threads > 64 || limits.search_threads > 64 { + return Err(FulltextError::invalid("thread limits must not exceed 64")); + } + let per_thread = limits.writer_memory_bytes / limits.indexing_threads; + if !(15_000_000..u32::MAX as usize).contains(&per_thread) { + return Err(FulltextError::invalid(format!( + "writerMemoryBytes/indexingThreads is {per_thread}; Tantivy requires 15000000..{}", + u32::MAX + ))); + } + if limits.max_batch_bytes == 0 || limits.max_batch_bytes > limits.max_queued_bytes { + return Err(FulltextError::invalid( + "maxBatchBytes must be greater than zero and no larger than maxQueuedBytes", + )); + } + Ok(config) +} + +struct Cursor<'a> { + bytes: &'a [u8], + offset: usize, +} + +impl<'a> Cursor<'a> { + fn new(bytes: &'a [u8], magic: [u8; 4]) -> Result { + if bytes.len() < 6 || bytes[..4] != magic { + return Err(FulltextError::invalid("invalid packed request magic")); + } + let version = u16::from_le_bytes([bytes[4], bytes[5]]); + if version != PROTOCOL_VERSION { + return Err(FulltextError::invalid(format!( + "unsupported packed request version {version}" + ))); + } + Ok(Self { bytes, offset: 6 }) + } + + fn take(&mut self, length: usize) -> Result<&'a [u8]> { + let end = self + .offset + .checked_add(length) + .ok_or_else(|| FulltextError::invalid("packed request length overflow"))?; + let value = self + .bytes + .get(self.offset..end) + .ok_or_else(|| FulltextError::invalid("packed request is truncated"))?; + self.offset = end; + Ok(value) + } + + fn u8(&mut self) -> Result { + Ok(self.take(1)?[0]) + } + + fn boolean(&mut self) -> Result { + match self.u8()? { + 0 => Ok(false), + 1 => Ok(true), + _ => Err(FulltextError::invalid("packed boolean must be zero or one")), + } + } + + fn u16(&mut self) -> Result { + let bytes = self.take(2)?; + Ok(u16::from_le_bytes([bytes[0], bytes[1]])) + } + + fn u32(&mut self) -> Result { + let bytes = self.take(4)?; + Ok(u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]])) + } + + fn u64_usize(&mut self) -> Result { + let bytes = self.take(8)?; + let value = u64::from_le_bytes([ + bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7], + ]); + usize::try_from(value).map_err(|_| FulltextError::invalid("numeric limit exceeds usize")) + } + + fn f32(&mut self) -> Result { + let bytes = self.take(4)?; + Ok(f32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]])) + } + + fn string(&mut self) -> Result { + let length = self.u32()? as usize; + if length > MAX_STRING_BYTES { + return Err(FulltextError::invalid("packed string exceeds 1 MiB")); + } + let bytes = self.take(length)?; + String::from_utf8(bytes.to_vec()).map_err(|_| FulltextError::invalid("packed string is not valid UTF-8")) + } + + fn finish(&self) -> Result<()> { + if self.offset == self.bytes.len() { + Ok(()) + } else { + Err(FulltextError::invalid("packed request has trailing bytes")) + } + } + + fn remaining(&self) -> usize { + self.bytes.len() - self.offset + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rejects_counts_before_allocating() { + let mut bytes = b"FTMB\x01\x00".to_vec(); + bytes.extend_from_slice(&u32::MAX.to_le_bytes()); + bytes.extend_from_slice(&0u32.to_le_bytes()); + assert_eq!(decode_batch(&bytes).unwrap_err().code, "E_INVALID_ARGUMENT"); + } + + #[test] + fn rejects_invalid_utf8() { + let mut bytes = b"FTSQ\x01\x00".to_vec(); + bytes.extend_from_slice(&1u32.to_le_bytes()); + bytes.push(0xff); + assert_eq!(decode_search(&bytes).unwrap_err().code, "E_INVALID_ARGUMENT"); + } + + #[test] + fn rejects_nested_counts_before_allocating() { + let mut fields = b"FTMB\x01\x00".to_vec(); + fields.extend_from_slice(&1u32.to_le_bytes()); + fields.extend_from_slice(&0u32.to_le_bytes()); + fields.extend_from_slice(&0u32.to_le_bytes()); + fields.extend_from_slice(&u16::MAX.to_le_bytes()); + assert_eq!(decode_batch(&fields).unwrap_err().code, "E_INVALID_ARGUMENT"); + + let mut values = b"FTMB\x01\x00".to_vec(); + values.extend_from_slice(&1u32.to_le_bytes()); + values.extend_from_slice(&0u32.to_le_bytes()); + values.extend_from_slice(&0u32.to_le_bytes()); + values.extend_from_slice(&1u16.to_le_bytes()); + values.extend_from_slice(&0u32.to_le_bytes()); + values.extend_from_slice(&u16::MAX.to_le_bytes()); + assert_eq!(decode_batch(&values).unwrap_err().code, "E_INVALID_ARGUMENT"); + } +} diff --git a/test/error-codes.test.mjs b/test/error-codes.test.mjs new file mode 100644 index 0000000..afdab19 --- /dev/null +++ b/test/error-codes.test.mjs @@ -0,0 +1,23 @@ +import assert from 'node:assert'; +import { readFileSync } from 'node:fs'; +import test from 'node:test'; + +test('keeps Rust and TypeScript error codes in parity', () => { + const rust = readFileSync(new URL('../src/error.rs', import.meta.url), 'utf8'); + const typescript = readFileSync(new URL('../ts/errors.ts', import.meta.url), 'utf8'); + const rustCodes = codesIn(constantBlock(rust, 'pub const ERROR_CODES', '];')); + const typescriptCodes = codesIn(constantBlock(typescript, 'const errorCodes', '] as const')); + assert.deepStrictEqual(rustCodes, typescriptCodes); +}); + +function codesIn(source) { + return [...source.matchAll(/["'](E_[A-Z_]+)["']/g)].map((match) => match[1]).sort(); +} + +function constantBlock(source, marker, terminator) { + const start = source.indexOf(marker); + assert.notStrictEqual(start, -1); + const end = source.indexOf(terminator, start); + assert.notStrictEqual(end, -1); + return source.slice(start, end); +} diff --git a/test/fixtures/native-crash-child.mjs b/test/fixtures/native-crash-child.mjs new file mode 100644 index 0000000..ba21c11 --- /dev/null +++ b/test/fixtures/native-crash-child.mjs @@ -0,0 +1,23 @@ +import { encodeMutationBatch, openNativeFullTextIndex } from '../../dist/native.js'; + +const [indexPath, mode] = process.argv.slice(2); +const index = await openNativeFullTextIndex({ + path: indexPath, + indexId: 'crash-products', + generation: 'generation-1', + fields: [{ name: 'title' }], + analyzer: 'english@1', + limits: { + indexingThreads: 1, + searchThreads: 1, + writerMemoryBytes: 15_000_000, + maxQueuedCommands: 8, + maxQueuedBytes: 1024 * 1024, + maxBatchBytes: 1024 * 1024, + }, +}); +await index.apply(encodeMutationBatch({ upserts: [{ id: 'product-1', fields: { title: 'durable product' } }] })); +if (mode === 'committed') { + await index.commit(); +} +process.exit(17); diff --git a/test/fixtures/native-worker-child.mjs b/test/fixtures/native-worker-child.mjs new file mode 100644 index 0000000..8d06dad --- /dev/null +++ b/test/fixtures/native-worker-child.mjs @@ -0,0 +1,45 @@ +import { parentPort, workerData } from 'node:worker_threads'; + +const { encodeMutationBatch, openNativeFullTextIndex } = await import(workerData.moduleUrl); +const openIndex = (indexPath) => + openNativeFullTextIndex({ + path: indexPath, + indexId: 'worker-products', + generation: 'generation-1', + fields: [{ name: 'title' }], + analyzer: 'english@1', + limits: { + indexingThreads: 2, + searchThreads: 2, + writerMemoryBytes: 30_000_000, + maxQueuedCommands: 8, + maxQueuedBytes: 16 * 1024 * 1024, + maxBatchBytes: 16 * 1024 * 1024, + }, + }); +if (workerData.mode === 'multiple') { + await Promise.all(workerData.indexPaths.map(openIndex)); + parentPort.postMessage('multiple-open'); + await stayAlive(); +} +const opening = openIndex(workerData.indexPath); +if (workerData.mode === 'opening') { + parentPort.postMessage('opening'); +} +const index = await opening; +if (workerData.mode === 'opening') { + parentPort.postMessage('opened'); + await stayAlive(); +} +const upserts = Array.from({ length: 50_000 }, (_, id) => ({ + id: String(id), + fields: { title: `worker-owned running product ${id}` }, +})); +const packed = encodeMutationBatch({ upserts }, 16 * 1024 * 1024); +parentPort.postMessage('applying'); +await index.apply(packed); +parentPort.postMessage('finished'); + +function stayAlive() { + return new Promise(() => setInterval(() => {}, 1_000)); +} diff --git a/test/native-crash.test.mjs b/test/native-crash.test.mjs new file mode 100644 index 0000000..0014c4b --- /dev/null +++ b/test/native-crash.test.mjs @@ -0,0 +1,42 @@ +import assert from 'node:assert'; +import { spawnSync } from 'node:child_process'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { fileURLToPath } from 'node:url'; + +import { openNativeFullTextIndex } from '@harperfast/fulltext/native'; + +for (const [mode, expected] of [ + ['committed', 1], + ['uncommitted', 0], +]) { + test(`reopens ${mode} state after abrupt process exit`, async (context) => { + const indexPath = mkdtempSync(path.join(tmpdir(), `harper-fulltext-${mode}-`)); + context.after(() => rmSync(indexPath, { recursive: true, force: true })); + const child = spawnSync( + process.execPath, + [fileURLToPath(new URL('./fixtures/native-crash-child.mjs', import.meta.url)), indexPath, mode], + { encoding: 'utf8', timeout: 30_000 }, + ); + assert.strictEqual(child.status, 17, child.stderr); + const index = await openNativeFullTextIndex({ + path: indexPath, + indexId: 'crash-products', + generation: 'generation-1', + fields: [{ name: 'title' }], + analyzer: 'english@1', + limits: { + indexingThreads: 1, + searchThreads: 1, + writerMemoryBytes: 15_000_000, + maxQueuedCommands: 8, + maxQueuedBytes: 1024 * 1024, + maxBatchBytes: 1024 * 1024, + }, + }); + assert.strictEqual((await index.search({ text: 'durable product', exactTotal: true })).total, expected); + await index.close(); + }); +} diff --git a/test/native-index.test.mjs b/test/native-index.test.mjs new file mode 100644 index 0000000..9029767 --- /dev/null +++ b/test/native-index.test.mjs @@ -0,0 +1,196 @@ +import assert from 'node:assert'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; + +import { encodeMutationBatch, openNativeFullTextIndex } from '@harperfast/fulltext/native'; + +function options(indexPath, overrides = {}) { + return { + path: indexPath, + indexId: 'products', + generation: 'generation-1', + fields: [{ name: 'title', weight: 3 }, { name: 'description' }], + analyzer: 'english@1', + limits: { + indexingThreads: 1, + searchThreads: 2, + writerMemoryBytes: 15_000_000, + maxQueuedCommands: 32, + maxQueuedBytes: 8 * 1024 * 1024, + maxBatchBytes: 8 * 1024 * 1024, + }, + ...overrides, + }; +} + +function temporaryIndex(context) { + const directory = mkdtempSync(path.join(tmpdir(), 'harper-fulltext-index-')); + context.after(() => rmSync(directory, { recursive: true, force: true })); + return directory; +} + +test('runs the public create, mutate, BM25 search, close, and reopen route', async (context) => { + const indexPath = temporaryIndex(context); + const config = options(indexPath); + let index = await openNativeFullTextIndex(config); + assert.strictEqual( + await index.apply( + encodeMutationBatch({ + upserts: [ + { + id: 'shoe-1', + fields: { title: 'Trail Running Shoes', description: 'red outdoor footwear' }, + }, + { id: 'rack-1', fields: { title: 'Wood Rack', description: 'shoe organizer' } }, + ], + }), + ), + 2, + ); + assert.strictEqual(index.status().uncommittedMutations, 2n); + await index.commit(); + await index.reload(); + const approximate = await index.search({ text: 'running shoes', limit: 1 }); + assert.strictEqual(approximate.total, 1); + assert.strictEqual(approximate.totalRelation, 'lower-bound'); + const result = await index.search({ text: 'running shoes', exactTotal: true }); + assert.strictEqual(result.total, 2); + assert.strictEqual(result.totalRelation, 'exact'); + assert.strictEqual(result.hits[0].id, 'shoe-1'); + assert(result.hits[0].score > result.hits[1].score); + await Promise.all([index.close(), index.close()]); + assert.strictEqual(index.status().state, 'closed'); + + index = await openNativeFullTextIndex(config); + assert.deepStrictEqual( + (await index.search({ text: 'running shoes', exactTotal: true })).hits.map((hit) => hit.id), + ['shoe-1', 'rack-1'], + ); + await index.apply(encodeMutationBatch({ deletes: ['shoe-1'] })); + await index.commit(); + await index.reload(); + const afterDelete = await index.search({ text: 'running shoes', exactTotal: true }); + assert.deepStrictEqual( + afterDelete.hits.map((hit) => hit.id), + ['rack-1'], + ); + assert(afterDelete.hits[0].score > 0); + await index.close(); +}); + +test('preserves clean-close state after a rejected batch', async (context) => { + const index = await openNativeFullTextIndex(options(temporaryIndex(context))); + await assert.rejects( + index.apply( + encodeMutationBatch({ + upserts: [ + { id: 'partial', fields: { title: 'must not survive' } }, + { id: 'bad', fields: { unknown: 'value' } }, + ], + }), + ), + (error) => error.code === 'E_INVALID_ARGUMENT', + ); + assert.strictEqual(index.status().uncommittedMutations, 0n); + await index.apply(encodeMutationBatch({ upserts: [{ id: 'valid', fields: { title: 'survives' } }] })); + await index.commit(); + await index.reload(); + assert.strictEqual((await index.search({ text: 'must survive', exactTotal: true })).total, 1); + await index.close(); +}); + +test('requires an explicit rollback when close would discard mutations', async (context) => { + const index = await openNativeFullTextIndex(options(temporaryIndex(context))); + await index.apply(encodeMutationBatch({ upserts: [{ id: 'one', fields: { title: 'one' } }] })); + await assert.rejects(index.close(), (error) => error.code === 'E_DIRTY_CLOSE'); + assert.strictEqual(index.status().state, 'open'); + await index.close({ mode: 'rollback' }); +}); + +test('rejects duplicate opens and persisted identity drift', async (context) => { + const indexPath = temporaryIndex(context); + const config = options(indexPath); + const first = await openNativeFullTextIndex(config); + await assert.rejects(openNativeFullTextIndex(config), (error) => error.code === 'E_DUPLICATE_OPEN'); + await first.close(); + await assert.rejects( + openNativeFullTextIndex(options(indexPath, { generation: 'generation-2' })), + (error) => error.code === 'E_IDENTITY_MISMATCH', + ); +}); + +test('keeps the JavaScript event loop responsive while indexing', async (context) => { + const index = await openNativeFullTextIndex(options(temporaryIndex(context))); + const upserts = Array.from({ length: 20_000 }, (_, id) => ({ + id: String(id), + fields: { title: `running shoe model ${id}`, description: 'lightweight outdoor product' }, + })); + let applySettled = false; + let heartbeatsWhilePending = 0; + const timer = setInterval(() => { + if (!applySettled) heartbeatsWhilePending++; + }, 5); + await index.apply(encodeMutationBatch({ upserts })).finally(() => { + applySettled = true; + }); + clearInterval(timer); + assert(heartbeatsWhilePending > 5, `indexing allowed only ${heartbeatsWhilePending} event-loop heartbeats`); + await index.close({ mode: 'rollback' }); +}); + +test('search completes while the writer is processing a large batch', async (context) => { + const config = options(temporaryIndex(context)); + config.limits = { + ...config.limits, + maxQueuedBytes: 16 * 1024 * 1024, + maxBatchBytes: 16 * 1024 * 1024, + }; + const index = await openNativeFullTextIndex(config); + await index.apply(encodeMutationBatch({ upserts: [{ id: 'visible', fields: { title: 'visible trail shoe' } }] })); + await index.commit(); + await index.reload(); + const packed = encodeMutationBatch( + { + upserts: Array.from({ length: 50_000 }, (_, id) => ({ + id: `pending-${id}`, + fields: { title: `pending catalog product ${id}`, description: 'large concurrent batch' }, + })), + }, + config.limits.maxBatchBytes, + ); + let applySettled = false; + const apply = index.apply(packed).finally(() => { + applySettled = true; + }); + const result = await index.search({ text: 'visible trail shoe', exactTotal: true }); + assert.strictEqual(result.hits[0].id, 'visible'); + assert.strictEqual(applySettled, false, 'search waited for the writer batch to finish'); + await apply; + await index.close({ mode: 'rollback' }); +}); + +test('rejects overload instead of blocking the JavaScript thread', async (context) => { + const indexPath = temporaryIndex(context); + const config = options(indexPath); + config.limits = { + ...config.limits, + maxQueuedCommands: 1, + maxQueuedBytes: 4 * 1024 * 1024, + maxBatchBytes: 4 * 1024 * 1024, + }; + const index = await openNativeFullTextIndex(config); + const packed = encodeMutationBatch( + { + upserts: Array.from({ length: 15_000 }, (_, id) => ({ + id: String(id), + fields: { title: `queued running product ${id}` }, + })), + }, + config.limits.maxBatchBytes, + ); + const settled = await Promise.allSettled(Array.from({ length: 12 }, () => index.apply(packed))); + assert(settled.some((result) => result.status === 'rejected' && result.reason.code === 'E_QUEUE_FULL')); + await index.close({ mode: 'rollback' }); +}); diff --git a/test/native-worker.test.mjs b/test/native-worker.test.mjs new file mode 100644 index 0000000..9369e46 --- /dev/null +++ b/test/native-worker.test.mjs @@ -0,0 +1,93 @@ +import assert from 'node:assert'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { Worker } from 'node:worker_threads'; + +import { openNativeFullTextIndex } from '@harperfast/fulltext/native'; + +test('worker termination detaches completions and releases its writer', async (context) => { + const indexPath = mkdtempSync(path.join(tmpdir(), 'harper-fulltext-worker-')); + context.after(() => rmSync(indexPath, { recursive: true, force: true })); + const worker = new Worker(new URL('./fixtures/native-worker-child.mjs', import.meta.url), { + workerData: { + indexPath, + moduleUrl: new URL('../dist/native.js', import.meta.url).href, + }, + }); + await new Promise((resolve, reject) => { + worker.once('error', reject); + worker.on('message', (message) => message === 'applying' && resolve()); + }); + await worker.terminate(); + const index = await waitForOpen(indexPath); + assert.strictEqual((await index.search({ text: 'worker owned product', exactTotal: true })).total, 0); + await index.close(); +}); + +test('worker termination during open releases its native cleanup hook', async (context) => { + const indexPath = mkdtempSync(path.join(tmpdir(), 'harper-fulltext-worker-open-')); + context.after(() => rmSync(indexPath, { recursive: true, force: true })); + const worker = new Worker(new URL('./fixtures/native-worker-child.mjs', import.meta.url), { + workerData: { + indexPath, + moduleUrl: new URL('../dist/native.js', import.meta.url).href, + mode: 'opening', + }, + }); + await new Promise((resolve, reject) => { + worker.once('error', reject); + worker.on('message', (message) => message === 'opening' && resolve()); + }); + await worker.terminate(); + const index = await waitForOpen(indexPath); + await index.close(); +}); + +test('worker termination releases every index in one Node environment', async (context) => { + const indexPaths = Array.from({ length: 3 }, () => mkdtempSync(path.join(tmpdir(), 'harper-fulltext-worker-many-'))); + context.after(() => indexPaths.forEach((indexPath) => rmSync(indexPath, { recursive: true, force: true }))); + const worker = new Worker(new URL('./fixtures/native-worker-child.mjs', import.meta.url), { + workerData: { + indexPaths, + moduleUrl: new URL('../dist/native.js', import.meta.url).href, + mode: 'multiple', + }, + }); + await new Promise((resolve, reject) => { + worker.once('error', reject); + worker.on('message', (message) => message === 'multiple-open' && resolve()); + }); + await worker.terminate(); + const indexes = await Promise.all(indexPaths.map(waitForOpen)); + await Promise.all(indexes.map((index) => index.close())); +}); + +async function waitForOpen(indexPath) { + const deadline = performance.now() + 10_000; + while (true) { + try { + return await openNativeFullTextIndex({ + path: indexPath, + indexId: 'worker-products', + generation: 'generation-1', + fields: [{ name: 'title' }], + analyzer: 'english@1', + limits: { + indexingThreads: 1, + searchThreads: 1, + writerMemoryBytes: 15_000_000, + maxQueuedCommands: 8, + maxQueuedBytes: 1024 * 1024, + maxBatchBytes: 1024 * 1024, + }, + }); + } catch (error) { + if (!['E_DUPLICATE_OPEN', 'E_LOCK_BUSY'].includes(error.code) || performance.now() >= deadline) { + throw error; + } + await new Promise((resolve) => setTimeout(resolve, 10)); + } + } +} diff --git a/test/native.test.mjs b/test/native.test.mjs index 1878c76..ede020f 100644 --- a/test/native.test.mjs +++ b/test/native.test.mjs @@ -1,8 +1,11 @@ import assert from 'node:assert'; -import { readFileSync } from 'node:fs'; +import { mkdtempSync, readFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; import test from 'node:test'; import { runtimeInfo } from '@harperfast/fulltext/native'; +import { decodeResponse, encodeOpen } from '../dist/codec.js'; import { normalizeNativeError } from '../dist/errors.js'; import { loadAddon, platformTriple } from '../dist/load-addon.js'; @@ -45,6 +48,58 @@ test('turns a panic into a coded terminal error', async () => { await assert.doesNotReject(runtimeInfo()); }); +test('default close tears down a poisoned native handle', async (context) => { + const indexPath = mkdtempSync(path.join(tmpdir(), 'harper-fulltext-poison-')); + context.after(() => rmSync(indexPath, { recursive: true, force: true })); + const addon = loadAddon(); + assert(addon.__testPoisonNativeHandle); + const config = { + path: indexPath, + indexId: 'poison-close', + generation: 'one', + fields: [{ name: 'title', weight: 1 }], + analyzer: 'english@1', + stopWords: true, + positions: true, + surfaceTerms: false, + limits: { + indexingThreads: 1, + searchThreads: 1, + writerMemoryBytes: 15_000_000, + maxQueuedCommands: 8, + maxQueuedBytes: 1024 * 1024, + maxBatchBytes: 1024 * 1024, + }, + }; + const opened = await invoke((callback) => addon.__nativeOpen(encodeOpen(config), callback)); + const handle = opened.u32(); + opened.finish(); + addon.__testPoisonNativeHandle(handle); + const closed = await invoke((callback) => addon.__nativeClose(handle, false, callback)); + closed.finish(); + const reopened = await invoke((callback) => addon.__nativeOpen(encodeOpen(config), callback)); + const reopenedHandle = reopened.u32(); + reopened.finish(); + const reclosed = await invoke((callback) => addon.__nativeClose(reopenedHandle, false, callback)); + reclosed.finish(); +}); + +function invoke(start) { + return new Promise((resolve, reject) => { + try { + start((response) => { + try { + resolve(decodeResponse(response)); + } catch (error) { + reject(error); + } + }); + } catch (error) { + reject(normalizeNativeError(error)); + } + }); +} + function tomlSection(manifest, name) { const sectionStart = manifest.indexOf(`[${name}]`); assert.notStrictEqual(sectionStart, -1, `Cargo.toml is missing [${name}]`); diff --git a/ts/codec.ts b/ts/codec.ts new file mode 100644 index 0000000..0f08a52 --- /dev/null +++ b/ts/codec.ts @@ -0,0 +1,274 @@ +import { FulltextError } from './errors.js'; + +const protocolVersion = 1; + +export interface PackedFieldConfig { + name: string; + weight: number; +} + +export interface PackedOpenConfig { + path: string; + indexId: string; + generation: string; + fields: PackedFieldConfig[]; + analyzer: string; + stopWords: boolean; + positions: boolean; + surfaceTerms: boolean; + limits: { + indexingThreads: number; + searchThreads: number; + writerMemoryBytes: number; + maxQueuedCommands: number; + maxQueuedBytes: number; + maxBatchBytes: number; + }; +} + +export interface PackedMutationBatch { + upserts: Array<{ id: string; fields: Record }>; + deletes: string[]; +} + +export interface PackedSearchRequest { + text: string; + operator: 'any' | 'all'; + fields: string[]; + offset: number; + limit: number; + exactTotal: boolean; +} + +export function encodeOpen(config: PackedOpenConfig): Buffer { + const writer = new ByteWriter(Number.MAX_SAFE_INTEGER); + writer.header('FTOP'); + writer.string(config.path); + writer.string(config.indexId); + writer.string(config.generation); + writer.string(config.analyzer); + writer.boolean(config.stopWords); + writer.boolean(config.positions); + writer.boolean(config.surfaceTerms); + writer.u16(config.fields.length, 'fields.length'); + for (const field of config.fields) { + writer.string(field.name); + writer.f32(field.weight, 'field.weight'); + } + writer.u16(config.limits.indexingThreads, 'limits.indexingThreads'); + writer.u16(config.limits.searchThreads, 'limits.searchThreads'); + writer.u64(config.limits.writerMemoryBytes, 'limits.writerMemoryBytes'); + writer.u32(config.limits.maxQueuedCommands, 'limits.maxQueuedCommands'); + writer.u64(config.limits.maxQueuedBytes, 'limits.maxQueuedBytes'); + writer.u64(config.limits.maxBatchBytes, 'limits.maxBatchBytes'); + return writer.finish(); +} + +export function encodeBatch(batch: PackedMutationBatch, maxBytes: number): Buffer { + const writer = new ByteWriter(maxBytes); + writer.header('FTMB'); + writer.u32(batch.upserts.length, 'upserts.length'); + writer.u32(batch.deletes.length, 'deletes.length'); + for (const upsert of batch.upserts) { + writer.string(upsert.id); + const fields = Object.entries(upsert.fields); + writer.u16(fields.length, 'upsert field count'); + for (const [name, value] of fields) { + writer.string(name); + const values = Array.isArray(value) ? value : [value]; + writer.u16(values.length, 'field value count'); + for (const entry of values) { + writer.string(entry); + } + } + } + for (const id of batch.deletes) { + writer.string(id); + } + return writer.finish(); +} + +export function encodeSearch(request: PackedSearchRequest): Buffer { + const writer = new ByteWriter(2 * 1024 * 1024); + writer.header('FTSQ'); + writer.string(request.text); + writer.u8(request.operator === 'all' ? 1 : 0, 'operator'); + writer.u16(request.fields.length, 'fields.length'); + for (const field of request.fields) { + writer.string(field); + } + writer.u32(request.offset, 'offset'); + writer.u32(request.limit, 'limit'); + writer.boolean(request.exactTotal); + return writer.finish(); +} + +export function decodeResponse(value: Buffer): Cursor { + const cursor = new Cursor(value); + if (cursor.text(4) !== 'FTRP' || cursor.u16() !== protocolVersion) { + throw new FulltextError('E_NATIVE_ABI_MISMATCH', 'Invalid native response envelope'); + } + const status = cursor.u8(); + if (status === 1) { + const code = cursor.string(); + const message = cursor.string(); + throw new FulltextError( + FulltextError.isCode(code) ? code : 'E_NATIVE_FAILURE', + FulltextError.isCode(code) ? message : `${code}: ${message}`, + ); + } + if (status !== 0) { + throw new FulltextError('E_NATIVE_FAILURE', `Unknown native response status ${status}`); + } + return cursor; +} + +export class Cursor { + readonly #buffer: Buffer; + #offset = 0; + + constructor(buffer: Buffer) { + this.#buffer = buffer; + } + + u8(): number { + return this.take(1)[0]; + } + + u16(): number { + const value = this.#buffer.readUInt16LE(this.#offset); + this.take(2); + return value; + } + + u32(): number { + const value = this.#buffer.readUInt32LE(this.#offset); + this.take(4); + return value; + } + + u64(): bigint { + const value = this.#buffer.readBigUInt64LE(this.#offset); + this.take(8); + return value; + } + + f32(): number { + const value = this.#buffer.readFloatLE(this.#offset); + this.take(4); + return value; + } + + string(): string { + return this.text(this.u32()); + } + + text(length: number): string { + return this.take(length).toString('utf8'); + } + + finish(): void { + if (this.#offset !== this.#buffer.length) { + throw new FulltextError('E_NATIVE_FAILURE', 'Native response has trailing bytes'); + } + } + + private take(length: number): Buffer { + const end = this.#offset + length; + if (!Number.isSafeInteger(end) || end > this.#buffer.length) { + throw new FulltextError('E_NATIVE_FAILURE', 'Native response is truncated'); + } + const value = this.#buffer.subarray(this.#offset, end); + this.#offset = end; + return value; + } +} + +class ByteWriter { + readonly #chunks: Buffer[] = []; + readonly #maxBytes: number; + #length = 0; + + constructor(maxBytes: number) { + if (!Number.isSafeInteger(maxBytes) || maxBytes <= 0) { + throw new FulltextError('E_INVALID_ARGUMENT', 'maxBytes must be a positive safe integer'); + } + this.#maxBytes = maxBytes; + } + + header(magic: string): void { + this.bytes(Buffer.from(magic, 'ascii')); + this.u16(protocolVersion, 'protocolVersion'); + } + + boolean(value: boolean): void { + this.u8(value ? 1 : 0, 'boolean'); + } + + u8(value: number, name: string): void { + this.integer(value, 0xff, name, 1, 'writeUInt8'); + } + + u16(value: number, name: string): void { + this.integer(value, 0xffff, name, 2, 'writeUInt16LE'); + } + + u32(value: number, name: string): void { + this.integer(value, 0xffffffff, name, 4, 'writeUInt32LE'); + } + + u64(value: number, name: string): void { + if (!Number.isSafeInteger(value) || value < 0) { + throw new FulltextError('E_INVALID_ARGUMENT', `${name} must be a non-negative safe integer`); + } + const buffer = Buffer.allocUnsafe(8); + buffer.writeBigUInt64LE(BigInt(value)); + this.bytes(buffer); + } + + f32(value: number, name: string): void { + if (!Number.isFinite(value) || value <= 0) { + throw new FulltextError('E_INVALID_ARGUMENT', `${name} must be finite and greater than zero`); + } + const buffer = Buffer.allocUnsafe(4); + buffer.writeFloatLE(value); + this.bytes(buffer); + } + + string(value: string): void { + if (typeof value !== 'string') { + throw new FulltextError('E_INVALID_ARGUMENT', 'packed string values must be strings'); + } + const bytes = Buffer.from(value, 'utf8'); + this.u32(bytes.length, 'string byte length'); + this.bytes(bytes); + } + + finish(): Buffer { + return Buffer.concat(this.#chunks, this.#length); + } + + private integer( + value: number, + maximum: number, + name: string, + width: number, + method: 'writeUInt8' | 'writeUInt16LE' | 'writeUInt32LE', + ): void { + if (!Number.isInteger(value) || value < 0 || value > maximum) { + throw new FulltextError('E_INVALID_ARGUMENT', `${name} is outside its packed integer range`); + } + const buffer = Buffer.allocUnsafe(width); + buffer[method](value, 0); + this.bytes(buffer); + } + + private bytes(value: Buffer): void { + const nextLength = this.#length + value.length; + if (!Number.isSafeInteger(nextLength) || nextLength > this.#maxBytes) { + throw new FulltextError('E_INVALID_ARGUMENT', `packed value exceeds ${this.#maxBytes} bytes`); + } + this.#length = nextLength; + this.#chunks.push(value); + } +} diff --git a/ts/errors.ts b/ts/errors.ts index 25f4223..660fb0b 100644 --- a/ts/errors.ts +++ b/ts/errors.ts @@ -5,6 +5,16 @@ const errorCodes = [ 'E_NATIVE_PANIC', 'E_POISONED', 'E_NATIVE_FAILURE', + 'E_CLOSED', + 'E_DIRTY_CLOSE', + 'E_DUPLICATE_OPEN', + 'E_IDENTITY_MISMATCH', + 'E_INCOMPLETE_CREATE', + 'E_INVALID_ARGUMENT', + 'E_LOCK_BUSY', + 'E_QUEUE_FULL', + 'E_SCHEMA_MISMATCH', + 'E_STORAGE', ] as const; export type FulltextErrorCode = (typeof errorCodes)[number]; @@ -17,6 +27,10 @@ export class FulltextError extends Error { this.name = 'FulltextError'; this.code = code; } + + static isCode(value: string): value is FulltextErrorCode { + return errorCodes.includes(value as FulltextErrorCode); + } } export function normalizeNativeError(error: unknown): FulltextError { @@ -36,7 +50,5 @@ function readErrorCode(error: unknown): FulltextErrorCode | undefined { return undefined; } const code = error.code; - return typeof code === 'string' && errorCodes.includes(code as FulltextErrorCode) - ? (code as FulltextErrorCode) - : undefined; + return typeof code === 'string' && FulltextError.isCode(code) ? (code as FulltextErrorCode) : undefined; } diff --git a/ts/load-addon.ts b/ts/load-addon.ts index a3c4e07..c5861a1 100644 --- a/ts/load-addon.ts +++ b/ts/load-addon.ts @@ -13,11 +13,21 @@ interface NativeRuntimeInfo { interface NativeAddonApi { runtimeInfo(): NativeRuntimeInfo; + __nativeOpen(config: Buffer, callback: NativeCallback): void; + __nativeApply(handle: number, batch: Buffer, callback: NativeCallback): void; + __nativeCommit(handle: number, callback: NativeCallback): void; + __nativeReload(handle: number, callback: NativeCallback): void; + __nativeSearch(handle: number, request: Buffer, callback: NativeCallback): void; + __nativeClose(handle: number, rollback: boolean, callback: NativeCallback): void; + __nativeStatus(handle: number): Buffer; __testCreateHandle?(): number; __testPanic?(id: number): void; __testCheck?(id: number): boolean; + __testPoisonNativeHandle?(handle: number): void; } +export type NativeCallback = (response: Buffer) => void; + const require = createRequire(import.meta.url); const expectedNativeAbiVersion = 1; let loadedAddon: NativeAddonApi | undefined; diff --git a/ts/native.ts b/ts/native.ts index f1e8664..28801d6 100644 --- a/ts/native.ts +++ b/ts/native.ts @@ -1,4 +1,5 @@ -import { normalizeNativeError } from './errors.js'; +import { FulltextError, normalizeNativeError } from './errors.js'; +import { Cursor, decodeResponse, encodeBatch, encodeOpen, encodeSearch } from './codec.js'; import { loadAddon } from './load-addon.js'; export { FulltextError } from './errors.js'; @@ -11,6 +12,204 @@ export interface RuntimeInfo { storageBackends: ReadonlyArray<'native'>; } +export interface NativeFullTextIndexOptions { + path: string; + indexId: string; + generation: string; + fields: Array<{ name: string; weight?: number }>; + analyzer: 'english@1'; + stopWords?: boolean; + positions?: boolean; + surfaceTerms?: boolean; + limits: { + indexingThreads: number; + searchThreads: number; + writerMemoryBytes: number; + maxQueuedCommands: number; + maxQueuedBytes: number; + maxBatchBytes: number; + }; +} + +export interface FullTextMutationBatch { + upserts?: Array<{ id: string; fields: Record }>; + deletes?: string[]; +} + +export interface SearchRequest { + text: string; + operator?: 'any' | 'all'; + fields?: string[]; + offset?: number; + limit?: number; + exactTotal?: boolean; +} + +export interface SearchResult { + total: number; + totalRelation: 'exact' | 'lower-bound'; + hits: Array<{ id: string; score: number }>; +} + +export interface FullTextStatus { + state: 'open' | 'closing' | 'closed' | 'poisoned'; + uncommittedMutations: bigint; + writerQueuedCommands: bigint; + writerQueuedBytes: bigint; + searchQueuedCommands: bigint; + searchQueuedBytes: bigint; + commitOpstamp: bigint; + metrics: { + writerQueueNanoseconds: bigint; + writerExecutionNanoseconds: bigint; + searchQueueNanoseconds: bigint; + searchExecutionNanoseconds: bigint; + }; +} + +export interface CloseOptions { + mode?: 'require-clean' | 'rollback'; +} + +export class NativeFullTextIndex { + readonly #handle: number; + #closed = false; + #closedStatus?: FullTextStatus; + #closePromise?: Promise; + + constructor(handle: number) { + this.#handle = handle; + } + + async apply(packedBatch: Uint8Array): Promise { + const cursor = await invoke((callback) => loadAddon().__nativeApply(this.#handle, asBuffer(packedBatch), callback)); + const count = safeNumber(cursor.u64(), 'mutation count'); + cursor.finish(); + return count; + } + + async commit(): Promise { + const cursor = await invoke((callback) => loadAddon().__nativeCommit(this.#handle, callback)); + const opstamp = cursor.u64(); + cursor.finish(); + return opstamp; + } + + async reload(): Promise { + const cursor = await invoke((callback) => loadAddon().__nativeReload(this.#handle, callback)); + cursor.finish(); + } + + async search(request: SearchRequest): Promise { + const cursor = await invoke((callback) => + loadAddon().__nativeSearch( + this.#handle, + encodeSearch({ + text: request.text, + operator: request.operator ?? 'any', + fields: request.fields ?? [], + offset: request.offset ?? 0, + limit: request.limit ?? 20, + exactTotal: request.exactTotal ?? false, + }), + callback, + ), + ); + const total = safeNumber(cursor.u64(), 'search total'); + const relation = cursor.u8(); + const hitCount = cursor.u32(); + const hits = Array.from({ length: hitCount }, () => ({ score: cursor.f32(), id: cursor.string() })); + cursor.finish(); + if (relation !== 0 && relation !== 1) { + throw new FulltextError('E_NATIVE_FAILURE', `Unknown total relation ${relation}`); + } + return { + total, + totalRelation: relation === 0 ? 'exact' : 'lower-bound', + hits, + }; + } + + status(): FullTextStatus { + if (this.#closedStatus) { + return this.#closedStatus; + } + try { + const cursor = decodeResponse(loadAddon().__nativeStatus(this.#handle)); + const state = cursor.u8(); + const status: FullTextStatus = { + state: ['open', 'closing', 'closed', 'poisoned'][state] as FullTextStatus['state'], + uncommittedMutations: cursor.u64(), + writerQueuedCommands: cursor.u64(), + writerQueuedBytes: cursor.u64(), + searchQueuedCommands: cursor.u64(), + searchQueuedBytes: cursor.u64(), + commitOpstamp: cursor.u64(), + metrics: { + writerQueueNanoseconds: cursor.u64(), + writerExecutionNanoseconds: cursor.u64(), + searchQueueNanoseconds: cursor.u64(), + searchExecutionNanoseconds: cursor.u64(), + }, + }; + cursor.finish(); + if (state > 3) { + throw new FulltextError('E_NATIVE_FAILURE', `Unknown native lifecycle state ${state}`); + } + return status; + } catch (error) { + throw normalizeNativeError(error); + } + } + + async close(options: CloseOptions = {}): Promise { + if (this.#closed) { + return; + } + if (this.#closePromise) { + return this.#closePromise; + } + const openStatus = this.status(); + this.#closePromise = (async () => { + const cursor = await invoke((callback) => + loadAddon().__nativeClose(this.#handle, options.mode === 'rollback', callback), + ); + cursor.finish(); + this.#closed = true; + this.#closedStatus = { ...openStatus, state: 'closed' }; + })(); + try { + await this.#closePromise; + } finally { + if (!this.#closed) { + this.#closePromise = undefined; + } + } + } +} + +export function encodeMutationBatch(batch: FullTextMutationBatch, maxBytes = 8 * 1024 * 1024): Uint8Array { + return encodeBatch({ upserts: batch.upserts ?? [], deletes: batch.deletes ?? [] }, maxBytes); +} + +export async function openNativeFullTextIndex(options: NativeFullTextIndexOptions): Promise { + const cursor = await invoke((callback) => + loadAddon().__nativeOpen( + encodeOpen({ + ...options, + fields: options.fields.map((field) => ({ name: field.name, weight: field.weight ?? 1 })), + stopWords: options.stopWords ?? true, + positions: options.positions ?? true, + surfaceTerms: options.surfaceTerms ?? false, + }), + callback, + ), + ); + const handle = cursor.u32(); + cursor.finish(); + return new NativeFullTextIndex(handle); +} + export async function runtimeInfo(): Promise { try { const info = loadAddon().runtimeInfo(); @@ -24,3 +223,30 @@ export async function runtimeInfo(): Promise { throw normalizeNativeError(error); } } + +function invoke(start: (callback: (response: Buffer) => void) => void): Promise { + return new Promise((resolve, reject) => { + try { + start((response) => { + try { + resolve(decodeResponse(response)); + } catch (error) { + reject(normalizeNativeError(error)); + } + }); + } catch (error) { + reject(normalizeNativeError(error)); + } + }); +} + +function asBuffer(value: Uint8Array): Buffer { + return Buffer.isBuffer(value) ? value : Buffer.from(value.buffer, value.byteOffset, value.byteLength); +} + +function safeNumber(value: bigint, name: string): number { + if (value > BigInt(Number.MAX_SAFE_INTEGER)) { + throw new FulltextError('E_NATIVE_FAILURE', `${name} exceeds JavaScript's safe integer range`); + } + return Number(value); +}