From e7c750d3a33a9425b7545ba4c03510dd0fb1e9b5 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Thu, 10 Sep 2026 14:58:35 -0600 Subject: [PATCH 1/6] Add experimental Harper-backed runtime --- Cargo.toml | 3 +- README.md | 27 +- dependencies.md | 9 +- package.json | 8 +- src/engine.rs | 52 ++- src/host_storage.rs | 262 +++++++++++--- src/lib.rs | 8 +- src/native.rs | 331 +++++++++++++++--- src/phase0.rs | 1 + src/protocol.rs | 101 +++++- test/fixtures/harper-worker-child.mjs | 101 ++++++ .../host-storage-transport-worker.mjs | 11 +- test/harper-index.test.mjs | 145 ++++++++ test/host-storage-transport.test.mjs | 38 ++ test/native.test.mjs | 2 +- test/package.release.test.mjs | 3 +- ts/codec.ts | 57 ++- ts/harper.ts | 177 ++++++++++ ts/invoke.ts | 19 + ts/load-addon.ts | 13 +- ts/native.ts | 23 +- 21 files changed, 1223 insertions(+), 168 deletions(-) create mode 100644 test/fixtures/harper-worker-child.mjs create mode 100644 test/harper-index.test.mjs create mode 100644 ts/harper.ts create mode 100644 ts/invoke.ts diff --git a/Cargo.toml b/Cargo.toml index 4dd39b2..b2ec4bd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,7 +19,8 @@ required-features = ["phase0"] default = [] node-api = ["dep:napi", "dep:napi-derive"] phase0 = ["dep:async-trait", "dep:stable_deref_trait"] -test-panic = ["node-api", "phase0"] +host-storage = ["node-api", "phase0"] +test-panic = ["host-storage"] [dependencies] async-trait = { version = "=0.1.92", optional = true } diff --git a/README.md b/README.md index e23e3cd..a61c1f1 100644 --- a/README.md +++ b/README.md @@ -1,10 +1,11 @@ # @harperfast/fulltext -Native Tantivy full-text indexing for Node.js, with a native filesystem backend and a planned -caller-owned rocksdb-js backend for Harper. +Native Tantivy full-text indexing for Node.js, with a standalone filesystem backend and an +experimental Harper-owned storage integration. 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. +index backed by `MmapDirectory`. Harper releases will use only the Harper integration backed by +Harper's existing RocksDB lifecycle; they will not use Tantivy's filesystem storage. ## Requirements @@ -67,15 +68,27 @@ handle's searches. ## Storage boundaries -The package is designed around two explicit entry points: +The package has two explicit entry points: - `@harperfast/fulltext/native` uses Tantivy's native directory implementation and has no rocksdb-js dependency. -- `@harperfast/fulltext/rocks` will use a caller-owned rocksdb-js database through a versioned - native capability lease. It is not exported until that contract is implemented and tested. +- `@harperfast/fulltext/harper` is an experimental integration surface that stores Tantivy objects + through a synchronous Harper-owned key-value view. It exists to prove and measure the real + derived-index path before Harper enables a customer-facing feature. There is no generic storage selector and no fallback between backends. Harper will consume only the -Rocks entry point. The fulltext addon will not link its own copy of RocksDB. +Harper entry point. The fulltext addon does not link RocksDB or depend on rocksdb-js; Harper owns the +database, durability, and store lifecycle. + +The Harper opener requires a process-lifetime store identity, persistent generation, byte namespace, +bounded transport limits, and a `HostStorage` implementation. `publish(payload)` commits the index +and opaque payload into one Tantivy `meta.json` generation, then reloads the local reader before it +resolves. Harper uses that payload for its derived-index cursor. `committedPayload` exposes the +payload recovered at open. + +The current Harper path is owner-worker-only. It does not yet provide non-owner read handles, +cross-worker refresh, generation retirement, or scheduled physical reclamation. Those lifecycle +pieces and representative performance results are required before release enablement. ## Development diff --git a/dependencies.md b/dependencies.md index 6fe3090..41b052f 100644 --- a/dependencies.md +++ b/dependencies.md @@ -15,8 +15,8 @@ reviewed deliberately. | `libloading` 0.8.9 | Windows runtime | Resolves the Node-API type-tag check from the host, matching napi-rs's Windows strategy. | | `stable_deref_trait` 1.2.1 | optional Phase 0 runtime | Lets Tantivy `OwnedBytes` retain and directly read provider-owned buffers. | -The Rust dependency graph must not include RocksDB. The future Rocks backend calls a C-ABI -capability table owned by rocksdb-js rather than linking a second RocksDB runtime. +The Rust dependency graph must not include RocksDB. The Harper integration reaches Harper-owned +storage through its narrow host interface rather than linking a second RocksDB runtime. napi-rs 2.16 generates an outer unwind boundary only for exports marked `catch_unwind`. Every fulltext function, method, and constructor uses that option to contain argument and result @@ -32,5 +32,6 @@ error codes and per-handle poison state for package-owned operations. | `prettier` 3.6.2 | development | Repository formatting checks. | | `typescript` 5.9.3 | development | Compiles the public façade and declarations. | -The native entry point has no production npm dependencies. rocksdb-js will be an optional peer -dependency only when the Rocks entry point is implemented. +Neither entry point has production npm dependencies. The Harper integration accepts a narrow host +storage interface and does not depend on rocksdb-js; Harper supplies the implementation backed by +its own RocksDB lifecycle. diff --git a/package.json b/package.json index 2bba787..6cfcc7b 100644 --- a/package.json +++ b/package.json @@ -7,6 +7,10 @@ "./native": { "types": "./dist/native.d.ts", "import": "./dist/native.js" + }, + "./harper": { + "types": "./dist/harper.d.ts", + "import": "./dist/harper.js" } }, "files": [ @@ -29,8 +33,8 @@ }, "scripts": { "build": "npm run build:typescript && npm run build:native", - "build:debug": "npm run build:typescript && napi build --platform --js false --dts ts/addon.d.ts --features node-api", - "build:native": "napi build --platform --js false --dts ts/addon.d.ts --release --features node-api", + "build:debug": "npm run build:typescript && napi build --platform --js false --dts ts/addon.d.ts --features host-storage", + "build:native": "napi build --platform --js false --dts ts/addon.d.ts --release --features host-storage", "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", diff --git a/src/engine.rs b/src/engine.rs index 92c0a99..2542045 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -17,6 +17,7 @@ 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"; +pub const MAX_COMMIT_PAYLOAD_BYTES: usize = 64 * 1024; #[derive(Clone)] pub struct Engine { @@ -141,6 +142,10 @@ impl Engine { .map_err(index_error) } + pub fn committed_payload(&self) -> Result> { + Ok(self.index.load_metas().map_err(index_error)?.payload) + } + 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)?; @@ -326,7 +331,20 @@ impl Writer { } pub fn commit(&mut self) -> Result { - self.inner.commit().map_err(index_error) + self.commit_with_payload(None) + } + + pub fn commit_with_payload(&mut self, payload: Option<&str>) -> Result { + if payload.is_some_and(|payload| payload.len() > MAX_COMMIT_PAYLOAD_BYTES) { + return Err(FulltextError::invalid(format!( + "commit payload exceeds {MAX_COMMIT_PAYLOAD_BYTES} UTF-8 bytes" + ))); + } + let mut commit = self.inner.prepare_commit().map_err(index_error)?; + if let Some(payload) = payload { + commit.set_payload(payload); + } + commit.commit().map_err(index_error) } pub fn rollback(&mut self) -> Result { @@ -429,7 +447,6 @@ mod tests { fn config() -> EngineConfig { EngineConfig { - path: "unused".to_owned(), index_id: "products".to_owned(), generation: "one".to_owned(), fields: vec![ @@ -550,6 +567,37 @@ mod tests { ); } + #[test] + fn commit_payload_survives_merge_and_reopen() { + let directory = RamDirectory::create(); + let config = config(); + let engine = Engine::open(directory.clone(), &config).unwrap(); + let mut writer = engine.writer(&config).unwrap(); + writer + .inner + .set_merge_policy(Box::new(tantivy::merge_policy::NoMergePolicy)); + writer.apply(batch()).unwrap(); + writer.commit_with_payload(Some("cursor-v1:42")).unwrap(); + writer + .apply(MutationBatch { + upserts: vec![crate::protocol::Upsert { + id: "three".to_owned(), + fields: vec![("title".to_owned(), vec!["Hiking Boots".to_owned()])], + }], + deletes: Vec::new(), + }) + .unwrap(); + writer.commit_with_payload(Some("cursor-v1:43")).unwrap(); + let segments = engine.index.searchable_segment_ids().unwrap(); + assert_eq!(segments.len(), 2); + writer.inner.merge(&segments).wait().unwrap(); + assert_eq!(engine.committed_payload().unwrap().as_deref(), Some("cursor-v1:43")); + writer.close().unwrap(); + + let reopened = Engine::open(directory, &config).unwrap(); + assert_eq!(reopened.committed_payload().unwrap().as_deref(), Some("cursor-v1:43")); + } + #[test] fn completes_an_interrupted_sidecar_first_create() { let directory = RamDirectory::create(); diff --git a/src/host_storage.rs b/src/host_storage.rs index c66d995..a624172 100644 --- a/src/host_storage.rs +++ b/src/host_storage.rs @@ -1,10 +1,16 @@ use std::collections::HashMap; use std::io; +#[cfg(feature = "test-panic")] use std::io::Write as IoWrite; +#[cfg(feature = "test-panic")] use std::panic::{catch_unwind, AssertUnwindSafe}; +#[cfg(feature = "test-panic")] use std::path::Path; -use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; +#[cfg(feature = "test-panic")] +use std::sync::atomic::AtomicU32; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak}; +#[cfg(feature = "test-panic")] use std::thread; use std::time::{Duration, Instant}; @@ -12,25 +18,33 @@ use napi::bindgen_prelude::Buffer; use napi::threadsafe_function::{ErrorStrategy, ThreadSafeCallContext, ThreadsafeFunction, ThreadsafeFunctionCallMode}; use napi::{Env, JsBuffer, JsFunction, JsUnknown, Status}; use napi_derive::napi; -use tantivy::directory::{Directory, OwnedBytes, TerminatingWrite}; +use tantivy::directory::OwnedBytes; +#[cfg(feature = "test-panic")] +use tantivy::directory::{Directory, TerminatingWrite}; use crate::boundary; -use crate::phase0::{ - reclaim_read_key_bytes, KvDirectory, KvStore, KvStoreIdentity, Mutation, ReclaimBudget, WritePolicy, CHUNK_SIZE, - RECLAIM_MAX_BATCH_REQUEST_BYTES, -}; +#[cfg(feature = "test-panic")] +use crate::phase0::ReclaimBudget; +#[cfg(any(test, feature = "test-panic"))] +use crate::phase0::{reclaim_read_key_bytes, RECLAIM_MAX_BATCH_REQUEST_BYTES}; +use crate::phase0::{KvDirectory, KvStore, KvStoreIdentity, Mutation, WritePolicy, CHUNK_SIZE}; +use crate::protocol::HostOpenConfig; type HostCallback = ThreadsafeFunction, ErrorStrategy::Fatal>; +#[cfg(feature = "test-panic")] type CompletionCallback = ThreadsafeFunction, ErrorStrategy::Fatal>; +#[cfg(feature = "test-panic")] static NEXT_TRANSPORT_HANDLE: AtomicU32 = AtomicU32::new(1); +static NEXT_REQUEST_ID: AtomicU64 = AtomicU64::new(1); +#[cfg(feature = "test-panic")] static HOST_TRANSPORTS: OnceLock>>> = OnceLock::new(); +static PENDING_TRANSPORTS: OnceLock>>> = OnceLock::new(); -struct HostTransport { - handler: HostCallback, +pub(crate) struct HostTransport { + handler: Mutex>, state: Mutex, capacity: Condvar, - next_request_id: AtomicU64, abandoned_waiters: AtomicU64, max_operations: usize, max_bytes: usize, @@ -50,7 +64,9 @@ struct TransportState { struct PendingRequest { retained_bytes: usize, + response_budget: usize, class: AdmissionClass, + entered: bool, response: Weak, } @@ -96,10 +112,9 @@ impl HostTransport { .unref(env) .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?; Ok(Self { - handler, + handler: Mutex::new(Some(handler)), state: Mutex::new(TransportState::default()), capacity: Condvar::new(), - next_request_id: AtomicU64::new(1), abandoned_waiters: AtomicU64::new(0), max_operations, max_bytes, @@ -115,41 +130,24 @@ impl HostTransport { deadline: Option, class: AdmissionClass, ) -> io::Result> { - let request_id = self.next_request_id.fetch_add(1, Ordering::Relaxed); - if request_id == 0 { - self.fail(io::ErrorKind::Other, "host storage request id space exhausted"); - return Err(io::Error::other("host storage request id space exhausted")); - } + let request_id = next_request_id()?; let response = Arc::new(ResponseSlot::new()); - self.admit(request_id, request.len(), response_budget, &response, deadline, class)?; - - let transport = Arc::downgrade(self); - let callback_response = response.clone(); - let max_response_bytes = response_budget; - let status = self.handler.call_with_return_value::( - request, - ThreadsafeFunctionCallMode::NonBlocking, - move |value| { - let completed = catch_unwind(AssertUnwindSafe(|| { - let result = response_bytes(value, max_response_bytes); - if let Some(transport) = transport.upgrade() { - transport.complete(request_id, result); - } else { - callback_response.complete(Err(io::Error::new( - io::ErrorKind::BrokenPipe, - "host storage transport was released", - ))); - } - })); - if completed.is_err() { - if let Some(transport) = transport.upgrade() { - transport.fail(io::ErrorKind::Other, "host storage completion panicked"); - } - callback_response.complete(Err(io::Error::other("host storage completion panicked"))); - } - Ok(()) - }, - ); + pending_registry().insert(request_id, Arc::downgrade(self)); + if let Err(error) = self.admit(request_id, request.len(), response_budget, &response, deadline, class) { + pending_registry().remove(&request_id); + return Err(error); + } + let mut dispatch = Vec::with_capacity(8 + request.len()); + dispatch.extend_from_slice(&request_id.to_le_bytes()); + dispatch.extend_from_slice(&request); + let status = self + .handler + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .as_ref() + .map_or(Status::Closing, |handler| { + handler.call(dispatch, ThreadsafeFunctionCallMode::NonBlocking) + }); if status != Status::Ok { if status == Status::Closing { self.fail(io::ErrorKind::BrokenPipe, "host storage transport is closed"); @@ -171,8 +169,26 @@ impl HostTransport { )) } - fn close(&self) { + pub(crate) fn close(&self) { self.fail(io::ErrorKind::BrokenPipe, "host storage transport is closed"); + if let Some(handler) = self + .handler + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take() + { + let _ = handler.abort(); + } + } + + pub(crate) fn wait_idle(&self) { + let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + while state.operations != 0 { + state = self + .capacity + .wait(state) + .unwrap_or_else(|poisoned| poisoned.into_inner()); + } } fn admit( @@ -244,13 +260,30 @@ impl HostTransport { request_id, PendingRequest { retained_bytes, + response_budget: response_bytes, class, + entered: false, response: Arc::downgrade(response), }, ); Ok(()) } + fn begin(&self, request_id: u64) -> bool { + let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + if state.closed.is_some() { + return false; + } + let Some(pending) = state.pending.get_mut(&request_id) else { + return false; + }; + if pending.entered { + return false; + } + pending.entered = true; + true + } + fn has_capacity(&self, state: &TransportState, retained_bytes: usize, class: AdmissionClass) -> io::Result { let cleanup_capacity = if class == AdmissionClass::Cleanup { let capacity = state.cleanup_capacity.ok_or_else(|| { @@ -284,6 +317,7 @@ impl HostTransport { <= self.max_bytes) } + #[cfg(feature = "test-panic")] fn configure_cleanup(&self, foreground_reserved_bytes: usize, max_cleanup_bytes: usize) -> io::Result<()> { if self.max_operations < 2 || foreground_reserved_bytes == 0 || max_cleanup_bytes == 0 { return Err(io::Error::new( @@ -319,6 +353,7 @@ impl HostTransport { } fn complete(&self, request_id: u64, result: io::Result>) { + pending_registry().remove(&request_id); let response = { let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); let Some(pending) = state.pending.remove(&request_id) else { @@ -339,7 +374,8 @@ impl HostTransport { if let Some(response) = pending.response.upgrade() { response.complete(Err(error())); } - for pending in remaining.into_values() { + for (request_id, pending) in remaining { + pending_registry().remove(&request_id); if let Some(response) = pending.response.upgrade() { response.complete(Err(error())); } @@ -361,6 +397,15 @@ impl HostTransport { } } + fn response_budget(&self, request_id: u64) -> Option { + self.state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .pending + .get(&request_id) + .map(|request| request.response_budget) + } + fn fail(&self, kind: io::ErrorKind, message: &str) { let pending = { let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); @@ -374,7 +419,8 @@ impl HostTransport { std::mem::take(&mut state.pending) }; self.capacity.notify_all(); - for request in pending.into_values() { + for (request_id, request) in pending { + pending_registry().remove(&request_id); if let Some(response) = request.response.upgrade() { response.complete(Err(io::Error::new(kind, message.to_owned()))); } @@ -447,6 +493,71 @@ fn response_bytes(value: JsUnknown, max_bytes: usize) -> io::Result> { Ok(buffer.as_ref().to_vec()) } +#[napi(catch_unwind, skip_typescript, js_name = "__hostStorageComplete")] +pub fn host_storage_complete(request_id: String, response: JsUnknown) -> boundary::Result { + boundary::run_stateless(|| { + let request_id = parse_request_id(&request_id)?; + let Some(transport) = pending_registry().get(&request_id).and_then(Weak::upgrade) else { + return Ok(false); + }; + let Some(response_budget) = transport.response_budget(request_id) else { + return Ok(false); + }; + transport.complete(request_id, response_bytes(response, response_budget)); + Ok(true) + })? +} + +#[napi(catch_unwind, skip_typescript, js_name = "__hostStorageBegin")] +pub fn host_storage_begin(request_id: String) -> boundary::Result { + boundary::run_stateless(|| { + let request_id = parse_request_id(&request_id)?; + Ok(pending_registry() + .get(&request_id) + .and_then(Weak::upgrade) + .is_some_and(|transport| transport.begin(request_id))) + })? +} + +#[napi(catch_unwind, skip_typescript, js_name = "__hostStorageFail")] +pub fn host_storage_fail(request_id: String, message: String) -> boundary::Result { + boundary::run_stateless(|| { + let request_id = parse_request_id(&request_id)?; + let Some(transport) = pending_registry().get(&request_id).and_then(Weak::upgrade) else { + return Ok(false); + }; + let mut end = message.len().min(4_096); + while !message.is_char_boundary(end) { + end -= 1; + } + let message = &message[..end]; + transport.complete(request_id, Err(io::Error::other(message.to_owned()))); + Ok(true) + })? +} + +fn parse_request_id(request_id: &str) -> boundary::Result { + request_id + .parse() + .map_err(|_| napi::Error::new("E_INVALID_ARGUMENT", "invalid host storage request id")) +} + +fn next_request_id() -> io::Result { + loop { + let request_id = NEXT_REQUEST_ID.load(Ordering::Relaxed); + if request_id == 0 { + return Err(io::Error::other("host storage request id space exhausted")); + } + let next = request_id.checked_add(1).unwrap_or(0); + if NEXT_REQUEST_ID + .compare_exchange_weak(request_id, next, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + return Ok(request_id); + } + } +} + const HOST_PROTOCOL_VERSION: u8 = 1; const OP_READ: u8 = 1; const OP_WRITE: u8 = 2; @@ -457,16 +568,20 @@ const VALUE_MISSING: u8 = 0; const VALUE_PRESENT: u8 = 1; const MUTATION_PUT: u8 = 1; const MUTATION_DELETE: u8 = 2; +#[cfg(any(test, feature = "test-panic"))] const HOST_PROTOCOL_HEADER_BYTES: usize = 2; +#[cfg(any(test, feature = "test-panic"))] const HOST_LENGTH_PREFIX_BYTES: usize = 4; const READ_RESPONSE_OVERHEAD: usize = 7; +#[cfg(any(test, feature = "test-panic"))] fn host_read_request_bytes(key_bytes: usize) -> io::Result { key_bytes .checked_add(HOST_PROTOCOL_HEADER_BYTES + HOST_LENGTH_PREFIX_BYTES) .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "host read request size overflow")) } +#[cfg(any(test, feature = "test-panic"))] fn minimum_cleanup_reservation( max_read_response_bytes: usize, max_control_response_bytes: usize, @@ -493,7 +608,7 @@ fn validate_cleanup_request(request_bytes: usize, limit: Option, operatio } #[derive(Clone)] -struct HostKvStore { +pub(crate) struct HostKvStore { transport: Arc, identity: KvStoreIdentity, max_read_response_bytes: usize, @@ -528,6 +643,7 @@ impl HostKvStore { }) } + #[cfg(feature = "test-panic")] fn minimum_cleanup_bytes(&self, namespace: &[u8], budget: ReclaimBudget) -> io::Result { minimum_cleanup_reservation( self.max_read_response_bytes, @@ -537,6 +653,7 @@ impl HostKvStore { ) } + #[cfg(feature = "test-panic")] fn cleanup_directory( &self, namespace: &[u8], @@ -621,6 +738,33 @@ impl HostKvStore { } } +pub(crate) fn open_directory( + env: &Env, + handler: JsFunction, + config: &HostOpenConfig, +) -> boundary::Result<(KvDirectory, Arc)> { + let transport = Arc::new(HostTransport::new( + env, + handler, + config.max_operations, + config.max_transport_bytes, + config.read_timeout, + )?); + let identity = KvStoreIdentity( + config.store_identity.0, + config.store_identity.1, + config.store_identity.2, + ); + let store = HostKvStore::new( + transport.clone(), + identity, + config.max_read_response_bytes, + config.max_control_response_bytes, + ) + .map_err(|error| napi::Error::new("E_INVALID_ARGUMENT", error.to_string()))?; + Ok((KvDirectory::with_namespace(store, &config.namespace), transport)) +} + impl KvStore for HostKvStore { fn identity(&self) -> KvStoreIdentity { self.identity @@ -789,6 +933,7 @@ impl ResponseDecoder { } } +#[cfg(feature = "test-panic")] fn registry() -> std::sync::MutexGuard<'static, HashMap>> { HOST_TRANSPORTS .get_or_init(Default::default) @@ -796,6 +941,14 @@ fn registry() -> std::sync::MutexGuard<'static, HashMap> .unwrap_or_else(|poisoned| poisoned.into_inner()) } +fn pending_registry() -> std::sync::MutexGuard<'static, HashMap>> { + PENDING_TRANSPORTS + .get_or_init(Default::default) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +#[cfg(feature = "test-panic")] fn completion(callback: JsFunction) -> boundary::Result { callback .create_threadsafe_function::, Buffer, _, ErrorStrategy::Fatal>( @@ -805,6 +958,7 @@ fn completion(callback: JsFunction) -> boundary::Result { .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string())) } +#[cfg(feature = "test-panic")] fn test_result(result: io::Result>) -> Vec { match result { Ok(bytes) => { @@ -823,6 +977,7 @@ fn test_result(result: io::Result>) -> Vec { } } +#[cfg(feature = "test-panic")] fn test_thread_result(operation: impl FnOnce() -> io::Result>) -> Vec { match catch_unwind(AssertUnwindSafe(operation)) { Ok(result) => test_result(result), @@ -830,6 +985,7 @@ fn test_thread_result(operation: impl FnOnce() -> io::Result>) -> Vec, @@ -968,13 +1124,7 @@ pub fn test_hold_host_transport_capacity( .get(&handle) .cloned() .ok_or_else(|| napi::Error::new("E_CLOSED", "unknown or closed host storage transport"))?; - let request_id = transport.next_request_id.fetch_add(1, Ordering::Relaxed); - if request_id == 0 { - return Err(napi::Error::new( - "E_NATIVE_FAILURE", - "host storage request id space exhausted", - )); - } + let request_id = next_request_id().map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?; let response = Arc::new(ResponseSlot::new()); transport .admit( diff --git a/src/lib.rs b/src/lib.rs index c092cb4..c94c7a8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -15,7 +15,7 @@ pub mod rocks_lease; #[cfg(feature = "node-api")] mod boundary; -#[cfg(feature = "test-panic")] +#[cfg(feature = "host-storage")] pub mod host_storage; #[cfg(feature = "node-api")] @@ -49,11 +49,15 @@ pub struct RuntimeInfo { #[cfg(feature = "node-api")] #[napi(catch_unwind, js_name = "runtimeInfo")] pub fn runtime_info() -> boundary::Result { + #[cfg(feature = "host-storage")] + let storage_backends = vec!["native".to_owned(), "harper".to_owned()]; + #[cfg(not(feature = "host-storage"))] + let storage_backends = vec!["native".to_owned()]; boundary::run_stateless(|| RuntimeInfo { package_version: env!("CARGO_PKG_VERSION").to_owned(), tantivy_version: TANTIVY_VERSION.to_owned(), native_abi_version: NATIVE_ABI_VERSION, - storage_backends: vec!["native".to_owned()], + storage_backends, }) } diff --git a/src/native.rs b/src/native.rs index ebbf069..da2810e 100644 --- a/src/native.rs +++ b/src/native.rs @@ -12,12 +12,17 @@ use napi::bindgen_prelude::Buffer; use napi::threadsafe_function::{ErrorStrategy, ThreadSafeCallContext, ThreadsafeFunction, ThreadsafeFunctionCallMode}; use napi::{Env, JsFunction}; use napi_derive::napi; +use tantivy::directory::Directory; use tantivy::directory::MmapDirectory; use tantivy::IndexReader; use crate::boundary; use crate::engine::{Engine, SearchResult, TotalRelation, Writer}; use crate::error::{FulltextError, Result}; +#[cfg(feature = "host-storage")] +use crate::host_storage::HostTransport; +#[cfg(feature = "host-storage")] +use crate::protocol::decode_host_open; use crate::protocol::{ decode_batch, decode_open, decode_search, validate_batch_header, validate_search_header, EngineConfig, }; @@ -34,7 +39,7 @@ static REGISTRY: OnceLock> = OnceLock::new(); #[derive(Default)] struct Registry { handles: HashMap>, - paths: HashMap, + identities: HashMap, opening: HashSet, cancelled: HashSet, environments: HashMap>, @@ -48,9 +53,19 @@ enum PathIdentity { Path(PathBuf), } +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +enum RuntimeIdentity { + Native(PathIdentity), + #[cfg(feature = "host-storage")] + Host { + store: (u64, u64, u64), + namespace: Vec, + }, +} + struct Runtime { handle: u32, - path_identity: PathIdentity, + identity: RuntimeIdentity, config: EngineConfig, engine: Arc, reader: Arc, @@ -66,6 +81,18 @@ struct Runtime { search_execution_nanoseconds: AtomicU64, search_threads: Mutex>>, closed: Arc, + #[cfg(feature = "host-storage")] + host_transport: Option>, +} + +struct RuntimeParts { + identity: RuntimeIdentity, + config: EngineConfig, + engine: Engine, + writer: Writer, + reader: IndexReader, + #[cfg(feature = "host-storage")] + host_transport: Option>, } struct CompletionSignal { @@ -75,7 +102,13 @@ struct CompletionSignal { struct EnvironmentState { alive: Arc, - handles: Mutex>>, + handles: Mutex>, +} + +struct TrackedHandle { + opening_done: Arc, + #[cfg(feature = "host-storage")] + host_transport: Option>, } struct QueueState { @@ -113,9 +146,13 @@ struct WriterCommand { enum WriterOperation { Apply(Vec), - Commit, + Commit(Option), + #[cfg(feature = "host-storage")] + Publish(String), Reload, - Close { rollback: bool }, + Close { + rollback: bool, + }, } enum WriterOutcome { @@ -154,6 +191,42 @@ pub fn native_open(env: Env, packed_config: Buffer, callback: JsFunction) -> bou })? } +#[cfg(feature = "host-storage")] +#[napi(catch_unwind, skip_typescript, js_name = "__harperOpen")] +pub fn harper_open(env: Env, packed_config: Buffer, handler: JsFunction, callback: JsFunction) -> boundary::Result<()> { + boundary::run_stateless(|| { + let config = decode_host_open(&packed_config).map_err(fulltext_napi_error)?; + 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)?; + let (directory, transport) = crate::host_storage::open_directory(&env, handler, &config)?; + registry().opening.insert(handle); + environment.track_host(handle, opening_done.clone(), transport.clone()); + let thread_opening_done = opening_done.clone(); + let thread_environment = environment.clone(); + if let Err(error) = thread::Builder::new() + .name(format!("fulltext-harper-open-{handle}")) + .spawn(move || { + open_host_on_thread( + handle, + config, + directory, + transport, + 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(|| { @@ -182,7 +255,29 @@ pub fn native_commit(handle: u32, callback: JsFunction) -> boundary::Result<()> let completion = completion(callback, runtime.environment.alive.clone())?; runtime.enqueue_writer( WriterCommand { - operation: WriterOperation::Commit, + operation: WriterOperation::Commit(None), + completion, + }, + 0, + ) + })? +} + +#[cfg(feature = "host-storage")] +#[napi(catch_unwind, skip_typescript, js_name = "__harperPublish")] +pub fn harper_publish(handle: u32, payload: String, callback: JsFunction) -> boundary::Result<()> { + boundary::run_stateless(|| { + if payload.len() > crate::engine::MAX_COMMIT_PAYLOAD_BYTES { + return Err(fulltext_napi_error(FulltextError::invalid(format!( + "commit payload exceeds {} UTF-8 bytes", + crate::engine::MAX_COMMIT_PAYLOAD_BYTES + )))); + } + let runtime = runtime(handle)?; + let completion = completion(callback, runtime.environment.alive.clone())?; + runtime.enqueue_writer( + WriterCommand { + operation: WriterOperation::Publish(payload), completion, }, 0, @@ -284,30 +379,22 @@ pub fn native_status(handle: u32) -> boundary::Result { } 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; + fn start(handle: u32, environment: Arc, parts: RuntimeParts) -> Result> { + let search_thread_count = parts.config.limits.search_threads; let writer_queue = Arc::new(BoundedQueue::new( - config.limits.max_queued_commands, - config.limits.max_queued_bytes, + parts.config.limits.max_queued_commands, + parts.config.limits.max_queued_bytes, )); let search_queue = Arc::new(BoundedQueue::new( - config.limits.max_queued_commands, - config.limits.max_queued_bytes, + parts.config.limits.max_queued_commands, + parts.config.limits.max_queued_bytes, )); let runtime = Arc::new(Self { handle, - path_identity, - config, - engine: Arc::new(engine), - reader: Arc::new(reader), + identity: parts.identity, + config: parts.config, + engine: Arc::new(parts.engine), + reader: Arc::new(parts.reader), writer_queue, search_queue, state: AtomicU8::new(STATE_OPEN), @@ -320,11 +407,13 @@ impl Runtime { search_execution_nanoseconds: AtomicU64::new(0), search_threads: Mutex::new(Vec::with_capacity(search_thread_count)), closed: Arc::new(CompletionSignal::new()), + #[cfg(feature = "host-storage")] + host_transport: parts.host_transport, }); let writer_runtime = runtime.clone(); thread::Builder::new() .name(format!("fulltext-writer-{handle}")) - .spawn(move || writer_loop(writer_runtime, writer)) + .spawn(move || writer_loop(writer_runtime, parts.writer)) .map_err(FulltextError::native)?; for worker in 0..search_thread_count { let search_runtime = runtime.clone(); @@ -385,8 +474,16 @@ impl Runtime { fn poison(&self, error: FulltextError) { self.state.store(STATE_POISONED, Ordering::Release); + let mut close = None; for command in self.writer_queue.drain() { - command.value.fail(error.clone()); + if matches!(&command.value.operation, WriterOperation::Close { .. }) && close.is_none() { + close = Some(command.value); + } else { + command.value.fail(error.clone()); + } + } + if let Some(close) = close { + let _ = self.writer_queue.push_force(close, 0); } for command in self.search_queue.close() { command.value.completion.failure(error.clone()); @@ -617,7 +714,29 @@ fn writer_loop(runtime: Arc, writer: Writer) { Err(error) => WriterOutcome::Continue(Err(error)), } } - WriterOperation::Commit => match active_writer_mut(&mut writer).and_then(Writer::commit) { + WriterOperation::Commit(payload) => { + match active_writer_mut(&mut writer).and_then(|writer| writer.commit_with_payload(payload.as_deref())) { + 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", + ), + ), + } + } + #[cfg(feature = "host-storage")] + WriterOperation::Publish(payload) => match active_writer_mut(&mut writer) + .and_then(|writer| writer.commit_with_payload(Some(&payload))) + .and_then(|opstamp| { + runtime.reader.reload().map_err(FulltextError::native)?; + Ok(opstamp) + }) { Ok(opstamp) => { runtime.uncommitted_mutations.store(0, Ordering::Release); runtime.commit_opstamp.store(opstamp, Ordering::Release); @@ -627,7 +746,7 @@ fn writer_loop(runtime: Arc, writer: Writer) { Err(error), FulltextError::new( "E_POISONED", - "a prior commit failed and the index generation is terminal", + "a publish failed after writer state changed and the index generation is terminal", ), ), }, @@ -659,9 +778,14 @@ fn writer_loop(runtime: Arc, writer: Writer) { for join in std::mem::take(&mut *lock(&runtime.search_threads)) { let _ = join.join(); } + #[cfg(feature = "host-storage")] + if let Some(transport) = &runtime.host_transport { + transport.wait_idle(); + transport.close(); + } runtime.writer_queue.close(); runtime.state.store(STATE_CLOSED, Ordering::Release); - release_runtime(runtime.handle, &runtime.path_identity, &runtime.environment); + release_runtime(runtime.handle, &runtime.identity, &runtime.environment); runtime.signal_closed(); WriterOutcome::Stop(close_result.map(|()| Vec::new())) } @@ -684,14 +808,22 @@ fn writer_loop(runtime: Arc, writer: Writer) { 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); + #[cfg(feature = "host-storage")] + if let Some(transport) = &runtime.host_transport { + transport.close(); + } + release_runtime(runtime.handle, &runtime.identity, &runtime.environment); runtime.signal_closed(); return; } } } runtime.state.store(STATE_CLOSED, Ordering::Release); - release_runtime(runtime.handle, &runtime.path_identity, &runtime.environment); + #[cfg(feature = "host-storage")] + if let Some(transport) = &runtime.host_transport { + transport.close(); + } + release_runtime(runtime.handle, &runtime.identity, &runtime.environment); runtime.signal_closed(); } @@ -764,9 +896,68 @@ fn open_on_thread( } 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 open = decode_open(&bytes)?; + let canonical = create_and_canonicalize(Path::new(&open.path))?; + let identity = RuntimeIdentity::Native(path_identity(&canonical)?); + let directory = MmapDirectory::open(&canonical).map_err(storage_error)?; + open_runtime_with_directory( + handle, + identity, + open.engine, + directory, + environment, + #[cfg(feature = "host-storage")] + None, + ) + .map(|_| ()) +} + +#[cfg(feature = "host-storage")] +fn open_host_on_thread( + handle: u32, + open: crate::protocol::HostOpenConfig, + directory: crate::phase0::KvDirectory, + transport: Arc, + completion: Completion, + opening_done: Arc, + environment: Arc, +) { + let identity = RuntimeIdentity::Host { + store: open.store_identity, + namespace: open.namespace.clone(), + }; + let result = catch_unwind(AssertUnwindSafe(|| { + open_runtime_with_directory( + handle, + identity, + open.engine, + directory, + environment.clone(), + Some(transport.clone()), + ) + })); + let opened = matches!(result, Ok(Ok(_))); + match result { + Ok(Ok(payload)) => completion.success(host_open_body(handle, payload.as_deref())), + Ok(Err(error)) => completion.failure(error), + Err(_) => completion.failure(FulltextError::new("E_NATIVE_PANIC", "Harper index open panicked")), + } + registry().opening.remove(&handle); + if !opened { + transport.close(); + environment.release(handle); + } + opening_done.signal(); +} + +fn open_runtime_with_directory( + handle: u32, + identity: RuntimeIdentity, + config: EngineConfig, + directory: D, + environment: Arc, + #[cfg(feature = "host-storage")] host_transport: Option>, +) -> Result> { { let mut registry = registry(); if registry.cancelled.remove(&handle) || !environment.alive.load(Ordering::Acquire) { @@ -776,27 +967,31 @@ fn open_runtime(handle: u32, bytes: Vec, environment: Arc) "Node environment closed during index open", )); } - if registry.paths.contains_key(&path_identity) { + if registry.identities.contains_key(&identity) { return Err(FulltextError::new( "E_DUPLICATE_OPEN", "the physical index is already open", )); } - registry.paths.insert(path_identity.clone(), handle); + registry.identities.insert(identity.clone(), handle); } let result = (|| { - let directory = MmapDirectory::open(&canonical).map_err(storage_error)?; let engine = Engine::open(directory, &config)?; + let committed_payload = engine.committed_payload()?; let writer = engine.writer(&config)?; let reader = engine.reader()?; let runtime = Runtime::start( handle, - path_identity.clone(), - config, - engine, - writer, - reader, environment.clone(), + RuntimeParts { + identity: identity.clone(), + config, + engine, + writer, + reader, + #[cfg(feature = "host-storage")] + host_transport, + }, )?; let mut registry = registry(); if registry.cancelled.remove(&handle) || !environment.alive.load(Ordering::Acquire) { @@ -810,10 +1005,10 @@ fn open_runtime(handle: u32, bytes: Vec, environment: Arc) } registry.handles.insert(handle, runtime); registry.opening.remove(&handle); - Ok(()) + Ok(committed_payload) })(); if result.is_err() { - release_runtime(handle, &path_identity, &environment); + release_runtime(handle, &identity, &environment); } result } @@ -895,11 +1090,17 @@ fn environment_state(env: &Env) -> boundary::Result> { fn finish_environment_cleanup(data: EnvironmentHookData) { data.environment.alive.store(false, Ordering::Release); let tracked = data.environment.take_handles(); + #[cfg(feature = "host-storage")] + for handle in tracked.values() { + if let Some(transport) = &handle.host_transport { + transport.close(); + } + } let waits = tracked .into_iter() - .map(|(handle, opening_done)| match cleanup_handle(handle) { + .map(|(handle, tracked)| match cleanup_handle(handle) { Some(runtime) => CleanupWait::Runtime(runtime), - None => CleanupWait::Opening(opening_done), + None => CleanupWait::Opening(tracked.opening_done), }) .collect::>(); let remove_environment = registry() @@ -922,14 +1123,32 @@ fn finish_environment_cleanup(data: EnvironmentHookData) { impl EnvironmentState { fn track(&self, handle: u32, opening_done: Arc) { - lock(&self.handles).insert(handle, opening_done); + lock(&self.handles).insert( + handle, + TrackedHandle { + opening_done, + #[cfg(feature = "host-storage")] + host_transport: None, + }, + ); + } + + #[cfg(feature = "host-storage")] + fn track_host(&self, handle: u32, opening_done: Arc, host_transport: Arc) { + lock(&self.handles).insert( + handle, + TrackedHandle { + opening_done, + host_transport: Some(host_transport), + }, + ); } fn release(&self, handle: u32) { lock(&self.handles).remove(&handle); } - fn take_handles(&self) -> HashMap> { + fn take_handles(&self) -> HashMap { mem::take(&mut *lock(&self.handles)) } } @@ -943,11 +1162,11 @@ impl CleanupWait { } } -fn release_runtime(handle: u32, identity: &PathIdentity, environment: &EnvironmentState) { +fn release_runtime(handle: u32, identity: &RuntimeIdentity, environment: &EnvironmentState) { let mut registry = registry(); registry.handles.remove(&handle); - if registry.paths.get(identity) == Some(&handle) { - registry.paths.remove(identity); + if registry.identities.get(identity) == Some(&handle) { + registry.identities.remove(identity); } drop(registry); environment.release(handle); @@ -1008,6 +1227,16 @@ fn u32_body(value: u32) -> Vec { value.to_le_bytes().to_vec() } +#[cfg(feature = "host-storage")] +fn host_open_body(handle: u32, payload: Option<&str>) -> Vec { + let mut bytes = u32_body(handle); + bytes.push(u8::from(payload.is_some())); + if let Some(payload) = payload { + push_string(&mut bytes, payload); + } + bytes +} + fn u64_body(value: u64) -> Vec { value.to_le_bytes().to_vec() } diff --git a/src/phase0.rs b/src/phase0.rs index a17fee9..8c77d59 100644 --- a/src/phase0.rs +++ b/src/phase0.rs @@ -93,6 +93,7 @@ pub trait KvStore: Clone + Send + Sync + 'static { pub(crate) const CHUNK_SIZE: usize = 256 * 1024; +#[cfg(any(test, feature = "test-panic"))] pub(crate) fn reclaim_read_key_bytes(namespace: &[u8]) -> usize { // Tail objects have the longest key read by reclamation. tail_key(namespace, u64::MAX, u64::MAX).len() diff --git a/src/protocol.rs b/src/protocol.rs index 38f0efe..65b08ff 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -22,7 +22,6 @@ pub struct Limits { #[derive(Clone, Debug, PartialEq)] pub struct EngineConfig { - pub path: String, pub index_id: String, pub generation: String, pub fields: Vec, @@ -33,6 +32,24 @@ pub struct EngineConfig { pub limits: Limits, } +#[derive(Clone, Debug, PartialEq)] +pub struct NativeOpenConfig { + pub path: String, + pub engine: EngineConfig, +} + +#[derive(Clone, Debug, PartialEq)] +pub struct HostOpenConfig { + pub engine: EngineConfig, + pub store_identity: (u64, u64, u64), + pub namespace: Vec, + pub max_operations: usize, + pub max_transport_bytes: usize, + pub read_timeout: std::time::Duration, + pub max_read_response_bytes: usize, + pub max_control_response_bytes: usize, +} + #[derive(Clone, Debug, PartialEq, Eq)] pub struct Upsert { pub id: String, @@ -61,9 +78,57 @@ pub struct SearchRequest { pub exact_total: bool, } -pub fn decode_open(bytes: &[u8]) -> Result { +pub fn decode_open(bytes: &[u8]) -> Result { let mut cursor = Cursor::new(bytes, *b"FTOP")?; let path = cursor.string()?; + let engine = decode_engine_config(&mut cursor)?; + cursor.finish()?; + if path.is_empty() { + return Err(FulltextError::invalid("path must not be empty")); + } + Ok(NativeOpenConfig { path, engine }) +} + +pub fn decode_host_open(bytes: &[u8]) -> Result { + let mut cursor = Cursor::new(bytes, *b"FTHO")?; + let store_identity = (cursor.u64()?, cursor.u64()?, cursor.u64()?); + if store_identity == (0, 0, 0) { + return Err(FulltextError::invalid("store identity must not be all zero")); + } + let namespace = cursor.bytes()?.to_vec(); + if namespace.is_empty() { + return Err(FulltextError::invalid("namespace must not be empty")); + } + let max_operations = cursor.u32()? as usize; + let max_transport_bytes = cursor.u64_usize()?; + let read_timeout_ms = cursor.u64()?; + let max_read_response_bytes = cursor.u64_usize()?; + let max_control_response_bytes = cursor.u64_usize()?; + let engine = decode_engine_config(&mut cursor)?; + cursor.finish()?; + if max_operations == 0 + || max_transport_bytes == 0 + || read_timeout_ms == 0 + || max_read_response_bytes == 0 + || max_control_response_bytes == 0 + { + return Err(FulltextError::invalid( + "host transport limits must be greater than zero", + )); + } + Ok(HostOpenConfig { + engine, + store_identity, + namespace, + max_operations, + max_transport_bytes, + read_timeout: std::time::Duration::from_millis(read_timeout_ms), + max_read_response_bytes, + max_control_response_bytes, + }) +} + +fn decode_engine_config(cursor: &mut Cursor<'_>) -> Result { let index_id = cursor.string()?; let generation = cursor.string()?; let analyzer = cursor.string()?; @@ -93,9 +158,7 @@ pub fn decode_open(bytes: &[u8]) -> Result { max_queued_bytes: cursor.u64_usize()?, max_batch_bytes: cursor.u64_usize()?, }; - cursor.finish()?; validate_config(EngineConfig { - path, index_id, generation, fields, @@ -215,10 +278,8 @@ pub fn decode_search(bytes: &[u8]) -> Result { } 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.is_empty() || config.generation.is_empty() { + return Err(FulltextError::invalid("indexId and generation must not be empty")); } if config.index_id.len() > 4_096 || config.generation.len() > 4_096 { return Err(FulltextError::invalid( @@ -315,11 +376,23 @@ impl<'a> Cursor<'a> { } fn u64_usize(&mut self) -> Result { + let value = self.u64()?; + usize::try_from(value).map_err(|_| FulltextError::invalid("numeric limit exceeds usize")) + } + + fn u64(&mut self) -> Result { let bytes = self.take(8)?; - let value = u64::from_le_bytes([ + Ok(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 bytes(&mut self) -> Result<&'a [u8]> { + let length = self.u32()? as usize; + if length > MAX_STRING_BYTES { + return Err(FulltextError::invalid("packed byte string exceeds 1 MiB")); + } + self.take(length) } fn f32(&mut self) -> Result { @@ -328,11 +401,7 @@ impl<'a> Cursor<'a> { } 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)?; + let bytes = self.bytes()?; String::from_utf8(bytes.to_vec()).map_err(|_| FulltextError::invalid("packed string is not valid UTF-8")) } diff --git a/test/fixtures/harper-worker-child.mjs b/test/fixtures/harper-worker-child.mjs new file mode 100644 index 0000000..449737f --- /dev/null +++ b/test/fixtures/harper-worker-child.mjs @@ -0,0 +1,101 @@ +import { parentPort, workerData } from 'node:worker_threads'; + +import { encodeMutationBatch, openHarperFullTextIndex } from '../../dist/harper.js'; + +const entries = new Map(); +const wait = new Int32Array(workerData.control); +let blocked = workerData.stage === 'open'; +let reported = false; + +function maybeBlock(operation) { + if (!blocked || (workerData.stage === 'read' && operation !== 'read')) return; + if (!reported) { + reported = true; + parentPort.postMessage('blocked'); + } + blocked = false; + Atomics.wait(wait, 0, 0, 250); + if (Atomics.load(wait, 0) === 1) throw new Error('host generation was revoked'); +} + +const storage = { + read(key) { + maybeBlock('read'); + const value = entries.get(key.toString('hex')); + return value && Buffer.from(value); + }, + write(mutations) { + maybeBlock('write'); + for (const mutation of mutations) { + const key = mutation.key.toString('hex'); + if (mutation.type === 'put') entries.set(key, Buffer.from(mutation.value)); + else entries.delete(key); + } + }, + sync() { + maybeBlock('sync'); + }, +}; + +const config = { + storage, + storeIdentity: [101n, 202n, 303n], + namespace: Buffer.from(`worker-${workerData.stage}`), + indexId: `worker-${workerData.stage}`, + generation: 'one', + fields: [{ name: 'title' }], + analyzer: 'english@1', + limits: { + indexingThreads: 1, + searchThreads: 1, + writerMemoryBytes: 15_000_000, + maxQueuedCommands: 8, + maxQueuedBytes: 32 * 1024 * 1024, + maxBatchBytes: 32 * 1024 * 1024, + }, + transport: { + maxOperations: 8, + maxBytes: 40 * 1024 * 1024, + readTimeoutMs: 30_000, + maxMutations: 4_096, + maxReadResponseBytes: 1024 * 1024, + maxControlResponseBytes: 1024 * 1024, + maxErrorBytes: 64 * 1024, + }, +}; + +try { + const index = await openHarperFullTextIndex(config); + if (workerData.stage === 'open') throw new Error('open unexpectedly passed the blocked storage call'); + const seed = encodeMutationBatch({ upserts: [{ id: 'seed', fields: { title: 'running shoe' } }] }); + if (workerData.stage === 'publish') { + await index.apply(seed); + blocked = true; + await index.publish('cursor'); + } else if (workerData.stage === 'read') { + await index.apply(seed); + await index.publish('cursor'); + blocked = true; + await index.search({ text: 'running' }); + } else if (workerData.stage === 'apply') { + blocked = false; + reported = true; + const batch = encodeMutationBatch( + { + upserts: Array.from({ length: 50_000 }, (_, id) => ({ + id: String(id), + fields: { title: `worker-owned running product ${id}` }, + })), + }, + config.limits.maxBatchBytes, + ); + const applying = index.apply(batch); + parentPort.postMessage('blocked'); + await applying; + } else { + throw new Error(`unknown worker stage ${workerData.stage}`); + } + parentPort.postMessage('unexpected-completion'); +} catch (error) { + parentPort.postMessage({ error: error instanceof Error ? error.message : String(error) }); +} diff --git a/test/fixtures/host-storage-transport-worker.mjs b/test/fixtures/host-storage-transport-worker.mjs index fe1b084..2d93f03 100644 --- a/test/fixtures/host-storage-transport-worker.mjs +++ b/test/fixtures/host-storage-transport-worker.mjs @@ -3,7 +3,16 @@ import { parentPort } from 'node:worker_threads'; import { loadAddon } from '../../dist/load-addon.js'; const addon = loadAddon(); -const handle = addon.__testOpenHostTransport((request) => request, 2, 1_024, 1_000); +const handle = addon.__testOpenHostTransport( + (dispatch) => { + const requestId = dispatch.readBigUInt64LE().toString(); + if (!addon.__hostStorageBegin(requestId)) return; + addon.__hostStorageComplete(requestId, dispatch.subarray(8)); + }, + 2, + 1_024, + 1_000, +); addon.__testConfigureHostTransportCleanup(handle, 256, 512); addon.__testHoldHostTransportCapacity(handle, 8, 128, true); addon.__testHoldHostTransportCapacity(handle, 10, 128, false); diff --git a/test/harper-index.test.mjs b/test/harper-index.test.mjs new file mode 100644 index 0000000..0c2de0e --- /dev/null +++ b/test/harper-index.test.mjs @@ -0,0 +1,145 @@ +import assert from 'node:assert'; +import test from 'node:test'; +import { Worker } from 'node:worker_threads'; + +import { encodeMutationBatch, openHarperFullTextIndex } from '@harperfast/fulltext/harper'; + +const readResponseBytes = 1024 * 1024; +const controlResponseBytes = 1024 * 1024; + +function createStorage() { + const entries = new Map(); + const calls = []; + return { + calls, + storage: { + read(key) { + calls.push('read'); + const value = entries.get(key.toString('hex')); + return value && Buffer.from(value); + }, + write(mutations, policy) { + calls.push(`write:${policy}`); + const next = new Map(entries); + for (const mutation of mutations) { + const key = mutation.key.toString('hex'); + if (mutation.type === 'put') next.set(key, Buffer.from(mutation.value)); + else next.delete(key); + } + entries.clear(); + for (const [key, value] of next) entries.set(key, value); + }, + sync() { + calls.push('sync'); + }, + }, + }; +} + +function options(storage, overrides = {}) { + return { + storage, + storeIdentity: [11n, 22n, 33n], + namespace: Buffer.from('products-title'), + indexId: 'products-title', + 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, + }, + transport: { + maxOperations: 32, + maxBytes: 40 * 1024 * 1024, + readTimeoutMs: 5_000, + maxMutations: 4_096, + maxReadResponseBytes: readResponseBytes, + maxControlResponseBytes: controlResponseBytes, + maxErrorBytes: 64 * 1024, + }, + ...overrides, + }; +} + +test('publishes, searches, closes, and reopens entirely through host storage', async () => { + const host = createStorage(); + const config = options(host.storage); + let index = await openHarperFullTextIndex(config); + assert.strictEqual(index.committedPayload, undefined); + 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' } }, + ], + }), + ); + await index.publish('cursor-v1:42'); + assert.strictEqual(index.committedPayload, 'cursor-v1:42'); + assert.deepStrictEqual( + (await index.search({ text: 'running shoes', exactTotal: true })).hits.map((hit) => hit.id), + ['shoe-1', 'rack-1'], + ); + await index.close(); + + const callsAfterClose = host.calls.length; + await new Promise((resolve) => setImmediate(resolve)); + assert.strictEqual(host.calls.length, callsAfterClose, 'storage was entered after close resolved'); + + index = await openHarperFullTextIndex(config); + assert.strictEqual(index.committedPayload, 'cursor-v1:42'); + assert.deepStrictEqual( + (await index.search({ text: 'running shoes', exactTotal: true })).hits.map((hit) => hit.id), + ['shoe-1', 'rack-1'], + ); + await index.close(); + assert(host.calls.includes('write:wal')); + assert(host.calls.includes('sync')); +}); + +test('rejects duplicate owner-writer opens for one process identity and namespace', async () => { + const host = createStorage(); + const config = options(host.storage); + const first = await openHarperFullTextIndex(config); + await assert.rejects(openHarperFullTextIndex(config), (error) => error.code === 'E_DUPLICATE_OPEN'); + await first.close(); +}); + +test('rejects an oversized cursor without changing or poisoning the generation', async () => { + const host = createStorage(); + const index = await openHarperFullTextIndex(options(host.storage)); + await assert.rejects(index.publish('x'.repeat(64 * 1024 + 1)), (error) => error.code === 'E_INVALID_ARGUMENT'); + assert.strictEqual(index.status().state, 'open'); + await index.close(); +}); + +for (const stage of ['open', 'read', 'apply', 'publish']) { + test(`worker termination safely drains a hosted runtime during ${stage}`, async () => { + const control = new SharedArrayBuffer(4); + const worker = new Worker(new URL('./fixtures/harper-worker-child.mjs', import.meta.url), { + workerData: { stage, control }, + }); + await new Promise((resolve, reject) => { + const timeout = setTimeout(() => reject(new Error(`${stage} did not enter host storage`)), 10_000); + worker.once('error', reject); + worker.on('message', (message) => { + if (message === 'blocked') { + clearTimeout(timeout); + resolve(); + } else if (message?.error) { + clearTimeout(timeout); + reject(new Error(message.error)); + } + }); + }); + const started = performance.now(); + Atomics.store(new Int32Array(control), 0, 1); + await worker.terminate(); + assert(performance.now() - started < 5_000, `${stage} teardown did not drain bounded storage work promptly`); + }); +} diff --git a/test/host-storage-transport.test.mjs b/test/host-storage-transport.test.mjs index a9242a6..744c36b 100644 --- a/test/host-storage-transport.test.mjs +++ b/test/host-storage-transport.test.mjs @@ -6,6 +6,8 @@ import { createHostStorageHandler } from '../dist/host-storage.js'; import { loadAddon } from '../dist/load-addon.js'; const addon = loadAddon(); +const openHostTransport = addon.__testOpenHostTransport.bind(addon); +addon.__testOpenHostTransport = (handler, ...limits) => openHostTransport(dispatchHostStorage(handler), ...limits); const readResponseBytes = 1024 * 1024; const controlResponseBytes = 64 * 1024; @@ -188,6 +190,30 @@ test('close wakes a request waiting indefinitely for foreground admission', asyn await assert.rejects(waiting, /host storage transport is closed/); }); +test('close fences a host callback already queued for JavaScript', async () => { + let calls = 0; + const handle = addon.__testOpenHostTransport( + (request) => { + calls++; + return request; + }, + 1, + 1_024, + 1_000, + ); + const pending = roundTrip(handle, Buffer.from('queued'), 128, false); + const pause = new Int32Array(new SharedArrayBuffer(4)); + const deadline = performance.now() + 1_000; + while (addon.__testHostTransportStats(handle)[0] !== '1') { + if (performance.now() >= deadline) throw new Error('request was not admitted'); + Atomics.wait(pause, 0, 0, 1); + } + assert.strictEqual(addon.__testCloseHostTransport(handle), true); + await assert.rejects(pending, /host storage transport is closed/); + await new Promise((resolve) => setImmediate(resolve)); + assert.strictEqual(calls, 0); +}); + test('invalid callback responses fail the request without terminating the process', async (context) => { for (const invalid of ['not a buffer', {}]) { const handle = addon.__testOpenHostTransport(() => invalid, 1, 1024, 1_000); @@ -445,3 +471,15 @@ function decodeHandlerError(response) { const length = response.readUInt32LE(2); return response.subarray(6, 6 + length).toString(); } + +function dispatchHostStorage(handler) { + return (dispatch) => { + const requestId = dispatch.readBigUInt64LE().toString(); + if (!addon.__hostStorageBegin(requestId)) return; + try { + addon.__hostStorageComplete(requestId, handler(dispatch.subarray(8))); + } catch (error) { + addon.__hostStorageFail(requestId, error instanceof Error ? error.message : String(error)); + } + }; +} diff --git a/test/native.test.mjs b/test/native.test.mjs index ede020f..d065fef 100644 --- a/test/native.test.mjs +++ b/test/native.test.mjs @@ -20,7 +20,7 @@ test('loads the artifact for the executing platform', async () => { packageVersion: packageManifest.version, tantivyVersion, nativeAbiVersion: 1, - storageBackends: ['native'], + storageBackends: ['native', 'harper'], }); assert.strictEqual(cargoPackageVersion, packageManifest.version); assert.match(platformTriple(), /^(darwin|linux|win32)-(arm64|x64)(-(gnu|musl|msvc))?$/); diff --git a/test/package.release.test.mjs b/test/package.release.test.mjs index 2f7ef7e..980abd3 100644 --- a/test/package.release.test.mjs +++ b/test/package.release.test.mjs @@ -32,6 +32,7 @@ test('the packed package loads without consumer lifecycle scripts', (context) => const { filename, files } = pack; const includedPaths = files.map((file) => file.path); assert(includedPaths.includes('dist/native.js')); + assert(includedPaths.includes('dist/harper.js')); assert(includedPaths.some((file) => /^fulltext\..+\.node$/.test(file))); assert(!includedPaths.some((file) => file.startsWith('src/') || file === 'ts/addon.d.ts')); assert.doesNotMatch(readFileSync(new URL('../ts/addon.d.ts', import.meta.url), 'utf8'), /__(?:test|phase0)/); @@ -67,7 +68,7 @@ test('the packed package loads without consumer lifecycle scripts', (context) => [ '--input-type=module', '--eval', - "import { createRequire } from 'node:module'; const addon = createRequire(import.meta.url)(process.argv[1]); if (Object.keys(addon).some(key => key.startsWith('__test') || key.startsWith('__phase0'))) process.exit(1); console.log(await import('@harperfast/fulltext/native').then(x => x.runtimeInfo()));", + "import { createRequire } from 'node:module'; const addon = createRequire(import.meta.url)(process.argv[1]); if (Object.keys(addon).some(key => key.startsWith('__test') || key.startsWith('__phase0'))) process.exit(1); await import('@harperfast/fulltext/harper'); console.log(await import('@harperfast/fulltext/native').then(x => x.runtimeInfo()));", installedAddonPath, ], { cwd: projectDirectory, encoding: 'utf8' }, diff --git a/ts/codec.ts b/ts/codec.ts index 0f08a52..8452259 100644 --- a/ts/codec.ts +++ b/ts/codec.ts @@ -7,8 +7,7 @@ export interface PackedFieldConfig { weight: number; } -export interface PackedOpenConfig { - path: string; +export interface PackedEngineConfig { indexId: string; generation: string; fields: PackedFieldConfig[]; @@ -26,6 +25,22 @@ export interface PackedOpenConfig { }; } +export interface PackedOpenConfig extends PackedEngineConfig { + path: string; +} + +export interface PackedHostOpenConfig extends PackedEngineConfig { + storeIdentity: readonly [bigint, bigint, bigint]; + namespace: Uint8Array; + transport: { + maxOperations: number; + maxBytes: number; + readTimeoutMs: number; + maxReadResponseBytes: number; + maxControlResponseBytes: number; + }; +} + export interface PackedMutationBatch { upserts: Array<{ id: string; fields: Record }>; deletes: string[]; @@ -44,6 +59,25 @@ export function encodeOpen(config: PackedOpenConfig): Buffer { const writer = new ByteWriter(Number.MAX_SAFE_INTEGER); writer.header('FTOP'); writer.string(config.path); + encodeEngine(writer, config); + return writer.finish(); +} + +export function encodeHostOpen(config: PackedHostOpenConfig): Buffer { + const writer = new ByteWriter(Number.MAX_SAFE_INTEGER); + writer.header('FTHO'); + for (const identity of config.storeIdentity) writer.u64BigInt(identity, 'storeIdentity'); + writer.byteString(config.namespace); + writer.u32(config.transport.maxOperations, 'transport.maxOperations'); + writer.u64(config.transport.maxBytes, 'transport.maxBytes'); + writer.u64(config.transport.readTimeoutMs, 'transport.readTimeoutMs'); + writer.u64(config.transport.maxReadResponseBytes, 'transport.maxReadResponseBytes'); + writer.u64(config.transport.maxControlResponseBytes, 'transport.maxControlResponseBytes'); + encodeEngine(writer, config); + return writer.finish(); +} + +function encodeEngine(writer: ByteWriter, config: PackedEngineConfig): void { writer.string(config.indexId); writer.string(config.generation); writer.string(config.analyzer); @@ -61,7 +95,6 @@ export function encodeOpen(config: PackedOpenConfig): Buffer { 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 { @@ -226,6 +259,15 @@ class ByteWriter { this.bytes(buffer); } + u64BigInt(value: bigint, name: string): void { + if (typeof value !== 'bigint' || value < 0n || value > 0xffff_ffff_ffff_ffffn) { + throw new FulltextError('E_INVALID_ARGUMENT', `${name} is outside its packed integer range`); + } + const buffer = Buffer.allocUnsafe(8); + buffer.writeBigUInt64LE(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`); @@ -244,6 +286,15 @@ class ByteWriter { this.bytes(bytes); } + byteString(value: Uint8Array): void { + if (!(value instanceof Uint8Array)) { + throw new FulltextError('E_INVALID_ARGUMENT', 'packed byte string values must be Uint8Array instances'); + } + const bytes = Buffer.from(value.buffer, value.byteOffset, value.byteLength); + this.u32(bytes.length, 'byte string length'); + this.bytes(bytes); + } + finish(): Buffer { return Buffer.concat(this.#chunks, this.#length); } diff --git a/ts/harper.ts b/ts/harper.ts new file mode 100644 index 0000000..d804607 --- /dev/null +++ b/ts/harper.ts @@ -0,0 +1,177 @@ +import { encodeHostOpen } from './codec.js'; +import { FulltextError } from './errors.js'; +import { createHostStorageHandler } from './host-storage.js'; +import type { HostStorage } from './host-storage.js'; +import { invoke } from './invoke.js'; +import { loadAddon } from './load-addon.js'; +import { NativeFullTextIndex } from './native.js'; +import type { + CloseOptions, + FullTextStatus, + NativeFullTextIndexOptions, + SearchRequest, + SearchResult, +} from './native.js'; + +export { encodeMutationBatch, FulltextError, runtimeInfo } from './native.js'; +export type { + CloseOptions, + FullTextMutationBatch, + FullTextStatus, + RuntimeInfo, + SearchRequest, + SearchResult, +} from './native.js'; +export type { HostStorage, HostStorageMutation, HostWritePolicy } from './host-storage.js'; + +const maxCommitPayloadBytes = 64 * 1024; + +export interface HarperFullTextIndexOptions extends Omit { + storage: HostStorage; + storeIdentity: readonly [bigint, bigint, bigint]; + namespace: Uint8Array; + transport: { + maxOperations: number; + maxBytes: number; + readTimeoutMs: number; + maxMutations: number; + maxReadResponseBytes: number; + maxControlResponseBytes: number; + maxErrorBytes: number; + }; +} + +class StorageGate implements HostStorage { + readonly #storage: HostStorage; + #active = true; + + constructor(storage: HostStorage) { + this.#storage = storage; + } + + read(key: Buffer): Buffer | undefined { + this.#requireActive(); + return this.#storage.read(key); + } + + write(mutations: Parameters[0], policy: Parameters[1]): void { + this.#requireActive(); + this.#storage.write(mutations, policy); + } + + sync(): void { + this.#requireActive(); + this.#storage.sync(); + } + + revoke(): void { + this.#active = false; + } + + #requireActive(): void { + if (!this.#active) throw new FulltextError('E_CLOSED', 'Harper storage generation is closed'); + } +} + +export class HarperFullTextIndex { + readonly #handle: number; + readonly #index: NativeFullTextIndex; + readonly #storageGate: StorageGate; + #committedPayload?: string; + + constructor(handle: number, committedPayload: string | undefined, storageGate: StorageGate) { + this.#handle = handle; + this.#index = new NativeFullTextIndex(handle); + this.#committedPayload = committedPayload; + this.#storageGate = storageGate; + } + + get committedPayload(): string | undefined { + return this.#committedPayload; + } + + apply(packedBatch: Uint8Array): Promise { + return this.#index.apply(packedBatch); + } + + async publish(payload: string): Promise { + if (typeof payload !== 'string') throw new FulltextError('E_INVALID_ARGUMENT', 'commit payload must be a string'); + if (Buffer.byteLength(payload) > maxCommitPayloadBytes) { + throw new FulltextError('E_INVALID_ARGUMENT', `commit payload exceeds ${maxCommitPayloadBytes} UTF-8 bytes`); + } + const cursor = await invoke((callback) => loadAddon().__harperPublish(this.#handle, payload, callback)); + const opstamp = cursor.u64(); + cursor.finish(); + this.#committedPayload = payload; + return opstamp; + } + + search(request: SearchRequest): Promise { + return this.#index.search(request); + } + + status(): FullTextStatus { + return this.#index.status(); + } + + async close(options: CloseOptions = {}): Promise { + await this.#index.close(options); + this.#storageGate.revoke(); + } +} + +export async function openHarperFullTextIndex(options: HarperFullTextIndexOptions): Promise { + const gate = new StorageGate(options.storage); + const handler = createHostStorageHandler(gate, { + maxMutations: options.transport.maxMutations, + maxReadResponseBytes: options.transport.maxReadResponseBytes, + maxControlResponseBytes: options.transport.maxControlResponseBytes, + maxErrorBytes: options.transport.maxErrorBytes, + }); + const dispatch = createStorageDispatcher(handler); + try { + const cursor = await invoke((callback) => + loadAddon().__harperOpen( + encodeHostOpen({ + storeIdentity: options.storeIdentity, + namespace: options.namespace, + indexId: options.indexId, + generation: options.generation, + fields: options.fields.map((field) => ({ name: field.name, weight: field.weight ?? 1 })), + analyzer: options.analyzer, + stopWords: options.stopWords ?? true, + positions: options.positions ?? true, + surfaceTerms: options.surfaceTerms ?? false, + limits: options.limits, + transport: options.transport, + }), + dispatch, + callback, + ), + ); + const handle = cursor.u32(); + const hasPayload = cursor.u8(); + if (hasPayload !== 0 && hasPayload !== 1) { + throw new FulltextError('E_NATIVE_FAILURE', `Unknown committed payload status ${hasPayload}`); + } + const committedPayload = hasPayload === 1 ? cursor.string() : undefined; + cursor.finish(); + return new HarperFullTextIndex(handle, committedPayload, gate); + } catch (error) { + gate.revoke(); + throw error; + } +} + +function createStorageDispatcher(handler: (request: Buffer) => Buffer): (dispatch: Buffer) => void { + return (dispatch) => { + if (dispatch.length < 8) return; + const requestId = dispatch.readBigUInt64LE().toString(); + if (!loadAddon().__hostStorageBegin(requestId)) return; + try { + loadAddon().__hostStorageComplete(requestId, handler(dispatch.subarray(8))); + } catch (error) { + loadAddon().__hostStorageFail(requestId, error instanceof Error ? error.message : String(error)); + } + }; +} diff --git a/ts/invoke.ts b/ts/invoke.ts new file mode 100644 index 0000000..818142c --- /dev/null +++ b/ts/invoke.ts @@ -0,0 +1,19 @@ +import { Cursor, decodeResponse } from './codec.js'; +import { normalizeNativeError } from './errors.js'; +import type { NativeCallback } from './load-addon.js'; + +export function invoke(start: (callback: NativeCallback) => void): Promise { + return new Promise((resolve, reject) => { + try { + start((response) => { + try { + resolve(decodeResponse(response)); + } catch (error) { + reject(normalizeNativeError(error)); + } + }); + } catch (error) { + reject(normalizeNativeError(error)); + } + }); +} diff --git a/ts/load-addon.ts b/ts/load-addon.ts index 96f9786..cd54e83 100644 --- a/ts/load-addon.ts +++ b/ts/load-addon.ts @@ -20,12 +20,17 @@ interface NativeAddonApi { __nativeSearch(handle: number, request: Buffer, callback: NativeCallback): void; __nativeClose(handle: number, rollback: boolean, callback: NativeCallback): void; __nativeStatus(handle: number): Buffer; + __harperOpen(config: Buffer, handler: (dispatch: Buffer) => void, callback: NativeCallback): void; + __harperPublish(handle: number, payload: string, callback: NativeCallback): void; + __hostStorageBegin(requestId: string): boolean; + __hostStorageComplete(requestId: string, response: unknown): boolean; + __hostStorageFail(requestId: string, message: string): boolean; __testCreateHandle?(): number; __testPanic?(id: number): void; __testCheck?(id: number): boolean; __testPoisonNativeHandle?(handle: number): void; __testOpenHostTransport?( - handler: (request: Buffer) => Buffer, + handler: (dispatch: Buffer) => void, maxOperations: number, maxBytes: number, readTimeoutMs: number, @@ -113,7 +118,11 @@ function validateAddon(addon: NativeAddonApi, artifactPath: string): void { `Fulltext native ABI ${info.nativeAbiVersion} from ${artifactPath} does not match ${expectedNativeAbiVersion}`, ); } - if (info.storageBackends.length !== 1 || info.storageBackends[0] !== 'native') { + if ( + info.storageBackends.length !== 2 || + info.storageBackends[0] !== 'native' || + info.storageBackends[1] !== 'harper' + ) { throw new FulltextError( 'E_NATIVE_CAPABILITY_MISMATCH', `Unexpected storage capabilities from ${artifactPath}: ${info.storageBackends.join(', ')}`, diff --git a/ts/native.ts b/ts/native.ts index 28801d6..8385085 100644 --- a/ts/native.ts +++ b/ts/native.ts @@ -1,5 +1,6 @@ import { FulltextError, normalizeNativeError } from './errors.js'; -import { Cursor, decodeResponse, encodeBatch, encodeOpen, encodeSearch } from './codec.js'; +import { decodeResponse, encodeBatch, encodeOpen, encodeSearch } from './codec.js'; +import { invoke } from './invoke.js'; import { loadAddon } from './load-addon.js'; export { FulltextError } from './errors.js'; @@ -9,7 +10,7 @@ export interface RuntimeInfo { packageVersion: string; tantivyVersion: string; nativeAbiVersion: number; - storageBackends: ReadonlyArray<'native'>; + storageBackends: ReadonlyArray<'native' | 'harper'>; } export interface NativeFullTextIndexOptions { @@ -217,29 +218,13 @@ export async function runtimeInfo(): Promise { packageVersion: info.packageVersion, tantivyVersion: info.tantivyVersion, nativeAbiVersion: info.nativeAbiVersion, - storageBackends: ['native'], + storageBackends: ['native', 'harper'], }; } catch (error) { 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); } From 31d7f904eb4dda1653cf0950869541b44d16102e Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Thu, 10 Sep 2026 15:29:47 -0600 Subject: [PATCH 2/6] Tighten hosted storage lifecycle --- README.md | 3 +- src/host_storage.rs | 159 ++++++++++++------ test/fixtures/harper-worker-child.mjs | 2 +- .../host-storage-transport-worker.mjs | 7 +- test/harper-index.test.mjs | 21 ++- test/host-storage-transport.test.mjs | 9 +- ts/harper.ts | 41 +++-- ts/host-storage.ts | 16 +- ts/load-addon.ts | 14 +- 9 files changed, 186 insertions(+), 86 deletions(-) diff --git a/README.md b/README.md index a61c1f1..ae0aa6e 100644 --- a/README.md +++ b/README.md @@ -84,7 +84,8 @@ The Harper opener requires a process-lifetime store identity, persistent generat bounded transport limits, and a `HostStorage` implementation. `publish(payload)` commits the index and opaque payload into one Tantivy `meta.json` generation, then reloads the local reader before it resolves. Harper uses that payload for its derived-index cursor. `committedPayload` exposes the -payload recovered at open. +payload recovered at open. Host storage methods are strictly synchronous; `write` and `sync` must +return `undefined`, and Promise-returning implementations are rejected rather than acknowledged. The current Harper path is owner-worker-only. It does not yet provide non-owner read handles, cross-worker refresh, generation retirement, or scheduled physical reclamation. Those lifecycle diff --git a/src/host_storage.rs b/src/host_storage.rs index a624172..e3ed6dd 100644 --- a/src/host_storage.rs +++ b/src/host_storage.rs @@ -30,21 +30,33 @@ use crate::phase0::{reclaim_read_key_bytes, RECLAIM_MAX_BATCH_REQUEST_BYTES}; use crate::phase0::{KvDirectory, KvStore, KvStoreIdentity, Mutation, WritePolicy, CHUNK_SIZE}; use crate::protocol::HostOpenConfig; -type HostCallback = ThreadsafeFunction, ErrorStrategy::Fatal>; #[cfg(feature = "test-panic")] type CompletionCallback = ThreadsafeFunction, ErrorStrategy::Fatal>; #[cfg(feature = "test-panic")] static NEXT_TRANSPORT_HANDLE: AtomicU32 = AtomicU32::new(1); -static NEXT_REQUEST_ID: AtomicU64 = AtomicU64::new(1); +static NEXT_TRANSPORT_ID: AtomicU64 = AtomicU64::new(1); #[cfg(feature = "test-panic")] static HOST_TRANSPORTS: OnceLock>>> = OnceLock::new(); -static PENDING_TRANSPORTS: OnceLock>>> = OnceLock::new(); +const TRANSPORT_SHARDS: usize = 64; +type TransportRegistry = HashMap>; +type TransportShards = [Mutex; TRANSPORT_SHARDS]; +static TRANSPORTS: OnceLock = OnceLock::new(); + +struct HostDispatch { + transport_id: u64, + request_id: u64, + request: Vec, +} + +type HostCallback = ThreadsafeFunction; pub(crate) struct HostTransport { + id: u64, handler: Mutex>, state: Mutex, capacity: Condvar, + next_request_id: AtomicU64, abandoned_waiters: AtomicU64, max_operations: usize, max_bytes: usize, @@ -103,18 +115,27 @@ impl HostTransport { } // Production construction must supply the total callback created by createHostStorageHandler. let mut handler = handler - .create_threadsafe_function::, Buffer, _, ErrorStrategy::Fatal>( + .create_threadsafe_function::( max_operations, - |context: ThreadSafeCallContext>| Ok(vec![Buffer::from(context.value)]), + |context: ThreadSafeCallContext| { + Ok(vec![ + Buffer::from(context.value.transport_id.to_le_bytes().to_vec()), + Buffer::from(context.value.request_id.to_le_bytes().to_vec()), + Buffer::from(context.value.request), + ]) + }, ) .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?; handler .unref(env) .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?; Ok(Self { + id: next_id(&NEXT_TRANSPORT_ID, "host storage transport") + .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?, handler: Mutex::new(Some(handler)), state: Mutex::new(TransportState::default()), capacity: Condvar::new(), + next_request_id: AtomicU64::new(1), abandoned_waiters: AtomicU64::new(0), max_operations, max_bytes, @@ -130,24 +151,25 @@ impl HostTransport { deadline: Option, class: AdmissionClass, ) -> io::Result> { - let request_id = next_request_id()?; + let request_id = next_id(&self.next_request_id, "host storage request")?; let response = Arc::new(ResponseSlot::new()); - pending_registry().insert(request_id, Arc::downgrade(self)); - if let Err(error) = self.admit(request_id, request.len(), response_budget, &response, deadline, class) { - pending_registry().remove(&request_id); - return Err(error); - } - let mut dispatch = Vec::with_capacity(8 + request.len()); - dispatch.extend_from_slice(&request_id.to_le_bytes()); - dispatch.extend_from_slice(&request); - let status = self + self.admit(request_id, request.len(), response_budget, &response, deadline, class)?; + let handler = self .handler .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) .as_ref() - .map_or(Status::Closing, |handler| { - handler.call(dispatch, ThreadsafeFunctionCallMode::NonBlocking) - }); + .cloned(); + let status = handler.map_or(Status::Closing, |handler| { + handler.call( + HostDispatch { + transport_id: self.id, + request_id, + request, + }, + ThreadsafeFunctionCallMode::NonBlocking, + ) + }); if status != Status::Ok { if status == Status::Closing { self.fail(io::ErrorKind::BrokenPipe, "host storage transport is closed"); @@ -170,6 +192,7 @@ impl HostTransport { } pub(crate) fn close(&self) { + unregister_transport(self.id); self.fail(io::ErrorKind::BrokenPipe, "host storage transport is closed"); if let Some(handler) = self .handler @@ -353,7 +376,6 @@ impl HostTransport { } fn complete(&self, request_id: u64, result: io::Result>) { - pending_registry().remove(&request_id); let response = { let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); let Some(pending) = state.pending.remove(&request_id) else { @@ -374,8 +396,7 @@ impl HostTransport { if let Some(response) = pending.response.upgrade() { response.complete(Err(error())); } - for (request_id, pending) in remaining { - pending_registry().remove(&request_id); + for pending in remaining.into_values() { if let Some(response) = pending.response.upgrade() { response.complete(Err(error())); } @@ -403,6 +424,7 @@ impl HostTransport { .unwrap_or_else(|poisoned| poisoned.into_inner()) .pending .get(&request_id) + .filter(|request| request.entered) .map(|request| request.response_budget) } @@ -419,8 +441,7 @@ impl HostTransport { std::mem::take(&mut state.pending) }; self.capacity.notify_all(); - for (request_id, request) in pending { - pending_registry().remove(&request_id); + for request in pending.into_values() { if let Some(response) = request.response.upgrade() { response.complete(Err(io::Error::new(kind, message.to_owned()))); } @@ -428,6 +449,12 @@ impl HostTransport { } } +impl Drop for HostTransport { + fn drop(&mut self) { + unregister_transport(self.id); + } +} + impl ResponseSlot { fn new() -> Self { Self { @@ -494,10 +521,11 @@ fn response_bytes(value: JsUnknown, max_bytes: usize) -> io::Result> { } #[napi(catch_unwind, skip_typescript, js_name = "__hostStorageComplete")] -pub fn host_storage_complete(request_id: String, response: JsUnknown) -> boundary::Result { +pub fn host_storage_complete(transport_id: Buffer, request_id: Buffer, response: JsUnknown) -> boundary::Result { boundary::run_stateless(|| { - let request_id = parse_request_id(&request_id)?; - let Some(transport) = pending_registry().get(&request_id).and_then(Weak::upgrade) else { + let transport_id = parse_id(&transport_id)?; + let request_id = parse_id(&request_id)?; + let Some(transport) = registered_transport(transport_id) else { return Ok(false); }; let Some(response_budget) = transport.response_budget(request_id) else { @@ -509,23 +537,25 @@ pub fn host_storage_complete(request_id: String, response: JsUnknown) -> boundar } #[napi(catch_unwind, skip_typescript, js_name = "__hostStorageBegin")] -pub fn host_storage_begin(request_id: String) -> boundary::Result { +pub fn host_storage_begin(transport_id: Buffer, request_id: Buffer) -> boundary::Result { boundary::run_stateless(|| { - let request_id = parse_request_id(&request_id)?; - Ok(pending_registry() - .get(&request_id) - .and_then(Weak::upgrade) - .is_some_and(|transport| transport.begin(request_id))) + let transport_id = parse_id(&transport_id)?; + let request_id = parse_id(&request_id)?; + Ok(registered_transport(transport_id).is_some_and(|transport| transport.begin(request_id))) })? } #[napi(catch_unwind, skip_typescript, js_name = "__hostStorageFail")] -pub fn host_storage_fail(request_id: String, message: String) -> boundary::Result { +pub fn host_storage_fail(transport_id: Buffer, request_id: Buffer, message: String) -> boundary::Result { boundary::run_stateless(|| { - let request_id = parse_request_id(&request_id)?; - let Some(transport) = pending_registry().get(&request_id).and_then(Weak::upgrade) else { + let transport_id = parse_id(&transport_id)?; + let request_id = parse_id(&request_id)?; + let Some(transport) = registered_transport(transport_id) else { return Ok(false); }; + if transport.response_budget(request_id).is_none() { + return Ok(false); + } let mut end = message.len().min(4_096); while !message.is_char_boundary(end) { end -= 1; @@ -536,24 +566,25 @@ pub fn host_storage_fail(request_id: String, message: String) -> boundary::Resul })? } -fn parse_request_id(request_id: &str) -> boundary::Result { - request_id - .parse() - .map_err(|_| napi::Error::new("E_INVALID_ARGUMENT", "invalid host storage request id")) +fn parse_id(id: &[u8]) -> boundary::Result { + let bytes: [u8; 8] = id + .try_into() + .map_err(|_| napi::Error::new("E_INVALID_ARGUMENT", "host storage id must contain eight bytes"))?; + Ok(u64::from_le_bytes(bytes)) } -fn next_request_id() -> io::Result { +fn next_id(counter: &AtomicU64, name: &str) -> io::Result { loop { - let request_id = NEXT_REQUEST_ID.load(Ordering::Relaxed); - if request_id == 0 { - return Err(io::Error::other("host storage request id space exhausted")); + let id = counter.load(Ordering::Relaxed); + if id == 0 { + return Err(io::Error::other(format!("{name} id space exhausted"))); } - let next = request_id.checked_add(1).unwrap_or(0); - if NEXT_REQUEST_ID - .compare_exchange_weak(request_id, next, Ordering::Relaxed, Ordering::Relaxed) + let next = id.checked_add(1).unwrap_or(0); + if counter + .compare_exchange_weak(id, next, Ordering::Relaxed, Ordering::Relaxed) .is_ok() { - return Ok(request_id); + return Ok(id); } } } @@ -762,6 +793,7 @@ pub(crate) fn open_directory( config.max_control_response_bytes, ) .map_err(|error| napi::Error::new("E_INVALID_ARGUMENT", error.to_string()))?; + register_transport(&transport); Ok((KvDirectory::with_namespace(store, &config.namespace), transport)) } @@ -941,11 +973,34 @@ fn registry() -> std::sync::MutexGuard<'static, HashMap> .unwrap_or_else(|poisoned| poisoned.into_inner()) } -fn pending_registry() -> std::sync::MutexGuard<'static, HashMap>> { - PENDING_TRANSPORTS - .get_or_init(Default::default) +fn transport_shard(transport_id: u64) -> &'static Mutex>> { + &TRANSPORTS.get_or_init(|| std::array::from_fn(|_| Mutex::new(HashMap::new()))) + [transport_id as usize % TRANSPORT_SHARDS] +} + +fn register_transport(transport: &Arc) { + transport_shard(transport.id) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .insert(transport.id, Arc::downgrade(transport)); +} + +fn registered_transport(transport_id: u64) -> Option> { + let mut shard = transport_shard(transport_id) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let transport = shard.get(&transport_id).and_then(Weak::upgrade); + if transport.is_none() { + shard.remove(&transport_id); + } + transport +} + +fn unregister_transport(transport_id: u64) { + transport_shard(transport_id) .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) + .remove(&transport_id); } #[cfg(feature = "test-panic")] @@ -1015,6 +1070,7 @@ pub fn test_open_host_transport( max_bytes as usize, Duration::from_millis(read_timeout_ms as u64), )?); + register_transport(&transport); registry().insert(handle, transport.clone()); if let Err(error) = env.add_async_cleanup_hook( CleanupTransport { @@ -1124,7 +1180,8 @@ pub fn test_hold_host_transport_capacity( .get(&handle) .cloned() .ok_or_else(|| napi::Error::new("E_CLOSED", "unknown or closed host storage transport"))?; - let request_id = next_request_id().map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?; + let request_id = next_id(&transport.next_request_id, "host storage request") + .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?; let response = Arc::new(ResponseSlot::new()); transport .admit( diff --git a/test/fixtures/harper-worker-child.mjs b/test/fixtures/harper-worker-child.mjs index 449737f..31e635e 100644 --- a/test/fixtures/harper-worker-child.mjs +++ b/test/fixtures/harper-worker-child.mjs @@ -14,7 +14,7 @@ function maybeBlock(operation) { parentPort.postMessage('blocked'); } blocked = false; - Atomics.wait(wait, 0, 0, 250); + Atomics.wait(wait, 0, 0); if (Atomics.load(wait, 0) === 1) throw new Error('host generation was revoked'); } diff --git a/test/fixtures/host-storage-transport-worker.mjs b/test/fixtures/host-storage-transport-worker.mjs index 2d93f03..a623f94 100644 --- a/test/fixtures/host-storage-transport-worker.mjs +++ b/test/fixtures/host-storage-transport-worker.mjs @@ -4,10 +4,9 @@ import { loadAddon } from '../../dist/load-addon.js'; const addon = loadAddon(); const handle = addon.__testOpenHostTransport( - (dispatch) => { - const requestId = dispatch.readBigUInt64LE().toString(); - if (!addon.__hostStorageBegin(requestId)) return; - addon.__hostStorageComplete(requestId, dispatch.subarray(8)); + (transportId, requestId, request) => { + if (!addon.__hostStorageBegin(transportId, requestId)) return; + addon.__hostStorageComplete(transportId, requestId, request); }, 2, 1_024, diff --git a/test/harper-index.test.mjs b/test/harper-index.test.mjs index 0c2de0e..565b421 100644 --- a/test/harper-index.test.mjs +++ b/test/harper-index.test.mjs @@ -118,6 +118,23 @@ test('rejects an oversized cursor without changing or poisoning the generation', await index.close(); }); +test('rejects asynchronous host storage before opening a runtime', async () => { + const host = createStorage(); + host.storage.sync = async () => {}; + await assert.rejects(openHarperFullTextIndex(options(host.storage)), (error) => error.code === 'E_INVALID_ARGUMENT'); + assert.deepStrictEqual(host.calls, []); +}); + +test('rejects a promise returned by a nominally synchronous storage method without wedging reopen', async () => { + const host = createStorage(); + const synchronousSync = host.storage.sync; + host.storage.sync = () => Promise.resolve(); + await assert.rejects(openHarperFullTextIndex(options(host.storage)), /must return undefined synchronously/); + host.storage.sync = synchronousSync; + const index = await openHarperFullTextIndex(options(host.storage)); + await index.close({ mode: 'rollback' }); +}); + for (const stage of ['open', 'read', 'apply', 'publish']) { test(`worker termination safely drains a hosted runtime during ${stage}`, async () => { const control = new SharedArrayBuffer(4); @@ -138,8 +155,8 @@ for (const stage of ['open', 'read', 'apply', 'publish']) { }); }); const started = performance.now(); - Atomics.store(new Int32Array(control), 0, 1); - await worker.terminate(); + const exitCode = await worker.terminate(); + assert(Number.isInteger(exitCode)); assert(performance.now() - started < 5_000, `${stage} teardown did not drain bounded storage work promptly`); }); } diff --git a/test/host-storage-transport.test.mjs b/test/host-storage-transport.test.mjs index 744c36b..784052f 100644 --- a/test/host-storage-transport.test.mjs +++ b/test/host-storage-transport.test.mjs @@ -473,13 +473,12 @@ function decodeHandlerError(response) { } function dispatchHostStorage(handler) { - return (dispatch) => { - const requestId = dispatch.readBigUInt64LE().toString(); - if (!addon.__hostStorageBegin(requestId)) return; + return (transportId, requestId, request) => { try { - addon.__hostStorageComplete(requestId, handler(dispatch.subarray(8))); + if (!addon.__hostStorageBegin(transportId, requestId)) return; + addon.__hostStorageComplete(transportId, requestId, handler(request)); } catch (error) { - addon.__hostStorageFail(requestId, error instanceof Error ? error.message : String(error)); + addon.__hostStorageFail(transportId, requestId, error instanceof Error ? error.message : String(error)); } }; } diff --git a/ts/harper.ts b/ts/harper.ts index d804607..5e6d8a8 100644 --- a/ts/harper.ts +++ b/ts/harper.ts @@ -54,14 +54,14 @@ class StorageGate implements HostStorage { return this.#storage.read(key); } - write(mutations: Parameters[0], policy: Parameters[1]): void { + write(mutations: Parameters[0], policy: Parameters[1]): undefined { this.#requireActive(); - this.#storage.write(mutations, policy); + return this.#storage.write(mutations, policy); } - sync(): void { + sync(): undefined { this.#requireActive(); - this.#storage.sync(); + return this.#storage.sync(); } revoke(): void { @@ -121,6 +121,7 @@ export class HarperFullTextIndex { } export async function openHarperFullTextIndex(options: HarperFullTextIndexOptions): Promise { + validateSynchronousStorage(options.storage); const gate = new StorageGate(options.storage); const handler = createHostStorageHandler(gate, { maxMutations: options.transport.maxMutations, @@ -129,6 +130,7 @@ export async function openHarperFullTextIndex(options: HarperFullTextIndexOption maxErrorBytes: options.transport.maxErrorBytes, }); const dispatch = createStorageDispatcher(handler); + let openedHandle: number | undefined; try { const cursor = await invoke((callback) => loadAddon().__harperOpen( @@ -150,6 +152,7 @@ export async function openHarperFullTextIndex(options: HarperFullTextIndexOption ), ); const handle = cursor.u32(); + openedHandle = handle; const hasPayload = cursor.u8(); if (hasPayload !== 0 && hasPayload !== 1) { throw new FulltextError('E_NATIVE_FAILURE', `Unknown committed payload status ${hasPayload}`); @@ -158,20 +161,36 @@ export async function openHarperFullTextIndex(options: HarperFullTextIndexOption cursor.finish(); return new HarperFullTextIndex(handle, committedPayload, gate); } catch (error) { + const handle = openedHandle; + if (handle !== undefined) { + await invoke((callback) => loadAddon().__nativeClose(handle, true, callback)).catch(() => undefined); + } gate.revoke(); throw error; } } -function createStorageDispatcher(handler: (request: Buffer) => Buffer): (dispatch: Buffer) => void { - return (dispatch) => { - if (dispatch.length < 8) return; - const requestId = dispatch.readBigUInt64LE().toString(); - if (!loadAddon().__hostStorageBegin(requestId)) return; +function validateSynchronousStorage(storage: HostStorage): void { + for (const name of ['read', 'write', 'sync'] as const) { + const method = storage?.[name]; + if (typeof method !== 'function') { + throw new FulltextError('E_INVALID_ARGUMENT', `host storage ${name} must be a function`); + } + if (method.constructor?.name === 'AsyncFunction') { + throw new FulltextError('E_INVALID_ARGUMENT', `host storage ${name} must be synchronous`); + } + } +} + +function createStorageDispatcher( + handler: (request: Buffer) => Buffer, +): (transportId: Buffer, requestId: Buffer, request: Buffer) => void { + return (transportId, requestId, request) => { try { - loadAddon().__hostStorageComplete(requestId, handler(dispatch.subarray(8))); + if (!loadAddon().__hostStorageBegin(transportId, requestId)) return; + loadAddon().__hostStorageComplete(transportId, requestId, handler(request)); } catch (error) { - loadAddon().__hostStorageFail(requestId, error instanceof Error ? error.message : String(error)); + loadAddon().__hostStorageFail(transportId, requestId, error instanceof Error ? error.message : String(error)); } }; } diff --git a/ts/host-storage.ts b/ts/host-storage.ts index 1953eb1..ea35985 100644 --- a/ts/host-storage.ts +++ b/ts/host-storage.ts @@ -17,10 +17,10 @@ export type HostStorageMutation = { type: 'put'; key: Buffer; value: Buffer } | export interface HostStorage { read(key: Buffer): Buffer | undefined; - /** Apply all mutations or none, satisfy the policy before returning, and throw only when none were applied. */ - write(mutations: Array, policy: HostWritePolicy): void; - /** Make prior writes durable before returning. */ - sync(): void; + /** Apply all mutations or none, satisfy the policy, and return undefined synchronously. */ + write(mutations: Array, policy: HostWritePolicy): undefined; + /** Make prior writes durable and return undefined synchronously. */ + sync(): undefined; } export interface HostStorageHandlerOptions { @@ -86,12 +86,16 @@ export function createHostStorageHandler( } } decoder.finish(); - storage.write(mutations, policy); + if (storage.write(mutations, policy) !== undefined) { + throw new Error('host storage write must return undefined synchronously'); + } return Buffer.from([protocolVersion, responseOk]); } if (operation === operationSync) { decoder.finish(); - storage.sync(); + if (storage.sync() !== undefined) { + throw new Error('host storage sync must return undefined synchronously'); + } return Buffer.from([protocolVersion, responseOk]); } throw new Error(`unknown host storage operation ${operation}`); diff --git a/ts/load-addon.ts b/ts/load-addon.ts index cd54e83..31f3293 100644 --- a/ts/load-addon.ts +++ b/ts/load-addon.ts @@ -20,17 +20,21 @@ interface NativeAddonApi { __nativeSearch(handle: number, request: Buffer, callback: NativeCallback): void; __nativeClose(handle: number, rollback: boolean, callback: NativeCallback): void; __nativeStatus(handle: number): Buffer; - __harperOpen(config: Buffer, handler: (dispatch: Buffer) => void, callback: NativeCallback): void; + __harperOpen( + config: Buffer, + handler: (transportId: Buffer, requestId: Buffer, request: Buffer) => void, + callback: NativeCallback, + ): void; __harperPublish(handle: number, payload: string, callback: NativeCallback): void; - __hostStorageBegin(requestId: string): boolean; - __hostStorageComplete(requestId: string, response: unknown): boolean; - __hostStorageFail(requestId: string, message: string): boolean; + __hostStorageBegin(transportId: Buffer, requestId: Buffer): boolean; + __hostStorageComplete(transportId: Buffer, requestId: Buffer, response: unknown): boolean; + __hostStorageFail(transportId: Buffer, requestId: Buffer, message: string): boolean; __testCreateHandle?(): number; __testPanic?(id: number): void; __testCheck?(id: number): boolean; __testPoisonNativeHandle?(handle: number): void; __testOpenHostTransport?( - handler: (dispatch: Buffer) => void, + handler: (transportId: Buffer, requestId: Buffer, request: Buffer) => void, maxOperations: number, maxBytes: number, readTimeoutMs: number, From cbe7cdec4de0e3c08576836a208a57a6520c98e0 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Thu, 10 Sep 2026 16:00:20 -0600 Subject: [PATCH 3/6] Address hosted runtime review findings --- src/host_storage.rs | 31 ++++++++++++++++++++------- src/phase0.rs | 2 +- test/fixtures/harper-worker-child.mjs | 12 +++++++++-- test/harper-index.test.mjs | 15 ++++++++++++- ts/harper.ts | 8 ++++++- ts/host-storage.ts | 2 +- 6 files changed, 56 insertions(+), 14 deletions(-) diff --git a/src/host_storage.rs b/src/host_storage.rs index e3ed6dd..0e138f1 100644 --- a/src/host_storage.rs +++ b/src/host_storage.rs @@ -25,7 +25,7 @@ use tantivy::directory::{Directory, TerminatingWrite}; use crate::boundary; #[cfg(feature = "test-panic")] use crate::phase0::ReclaimBudget; -#[cfg(any(test, feature = "test-panic"))] +#[cfg(any(feature = "test-panic", all(test, feature = "host-storage")))] use crate::phase0::{reclaim_read_key_bytes, RECLAIM_MAX_BATCH_REQUEST_BYTES}; use crate::phase0::{KvDirectory, KvStore, KvStoreIdentity, Mutation, WritePolicy, CHUNK_SIZE}; use crate::protocol::HostOpenConfig; @@ -44,9 +44,9 @@ type TransportShards = [Mutex; TRANSPORT_SHARDS]; static TRANSPORTS: OnceLock = OnceLock::new(); struct HostDispatch { + // napi 2.16 can abandon queued TSFN data during environment teardown, so keep this payload fixed-size. transport_id: u64, request_id: u64, - request: Vec, } type HostCallback = ThreadsafeFunction; @@ -79,6 +79,7 @@ struct PendingRequest { response_budget: usize, class: AdmissionClass, entered: bool, + request: Option>, response: Weak, } @@ -118,10 +119,13 @@ impl HostTransport { .create_threadsafe_function::( max_operations, |context: ThreadSafeCallContext| { + let request = registered_transport(context.value.transport_id) + .and_then(|transport| transport.take_request(context.value.request_id)) + .unwrap_or_default(); Ok(vec![ Buffer::from(context.value.transport_id.to_le_bytes().to_vec()), Buffer::from(context.value.request_id.to_le_bytes().to_vec()), - Buffer::from(context.value.request), + Buffer::from(request), ]) }, ) @@ -153,7 +157,7 @@ impl HostTransport { ) -> io::Result> { let request_id = next_id(&self.next_request_id, "host storage request")?; let response = Arc::new(ResponseSlot::new()); - self.admit(request_id, request.len(), response_budget, &response, deadline, class)?; + self.admit(request_id, Some(request), response_budget, &response, deadline, class)?; let handler = self .handler .lock() @@ -165,7 +169,6 @@ impl HostTransport { HostDispatch { transport_id: self.id, request_id, - request, }, ThreadsafeFunctionCallMode::NonBlocking, ) @@ -217,13 +220,15 @@ impl HostTransport { fn admit( &self, request_id: u64, - request_bytes: usize, + request: Option>, response_bytes: usize, response: &Arc, deadline: Option, class: AdmissionClass, ) -> io::Result<()> { - let retained_bytes = request_bytes + let retained_bytes = request + .as_ref() + .map_or(0, Vec::len) .checked_add(response_bytes) .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "host storage byte reservation overflow"))?; if response_bytes == 0 || retained_bytes > self.max_bytes { @@ -286,12 +291,22 @@ impl HostTransport { response_budget: response_bytes, class, entered: false, + request, response: Arc::downgrade(response), }, ); Ok(()) } + fn take_request(&self, request_id: u64) -> Option> { + self.state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .pending + .get_mut(&request_id) + .and_then(|pending| pending.request.take()) + } + fn begin(&self, request_id: u64) -> bool { let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); if state.closed.is_some() { @@ -1186,7 +1201,7 @@ pub fn test_hold_host_transport_capacity( transport .admit( request_id, - request_bytes as usize, + Some(vec![0; request_bytes as usize]), response_bytes as usize, &response, Some(Instant::now()), diff --git a/src/phase0.rs b/src/phase0.rs index 8c77d59..d77a5d6 100644 --- a/src/phase0.rs +++ b/src/phase0.rs @@ -93,7 +93,7 @@ pub trait KvStore: Clone + Send + Sync + 'static { pub(crate) const CHUNK_SIZE: usize = 256 * 1024; -#[cfg(any(test, feature = "test-panic"))] +#[cfg(any(feature = "test-panic", all(test, feature = "host-storage")))] pub(crate) fn reclaim_read_key_bytes(namespace: &[u8]) -> usize { // Tail objects have the longest key read by reclamation. tail_key(namespace, u64::MAX, u64::MAX).len() diff --git a/test/fixtures/harper-worker-child.mjs b/test/fixtures/harper-worker-child.mjs index 31e635e..7fdcafe 100644 --- a/test/fixtures/harper-worker-child.mjs +++ b/test/fixtures/harper-worker-child.mjs @@ -14,8 +14,10 @@ function maybeBlock(operation) { parentPort.postMessage('blocked'); } blocked = false; - Atomics.wait(wait, 0, 0); - if (Atomics.load(wait, 0) === 1) throw new Error('host generation was revoked'); + if (Atomics.wait(wait, 0, 0, 15_000) !== 'ok') { + throw new Error('blocked host storage test exceeded its failure bound'); + } + throw new Error('host generation was revoked'); } const storage = { @@ -90,6 +92,12 @@ try { config.limits.maxBatchBytes, ); const applying = index.apply(batch); + while (index.status().writerQueuedCommands !== 0n) { + await new Promise((resolve) => setImmediate(resolve)); + } + if (index.status().uncommittedMutations !== 0n) { + throw new Error('apply completed before the worker termination checkpoint'); + } parentPort.postMessage('blocked'); await applying; } else { diff --git a/test/harper-index.test.mjs b/test/harper-index.test.mjs index 565b421..e87ab8c 100644 --- a/test/harper-index.test.mjs +++ b/test/harper-index.test.mjs @@ -110,6 +110,19 @@ test('rejects duplicate owner-writer opens for one process identity and namespac await first.close(); }); +test('keeps the newest payload when publishes are issued concurrently', async () => { + const host = createStorage(); + const config = options(host.storage); + let index = await openHarperFullTextIndex(config); + await Promise.all([index.publish('cursor-v1:1'), index.publish('cursor-v1:2')]); + assert.strictEqual(index.committedPayload, 'cursor-v1:2'); + await index.close(); + + index = await openHarperFullTextIndex(config); + assert.strictEqual(index.committedPayload, 'cursor-v1:2'); + await index.close(); +}); + test('rejects an oversized cursor without changing or poisoning the generation', async () => { const host = createStorage(); const index = await openHarperFullTextIndex(options(host.storage)); @@ -136,7 +149,7 @@ test('rejects a promise returned by a nominally synchronous storage method witho }); for (const stage of ['open', 'read', 'apply', 'publish']) { - test(`worker termination safely drains a hosted runtime during ${stage}`, async () => { + test(`worker termination safely drains a hosted runtime during ${stage}`, { timeout: 20_000 }, async () => { const control = new SharedArrayBuffer(4); const worker = new Worker(new URL('./fixtures/harper-worker-child.mjs', import.meta.url), { workerData: { stage, control }, diff --git a/ts/harper.ts b/ts/harper.ts index 5e6d8a8..2ae1b00 100644 --- a/ts/harper.ts +++ b/ts/harper.ts @@ -78,6 +78,8 @@ export class HarperFullTextIndex { readonly #index: NativeFullTextIndex; readonly #storageGate: StorageGate; #committedPayload?: string; + #nextPublishSequence = 0n; + #publishedSequence = 0n; constructor(handle: number, committedPayload: string | undefined, storageGate: StorageGate) { this.#handle = handle; @@ -99,10 +101,14 @@ export class HarperFullTextIndex { if (Buffer.byteLength(payload) > maxCommitPayloadBytes) { throw new FulltextError('E_INVALID_ARGUMENT', `commit payload exceeds ${maxCommitPayloadBytes} UTF-8 bytes`); } + const sequence = ++this.#nextPublishSequence; const cursor = await invoke((callback) => loadAddon().__harperPublish(this.#handle, payload, callback)); const opstamp = cursor.u64(); cursor.finish(); - this.#committedPayload = payload; + if (sequence > this.#publishedSequence) { + this.#publishedSequence = sequence; + this.#committedPayload = payload; + } return opstamp; } diff --git a/ts/host-storage.ts b/ts/host-storage.ts index ea35985..52e0961 100644 --- a/ts/host-storage.ts +++ b/ts/host-storage.ts @@ -17,7 +17,7 @@ export type HostStorageMutation = { type: 'put'; key: Buffer; value: Buffer } | export interface HostStorage { read(key: Buffer): Buffer | undefined; - /** Apply all mutations or none, satisfy the policy, and return undefined synchronously. */ + /** Apply all mutations or none, satisfy the policy, return undefined synchronously, and throw only if none applied. */ write(mutations: Array, policy: HostWritePolicy): undefined; /** Make prior writes durable and return undefined synchronously. */ sync(): undefined; From 1b78b2ba10bcdf2a5d7817dd7a2e17857c77b5c3 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Thu, 10 Sep 2026 16:23:49 -0600 Subject: [PATCH 4/6] Resolve hosted runtime review findings --- README.md | 6 +- src/host_storage.rs | 84 +++++++------------ src/native.rs | 11 ++- .../host-storage-transport-worker.mjs | 5 +- test/harper-index.test.mjs | 65 +++++++++----- test/host-storage-transport.test.mjs | 11 ++- ts/harper.ts | 32 +++++-- ts/load-addon.ts | 13 +-- 8 files changed, 125 insertions(+), 102 deletions(-) diff --git a/README.md b/README.md index ae0aa6e..d825b85 100644 --- a/README.md +++ b/README.md @@ -84,8 +84,10 @@ The Harper opener requires a process-lifetime store identity, persistent generat bounded transport limits, and a `HostStorage` implementation. `publish(payload)` commits the index and opaque payload into one Tantivy `meta.json` generation, then reloads the local reader before it resolves. Harper uses that payload for its derived-index cursor. `committedPayload` exposes the -payload recovered at open. Host storage methods are strictly synchronous; `write` and `sync` must -return `undefined`, and Promise-returning implementations are rejected rather than acknowledged. +payload recovered at open or the newest successful publish. If a publish poisons the generation, +its durable outcome can be ambiguous, so the getter throws until the index is reopened. Host storage +methods are strictly synchronous; `write` and `sync` must return `undefined`, and Promise-returning +implementations are rejected rather than acknowledged. The current Harper path is owner-worker-only. It does not yet provide non-owner read handles, cross-worker refresh, generation retirement, or scheduled physical reclamation. Those lifecycle diff --git a/src/host_storage.rs b/src/host_storage.rs index 0e138f1..057625b 100644 --- a/src/host_storage.rs +++ b/src/host_storage.rs @@ -120,13 +120,12 @@ impl HostTransport { max_operations, |context: ThreadSafeCallContext| { let request = registered_transport(context.value.transport_id) - .and_then(|transport| transport.take_request(context.value.request_id)) + .and_then(|transport| transport.begin(context.value.request_id)) .unwrap_or_default(); - Ok(vec![ - Buffer::from(context.value.transport_id.to_le_bytes().to_vec()), - Buffer::from(context.value.request_id.to_le_bytes().to_vec()), - Buffer::from(request), - ]) + let mut dispatch_id = Vec::with_capacity(16); + dispatch_id.extend_from_slice(&context.value.transport_id.to_le_bytes()); + dispatch_id.extend_from_slice(&context.value.request_id.to_le_bytes()); + Ok(vec![Buffer::from(dispatch_id), Buffer::from(request)]) }, ) .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?; @@ -298,28 +297,18 @@ impl HostTransport { Ok(()) } - fn take_request(&self, request_id: u64) -> Option> { - self.state - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) - .pending - .get_mut(&request_id) - .and_then(|pending| pending.request.take()) - } - - fn begin(&self, request_id: u64) -> bool { + fn begin(&self, request_id: u64) -> Option> { let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); if state.closed.is_some() { - return false; + return None; } - let Some(pending) = state.pending.get_mut(&request_id) else { - return false; - }; + let pending = state.pending.get_mut(&request_id)?; if pending.entered { - return false; + return None; } + let request = pending.request.take()?; pending.entered = true; - true + Some(request) } fn has_capacity(&self, state: &TransportState, retained_bytes: usize, class: AdmissionClass) -> io::Result { @@ -390,11 +379,11 @@ impl HostTransport { } } - fn complete(&self, request_id: u64, result: io::Result>) { + fn complete(&self, request_id: u64, result: io::Result>) -> bool { let response = { let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); let Some(pending) = state.pending.remove(&request_id) else { - return; + return false; }; let cleanup_underflow = pending.class == AdmissionClass::Cleanup && (state.cleanup_operations == 0 || state.cleanup_bytes < pending.retained_bytes); @@ -416,7 +405,7 @@ impl HostTransport { response.complete(Err(error())); } } - return; + return true; } let response = pending.response.upgrade(); state.operations -= 1; @@ -431,6 +420,7 @@ impl HostTransport { if let Some(response) = response { response.complete(result); } + true } fn response_budget(&self, request_id: u64) -> Option { @@ -535,57 +525,47 @@ fn response_bytes(value: JsUnknown, max_bytes: usize) -> io::Result> { Ok(buffer.as_ref().to_vec()) } +// These internal exports fence lifecycle but do not authenticate callers; the package trusts process-local JavaScript. #[napi(catch_unwind, skip_typescript, js_name = "__hostStorageComplete")] -pub fn host_storage_complete(transport_id: Buffer, request_id: Buffer, response: JsUnknown) -> boundary::Result { +pub fn host_storage_complete(dispatch_id: Buffer, response: JsUnknown) -> boundary::Result { boundary::run_stateless(|| { - let transport_id = parse_id(&transport_id)?; - let request_id = parse_id(&request_id)?; + let (transport_id, request_id) = parse_dispatch_id(&dispatch_id)?; let Some(transport) = registered_transport(transport_id) else { return Ok(false); }; let Some(response_budget) = transport.response_budget(request_id) else { return Ok(false); }; - transport.complete(request_id, response_bytes(response, response_budget)); - Ok(true) - })? -} - -#[napi(catch_unwind, skip_typescript, js_name = "__hostStorageBegin")] -pub fn host_storage_begin(transport_id: Buffer, request_id: Buffer) -> boundary::Result { - boundary::run_stateless(|| { - let transport_id = parse_id(&transport_id)?; - let request_id = parse_id(&request_id)?; - Ok(registered_transport(transport_id).is_some_and(|transport| transport.begin(request_id))) + Ok(transport.complete(request_id, response_bytes(response, response_budget))) })? } #[napi(catch_unwind, skip_typescript, js_name = "__hostStorageFail")] -pub fn host_storage_fail(transport_id: Buffer, request_id: Buffer, message: String) -> boundary::Result { +pub fn host_storage_fail(dispatch_id: Buffer, message: String) -> boundary::Result { boundary::run_stateless(|| { - let transport_id = parse_id(&transport_id)?; - let request_id = parse_id(&request_id)?; + let (transport_id, request_id) = parse_dispatch_id(&dispatch_id)?; let Some(transport) = registered_transport(transport_id) else { return Ok(false); }; - if transport.response_budget(request_id).is_none() { - return Ok(false); - } let mut end = message.len().min(4_096); while !message.is_char_boundary(end) { end -= 1; } let message = &message[..end]; - transport.complete(request_id, Err(io::Error::other(message.to_owned()))); - Ok(true) + Ok(transport.complete(request_id, Err(io::Error::other(message.to_owned())))) })? } -fn parse_id(id: &[u8]) -> boundary::Result { - let bytes: [u8; 8] = id - .try_into() - .map_err(|_| napi::Error::new("E_INVALID_ARGUMENT", "host storage id must contain eight bytes"))?; - Ok(u64::from_le_bytes(bytes)) +fn parse_dispatch_id(id: &[u8]) -> boundary::Result<(u64, u64)> { + if id.len() != 16 { + return Err(napi::Error::new( + "E_INVALID_ARGUMENT", + "host storage dispatch id must contain sixteen bytes", + )); + } + let transport_id = u64::from_le_bytes(id[..8].try_into().expect("dispatch id length checked")); + let request_id = u64::from_le_bytes(id[8..].try_into().expect("dispatch id length checked")); + Ok((transport_id, request_id)) } fn next_id(counter: &AtomicU64, name: &str) -> io::Result { diff --git a/src/native.rs b/src/native.rs index da2810e..c7f13b3 100644 --- a/src/native.rs +++ b/src/native.rs @@ -275,12 +275,13 @@ pub fn harper_publish(handle: u32, payload: String, callback: JsFunction) -> bou } let runtime = runtime(handle)?; let completion = completion(callback, runtime.environment.alive.clone())?; + let bytes = payload.len(); runtime.enqueue_writer( WriterCommand { operation: WriterOperation::Publish(payload), completion, }, - 0, + bytes, ) })? } @@ -483,7 +484,7 @@ impl Runtime { } } if let Some(close) = close { - let _ = self.writer_queue.push_force(close, 0); + let _ = self.writer_queue.push_force(close.force_rollback(), 0); } for command in self.search_queue.close() { command.value.completion.failure(error.clone()); @@ -685,6 +686,12 @@ impl WriterCommand { fn fail(self, error: FulltextError) { self.completion.failure(error); } + + fn force_rollback(mut self) -> Self { + debug_assert!(matches!(self.operation, WriterOperation::Close { .. })); + self.operation = WriterOperation::Close { rollback: true }; + self + } } fn writer_loop(runtime: Arc, writer: Writer) { diff --git a/test/fixtures/host-storage-transport-worker.mjs b/test/fixtures/host-storage-transport-worker.mjs index a623f94..cbdc8b4 100644 --- a/test/fixtures/host-storage-transport-worker.mjs +++ b/test/fixtures/host-storage-transport-worker.mjs @@ -4,9 +4,8 @@ import { loadAddon } from '../../dist/load-addon.js'; const addon = loadAddon(); const handle = addon.__testOpenHostTransport( - (transportId, requestId, request) => { - if (!addon.__hostStorageBegin(transportId, requestId)) return; - addon.__hostStorageComplete(transportId, requestId, request); + (dispatchId, request) => { + addon.__hostStorageComplete(dispatchId, request); }, 2, 1_024, diff --git a/test/harper-index.test.mjs b/test/harper-index.test.mjs index e87ab8c..6c6a0fc 100644 --- a/test/harper-index.test.mjs +++ b/test/harper-index.test.mjs @@ -6,40 +6,44 @@ import { encodeMutationBatch, openHarperFullTextIndex } from '@harperfast/fullte const readResponseBytes = 1024 * 1024; const controlResponseBytes = 1024 * 1024; +const storeIdentities = new WeakMap(); +let nextStoreIdentity = 0n; function createStorage() { const entries = new Map(); const calls = []; + const storage = { + read(key) { + calls.push('read'); + const value = entries.get(key.toString('hex')); + return value && Buffer.from(value); + }, + write(mutations, policy) { + calls.push(`write:${policy}`); + const next = new Map(entries); + for (const mutation of mutations) { + const key = mutation.key.toString('hex'); + if (mutation.type === 'put') next.set(key, Buffer.from(mutation.value)); + else next.delete(key); + } + entries.clear(); + for (const [key, value] of next) entries.set(key, value); + }, + sync() { + calls.push('sync'); + }, + }; + storeIdentities.set(storage, [11n, 22n, ++nextStoreIdentity]); return { calls, - storage: { - read(key) { - calls.push('read'); - const value = entries.get(key.toString('hex')); - return value && Buffer.from(value); - }, - write(mutations, policy) { - calls.push(`write:${policy}`); - const next = new Map(entries); - for (const mutation of mutations) { - const key = mutation.key.toString('hex'); - if (mutation.type === 'put') next.set(key, Buffer.from(mutation.value)); - else next.delete(key); - } - entries.clear(); - for (const [key, value] of next) entries.set(key, value); - }, - sync() { - calls.push('sync'); - }, - }, + storage, }; } function options(storage, overrides = {}) { return { storage, - storeIdentity: [11n, 22n, 33n], + storeIdentity: storeIdentities.get(storage), namespace: Buffer.from('products-title'), indexId: 'products-title', generation: 'generation-1', @@ -123,6 +127,23 @@ test('keeps the newest payload when publishes are issued concurrently', async () await index.close(); }); +test('marks the committed payload unknown when a publish poisons the generation', async () => { + const host = createStorage(); + const index = await openHarperFullTextIndex(options(host.storage)); + await index.apply(encodeMutationBatch({ upserts: [{ id: 'shoe-1', fields: { title: 'running shoe' } }] })); + const sync = host.storage.sync; + host.storage.sync = () => { + throw new Error('injected durability failure'); + }; + await assert.rejects(index.publish('cursor-v1:1'), /injected durability failure/); + assert.throws( + () => index.committedPayload, + (error) => error.code === 'E_POISONED', + ); + host.storage.sync = sync; + await index.close({ mode: 'rollback' }); +}); + test('rejects an oversized cursor without changing or poisoning the generation', async () => { const host = createStorage(); const index = await openHarperFullTextIndex(options(host.storage)); diff --git a/test/host-storage-transport.test.mjs b/test/host-storage-transport.test.mjs index 784052f..e9ded6d 100644 --- a/test/host-storage-transport.test.mjs +++ b/test/host-storage-transport.test.mjs @@ -473,12 +473,15 @@ function decodeHandlerError(response) { } function dispatchHostStorage(handler) { - return (transportId, requestId, request) => { + return (dispatchId, request) => { try { - if (!addon.__hostStorageBegin(transportId, requestId)) return; - addon.__hostStorageComplete(transportId, requestId, handler(request)); + addon.__hostStorageComplete(dispatchId, handler(request)); } catch (error) { - addon.__hostStorageFail(transportId, requestId, error instanceof Error ? error.message : String(error)); + try { + addon.__hostStorageFail(dispatchId, error instanceof Error ? error.message : String(error)); + } catch { + // Transport teardown resolves any request that cannot be failed here. + } } }; } diff --git a/ts/harper.ts b/ts/harper.ts index 2ae1b00..d7bb068 100644 --- a/ts/harper.ts +++ b/ts/harper.ts @@ -78,6 +78,7 @@ export class HarperFullTextIndex { readonly #index: NativeFullTextIndex; readonly #storageGate: StorageGate; #committedPayload?: string; + #payloadKnown = true; #nextPublishSequence = 0n; #publishedSequence = 0n; @@ -89,6 +90,9 @@ export class HarperFullTextIndex { } get committedPayload(): string | undefined { + if (!this.#payloadKnown) { + throw new FulltextError('E_POISONED', 'committed payload is unknown until the index is reopened'); + } return this.#committedPayload; } @@ -102,7 +106,17 @@ export class HarperFullTextIndex { throw new FulltextError('E_INVALID_ARGUMENT', `commit payload exceeds ${maxCommitPayloadBytes} UTF-8 bytes`); } const sequence = ++this.#nextPublishSequence; - const cursor = await invoke((callback) => loadAddon().__harperPublish(this.#handle, payload, callback)); + let cursor; + try { + cursor = await invoke((callback) => loadAddon().__harperPublish(this.#handle, payload, callback)); + } catch (error) { + try { + this.#payloadKnown = this.#index.status().state !== 'poisoned'; + } catch { + this.#payloadKnown = false; + } + throw error; + } const opstamp = cursor.u64(); cursor.finish(); if (sequence > this.#publishedSequence) { @@ -188,15 +202,17 @@ function validateSynchronousStorage(storage: HostStorage): void { } } -function createStorageDispatcher( - handler: (request: Buffer) => Buffer, -): (transportId: Buffer, requestId: Buffer, request: Buffer) => void { - return (transportId, requestId, request) => { +function createStorageDispatcher(handler: (request: Buffer) => Buffer): (dispatchId: Buffer, request: Buffer) => void { + const addon = loadAddon(); + return (dispatchId, request) => { try { - if (!loadAddon().__hostStorageBegin(transportId, requestId)) return; - loadAddon().__hostStorageComplete(transportId, requestId, handler(request)); + addon.__hostStorageComplete(dispatchId, handler(request)); } catch (error) { - loadAddon().__hostStorageFail(transportId, requestId, error instanceof Error ? error.message : String(error)); + try { + addon.__hostStorageFail(dispatchId, error instanceof Error ? error.message : String(error)); + } catch { + // Transport teardown resolves any request that cannot be failed here. + } } }; } diff --git a/ts/load-addon.ts b/ts/load-addon.ts index 31f3293..02ee959 100644 --- a/ts/load-addon.ts +++ b/ts/load-addon.ts @@ -20,21 +20,16 @@ interface NativeAddonApi { __nativeSearch(handle: number, request: Buffer, callback: NativeCallback): void; __nativeClose(handle: number, rollback: boolean, callback: NativeCallback): void; __nativeStatus(handle: number): Buffer; - __harperOpen( - config: Buffer, - handler: (transportId: Buffer, requestId: Buffer, request: Buffer) => void, - callback: NativeCallback, - ): void; + __harperOpen(config: Buffer, handler: (dispatchId: Buffer, request: Buffer) => void, callback: NativeCallback): void; __harperPublish(handle: number, payload: string, callback: NativeCallback): void; - __hostStorageBegin(transportId: Buffer, requestId: Buffer): boolean; - __hostStorageComplete(transportId: Buffer, requestId: Buffer, response: unknown): boolean; - __hostStorageFail(transportId: Buffer, requestId: Buffer, message: string): boolean; + __hostStorageComplete(dispatchId: Buffer, response: unknown): boolean; + __hostStorageFail(dispatchId: Buffer, message: string): boolean; __testCreateHandle?(): number; __testPanic?(id: number): void; __testCheck?(id: number): boolean; __testPoisonNativeHandle?(handle: number): void; __testOpenHostTransport?( - handler: (transportId: Buffer, requestId: Buffer, request: Buffer) => void, + handler: (dispatchId: Buffer, request: Buffer) => void, maxOperations: number, maxBytes: number, readTimeoutMs: number, From d6768ac2125cc74b5a0dc2dac3678ef974c148aa Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Thu, 10 Sep 2026 16:36:41 -0600 Subject: [PATCH 5/6] Make hosted failure state explicit --- src/host_storage.rs | 6 ++++-- test/fixtures/host-storage-transport-worker.mjs | 1 + test/host-storage-transport.test.mjs | 1 + ts/harper.ts | 11 ++++------- 4 files changed, 10 insertions(+), 9 deletions(-) diff --git a/src/host_storage.rs b/src/host_storage.rs index 057625b..c3d3acc 100644 --- a/src/host_storage.rs +++ b/src/host_storage.rs @@ -119,9 +119,11 @@ impl HostTransport { .create_threadsafe_function::( max_operations, |context: ThreadSafeCallContext| { - let request = registered_transport(context.value.transport_id) + let Some(request) = registered_transport(context.value.transport_id) .and_then(|transport| transport.begin(context.value.request_id)) - .unwrap_or_default(); + else { + return Ok(vec![Buffer::default(), Buffer::default()]); + }; let mut dispatch_id = Vec::with_capacity(16); dispatch_id.extend_from_slice(&context.value.transport_id.to_le_bytes()); dispatch_id.extend_from_slice(&context.value.request_id.to_le_bytes()); diff --git a/test/fixtures/host-storage-transport-worker.mjs b/test/fixtures/host-storage-transport-worker.mjs index cbdc8b4..1d9be88 100644 --- a/test/fixtures/host-storage-transport-worker.mjs +++ b/test/fixtures/host-storage-transport-worker.mjs @@ -5,6 +5,7 @@ import { loadAddon } from '../../dist/load-addon.js'; const addon = loadAddon(); const handle = addon.__testOpenHostTransport( (dispatchId, request) => { + if (dispatchId.length === 0) return; addon.__hostStorageComplete(dispatchId, request); }, 2, diff --git a/test/host-storage-transport.test.mjs b/test/host-storage-transport.test.mjs index e9ded6d..1761c87 100644 --- a/test/host-storage-transport.test.mjs +++ b/test/host-storage-transport.test.mjs @@ -474,6 +474,7 @@ function decodeHandlerError(response) { function dispatchHostStorage(handler) { return (dispatchId, request) => { + if (dispatchId.length === 0) return; try { addon.__hostStorageComplete(dispatchId, handler(request)); } catch (error) { diff --git a/ts/harper.ts b/ts/harper.ts index d7bb068..a2b172c 100644 --- a/ts/harper.ts +++ b/ts/harper.ts @@ -78,9 +78,9 @@ export class HarperFullTextIndex { readonly #index: NativeFullTextIndex; readonly #storageGate: StorageGate; #committedPayload?: string; - #payloadKnown = true; #nextPublishSequence = 0n; #publishedSequence = 0n; + #uncertainPublishSequence = 0n; constructor(handle: number, committedPayload: string | undefined, storageGate: StorageGate) { this.#handle = handle; @@ -90,7 +90,7 @@ export class HarperFullTextIndex { } get committedPayload(): string | undefined { - if (!this.#payloadKnown) { + if (this.#uncertainPublishSequence > this.#publishedSequence) { throw new FulltextError('E_POISONED', 'committed payload is unknown until the index is reopened'); } return this.#committedPayload; @@ -110,11 +110,7 @@ export class HarperFullTextIndex { try { cursor = await invoke((callback) => loadAddon().__harperPublish(this.#handle, payload, callback)); } catch (error) { - try { - this.#payloadKnown = this.#index.status().state !== 'poisoned'; - } catch { - this.#payloadKnown = false; - } + if (sequence > this.#uncertainPublishSequence) this.#uncertainPublishSequence = sequence; throw error; } const opstamp = cursor.u64(); @@ -205,6 +201,7 @@ function validateSynchronousStorage(storage: HostStorage): void { function createStorageDispatcher(handler: (request: Buffer) => Buffer): (dispatchId: Buffer, request: Buffer) => void { const addon = loadAddon(); return (dispatchId, request) => { + if (dispatchId.length === 0) return; try { addon.__hostStorageComplete(dispatchId, handler(request)); } catch (error) { From d7321680efa60dd948aec79a6383daf67f35e9f4 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Thu, 10 Sep 2026 16:40:15 -0600 Subject: [PATCH 6/6] Guard publish response decoding --- ts/harper.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/ts/harper.ts b/ts/harper.ts index a2b172c..67646f7 100644 --- a/ts/harper.ts +++ b/ts/harper.ts @@ -106,15 +106,15 @@ export class HarperFullTextIndex { throw new FulltextError('E_INVALID_ARGUMENT', `commit payload exceeds ${maxCommitPayloadBytes} UTF-8 bytes`); } const sequence = ++this.#nextPublishSequence; - let cursor; + let opstamp: bigint; try { - cursor = await invoke((callback) => loadAddon().__harperPublish(this.#handle, payload, callback)); + const cursor = await invoke((callback) => loadAddon().__harperPublish(this.#handle, payload, callback)); + opstamp = cursor.u64(); + cursor.finish(); } catch (error) { if (sequence > this.#uncertainPublishSequence) this.#uncertainPublishSequence = sequence; throw error; } - const opstamp = cursor.u64(); - cursor.finish(); if (sequence > this.#publishedSequence) { this.#publishedSequence = sequence; this.#committedPayload = payload;