From e696b8483a002c3fffee3a4782dcfd07968e5ac8 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Tue, 8 Sep 2026 15:31:42 -0600 Subject: [PATCH 1/7] feat: prove host-backed Tantivy directory path --- src/host_storage.rs | 685 +++++++++++++++++++++++++++ src/lib.rs | 6 + test/host-storage-transport.test.mjs | 166 +++++++ ts/host-storage.ts | 160 +++++++ ts/load-addon.ts | 9 + 5 files changed, 1026 insertions(+) create mode 100644 src/host_storage.rs create mode 100644 test/host-storage-transport.test.mjs create mode 100644 ts/host-storage.ts diff --git a/src/host_storage.rs b/src/host_storage.rs new file mode 100644 index 0000000..f57d496 --- /dev/null +++ b/src/host_storage.rs @@ -0,0 +1,685 @@ +use std::collections::HashMap; +use std::io; +use std::ops::Deref; +use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; +use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak}; +use std::thread; +use std::time::Duration; + +use napi::bindgen_prelude::Buffer; +use napi::threadsafe_function::{ErrorStrategy, ThreadSafeCallContext, ThreadsafeFunction, ThreadsafeFunctionCallMode}; +use napi::{Env, JsBuffer, JsFunction, JsUnknown, Status, ValueType}; +use napi_derive::napi; +use tantivy::directory::OwnedBytes; + +use crate::boundary; +use crate::phase0::{KvDirectory, KvStore, KvStoreIdentity, Mutation, WritePolicy}; + +type HostCallback = ThreadsafeFunction, ErrorStrategy::Fatal>; +type CompletionCallback = ThreadsafeFunction, ErrorStrategy::Fatal>; + +static NEXT_TRANSPORT_HANDLE: AtomicU32 = AtomicU32::new(1); +static HOST_TRANSPORTS: OnceLock>>> = OnceLock::new(); + +struct HostTransport { + handler: HostCallback, + state: Mutex, + next_request_id: AtomicU64, + max_operations: usize, + max_bytes: usize, + timeout: Duration, +} + +#[derive(Default)] +struct TransportState { + closed: Option, + operations: usize, + bytes: usize, + pending: HashMap, +} + +struct PendingRequest { + retained_bytes: usize, + completed: bool, + response: Weak, +} + +struct ResponseSlot { + result: Mutex>>>, + ready: Condvar, +} + +impl HostTransport { + fn new(handler: JsFunction, max_operations: usize, max_bytes: usize, timeout: Duration) -> boundary::Result { + if max_operations == 0 || max_bytes == 0 || timeout.is_zero() { + return Err(napi::Error::new( + "E_INVALID_ARGUMENT", + "host transport limits must be greater than zero", + )); + } + let handler = handler + .create_threadsafe_function::, Buffer, _, ErrorStrategy::Fatal>( + max_operations, + |context: ThreadSafeCallContext>| Ok(vec![Buffer::from(context.value)]), + ) + .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?; + Ok(Self { + handler, + state: Mutex::new(TransportState::default()), + next_request_id: AtomicU64::new(1), + max_operations, + max_bytes, + timeout, + }) + } + + fn round_trip(self: &Arc, request: Vec, response_budget: usize) -> 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 response = Arc::new(ResponseSlot::new()); + self.admit(request_id, request.len(), response_budget, &response)?; + + 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 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", + ))); + } + Ok(()) + }, + ); + if status != Status::Ok { + self.complete( + request_id, + Err(io::Error::new( + io::ErrorKind::WouldBlock, + format!("host storage callback rejected request: {status:?}"), + )), + ); + } + + if let Some(result) = response.wait(self.timeout) { + return match result { + Ok(bytes) => Ok(TransportResponse { + bytes, + transport: Arc::downgrade(self), + request_id, + }), + Err(error) => { + self.release(request_id); + Err(error) + } + }; + } + self.fail(io::ErrorKind::TimedOut, "host storage request timed out"); + let _ = response.take(); + Err(io::Error::new( + io::ErrorKind::TimedOut, + "host storage request timed out", + )) + } + + fn close(&self) { + self.fail(io::ErrorKind::BrokenPipe, "host storage transport is closed"); + } + + fn admit( + &self, + request_id: u64, + request_bytes: usize, + response_bytes: usize, + response: &Arc, + ) -> io::Result<()> { + let retained_bytes = request_bytes + .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 { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "host storage request and response reservation exceeds the byte limit", + )); + } + let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + if let Some(error) = &state.closed { + return Err(io::Error::new(io::ErrorKind::BrokenPipe, error.clone())); + } + if state.operations >= self.max_operations || state.bytes.saturating_add(retained_bytes) > self.max_bytes { + return Err(io::Error::new( + io::ErrorKind::WouldBlock, + "host storage transport is at capacity", + )); + } + state.operations += 1; + state.bytes += retained_bytes; + state.pending.insert( + request_id, + PendingRequest { + retained_bytes, + completed: false, + response: Arc::downgrade(response), + }, + ); + Ok(()) + } + + fn complete(&self, request_id: u64, result: io::Result>) { + let response = { + let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let Some(pending) = state.pending.get_mut(&request_id) else { + return; + }; + if pending.completed { + return; + } + let response = pending.response.upgrade(); + pending.completed = true; + response + }; + if let Some(response) = response { + response.complete(result); + } + } + + fn release(&self, request_id: u64) { + let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + if let Some(pending) = state.pending.remove(&request_id) { + state.operations -= 1; + state.bytes -= pending.retained_bytes; + } + } + + fn fail(&self, kind: io::ErrorKind, message: &str) { + let pending = { + let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + if state.closed.is_none() { + state.closed = Some(message.to_owned()); + } + state.operations = 0; + state.bytes = 0; + std::mem::take(&mut state.pending) + }; + for request in pending.into_values() { + if let Some(response) = request.response.upgrade() { + response.complete(Err(io::Error::new(kind, message.to_owned()))); + } + } + } +} + +struct TransportResponse { + bytes: Vec, + transport: Weak, + request_id: u64, +} + +impl TransportResponse { + fn into_bytes(mut self) -> Vec { + std::mem::take(&mut self.bytes) + } +} + +impl Deref for TransportResponse { + type Target = [u8]; + + fn deref(&self) -> &Self::Target { + &self.bytes + } +} + +impl Drop for TransportResponse { + fn drop(&mut self) { + if let Some(transport) = self.transport.upgrade() { + transport.release(self.request_id); + } + } +} + +impl ResponseSlot { + fn new() -> Self { + Self { + result: Mutex::new(None), + ready: Condvar::new(), + } + } + + fn complete(&self, result: io::Result>) { + let mut slot = self.result.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + if slot.is_none() { + *slot = Some(result); + self.ready.notify_all(); + } + } + + fn wait(&self, timeout: Duration) -> Option>> { + let slot = self.result.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let (mut slot, _) = self + .ready + .wait_timeout_while(slot, timeout, |result| result.is_none()) + .unwrap_or_else(|poisoned| poisoned.into_inner()); + slot.take() + } + + fn take(&self) -> Option>> { + self.result + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take() + } +} + +fn response_bytes(value: JsUnknown, max_bytes: usize) -> io::Result> { + if value.get_type().map_err(|error| io::Error::other(error.to_string()))? != ValueType::Object { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "host storage callback must return a Buffer", + )); + } + let buffer: JsBuffer = unsafe { value.cast() }; + let buffer = buffer + .into_value() + .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "host storage callback must return a Buffer"))?; + if buffer.len() > max_bytes { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "host storage callback response exceeds the byte limit", + )); + } + Ok(buffer.as_ref().to_vec()) +} + +const HOST_PROTOCOL_VERSION: u8 = 1; +const OP_READ: u8 = 1; +const OP_WRITE: u8 = 2; +const OP_SYNC: u8 = 3; +const RESPONSE_OK: u8 = 0; +const RESPONSE_ERROR: u8 = 1; +const VALUE_MISSING: u8 = 0; +const VALUE_PRESENT: u8 = 1; +const MUTATION_PUT: u8 = 1; +const MUTATION_DELETE: u8 = 2; + +#[derive(Clone)] +struct HostKvStore { + transport: Arc, + identity: KvStoreIdentity, + max_read_response_bytes: usize, + max_control_response_bytes: usize, +} + +impl HostKvStore { + fn new( + transport: Arc, + identity: KvStoreIdentity, + max_read_response_bytes: usize, + max_control_response_bytes: usize, + ) -> Self { + Self { + transport, + identity, + max_read_response_bytes, + max_control_response_bytes, + } + } + + fn request(&self, request: Vec, response_budget: usize) -> io::Result { + let response = self.transport.round_trip(request, response_budget)?; + let mut decoder = ResponseDecoder::new(response); + if decoder.u8()? != HOST_PROTOCOL_VERSION { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "host storage response has an unsupported protocol version", + )); + } + match decoder.u8()? { + RESPONSE_OK => Ok(decoder), + RESPONSE_ERROR => { + let message = decoder.bytes()?.to_vec(); + decoder.finish()?; + Err(io::Error::other(String::from_utf8_lossy(&message).into_owned())) + } + _ => Err(io::Error::new( + io::ErrorKind::InvalidData, + "host storage response has an unknown status", + )), + } + } +} + +impl KvStore for HostKvStore { + fn identity(&self) -> KvStoreIdentity { + self.identity + } + + fn read(&self, key: &[u8]) -> io::Result> { + let mut request = RequestEncoder::new(OP_READ); + request.bytes(key)?; + let mut response = self.request(request.finish(), self.max_read_response_bytes)?; + let value = match response.u8()? { + VALUE_MISSING => None, + VALUE_PRESENT => Some(OwnedBytes::new(response.bytes()?.to_vec())), + _ => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "host storage read response has an unknown value status", + )) + } + }; + response.finish()?; + Ok(value) + } + + fn write(&self, mutations: &[Mutation], policy: WritePolicy) -> io::Result<()> { + let mut request = RequestEncoder::new(OP_WRITE); + request.u8(match policy { + WritePolicy::WAL => 1, + WritePolicy::WAL_SYNC => 2, + WritePolicy::NO_WAL => 3, + _ => return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid write policy")), + }); + request.u32(mutations.len())?; + for mutation in mutations { + match mutation { + Mutation::Put(key, value) => { + request.u8(MUTATION_PUT); + request.bytes(key)?; + request.bytes(value)?; + } + Mutation::Delete(key) => { + request.u8(MUTATION_DELETE); + request.bytes(key)?; + } + } + } + self.request(request.finish(), self.max_control_response_bytes)? + .finish() + } + + fn sync(&self) -> io::Result<()> { + self.request(RequestEncoder::new(OP_SYNC).finish(), self.max_control_response_bytes)? + .finish() + } +} + +struct RequestEncoder(Vec); + +impl RequestEncoder { + fn new(operation: u8) -> Self { + Self(vec![HOST_PROTOCOL_VERSION, operation]) + } + + fn u8(&mut self, value: u8) { + self.0.push(value); + } + + fn u32(&mut self, value: usize) -> io::Result<()> { + let value = + u32::try_from(value).map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "value is too large"))?; + self.0.extend_from_slice(&value.to_le_bytes()); + Ok(()) + } + + fn bytes(&mut self, value: &[u8]) -> io::Result<()> { + self.u32(value.len())?; + self.0.extend_from_slice(value); + Ok(()) + } + + fn finish(self) -> Vec { + self.0 + } +} + +struct ResponseDecoder { + bytes: TransportResponse, + offset: usize, +} + +impl ResponseDecoder { + fn new(bytes: TransportResponse) -> Self { + Self { bytes, offset: 0 } + } + + fn u8(&mut self) -> io::Result { + let value = *self + .bytes + .get(self.offset) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "host storage response is truncated"))?; + self.offset += 1; + Ok(value) + } + + fn bytes(&mut self) -> io::Result<&[u8]> { + let length_end = self + .offset + .checked_add(4) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "host storage response length overflow"))?; + let length_bytes: [u8; 4] = self + .bytes + .get(self.offset..length_end) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "host storage response is truncated"))? + .try_into() + .unwrap(); + self.offset = length_end; + let length = u32::from_le_bytes(length_bytes) as usize; + let end = self + .offset + .checked_add(length) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "host storage response length overflow"))?; + let value = self + .bytes + .get(self.offset..end) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "host storage response is truncated"))?; + self.offset = end; + Ok(value) + } + + fn finish(self) -> io::Result<()> { + if self.offset == self.bytes.len() { + Ok(()) + } else { + Err(io::Error::new( + io::ErrorKind::InvalidData, + "host storage response has trailing bytes", + )) + } + } +} + +fn registry() -> std::sync::MutexGuard<'static, HashMap>> { + HOST_TRANSPORTS + .get_or_init(Default::default) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn completion(callback: JsFunction) -> boundary::Result { + callback + .create_threadsafe_function::, Buffer, _, ErrorStrategy::Fatal>( + 0, + |context: ThreadSafeCallContext>| Ok(vec![Buffer::from(context.value)]), + ) + .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string())) +} + +fn test_result(result: io::Result>) -> Vec { + match result { + Ok(bytes) => { + let mut encoded = Vec::with_capacity(bytes.len() + 1); + encoded.push(0); + encoded.extend_from_slice(&bytes); + encoded + } + Err(error) => { + let message = error.to_string(); + let mut encoded = Vec::with_capacity(message.len() + 1); + encoded.push(1); + encoded.extend_from_slice(message.as_bytes()); + encoded + } + } +} + +struct CleanupTransport { + handle: u32, + transport: Weak, +} + +#[cfg(feature = "test-panic")] +#[napi(catch_unwind, skip_typescript, js_name = "__testOpenHostTransport")] +pub fn test_open_host_transport( + env: Env, + handler: JsFunction, + max_operations: u32, + max_bytes: u32, + timeout_ms: u32, +) -> boundary::Result { + boundary::run_stateless(|| { + let handle = NEXT_TRANSPORT_HANDLE.fetch_add(1, Ordering::Relaxed); + if handle == 0 { + return Err(napi::Error::new( + "E_NATIVE_FAILURE", + "host transport handle space exhausted", + )); + } + let transport = Arc::new(HostTransport::new( + handler, + max_operations as usize, + max_bytes as usize, + Duration::from_millis(timeout_ms as u64), + )?); + registry().insert(handle, transport.clone()); + if let Err(error) = env.add_async_cleanup_hook( + CleanupTransport { + handle, + transport: Arc::downgrade(&transport), + }, + |cleanup| { + if let Some(transport) = cleanup.transport.upgrade() { + transport.close(); + } + registry().remove(&cleanup.handle); + }, + ) { + registry().remove(&handle); + return Err(napi::Error::new("E_NATIVE_FAILURE", error.to_string())); + } + Ok(handle) + })? +} + +#[cfg(feature = "test-panic")] +#[napi(catch_unwind, skip_typescript, js_name = "__testHostRoundTrip")] +pub fn test_host_round_trip( + handle: u32, + request: Buffer, + response_budget: u32, + callback: JsFunction, +) -> boundary::Result<()> { + boundary::run_stateless(|| { + let transport = registry() + .get(&handle) + .cloned() + .ok_or_else(|| napi::Error::new("E_CLOSED", "unknown or closed host storage transport"))?; + let completion = completion(callback)?; + thread::Builder::new() + .name(format!("fulltext-host-storage-test-{handle}")) + .spawn(move || { + let result = test_result( + transport + .round_trip(request.to_vec(), response_budget as usize) + .map(TransportResponse::into_bytes), + ); + let _ = completion.call(result, ThreadsafeFunctionCallMode::NonBlocking); + }) + .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?; + Ok(()) + })? +} + +#[cfg(feature = "test-panic")] +#[napi(catch_unwind, skip_typescript, js_name = "__testVerifyTantivyOnHostTransport")] +pub fn test_verify_tantivy_on_host_transport(handle: u32, callback: JsFunction) -> boundary::Result<()> { + boundary::run_stateless(|| { + let transport = registry() + .get(&handle) + .cloned() + .ok_or_else(|| napi::Error::new("E_CLOSED", "unknown or closed host storage transport"))?; + let completion = completion(callback)?; + thread::Builder::new() + .name(format!("fulltext-host-directory-test-{handle}")) + .spawn(move || { + let store = HostKvStore::new( + transport, + KvStoreIdentity(1, handle as u64, 1), + 32 * 1024 * 1024, + 64 * 1024, + ); + let run = NEXT_TRANSPORT_HANDLE.fetch_add(1, Ordering::Relaxed); + let case = AtomicU32::new(0); + let result = crate::directory_harness::verify_directory_contract(|| { + let namespace = format!("host-contract/{run}/{}", case.fetch_add(1, Ordering::Relaxed)); + KvDirectory::with_namespace(store.clone(), namespace.as_bytes()) + }) + .and_then(|_| { + let namespace = format!("host-lifecycle/{run}"); + crate::directory_harness::verify_tantivy_lifecycle(KvDirectory::with_namespace( + store, + namespace.as_bytes(), + )) + }) + .map(|_| Vec::new()) + .map_err(io::Error::other); + let _ = completion.call(test_result(result), ThreadsafeFunctionCallMode::NonBlocking); + }) + .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?; + Ok(()) + })? +} + +#[cfg(feature = "test-panic")] +#[napi(catch_unwind, skip_typescript, js_name = "__testCloseHostTransport")] +pub fn test_close_host_transport(handle: u32) -> boundary::Result { + boundary::run_stateless(|| { + let transport = registry().remove(&handle); + if let Some(transport) = &transport { + transport.close(); + } + transport.is_some() + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn closing_wakes_every_pending_request() { + let response = Arc::new(ResponseSlot::new()); + response.complete(Err(io::Error::new(io::ErrorKind::BrokenPipe, "closed"))); + assert_eq!( + response.wait(Duration::ZERO).unwrap().unwrap_err().kind(), + io::ErrorKind::BrokenPipe + ); + } + + #[test] + fn response_slot_accepts_only_the_first_completion() { + let response = ResponseSlot::new(); + response.complete(Ok(vec![1])); + response.complete(Ok(vec![2])); + assert_eq!(response.wait(Duration::ZERO).unwrap().unwrap(), vec![1]); + } +} diff --git a/src/lib.rs b/src/lib.rs index 86ed122..202a385 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -15,6 +15,12 @@ pub mod rocks_lease; #[cfg(feature = "node-api")] mod boundary; +#[cfg(feature = "test-panic")] +pub mod host_storage; + +#[cfg(all(feature = "node-api", feature = "phase0", not(feature = "test-panic")))] +mod host_storage; + #[cfg(feature = "node-api")] pub mod native; diff --git a/test/host-storage-transport.test.mjs b/test/host-storage-transport.test.mjs new file mode 100644 index 0000000..543e33d --- /dev/null +++ b/test/host-storage-transport.test.mjs @@ -0,0 +1,166 @@ +import assert from 'node:assert'; +import test from 'node:test'; + +import { createHostStorageHandler } from '../dist/host-storage.js'; +import { loadAddon } from '../dist/load-addon.js'; + +const addon = loadAddon(); + +test('host storage transport round-trips bytes without blocking JavaScript', async (context) => { + const requests = []; + const handle = addon.__testOpenHostTransport( + (request) => { + requests.push(Buffer.from(request)); + return Buffer.concat([Buffer.from('response:'), request]); + }, + 4, + 1024, + 1_000, + ); + context.after(() => addon.__testCloseHostTransport(handle)); + + const response = await roundTrip(handle, Buffer.from('read:key')); + assert.deepStrictEqual(requests, [Buffer.from('read:key')]); + assert.deepStrictEqual(response, Buffer.from('response:read:key')); +}); + +test('host storage transport rejects work beyond its operation budget', async (context) => { + const handle = addon.__testOpenHostTransport( + (request) => { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 30); + return request; + }, + 1, + 1024, + 1_000, + ); + context.after(() => addon.__testCloseHostTransport(handle)); + + const first = roundTrip(handle, Buffer.from('first')); + const second = roundTrip(handle, Buffer.from('second')); + const results = await Promise.allSettled([first, second]); + assert.strictEqual(results.filter(({ status }) => status === 'fulfilled').length, 1); + assert.strictEqual(results.filter(({ status }) => status === 'rejected').length, 1); + assert.match(results.find(({ status }) => status === 'rejected').reason.message, /at capacity/); +}); + +test('host storage transport reserves response capacity before dispatch', async (context) => { + let calls = 0; + const handle = addon.__testOpenHostTransport( + (request) => { + calls++; + return request; + }, + 1, + 16, + 1_000, + ); + context.after(() => addon.__testCloseHostTransport(handle)); + + await assert.rejects(roundTrip(handle, Buffer.alloc(8), 9), /reservation exceeds/); + assert.strictEqual(calls, 0); +}); + +test('a timed-out host request closes the transport and fences late completion', async (context) => { + const handle = addon.__testOpenHostTransport( + (request) => { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 40); + return request; + }, + 1, + 1024, + 5, + ); + context.after(() => addon.__testCloseHostTransport(handle)); + + await assert.rejects(roundTrip(handle, Buffer.from('slow')), /timed out/); + await assert.rejects(roundTrip(handle, Buffer.from('later')), /closed|timed out/); +}); + +test('invalid callback responses fail the request without terminating the process', async (context) => { + const handle = addon.__testOpenHostTransport(() => 'not a buffer', 1, 1024, 1_000); + context.after(() => addon.__testCloseHostTransport(handle)); + + await assert.rejects(roundTrip(handle, Buffer.from('read')), /must return a Buffer/); +}); + +test('KvDirectory and Tantivy operate through the host storage transport', async (context) => { + const entries = new Map(); + const requests = []; + const storage = { + read(key) { + requests.push('read'); + return entries.get(key.toString('hex')); + }, + write(mutations, policy) { + requests.push(`write:${policy}`); + 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() { + requests.push('sync'); + }, + }; + const handler = createHostStorageHandler(storage, { + maxMutations: 1_024, + maxReadResponseBytes: 32 * 1024 * 1024, + maxErrorBytes: 64 * 1024, + }); + const handle = addon.__testOpenHostTransport(handler, 32, 64 * 1024 * 1024, 5_000); + context.after(() => addon.__testCloseHostTransport(handle)); + + await verifyTantivy(handle); + assert.ok(requests.includes('read'), 'read requests were issued'); + assert.ok( + requests.some((request) => request.startsWith('write:')), + 'atomic write batches were issued', + ); + assert.ok(requests.includes('sync'), 'directory sync requests were issued'); + assert.ok(entries.size > 0, 'Tantivy state remains in host storage for reopen'); +}); + +test('host storage failures cross the native boundary without escaping JavaScript', async (context) => { + const storage = { + read() { + throw new Error('injected host read failure'); + }, + write() {}, + sync() {}, + }; + const handler = createHostStorageHandler(storage, { + maxMutations: 16, + maxReadResponseBytes: 1_024, + maxErrorBytes: 1_024, + }); + const handle = addon.__testOpenHostTransport(handler, 4, 64 * 1024 * 1024, 1_000); + context.after(() => addon.__testCloseHostTransport(handle)); + + await assert.rejects(verifyTantivy(handle), /injected host read failure/); +}); + +function roundTrip(handle, request, responseBudget = 128) { + return new Promise((resolve, reject) => { + addon.__testHostRoundTrip(handle, request, responseBudget, (encoded) => { + if (encoded[0] === 0) { + resolve(encoded.subarray(1)); + } else { + reject(new Error(encoded.subarray(1).toString())); + } + }); + }); +} + +function verifyTantivy(handle) { + return new Promise((resolve, reject) => { + addon.__testVerifyTantivyOnHostTransport(handle, (encoded) => { + if (encoded[0] === 0) { + resolve(); + } else { + reject(new Error(encoded.subarray(1).toString())); + } + }); + }); +} diff --git a/ts/host-storage.ts b/ts/host-storage.ts new file mode 100644 index 0000000..8ca43b6 --- /dev/null +++ b/ts/host-storage.ts @@ -0,0 +1,160 @@ +const protocolVersion = 1; +const operationRead = 1; +const operationWrite = 2; +const operationSync = 3; +const responseOk = 0; +const responseError = 1; +const valueMissing = 0; +const valuePresent = 1; +const mutationPut = 1; +const mutationDelete = 2; +const minimumMutationBytes = 5; +const fallbackError = Buffer.from([protocolVersion, responseError, 0, 0, 0, 0]); + +export type HostWritePolicy = 'wal' | 'wal-sync' | 'no-wal'; + +export type HostStorageMutation = { type: 'put'; key: Buffer; value: Buffer } | { type: 'delete'; key: Buffer }; + +export interface HostStorage { + read(key: Buffer): Buffer | undefined; + /** Mutation buffers are borrowed for this call and must not be retained. */ + write(mutations: Array, policy: HostWritePolicy): void; + sync(): void; +} + +export interface HostStorageHandlerOptions { + maxMutations: number; + maxReadResponseBytes: number; + maxErrorBytes: number; +} + +export function createHostStorageHandler( + storage: HostStorage, + { maxMutations, maxReadResponseBytes, maxErrorBytes }: HostStorageHandlerOptions, +): (request: Buffer) => Buffer { + validateLimit(maxMutations, 'maxMutations'); + validateLimit(maxReadResponseBytes, 'maxReadResponseBytes'); + validateLimit(maxErrorBytes, 'maxErrorBytes'); + if (maxReadResponseBytes < 7 || maxReadResponseBytes < maxErrorBytes) { + throw new TypeError('maxReadResponseBytes must be at least 7 and no smaller than maxErrorBytes'); + } + return (request) => { + try { + const decoder = new RequestDecoder(request); + if (decoder.u8() !== protocolVersion) throw new Error('unsupported host storage protocol version'); + const operation = decoder.u8(); + if (operation === operationRead) { + const key = decoder.bytes(); + decoder.finish(); + const value = storage.read(key); + if (value === undefined) return Buffer.from([protocolVersion, responseOk, valueMissing]); + if (!Buffer.isBuffer(value)) throw new Error('host storage read must return a Buffer or undefined'); + const responseBytes = 7 + value.length; + if (responseBytes > maxReadResponseBytes) throw new Error('host storage read response exceeds its byte limit'); + const response = Buffer.allocUnsafe(responseBytes); + response[0] = protocolVersion; + response[1] = responseOk; + response[2] = valuePresent; + response.writeUInt32LE(value.length, 3); + value.copy(response, 7); + return response; + } + if (operation === operationWrite) { + const policy = decodeWritePolicy(decoder.u8()); + const count = decoder.u32(); + if (count > maxMutations || count > Math.floor(decoder.remaining / minimumMutationBytes)) { + throw new Error('host storage mutation count exceeds its limit'); + } + const mutations = new Array(count); + for (let index = 0; index < count; index++) { + const type = decoder.u8(); + const key = decoder.bytes(); + if (type === mutationPut) { + mutations[index] = { type: 'put', key, value: decoder.bytes() }; + } else if (type === mutationDelete) { + mutations[index] = { type: 'delete', key }; + } else { + throw new Error(`unknown host storage mutation type ${type}`); + } + } + decoder.finish(); + storage.write(mutations, policy); + return Buffer.from([protocolVersion, responseOk]); + } + if (operation === operationSync) { + decoder.finish(); + storage.sync(); + return Buffer.from([protocolVersion, responseOk]); + } + throw new Error(`unknown host storage operation ${operation}`); + } catch (error) { + try { + return encodeError(error, maxErrorBytes); + } catch { + return fallbackError; + } + } + }; +} + +function decodeWritePolicy(encoded: number): HostWritePolicy { + if (encoded === 1) return 'wal'; + if (encoded === 2) return 'wal-sync'; + if (encoded === 3) return 'no-wal'; + throw new Error(`unknown host storage write policy ${encoded}`); +} + +function encodeError(error: unknown, maxErrorBytes: number): Buffer { + const encoded = Buffer.from(error instanceof Error ? error.message : String(error)); + const length = Math.min(encoded.length, maxErrorBytes - 6); + const response = Buffer.allocUnsafe(6 + length); + response[0] = protocolVersion; + response[1] = responseError; + response.writeUInt32LE(length, 2); + encoded.copy(response, 6, 0, length); + return response; +} + +function validateLimit(value: number, name: string): void { + if (!Number.isSafeInteger(value) || value <= 0 || (name === 'maxErrorBytes' && value < 6)) { + throw new TypeError(`${name} must be a positive safe integer${name === 'maxErrorBytes' ? ' of at least 6' : ''}`); + } +} + +class RequestDecoder { + readonly #bytes: Buffer; + #offset = 0; + + constructor(bytes: Buffer) { + if (!Buffer.isBuffer(bytes)) throw new TypeError('host storage request must be a Buffer'); + this.#bytes = bytes; + } + + get remaining(): number { + return this.#bytes.length - this.#offset; + } + + u8(): number { + if (this.remaining < 1) throw new Error('host storage request is truncated'); + return this.#bytes[this.#offset++]; + } + + u32(): number { + if (this.remaining < 4) throw new Error('host storage request is truncated'); + const value = this.#bytes.readUInt32LE(this.#offset); + this.#offset += 4; + return value; + } + + bytes(): Buffer { + const length = this.u32(); + if (length > this.remaining) throw new Error('host storage request is truncated'); + const start = this.#offset; + this.#offset += length; + return this.#bytes.subarray(start, this.#offset); + } + + finish(): void { + if (this.remaining !== 0) throw new Error('host storage request has trailing bytes'); + } +} diff --git a/ts/load-addon.ts b/ts/load-addon.ts index c5861a1..d401029 100644 --- a/ts/load-addon.ts +++ b/ts/load-addon.ts @@ -24,6 +24,15 @@ interface NativeAddonApi { __testPanic?(id: number): void; __testCheck?(id: number): boolean; __testPoisonNativeHandle?(handle: number): void; + __testOpenHostTransport?( + handler: (request: Buffer) => Buffer, + maxOperations: number, + maxBytes: number, + timeoutMs: number, + ): number; + __testHostRoundTrip?(handle: number, request: Buffer, responseBudget: number, callback: NativeCallback): void; + __testVerifyTantivyOnHostTransport?(handle: number, callback: NativeCallback): void; + __testCloseHostTransport?(handle: number): boolean; } export type NativeCallback = (response: Buffer) => void; From 3d34a262a708c3b63d472d7191a72bbbbff58884 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Tue, 8 Sep 2026 16:08:21 -0600 Subject: [PATCH 2/7] fix: harden host storage boundary --- src/host_storage.rs | 151 +++++++++++++++++++-------- src/lib.rs | 3 - test/host-storage-transport.test.mjs | 50 ++++++--- ts/host-storage.ts | 12 ++- ts/load-addon.ts | 15 ++- 5 files changed, 166 insertions(+), 65 deletions(-) diff --git a/src/host_storage.rs b/src/host_storage.rs index f57d496..d09d3bd 100644 --- a/src/host_storage.rs +++ b/src/host_storage.rs @@ -1,6 +1,7 @@ use std::collections::HashMap; use std::io; use std::ops::Deref; +use std::panic::{catch_unwind, AssertUnwindSafe}; use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak}; use std::thread; @@ -8,7 +9,7 @@ use std::time::Duration; use napi::bindgen_prelude::Buffer; use napi::threadsafe_function::{ErrorStrategy, ThreadSafeCallContext, ThreadsafeFunction, ThreadsafeFunctionCallMode}; -use napi::{Env, JsBuffer, JsFunction, JsUnknown, Status, ValueType}; +use napi::{Env, JsBuffer, JsFunction, JsUnknown, Status}; use napi_derive::napi; use tantivy::directory::OwnedBytes; @@ -73,7 +74,13 @@ impl HostTransport { }) } - fn round_trip(self: &Arc, request: Vec, response_budget: usize) -> io::Result { + // Completion requires the owning JavaScript environment to run, so callers must be native worker threads. + fn round_trip( + self: &Arc, + request: Vec, + response_budget: usize, + terminal_on_timeout: bool, + ) -> 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"); @@ -124,7 +131,11 @@ impl HostTransport { } }; } - self.fail(io::ErrorKind::TimedOut, "host storage request timed out"); + if terminal_on_timeout { + self.fail(io::ErrorKind::TimedOut, "host storage request timed out"); + } else { + self.release(request_id); + } let _ = response.take(); Err(io::Error::new( io::ErrorKind::TimedOut, @@ -239,6 +250,9 @@ impl Deref for TransportResponse { } } +// Moving this wrapper never moves the Vec allocation referenced by Deref. +unsafe impl stable_deref_trait::StableDeref for TransportResponse {} + impl Drop for TransportResponse { fn drop(&mut self) { if let Some(transport) = self.transport.upgrade() { @@ -281,7 +295,7 @@ impl ResponseSlot { } fn response_bytes(value: JsUnknown, max_bytes: usize) -> io::Result> { - if value.get_type().map_err(|error| io::Error::other(error.to_string()))? != ValueType::Object { + if !value.is_buffer().map_err(|error| io::Error::other(error.to_string()))? { return Err(io::Error::new( io::ErrorKind::InvalidData, "host storage callback must return a Buffer", @@ -335,7 +349,22 @@ impl HostKvStore { } fn request(&self, request: Vec, response_budget: usize) -> io::Result { - let response = self.transport.round_trip(request, response_budget)?; + self.request_with_timeout_policy(request, response_budget, false) + } + + fn request_terminal(&self, request: Vec, response_budget: usize) -> io::Result { + self.request_with_timeout_policy(request, response_budget, true) + } + + fn request_with_timeout_policy( + &self, + request: Vec, + response_budget: usize, + terminal_on_timeout: bool, + ) -> io::Result { + let response = self + .transport + .round_trip(request, response_budget, terminal_on_timeout)?; let mut decoder = ResponseDecoder::new(response); if decoder.u8()? != HOST_PROTOCOL_VERSION { return Err(io::Error::new( @@ -368,8 +397,11 @@ impl KvStore for HostKvStore { request.bytes(key)?; let mut response = self.request(request.finish(), self.max_read_response_bytes)?; let value = match response.u8()? { - VALUE_MISSING => None, - VALUE_PRESENT => Some(OwnedBytes::new(response.bytes()?.to_vec())), + VALUE_MISSING => { + response.finish()?; + None + } + VALUE_PRESENT => Some(response.into_owned_bytes()?), _ => { return Err(io::Error::new( io::ErrorKind::InvalidData, @@ -377,7 +409,6 @@ impl KvStore for HostKvStore { )) } }; - response.finish()?; Ok(value) } @@ -403,12 +434,12 @@ impl KvStore for HostKvStore { } } } - self.request(request.finish(), self.max_control_response_bytes)? + self.request_terminal(request.finish(), self.max_control_response_bytes)? .finish() } fn sync(&self) -> io::Result<()> { - self.request(RequestEncoder::new(OP_SYNC).finish(), self.max_control_response_bytes)? + self.request_terminal(RequestEncoder::new(OP_SYNC).finish(), self.max_control_response_bytes)? .finish() } } @@ -462,6 +493,23 @@ impl ResponseDecoder { } fn bytes(&mut self) -> io::Result<&[u8]> { + let range = self.byte_range()?; + Ok(&self.bytes[range]) + } + + fn into_owned_bytes(mut self) -> io::Result { + let range = self.byte_range()?; + if self.offset != self.bytes.len() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "host storage response has trailing bytes", + )); + } + let bytes = OwnedBytes::new(self.bytes); + Ok(bytes.slice(range)) + } + + fn byte_range(&mut self) -> io::Result> { let length_end = self .offset .checked_add(4) @@ -478,12 +526,15 @@ impl ResponseDecoder { .offset .checked_add(length) .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "host storage response length overflow"))?; - let value = self - .bytes - .get(self.offset..end) - .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "host storage response is truncated"))?; + if end > self.bytes.len() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "host storage response is truncated", + )); + } + let start = self.offset; self.offset = end; - Ok(value) + Ok(start..end) } fn finish(self) -> io::Result<()> { @@ -532,6 +583,13 @@ fn test_result(result: io::Result>) -> Vec { } } +fn test_thread_result(operation: impl FnOnce() -> io::Result>) -> Vec { + match catch_unwind(AssertUnwindSafe(operation)) { + Ok(result) => test_result(result), + Err(_) => test_result(Err(io::Error::other("native host storage test panicked"))), + } +} + struct CleanupTransport { handle: u32, transport: Weak, @@ -586,6 +644,7 @@ pub fn test_host_round_trip( handle: u32, request: Buffer, response_budget: u32, + terminal_on_timeout: bool, callback: JsFunction, ) -> boundary::Result<()> { boundary::run_stateless(|| { @@ -594,14 +653,15 @@ pub fn test_host_round_trip( .cloned() .ok_or_else(|| napi::Error::new("E_CLOSED", "unknown or closed host storage transport"))?; let completion = completion(callback)?; + let request = request.to_vec(); thread::Builder::new() .name(format!("fulltext-host-storage-test-{handle}")) .spawn(move || { - let result = test_result( + let result = test_thread_result(|| { transport - .round_trip(request.to_vec(), response_budget as usize) - .map(TransportResponse::into_bytes), - ); + .round_trip(request, response_budget as usize, terminal_on_timeout) + .map(TransportResponse::into_bytes) + }); let _ = completion.call(result, ThreadsafeFunctionCallMode::NonBlocking); }) .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?; @@ -611,7 +671,12 @@ pub fn test_host_round_trip( #[cfg(feature = "test-panic")] #[napi(catch_unwind, skip_typescript, js_name = "__testVerifyTantivyOnHostTransport")] -pub fn test_verify_tantivy_on_host_transport(handle: u32, callback: JsFunction) -> boundary::Result<()> { +pub fn test_verify_tantivy_on_host_transport( + handle: u32, + max_read_response_bytes: u32, + max_control_response_bytes: u32, + callback: JsFunction, +) -> boundary::Result<()> { boundary::run_stateless(|| { let transport = registry() .get(&handle) @@ -621,28 +686,30 @@ pub fn test_verify_tantivy_on_host_transport(handle: u32, callback: JsFunction) thread::Builder::new() .name(format!("fulltext-host-directory-test-{handle}")) .spawn(move || { - let store = HostKvStore::new( - transport, - KvStoreIdentity(1, handle as u64, 1), - 32 * 1024 * 1024, - 64 * 1024, - ); - let run = NEXT_TRANSPORT_HANDLE.fetch_add(1, Ordering::Relaxed); - let case = AtomicU32::new(0); - let result = crate::directory_harness::verify_directory_contract(|| { - let namespace = format!("host-contract/{run}/{}", case.fetch_add(1, Ordering::Relaxed)); - KvDirectory::with_namespace(store.clone(), namespace.as_bytes()) - }) - .and_then(|_| { - let namespace = format!("host-lifecycle/{run}"); - crate::directory_harness::verify_tantivy_lifecycle(KvDirectory::with_namespace( - store, - namespace.as_bytes(), - )) - }) - .map(|_| Vec::new()) - .map_err(io::Error::other); - let _ = completion.call(test_result(result), ThreadsafeFunctionCallMode::NonBlocking); + let result = test_thread_result(|| { + let store = HostKvStore::new( + transport, + KvStoreIdentity(1, handle as u64, 1), + max_read_response_bytes as usize, + max_control_response_bytes as usize, + ); + let run = NEXT_TRANSPORT_HANDLE.fetch_add(1, Ordering::Relaxed); + let case = AtomicU32::new(0); + crate::directory_harness::verify_directory_contract(|| { + let namespace = format!("host-contract/{run}/{}", case.fetch_add(1, Ordering::Relaxed)); + KvDirectory::with_namespace(store.clone(), namespace.as_bytes()) + }) + .and_then(|_| { + let namespace = format!("host-lifecycle/{run}"); + crate::directory_harness::verify_tantivy_lifecycle(KvDirectory::with_namespace( + store, + namespace.as_bytes(), + )) + }) + .map(|_| Vec::new()) + .map_err(io::Error::other) + }); + let _ = completion.call(result, ThreadsafeFunctionCallMode::NonBlocking); }) .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?; Ok(()) diff --git a/src/lib.rs b/src/lib.rs index 202a385..e751c89 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -18,9 +18,6 @@ mod boundary; #[cfg(feature = "test-panic")] pub mod host_storage; -#[cfg(all(feature = "node-api", feature = "phase0", not(feature = "test-panic")))] -mod host_storage; - #[cfg(feature = "node-api")] pub mod native; diff --git a/test/host-storage-transport.test.mjs b/test/host-storage-transport.test.mjs index 543e33d..159744e 100644 --- a/test/host-storage-transport.test.mjs +++ b/test/host-storage-transport.test.mjs @@ -5,8 +5,10 @@ import { createHostStorageHandler } from '../dist/host-storage.js'; import { loadAddon } from '../dist/load-addon.js'; const addon = loadAddon(); +const readResponseBytes = 1024 * 1024; +const controlResponseBytes = 64 * 1024; -test('host storage transport round-trips bytes without blocking JavaScript', async (context) => { +test('host storage transport round-trips bytes intact', async (context) => { const requests = []; const handle = addon.__testOpenHostTransport( (request) => { @@ -73,15 +75,33 @@ test('a timed-out host request closes the transport and fences late completion', ); context.after(() => addon.__testCloseHostTransport(handle)); - await assert.rejects(roundTrip(handle, Buffer.from('slow')), /timed out/); + await assert.rejects(roundTrip(handle, Buffer.from('slow'), 128, true), /timed out/); await assert.rejects(roundTrip(handle, Buffer.from('later')), /closed|timed out/); }); -test('invalid callback responses fail the request without terminating the process', async (context) => { - const handle = addon.__testOpenHostTransport(() => 'not a buffer', 1, 1024, 1_000); +test('a read timeout fences only that request', async (context) => { + let calls = 0; + const handle = addon.__testOpenHostTransport( + (request) => { + if (calls++ === 0) Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 40); + return request; + }, + 1, + 1_024, + 5, + ); context.after(() => addon.__testCloseHostTransport(handle)); - await assert.rejects(roundTrip(handle, Buffer.from('read')), /must return a Buffer/); + await assert.rejects(roundTrip(handle, Buffer.from('slow')), /timed out/); + assert.deepStrictEqual(await roundTrip(handle, Buffer.from('recovered')), Buffer.from('recovered')); +}); + +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); + context.after(() => addon.__testCloseHostTransport(handle)); + await assert.rejects(roundTrip(handle, Buffer.from('read')), /must return a Buffer/); + } }); test('KvDirectory and Tantivy operate through the host storage transport', async (context) => { @@ -106,18 +126,17 @@ test('KvDirectory and Tantivy operate through the host storage transport', async }; const handler = createHostStorageHandler(storage, { maxMutations: 1_024, - maxReadResponseBytes: 32 * 1024 * 1024, - maxErrorBytes: 64 * 1024, + maxReadResponseBytes: readResponseBytes, + maxControlResponseBytes: controlResponseBytes, + maxErrorBytes: controlResponseBytes, }); - const handle = addon.__testOpenHostTransport(handler, 32, 64 * 1024 * 1024, 5_000); + const handle = addon.__testOpenHostTransport(handler, 32, 40 * 1024 * 1024, 5_000); context.after(() => addon.__testCloseHostTransport(handle)); await verifyTantivy(handle); assert.ok(requests.includes('read'), 'read requests were issued'); - assert.ok( - requests.some((request) => request.startsWith('write:')), - 'atomic write batches were issued', - ); + assert.ok(requests.includes('write:wal'), 'ordinary objects use WAL writes'); + assert.ok(requests.includes('write:wal-sync'), 'metadata publication uses a synchronous WAL write'); assert.ok(requests.includes('sync'), 'directory sync requests were issued'); assert.ok(entries.size > 0, 'Tantivy state remains in host storage for reopen'); }); @@ -133,6 +152,7 @@ test('host storage failures cross the native boundary without escaping JavaScrip const handler = createHostStorageHandler(storage, { maxMutations: 16, maxReadResponseBytes: 1_024, + maxControlResponseBytes: 1_024, maxErrorBytes: 1_024, }); const handle = addon.__testOpenHostTransport(handler, 4, 64 * 1024 * 1024, 1_000); @@ -141,9 +161,9 @@ test('host storage failures cross the native boundary without escaping JavaScrip await assert.rejects(verifyTantivy(handle), /injected host read failure/); }); -function roundTrip(handle, request, responseBudget = 128) { +function roundTrip(handle, request, responseBudget = 128, terminalOnTimeout = false) { return new Promise((resolve, reject) => { - addon.__testHostRoundTrip(handle, request, responseBudget, (encoded) => { + addon.__testHostRoundTrip(handle, request, responseBudget, terminalOnTimeout, (encoded) => { if (encoded[0] === 0) { resolve(encoded.subarray(1)); } else { @@ -155,7 +175,7 @@ function roundTrip(handle, request, responseBudget = 128) { function verifyTantivy(handle) { return new Promise((resolve, reject) => { - addon.__testVerifyTantivyOnHostTransport(handle, (encoded) => { + addon.__testVerifyTantivyOnHostTransport(handle, readResponseBytes, controlResponseBytes, (encoded) => { if (encoded[0] === 0) { resolve(); } else { diff --git a/ts/host-storage.ts b/ts/host-storage.ts index 8ca43b6..fa54a9e 100644 --- a/ts/host-storage.ts +++ b/ts/host-storage.ts @@ -25,18 +25,24 @@ export interface HostStorage { export interface HostStorageHandlerOptions { maxMutations: number; maxReadResponseBytes: number; + maxControlResponseBytes: number; maxErrorBytes: number; } export function createHostStorageHandler( storage: HostStorage, - { maxMutations, maxReadResponseBytes, maxErrorBytes }: HostStorageHandlerOptions, + { maxMutations, maxReadResponseBytes, maxControlResponseBytes, maxErrorBytes }: HostStorageHandlerOptions, ): (request: Buffer) => Buffer { validateLimit(maxMutations, 'maxMutations'); validateLimit(maxReadResponseBytes, 'maxReadResponseBytes'); + validateLimit(maxControlResponseBytes, 'maxControlResponseBytes'); validateLimit(maxErrorBytes, 'maxErrorBytes'); - if (maxReadResponseBytes < 7 || maxReadResponseBytes < maxErrorBytes) { - throw new TypeError('maxReadResponseBytes must be at least 7 and no smaller than maxErrorBytes'); + if ( + maxReadResponseBytes < 7 || + maxControlResponseBytes < 2 || + maxErrorBytes > Math.min(maxReadResponseBytes, maxControlResponseBytes) + ) { + throw new TypeError('response byte limits are inconsistent'); } return (request) => { try { diff --git a/ts/load-addon.ts b/ts/load-addon.ts index d401029..655efe0 100644 --- a/ts/load-addon.ts +++ b/ts/load-addon.ts @@ -30,8 +30,19 @@ interface NativeAddonApi { maxBytes: number, timeoutMs: number, ): number; - __testHostRoundTrip?(handle: number, request: Buffer, responseBudget: number, callback: NativeCallback): void; - __testVerifyTantivyOnHostTransport?(handle: number, callback: NativeCallback): void; + __testHostRoundTrip?( + handle: number, + request: Buffer, + responseBudget: number, + terminalOnTimeout: boolean, + callback: NativeCallback, + ): void; + __testVerifyTantivyOnHostTransport?( + handle: number, + maxReadResponseBytes: number, + maxControlResponseBytes: number, + callback: NativeCallback, + ): void; __testCloseHostTransport?(handle: number): boolean; } From f7d1b725d163eecacd0d27a35ace9fff8b38cb96 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Tue, 8 Sep 2026 16:12:52 -0600 Subject: [PATCH 3/7] fix: retain actual host response bytes --- src/host_storage.rs | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/src/host_storage.rs b/src/host_storage.rs index d09d3bd..8faccab 100644 --- a/src/host_storage.rs +++ b/src/host_storage.rs @@ -189,13 +189,19 @@ impl HostTransport { fn complete(&self, request_id: u64, result: io::Result>) { let response = { let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); - let Some(pending) = state.pending.get_mut(&request_id) else { + let Some(pending) = state.pending.get(&request_id) else { return; }; if pending.completed { return; } let response = pending.response.upgrade(); + let reserved_bytes = pending.retained_bytes; + let response_bytes = result.as_ref().map_or(0, Vec::len); + state.operations -= 1; + state.bytes = state.bytes - reserved_bytes + response_bytes; + let pending = state.pending.get_mut(&request_id).unwrap(); + pending.retained_bytes = response_bytes; pending.completed = true; response }; @@ -207,7 +213,9 @@ impl HostTransport { fn release(&self, request_id: u64) { let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); if let Some(pending) = state.pending.remove(&request_id) { - state.operations -= 1; + if !pending.completed { + state.operations -= 1; + } state.bytes -= pending.retained_bytes; } } From e65566b5a6fa0024bde0fb520a399f52bd704ea1 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Tue, 8 Sep 2026 16:35:26 -0600 Subject: [PATCH 4/7] fix: make host mutations outcome-safe --- src/host_storage.rs | 159 +++++++++++++++++++-------- test/host-storage-transport.test.mjs | 64 +++++++++-- ts/host-storage.ts | 3 +- ts/load-addon.ts | 2 +- 4 files changed, 169 insertions(+), 59 deletions(-) diff --git a/src/host_storage.rs b/src/host_storage.rs index 8faccab..5332f5a 100644 --- a/src/host_storage.rs +++ b/src/host_storage.rs @@ -5,7 +5,7 @@ use std::panic::{catch_unwind, AssertUnwindSafe}; use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak}; use std::thread; -use std::time::Duration; +use std::time::{Duration, Instant}; use napi::bindgen_prelude::Buffer; use napi::threadsafe_function::{ErrorStrategy, ThreadSafeCallContext, ThreadsafeFunction, ThreadsafeFunctionCallMode}; @@ -25,6 +25,7 @@ static HOST_TRANSPORTS: OnceLock>>> = Once struct HostTransport { handler: HostCallback, state: Mutex, + capacity: Condvar, next_request_id: AtomicU64, max_operations: usize, max_bytes: usize, @@ -51,22 +52,32 @@ struct ResponseSlot { } impl HostTransport { - fn new(handler: JsFunction, max_operations: usize, max_bytes: usize, timeout: Duration) -> boundary::Result { + fn new( + env: &Env, + handler: JsFunction, + max_operations: usize, + max_bytes: usize, + timeout: Duration, + ) -> boundary::Result { if max_operations == 0 || max_bytes == 0 || timeout.is_zero() { return Err(napi::Error::new( "E_INVALID_ARGUMENT", "host transport limits must be greater than zero", )); } - let handler = handler + let mut handler = handler .create_threadsafe_function::, Buffer, _, ErrorStrategy::Fatal>( max_operations, |context: ThreadSafeCallContext>| Ok(vec![Buffer::from(context.value)]), ) .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 { handler, state: Mutex::new(TransportState::default()), + capacity: Condvar::new(), next_request_id: AtomicU64::new(1), max_operations, max_bytes, @@ -79,7 +90,7 @@ impl HostTransport { self: &Arc, request: Vec, response_budget: usize, - terminal_on_timeout: bool, + deadline: Option, ) -> io::Result { let request_id = self.next_request_id.fetch_add(1, Ordering::Relaxed); if request_id == 0 { @@ -87,7 +98,7 @@ impl HostTransport { return Err(io::Error::other("host storage request id space exhausted")); } let response = Arc::new(ResponseSlot::new()); - self.admit(request_id, request.len(), response_budget, &response)?; + self.admit(request_id, request.len(), response_budget, &response, deadline)?; let transport = Arc::downgrade(self); let callback_response = response.clone(); @@ -96,14 +107,19 @@ impl HostTransport { request, ThreadsafeFunctionCallMode::NonBlocking, move |value| { - 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", - ))); + 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() { + callback_response.complete(Err(io::Error::other("host storage completion panicked"))); } Ok(()) }, @@ -118,7 +134,7 @@ impl HostTransport { ); } - if let Some(result) = response.wait(self.timeout) { + if let Some(result) = response.wait(deadline) { return match result { Ok(bytes) => Ok(TransportResponse { bytes, @@ -131,11 +147,7 @@ impl HostTransport { } }; } - if terminal_on_timeout { - self.fail(io::ErrorKind::TimedOut, "host storage request timed out"); - } else { - self.release(request_id); - } + self.release(request_id); let _ = response.take(); Err(io::Error::new( io::ErrorKind::TimedOut, @@ -153,6 +165,7 @@ impl HostTransport { request_bytes: usize, response_bytes: usize, response: &Arc, + deadline: Option, ) -> io::Result<()> { let retained_bytes = request_bytes .checked_add(response_bytes) @@ -164,14 +177,39 @@ impl HostTransport { )); } let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); - if let Some(error) = &state.closed { - return Err(io::Error::new(io::ErrorKind::BrokenPipe, error.clone())); - } - if state.operations >= self.max_operations || state.bytes.saturating_add(retained_bytes) > self.max_bytes { - return Err(io::Error::new( - io::ErrorKind::WouldBlock, - "host storage transport is at capacity", - )); + loop { + if let Some(error) = &state.closed { + return Err(io::Error::new(io::ErrorKind::BrokenPipe, error.clone())); + } + if state.operations < self.max_operations && state.bytes.saturating_add(retained_bytes) <= self.max_bytes { + break; + } + state = match deadline { + Some(deadline) => { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "host storage request timed out", + )); + } + let (state, wait) = self + .capacity + .wait_timeout(state, remaining) + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if wait.timed_out() { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "host storage request timed out", + )); + } + state + } + None => self + .capacity + .wait(state) + .unwrap_or_else(|poisoned| poisoned.into_inner()), + }; } state.operations += 1; state.bytes += retained_bytes; @@ -203,6 +241,7 @@ impl HostTransport { let pending = state.pending.get_mut(&request_id).unwrap(); pending.retained_bytes = response_bytes; pending.completed = true; + self.capacity.notify_all(); response }; if let Some(response) = response { @@ -217,6 +256,7 @@ impl HostTransport { state.operations -= 1; } state.bytes -= pending.retained_bytes; + self.capacity.notify_all(); } } @@ -230,6 +270,7 @@ impl HostTransport { state.bytes = 0; std::mem::take(&mut state.pending) }; + self.capacity.notify_all(); for request in pending.into_values() { if let Some(response) = request.response.upgrade() { response.complete(Err(io::Error::new(kind, message.to_owned()))); @@ -285,12 +326,31 @@ impl ResponseSlot { } } - fn wait(&self, timeout: Duration) -> Option>> { - let slot = self.result.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); - let (mut slot, _) = self - .ready - .wait_timeout_while(slot, timeout, |result| result.is_none()) - .unwrap_or_else(|poisoned| poisoned.into_inner()); + fn wait(&self, deadline: Option) -> Option>> { + let mut slot = self.result.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + match deadline { + Some(deadline) => { + while slot.is_none() { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return None; + } + let (next, wait) = self + .ready + .wait_timeout(slot, remaining) + .unwrap_or_else(|poisoned| poisoned.into_inner()); + slot = next; + if wait.timed_out() && slot.is_none() { + return None; + } + } + } + None => { + while slot.is_none() { + slot = self.ready.wait(slot).unwrap_or_else(|poisoned| poisoned.into_inner()); + } + } + } slot.take() } @@ -357,22 +417,25 @@ impl HostKvStore { } fn request(&self, request: Vec, response_budget: usize) -> io::Result { - self.request_with_timeout_policy(request, response_budget, false) + // Read deadlines cover admission and host execution; a timed-out read has no storage side effect. + let deadline = Instant::now() + .checked_add(self.transport.timeout) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "host storage timeout is too large"))?; + self.request_with_deadline(request, response_budget, Some(deadline)) } - fn request_terminal(&self, request: Vec, response_budget: usize) -> io::Result { - self.request_with_timeout_policy(request, response_budget, true) + fn request_mutation(&self, request: Vec, response_budget: usize) -> io::Result { + // A dispatched JavaScript mutation cannot be canceled, so wait for its definitive result. + self.request_with_deadline(request, response_budget, None) } - fn request_with_timeout_policy( + fn request_with_deadline( &self, request: Vec, response_budget: usize, - terminal_on_timeout: bool, + deadline: Option, ) -> io::Result { - let response = self - .transport - .round_trip(request, response_budget, terminal_on_timeout)?; + let response = self.transport.round_trip(request, response_budget, deadline)?; let mut decoder = ResponseDecoder::new(response); if decoder.u8()? != HOST_PROTOCOL_VERSION { return Err(io::Error::new( @@ -442,12 +505,12 @@ impl KvStore for HostKvStore { } } } - self.request_terminal(request.finish(), self.max_control_response_bytes)? + self.request_mutation(request.finish(), self.max_control_response_bytes)? .finish() } fn sync(&self) -> io::Result<()> { - self.request_terminal(RequestEncoder::new(OP_SYNC).finish(), self.max_control_response_bytes)? + self.request_mutation(RequestEncoder::new(OP_SYNC).finish(), self.max_control_response_bytes)? .finish() } } @@ -621,6 +684,7 @@ pub fn test_open_host_transport( )); } let transport = Arc::new(HostTransport::new( + &env, handler, max_operations as usize, max_bytes as usize, @@ -652,7 +716,7 @@ pub fn test_host_round_trip( handle: u32, request: Buffer, response_budget: u32, - terminal_on_timeout: bool, + use_timeout: bool, callback: JsFunction, ) -> boundary::Result<()> { boundary::run_stateless(|| { @@ -666,8 +730,9 @@ pub fn test_host_round_trip( .name(format!("fulltext-host-storage-test-{handle}")) .spawn(move || { let result = test_thread_result(|| { + let deadline = use_timeout.then(|| Instant::now() + transport.timeout); transport - .round_trip(request, response_budget as usize, terminal_on_timeout) + .round_trip(request, response_budget as usize, deadline) .map(TransportResponse::into_bytes) }); let _ = completion.call(result, ThreadsafeFunctionCallMode::NonBlocking); @@ -745,7 +810,7 @@ mod tests { let response = Arc::new(ResponseSlot::new()); response.complete(Err(io::Error::new(io::ErrorKind::BrokenPipe, "closed"))); assert_eq!( - response.wait(Duration::ZERO).unwrap().unwrap_err().kind(), + response.wait(Some(Instant::now())).unwrap().unwrap_err().kind(), io::ErrorKind::BrokenPipe ); } @@ -755,6 +820,6 @@ mod tests { let response = ResponseSlot::new(); response.complete(Ok(vec![1])); response.complete(Ok(vec![2])); - assert_eq!(response.wait(Duration::ZERO).unwrap().unwrap(), vec![1]); + assert_eq!(response.wait(Some(Instant::now())).unwrap().unwrap(), vec![1]); } } diff --git a/test/host-storage-transport.test.mjs b/test/host-storage-transport.test.mjs index 159744e..745ac79 100644 --- a/test/host-storage-transport.test.mjs +++ b/test/host-storage-transport.test.mjs @@ -26,7 +26,7 @@ test('host storage transport round-trips bytes intact', async (context) => { assert.deepStrictEqual(response, Buffer.from('response:read:key')); }); -test('host storage transport rejects work beyond its operation budget', async (context) => { +test('host storage transport backpressures work beyond its operation budget', async (context) => { const handle = addon.__testOpenHostTransport( (request) => { Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 30); @@ -40,10 +40,24 @@ test('host storage transport rejects work beyond its operation budget', async (c const first = roundTrip(handle, Buffer.from('first')); const second = roundTrip(handle, Buffer.from('second')); - const results = await Promise.allSettled([first, second]); - assert.strictEqual(results.filter(({ status }) => status === 'fulfilled').length, 1); - assert.strictEqual(results.filter(({ status }) => status === 'rejected').length, 1); - assert.match(results.find(({ status }) => status === 'rejected').reason.message, /at capacity/); + assert.deepStrictEqual(await Promise.all([first, second]), [Buffer.from('first'), Buffer.from('second')]); +}); + +test('host storage transport backpressures work beyond its byte budget', async (context) => { + const handle = addon.__testOpenHostTransport( + (request) => { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 30); + return request; + }, + 2, + 136, + 1_000, + ); + context.after(() => addon.__testCloseHostTransport(handle)); + + const first = roundTrip(handle, Buffer.from('first')); + const second = roundTrip(handle, Buffer.from('second')); + assert.deepStrictEqual(await Promise.all([first, second]), [Buffer.from('first'), Buffer.from('second')]); }); test('host storage transport reserves response capacity before dispatch', async (context) => { @@ -63,7 +77,7 @@ test('host storage transport reserves response capacity before dispatch', async assert.strictEqual(calls, 0); }); -test('a timed-out host request closes the transport and fences late completion', async (context) => { +test('a host request can wait for a definitive result without a deadline', async (context) => { const handle = addon.__testOpenHostTransport( (request) => { Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 40); @@ -75,8 +89,7 @@ test('a timed-out host request closes the transport and fences late completion', ); context.after(() => addon.__testCloseHostTransport(handle)); - await assert.rejects(roundTrip(handle, Buffer.from('slow'), 128, true), /timed out/); - await assert.rejects(roundTrip(handle, Buffer.from('later')), /closed|timed out/); + assert.deepStrictEqual(await roundTrip(handle, Buffer.from('slow'), 128, false), Buffer.from('slow')); }); test('a read timeout fences only that request', async (context) => { @@ -161,9 +174,33 @@ test('host storage failures cross the native boundary without escaping JavaScrip await assert.rejects(verifyTantivy(handle), /injected host read failure/); }); -function roundTrip(handle, request, responseBudget = 128, terminalOnTimeout = false) { +test('host storage handler preserves no-WAL policy and rejects malformed frames', () => { + const policies = []; + const handler = createHostStorageHandler( + { + read() {}, + write(_mutations, policy) { + policies.push(policy); + }, + sync() {}, + }, + { + maxMutations: 4, + maxReadResponseBytes: 64, + maxControlResponseBytes: 64, + maxErrorBytes: 64, + }, + ); + + assert.deepStrictEqual(handler(Buffer.from([1, 2, 3, 1, 0, 0, 0, 2, 1, 0, 0, 0, 97])), Buffer.from([1, 0])); + assert.deepStrictEqual(policies, ['no-wal']); + assert.match(decodeHandlerError(handler(Buffer.from([2, 3]))), /unsupported.*protocol version/); + assert.match(decodeHandlerError(handler(Buffer.from([1, 1, 4, 0, 0]))), /truncated/); +}); + +function roundTrip(handle, request, responseBudget = 128, useTimeout = true) { return new Promise((resolve, reject) => { - addon.__testHostRoundTrip(handle, request, responseBudget, terminalOnTimeout, (encoded) => { + addon.__testHostRoundTrip(handle, request, responseBudget, useTimeout, (encoded) => { if (encoded[0] === 0) { resolve(encoded.subarray(1)); } else { @@ -184,3 +221,10 @@ function verifyTantivy(handle) { }); }); } + +function decodeHandlerError(response) { + assert.strictEqual(response[0], 1); + assert.strictEqual(response[1], 1); + const length = response.readUInt32LE(2); + return response.subarray(6, 6 + length).toString(); +} diff --git a/ts/host-storage.ts b/ts/host-storage.ts index fa54a9e..c434073 100644 --- a/ts/host-storage.ts +++ b/ts/host-storage.ts @@ -17,8 +17,9 @@ export type HostStorageMutation = { type: 'put'; key: Buffer; value: Buffer } | export interface HostStorage { read(key: Buffer): Buffer | undefined; - /** Mutation buffers are borrowed for this call and must not be retained. */ + /** Apply the batch atomically and satisfy the write policy before returning. Mutation buffers are borrowed. */ write(mutations: Array, policy: HostWritePolicy): void; + /** Make prior writes durable before returning. */ sync(): void; } diff --git a/ts/load-addon.ts b/ts/load-addon.ts index 655efe0..220bc04 100644 --- a/ts/load-addon.ts +++ b/ts/load-addon.ts @@ -34,7 +34,7 @@ interface NativeAddonApi { handle: number, request: Buffer, responseBudget: number, - terminalOnTimeout: boolean, + useTimeout: boolean, callback: NativeCallback, ): void; __testVerifyTantivyOnHostTransport?( From 5546e60c08f255a3981502cd935a7b4f35a39cab Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Tue, 8 Sep 2026 16:58:13 -0600 Subject: [PATCH 5/7] fix: scope transport accounting to in-flight work --- src/host_storage.rs | 89 +++++++++------------------- test/host-storage-transport.test.mjs | 28 +++++++++ ts/host-storage.ts | 1 + 3 files changed, 57 insertions(+), 61 deletions(-) diff --git a/src/host_storage.rs b/src/host_storage.rs index 5332f5a..a723e3a 100644 --- a/src/host_storage.rs +++ b/src/host_storage.rs @@ -1,6 +1,5 @@ use std::collections::HashMap; use std::io; -use std::ops::Deref; use std::panic::{catch_unwind, AssertUnwindSafe}; use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak}; @@ -42,7 +41,6 @@ struct TransportState { struct PendingRequest { retained_bytes: usize, - completed: bool, response: Weak, } @@ -91,7 +89,7 @@ impl HostTransport { request: Vec, response_budget: usize, deadline: Option, - ) -> io::Result { + ) -> 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"); @@ -135,17 +133,7 @@ impl HostTransport { } if let Some(result) = response.wait(deadline) { - return match result { - Ok(bytes) => Ok(TransportResponse { - bytes, - transport: Arc::downgrade(self), - request_id, - }), - Err(error) => { - self.release(request_id); - Err(error) - } - }; + return result; } self.release(request_id); let _ = response.take(); @@ -217,7 +205,6 @@ impl HostTransport { request_id, PendingRequest { retained_bytes, - completed: false, response: Arc::downgrade(response), }, ); @@ -227,20 +214,30 @@ impl HostTransport { fn complete(&self, request_id: u64, result: io::Result>) { let response = { let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); - let Some(pending) = state.pending.get(&request_id) else { + let Some(pending) = state.pending.remove(&request_id) else { return; }; - if pending.completed { + if state.operations == 0 || state.bytes < pending.retained_bytes { + state.closed = Some("host storage transport accounting failed".to_owned()); + state.operations = 0; + state.bytes = 0; + let remaining = std::mem::take(&mut state.pending); + drop(state); + self.capacity.notify_all(); + let error = || io::Error::other("host storage transport accounting failed"); + if let Some(response) = pending.response.upgrade() { + response.complete(Err(error())); + } + for pending in remaining.into_values() { + if let Some(response) = pending.response.upgrade() { + response.complete(Err(error())); + } + } return; } let response = pending.response.upgrade(); - let reserved_bytes = pending.retained_bytes; - let response_bytes = result.as_ref().map_or(0, Vec::len); state.operations -= 1; - state.bytes = state.bytes - reserved_bytes + response_bytes; - let pending = state.pending.get_mut(&request_id).unwrap(); - pending.retained_bytes = response_bytes; - pending.completed = true; + state.bytes -= pending.retained_bytes; self.capacity.notify_all(); response }; @@ -252,9 +249,12 @@ impl HostTransport { fn release(&self, request_id: u64) { let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); if let Some(pending) = state.pending.remove(&request_id) { - if !pending.completed { - state.operations -= 1; + if state.operations == 0 || state.bytes < pending.retained_bytes { + drop(state); + self.fail(io::ErrorKind::Other, "host storage transport accounting failed"); + return; } + state.operations -= 1; state.bytes -= pending.retained_bytes; self.capacity.notify_all(); } @@ -279,37 +279,6 @@ impl HostTransport { } } -struct TransportResponse { - bytes: Vec, - transport: Weak, - request_id: u64, -} - -impl TransportResponse { - fn into_bytes(mut self) -> Vec { - std::mem::take(&mut self.bytes) - } -} - -impl Deref for TransportResponse { - type Target = [u8]; - - fn deref(&self) -> &Self::Target { - &self.bytes - } -} - -// Moving this wrapper never moves the Vec allocation referenced by Deref. -unsafe impl stable_deref_trait::StableDeref for TransportResponse {} - -impl Drop for TransportResponse { - fn drop(&mut self) { - if let Some(transport) = self.transport.upgrade() { - transport.release(self.request_id); - } - } -} - impl ResponseSlot { fn new() -> Self { Self { @@ -545,12 +514,12 @@ impl RequestEncoder { } struct ResponseDecoder { - bytes: TransportResponse, + bytes: Vec, offset: usize, } impl ResponseDecoder { - fn new(bytes: TransportResponse) -> Self { + fn new(bytes: Vec) -> Self { Self { bytes, offset: 0 } } @@ -731,9 +700,7 @@ pub fn test_host_round_trip( .spawn(move || { let result = test_thread_result(|| { let deadline = use_timeout.then(|| Instant::now() + transport.timeout); - transport - .round_trip(request, response_budget as usize, deadline) - .map(TransportResponse::into_bytes) + transport.round_trip(request, response_budget as usize, deadline) }); let _ = completion.call(result, ThreadsafeFunctionCallMode::NonBlocking); }) diff --git a/test/host-storage-transport.test.mjs b/test/host-storage-transport.test.mjs index 745ac79..6327e33 100644 --- a/test/host-storage-transport.test.mjs +++ b/test/host-storage-transport.test.mjs @@ -117,6 +117,14 @@ test('invalid callback responses fail the request without terminating the proces } }); +test('malformed host protocol responses fail the Tantivy operation', async (context) => { + for (const invalid of [Buffer.from([2, 0]), Buffer.from([1])]) { + const handle = addon.__testOpenHostTransport(() => invalid, 4, 64 * 1024 * 1024, 1_000); + context.after(() => addon.__testCloseHostTransport(handle)); + await assert.rejects(verifyTantivy(handle), /unsupported protocol version|truncated/); + } +}); + test('KvDirectory and Tantivy operate through the host storage transport', async (context) => { const entries = new Map(); const requests = []; @@ -174,6 +182,26 @@ test('host storage failures cross the native boundary without escaping JavaScrip await assert.rejects(verifyTantivy(handle), /injected host read failure/); }); +test('host mutation failures return a definitive result without timing out', async (context) => { + const storage = { + read() {}, + write() { + throw new Error('injected host write failure'); + }, + sync() {}, + }; + const handler = createHostStorageHandler(storage, { + maxMutations: 16, + maxReadResponseBytes: 1_024, + maxControlResponseBytes: 1_024, + maxErrorBytes: 1_024, + }); + const handle = addon.__testOpenHostTransport(handler, 4, 64 * 1024 * 1024, 5); + context.after(() => addon.__testCloseHostTransport(handle)); + + await assert.rejects(verifyTantivy(handle), /injected host write failure/); +}); + test('host storage handler preserves no-WAL policy and rejects malformed frames', () => { const policies = []; const handler = createHostStorageHandler( diff --git a/ts/host-storage.ts b/ts/host-storage.ts index c434073..6fa79cc 100644 --- a/ts/host-storage.ts +++ b/ts/host-storage.ts @@ -30,6 +30,7 @@ export interface HostStorageHandlerOptions { maxErrorBytes: number; } +/** Creates the exception-safe callback required by the native host transport. */ export function createHostStorageHandler( storage: HostStorage, { maxMutations, maxReadResponseBytes, maxControlResponseBytes, maxErrorBytes }: HostStorageHandlerOptions, From 007daaf9f82fd0d7d3f1cf00809a810327906e0a Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Tue, 8 Sep 2026 17:07:38 -0600 Subject: [PATCH 6/7] fix: fail closed on host completion panic --- src/host_storage.rs | 20 ++++++++++++-------- ts/host-storage.ts | 4 ++-- ts/load-addon.ts | 2 +- 3 files changed, 15 insertions(+), 11 deletions(-) diff --git a/src/host_storage.rs b/src/host_storage.rs index a723e3a..8136bf7 100644 --- a/src/host_storage.rs +++ b/src/host_storage.rs @@ -28,7 +28,7 @@ struct HostTransport { next_request_id: AtomicU64, max_operations: usize, max_bytes: usize, - timeout: Duration, + read_timeout: Duration, } #[derive(Default)] @@ -55,14 +55,15 @@ impl HostTransport { handler: JsFunction, max_operations: usize, max_bytes: usize, - timeout: Duration, + read_timeout: Duration, ) -> boundary::Result { - if max_operations == 0 || max_bytes == 0 || timeout.is_zero() { + if max_operations == 0 || max_bytes == 0 || read_timeout.is_zero() { return Err(napi::Error::new( "E_INVALID_ARGUMENT", "host transport limits must be greater than zero", )); } + // Production construction must supply the total callback created by createHostStorageHandler. let mut handler = handler .create_threadsafe_function::, Buffer, _, ErrorStrategy::Fatal>( max_operations, @@ -79,7 +80,7 @@ impl HostTransport { next_request_id: AtomicU64::new(1), max_operations, max_bytes, - timeout, + read_timeout, }) } @@ -117,6 +118,9 @@ impl HostTransport { } })); 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(()) @@ -388,7 +392,7 @@ impl HostKvStore { fn request(&self, request: Vec, response_budget: usize) -> io::Result { // Read deadlines cover admission and host execution; a timed-out read has no storage side effect. let deadline = Instant::now() - .checked_add(self.transport.timeout) + .checked_add(self.transport.read_timeout) .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "host storage timeout is too large"))?; self.request_with_deadline(request, response_budget, Some(deadline)) } @@ -642,7 +646,7 @@ pub fn test_open_host_transport( handler: JsFunction, max_operations: u32, max_bytes: u32, - timeout_ms: u32, + read_timeout_ms: u32, ) -> boundary::Result { boundary::run_stateless(|| { let handle = NEXT_TRANSPORT_HANDLE.fetch_add(1, Ordering::Relaxed); @@ -657,7 +661,7 @@ pub fn test_open_host_transport( handler, max_operations as usize, max_bytes as usize, - Duration::from_millis(timeout_ms as u64), + Duration::from_millis(read_timeout_ms as u64), )?); registry().insert(handle, transport.clone()); if let Err(error) = env.add_async_cleanup_hook( @@ -699,7 +703,7 @@ pub fn test_host_round_trip( .name(format!("fulltext-host-storage-test-{handle}")) .spawn(move || { let result = test_thread_result(|| { - let deadline = use_timeout.then(|| Instant::now() + transport.timeout); + let deadline = use_timeout.then(|| Instant::now() + transport.read_timeout); transport.round_trip(request, response_budget as usize, deadline) }); let _ = completion.call(result, ThreadsafeFunctionCallMode::NonBlocking); diff --git a/ts/host-storage.ts b/ts/host-storage.ts index 6fa79cc..fe42fd7 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 the batch atomically and satisfy the write policy before returning. Mutation buffers are borrowed. */ + /** 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; @@ -30,7 +30,7 @@ export interface HostStorageHandlerOptions { maxErrorBytes: number; } -/** Creates the exception-safe callback required by the native host transport. */ +/** Creates the total callback required by the native transport; storage exceptions become protocol errors. */ export function createHostStorageHandler( storage: HostStorage, { maxMutations, maxReadResponseBytes, maxControlResponseBytes, maxErrorBytes }: HostStorageHandlerOptions, diff --git a/ts/load-addon.ts b/ts/load-addon.ts index 220bc04..fdc4ef9 100644 --- a/ts/load-addon.ts +++ b/ts/load-addon.ts @@ -28,7 +28,7 @@ interface NativeAddonApi { handler: (request: Buffer) => Buffer, maxOperations: number, maxBytes: number, - timeoutMs: number, + readTimeoutMs: number, ): number; __testHostRoundTrip?( handle: number, From 6f324a3cafc1018f81cddcea6c6c275754c08578 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Tue, 8 Sep 2026 22:12:17 -0600 Subject: [PATCH 7/7] fix: separate host writes from durability barriers --- src/host_storage.rs | 12 ++++++--- test/host-storage-transport.test.mjs | 40 ++++++++++++++++++++++++++-- ts/host-storage.ts | 3 +-- 3 files changed, 48 insertions(+), 7 deletions(-) diff --git a/src/host_storage.rs b/src/host_storage.rs index 8136bf7..de5c388 100644 --- a/src/host_storage.rs +++ b/src/host_storage.rs @@ -458,9 +458,10 @@ impl KvStore for HostKvStore { fn write(&self, mutations: &[Mutation], policy: WritePolicy) -> io::Result<()> { let mut request = RequestEncoder::new(OP_WRITE); + let requires_sync = policy == WritePolicy::WAL_SYNC; + // Harper exposes an atomic WAL batch and a separate database durability barrier. request.u8(match policy { - WritePolicy::WAL => 1, - WritePolicy::WAL_SYNC => 2, + WritePolicy::WAL | WritePolicy::WAL_SYNC => 1, WritePolicy::NO_WAL => 3, _ => return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid write policy")), }); @@ -479,7 +480,12 @@ impl KvStore for HostKvStore { } } self.request_mutation(request.finish(), self.max_control_response_bytes)? - .finish() + .finish()?; + if requires_sync { + // A barrier failure leaves the write outcome known-applied; the caller poisons the generation. + self.sync()?; + } + Ok(()) } fn sync(&self) -> io::Result<()> { diff --git a/test/host-storage-transport.test.mjs b/test/host-storage-transport.test.mjs index 6327e33..ee965b1 100644 --- a/test/host-storage-transport.test.mjs +++ b/test/host-storage-transport.test.mjs @@ -157,11 +157,46 @@ test('KvDirectory and Tantivy operate through the host storage transport', async await verifyTantivy(handle); assert.ok(requests.includes('read'), 'read requests were issued'); assert.ok(requests.includes('write:wal'), 'ordinary objects use WAL writes'); - assert.ok(requests.includes('write:wal-sync'), 'metadata publication uses a synchronous WAL write'); - assert.ok(requests.includes('sync'), 'directory sync requests were issued'); + assert.ok(!requests.includes('write:wal-sync'), 'the host is not asked for an unsupported per-write sync primitive'); + assert.ok(requests.includes('sync'), 'metadata publication and directory sync use explicit durability barriers'); assert.ok(entries.size > 0, 'Tantivy state remains in host storage for reopen'); }); +test('a failed durability barrier does not pretend the preceding atomic write rolled back', async (context) => { + const entries = new Map(); + let writes = 0; + let syncs = 0; + const storage = { + read(key) { + return entries.get(key.toString('hex')); + }, + write(mutations, policy) { + assert.strictEqual(policy, 'wal'); + writes++; + 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() { + if (++syncs === 1) throw new Error('injected durability failure'); + }, + }; + const handler = createHostStorageHandler(storage, { + maxMutations: 1_024, + maxReadResponseBytes: readResponseBytes, + maxControlResponseBytes: controlResponseBytes, + maxErrorBytes: controlResponseBytes, + }); + const handle = addon.__testOpenHostTransport(handler, 32, 40 * 1024 * 1024, 5_000); + context.after(() => addon.__testCloseHostTransport(handle)); + + await assert.rejects(verifyTantivy(handle), /injected durability failure/); + assert.ok(writes > 0, 'the atomic WAL write completed before its durability barrier failed'); + assert.ok(entries.size > 0, 'the transport does not report that applied writes were rolled back'); +}); + test('host storage failures cross the native boundary without escaping JavaScript', async (context) => { const storage = { read() { @@ -222,6 +257,7 @@ test('host storage handler preserves no-WAL policy and rejects malformed frames' assert.deepStrictEqual(handler(Buffer.from([1, 2, 3, 1, 0, 0, 0, 2, 1, 0, 0, 0, 97])), Buffer.from([1, 0])); assert.deepStrictEqual(policies, ['no-wal']); + assert.match(decodeHandlerError(handler(Buffer.from([1, 2, 2, 0, 0, 0, 0]))), /unknown.*write policy/); assert.match(decodeHandlerError(handler(Buffer.from([2, 3]))), /unsupported.*protocol version/); assert.match(decodeHandlerError(handler(Buffer.from([1, 1, 4, 0, 0]))), /truncated/); }); diff --git a/ts/host-storage.ts b/ts/host-storage.ts index fe42fd7..1953eb1 100644 --- a/ts/host-storage.ts +++ b/ts/host-storage.ts @@ -11,7 +11,7 @@ const mutationDelete = 2; const minimumMutationBytes = 5; const fallbackError = Buffer.from([protocolVersion, responseError, 0, 0, 0, 0]); -export type HostWritePolicy = 'wal' | 'wal-sync' | 'no-wal'; +export type HostWritePolicy = 'wal' | 'no-wal'; export type HostStorageMutation = { type: 'put'; key: Buffer; value: Buffer } | { type: 'delete'; key: Buffer }; @@ -107,7 +107,6 @@ export function createHostStorageHandler( function decodeWritePolicy(encoded: number): HostWritePolicy { if (encoded === 1) return 'wal'; - if (encoded === 2) return 'wal-sync'; if (encoded === 3) return 'no-wal'; throw new Error(`unknown host storage write policy ${encoded}`); }