diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md index 714193c..0f89204 100644 --- a/docs/reclamation-plan.md +++ b/docs/reclamation-plan.md @@ -281,13 +281,61 @@ stale empty hint. A namespace proven absent after bounded format validation retu the default read budget covers format validation plus one cold 64-shard sweep for a present namespace. -Cleanup runs on a dedicated native task, not the JavaScript service thread or writer actor. Host -callbacks still execute on JavaScript, so cleanup has a low-priority admission class that cannot -take the last foreground transport slot. Each admission bounds point reads, delete mutations, -request bytes, and elapsed time checked between storage operations. One admitted synchronous host -operation cannot be canceled and may exceed the elapsed budget. Panics are caught at the task -boundary; terminal failure, queue depth, pinned skips, and no-progress state are observable by -Harper instead of silently disabling reclamation. +Cleanup runs on one dedicated native task per index, not the JavaScript service thread or writer +actor. The task receives a low-priority clone of the foreground `HostKvStore`; the clone changes +only an internal admission-class field and retains the exact `KvStoreIdentity`, transport, and +namespace. Constructing a second identity for cleanup would split `DirectoryState`, hide foreground +reader pins from the reclaimer, and is rejected by an end-to-end retained-reader test. Priority is +not added to `KvStore`, the host protocol, or JavaScript storage operations. + +The production topology gives each open index generation one `HostTransport`; multiple indexes own +independent transports and native tasks. The transport enforces separate operation and byte ceilings +for low-priority work. Cleanup may have exactly one operation in flight, leaving +`max_operations - 1` slots for the index's configured foreground concurrency, and may use only the +byte ceiling left after its configured foreground reserve. The owner derives that reserve from the +fixed chunk size, bounded directory-key encoding, configured read/control response limits, and the +index's worker limits. Cleanup admission is enabled only when the low-priority view is requested, at +which point validation rejects a transport that cannot hold its foreground reserve, the configured +bounded read and mutation request sizes, and the store's read/control response reservations; small +transports used without cleanup retain their current behavior. + +Cleanup-directory construction derives the read-request bound from its namespace and the format key +high-water, so a store view cannot be paired with a different namespace after sizing. The +low-priority store retains both request bounds and rejects an encoded cleanup request that exceeds +either one before transport admission; these are enforced limits rather than sizing hints. + +A statically valid cleanup request that does not fit current occupancy returns `WouldBlock` without +joining the condition-variable wait queue. Permanent configuration and request-size failures are +classified before occupancy so they cannot masquerade as healthy deferral. Once dispatched, a +mutation still waits for a definitive host result because canceling an +unknown write outcome would violate queue progress atomicity. `WouldBlock` from this explicit +admission path is reported as deferred work; a threadsafe-function queue-full result is an accounting +failure and closes the transport rather than masquerading as deferral. `BrokenPipe` after close is +reported as shutdown, while malformed persisted data remains terminal generation corruption. + +Host callbacks still execute FIFO on JavaScript, so capacity headroom guarantees foreground +admission but not foreground latency. Cleanup therefore has one in-flight host operation, and the +task constructs each `ReclaimBudget` with per-admission mutation and encoded-request caps below the +directory format's hard 512-mutation and 64 KiB limits. The directory builds batches against that +budget; the transport does not reject a batch after it has been built. The production budget is +selected from the JavaScript-handler and Harper RocksDB batch-latency benchmark before cleanup is +enabled, then remains internal runtime policy rather than customer schema. On fixed hardware, three +alternating runs must show no more than 5% foreground directory-operation p99 regression with +sustained cleanup, and the catalog search benchmark must remain below the 50 ms p99 objective. If no +measured budget meets both gates, cleanup moves to a separate host callback channel before release. +Harper may additionally trigger admissions during idle periods, but higher-layer timing is an +optimization rather than the correctness boundary. + +Each cleanup admission also bounds total point reads, delete mutations, request bytes, and elapsed +time checked between storage operations. One admitted synchronous host operation cannot be canceled +and may finish after the elapsed target. A foreground read that times out stops waiting but its +operation and byte reservation remain charged until the already-dispatched callback completes or +the transport closes; the environment cleanup hook closes the transport before its callback is +dropped. Admission timeout and post-dispatch response timeout have distinct messages, and abandoned +waiter count plus still-charged operations and bytes are included in transport health. Cleanup cannot +consume capacity that is still live but no longer has a waiting caller. Panics are caught at the task +boundary. Terminal failure, queue depth, pinned skips, transport deferrals, no-progress state, and +the last terminal error are observable by Harper instead of silently disabling reclamation. Binding, queue-entry, and progress decoding validates versions, lengths, numeric ranges, and a configured maximum total extent before allocating or scheduling work. An undecodable binding is not @@ -319,6 +367,13 @@ generation from source data instead of masking the broken atomicity invariant. | Object-id allocation | Updating the counter with every binding creation serializes file creation across a host round trip. The allocator reserves fixed durable strides, reducing that shared operation to one per stride; unused ids after failure or restart are harmless. | | Queue-tail reads | Cache the next sequence and reread only after errors, or point-read it before every enqueue. The durable point read is chosen because slice 3 enqueues only file deletion rather than indexed documents, removes cache recovery state, and is measured explicitly. | | Cleanup priority | Cleanup leaves pinned entries durable and skips them with a process-local sequence cursor. Later entries may complete out of order, while the durable head advances only across observed holes. This lets the consumer advance without a second persistent queue or holding a foreground enqueue mutex or counter across low-priority host I/O. | +| Transport admission | A priority field on a same-identity `HostKvStore` clone is chosen over a `KvStore` method or thread-local state. The shared transport owns both operation and byte accounting, low-priority work fails fast rather than waiting, and one foreground operation plus its maximum byte reservation remains available. This keeps the storage protocol frozen and prevents a cleanup-only store from splitting reader-pin state. | +| Separate callback queue | A second cleanup `ThreadsafeFunction` would isolate native admission queues but still executes on the same JavaScript event loop and adds another host lifecycle contract. It is retained as the fallback only if measured per-call caps on the shared queue cannot meet foreground p99. | +| Higher-layer scheduling | Harper may request cleanup when its own load is low, but it cannot observe all native and host callback arrivals and the standalone library cannot depend on Harper's scheduler. Idle-time triggering may reduce contention but cannot replace transport-enforced headroom. | +| Accept and detect | Fair admission followed by disabling cleanup after latency rises reacts only after foreground work has already queued behind an uncancellable mutation. Metrics remain required, but detection alone is rejected as the protection mechanism. | +| Cleanup work unit | One in-flight cleanup operation plus measured mutation and encoded-request caps is chosen. The existing 512-mutation and 64 KiB values remain format safety ceilings, not latency targets; production defaults are fixed only after benchmarking the JavaScript handler and Harper RocksDB batch together. | +| External RocksDB drain | Harper could interpret and drain the FIFO without native transport round trips, but Harper cannot see the process-local reader pins that protect open Tantivy slices. An external drainer could therefore delete bytes beneath a live handle and is rejected. | +| Foreground piggyback | Spending a small reclamation budget after each foreground publication would rate cleanup with garbage creation and avoid a second admission class. It puts uncancellable cleanup writes directly on the writer actor's latency path and cannot drain an idle backlog, so it is rejected. | | Chosen | Binding high-waters plus one transition-fed durable FIFO per shard, sequence-keyed progress, object-id pins, and bounded low-priority draining. | ## Persistence format and delivery sequence @@ -339,9 +394,12 @@ Then deliver reclamation in reviewable slices: 4. add object-id/revision reader pins and coordinate registration with deletion through the hash-sharded registration fence; 5. add bounded sparse FIFO draining and its synchronous admission API; -6. add low-priority scheduling, failure observability, close fencing, crash recovery, and Harper's - exclusive-owner integration. Measure tail accumulation and add tail-supersession entries only if - that data justifies publication-path cost. +6. first add priority-aware operation and byte admission to the existing host transport, including + same-identity store views, fail-fast cleanup admission, live timeout accounting, and the + JavaScript/RocksDB batch-latency measurement that fixes the per-call caps; then add the dedicated + cleanup task, failure observability, and close fencing; finally wire Harper's exclusive-owner and + crash-recovery lifecycle. Measure tail accumulation and add tail-supersession entries only if that + data justifies publication-path cost. The production Harper package remains disabled until the cleanup lifecycle fence and Harper host- storage integration both pass. No native RocksDB storage provider is added to the public library. @@ -415,11 +473,27 @@ Efficacy is measured as well as safety: after repeated real Tantivy merge/delete key count and payload bytes must return to a bound proportional to the live index rather than bytes ever written. `CountingKv` asserts per-entry point-read and mutation cost does not grow with object ids ever allocated, plus request-byte and time-admission bounds. Measure real Tantivy tail revisions -per file before retaining the tail-only path. Add a hot-path regression test showing a foreground -read still gains transport capacity while cleanup waits, plus open/drop churn that keeps weak-pin -memory bounded. The same reclamation harness runs through host transport; Harper separately verifies -owner loss with a second worker, close drain, cleanup health reporting, backup/restore, and -derived-index replay coordination. +per file before retaining the tail-only path. The transport suite fills cleanup-eligible operation +slots and bytes independently and proves that one maximum foreground reservation still admits. It +also proves that an unsatisfiable cleanup reservation fails immediately, a timed-out foreground +callback remains charged until completion, close wakes an admission waiter, and worker teardown +removes a transport while foreground and cleanup reservations are charged. Teardown of a dispatched +low-priority callback moves with the dedicated cleanup task that creates that callback. + +The same reclamation harness runs through foreground and low-priority views of one host transport. +It verifies identical `KvStoreIdentity`, retained-reader protection across those views, and physical +removal of both pinned-then-released and immediately reclaimable object payloads. Durable resume +after interruption is added with the dedicated cleanup task. A latency benchmark measures +foreground reads while cleanup issues one host operation at each candidate mutation/request cap; +capacity-only assertions do not qualify the shared callback queue. Open/drop churn separately keeps +weak-pin memory bounded. Harper verifies owner loss with a second worker, close drain, cleanup health +reporting, backup/restore, and derived-index replay coordination. + +Priority-selection hooks remain test-only and absent from the packed release artifact. The existing +single-slot timeout test is rewritten for conservative accounting: a second request cannot reuse the +timed-out request's charge until the callback finishes, and admission versus response timeout text is +asserted separately. Environment teardown proves the async cleanup hook closes and removes a +transport while both admission classes are charged. The FIFO tests include parallel enqueue on different shards, sequence/batch failure, pinned-entry skip cost, out-of-order completion and bounded head compaction, corrupt binding and entry handling, diff --git a/src/host_storage.rs b/src/host_storage.rs index cdd8e9b..c66d995 100644 --- a/src/host_storage.rs +++ b/src/host_storage.rs @@ -1,6 +1,8 @@ use std::collections::HashMap; use std::io; +use std::io::Write as IoWrite; use std::panic::{catch_unwind, AssertUnwindSafe}; +use std::path::Path; use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak}; use std::thread; @@ -10,10 +12,13 @@ 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::OwnedBytes; +use tantivy::directory::{Directory, OwnedBytes, TerminatingWrite}; use crate::boundary; -use crate::phase0::{KvDirectory, KvStore, KvStoreIdentity, Mutation, WritePolicy, CHUNK_SIZE}; +use crate::phase0::{ + reclaim_read_key_bytes, KvDirectory, KvStore, KvStoreIdentity, Mutation, ReclaimBudget, WritePolicy, CHUNK_SIZE, + RECLAIM_MAX_BATCH_REQUEST_BYTES, +}; type HostCallback = ThreadsafeFunction, ErrorStrategy::Fatal>; type CompletionCallback = ThreadsafeFunction, ErrorStrategy::Fatal>; @@ -26,6 +31,7 @@ struct HostTransport { state: Mutex, capacity: Condvar, next_request_id: AtomicU64, + abandoned_waiters: AtomicU64, max_operations: usize, max_bytes: usize, read_timeout: Duration, @@ -36,14 +42,30 @@ struct TransportState { closed: Option, operations: usize, bytes: usize, + cleanup_operations: usize, + cleanup_bytes: usize, + cleanup_capacity: Option, pending: HashMap, } struct PendingRequest { retained_bytes: usize, + class: AdmissionClass, response: Weak, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum AdmissionClass { + Foreground, + Cleanup, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct CleanupCapacity { + foreground_reserved_bytes: usize, + max_bytes: usize, +} + struct ResponseSlot { result: Mutex>>>, ready: Condvar, @@ -78,6 +100,7 @@ impl HostTransport { state: Mutex::new(TransportState::default()), capacity: Condvar::new(), next_request_id: AtomicU64::new(1), + abandoned_waiters: AtomicU64::new(0), max_operations, max_bytes, read_timeout, @@ -90,6 +113,7 @@ impl HostTransport { request: Vec, response_budget: usize, deadline: Option, + class: AdmissionClass, ) -> io::Result> { let request_id = self.next_request_id.fetch_add(1, Ordering::Relaxed); if request_id == 0 { @@ -97,7 +121,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, deadline)?; + self.admit(request_id, request.len(), response_budget, &response, deadline, class)?; let transport = Arc::downgrade(self); let callback_response = response.clone(); @@ -127,23 +151,23 @@ impl HostTransport { }, ); if status != Status::Ok { - self.complete( - request_id, - Err(io::Error::new( - io::ErrorKind::WouldBlock, - format!("host storage callback rejected request: {status:?}"), - )), - ); + if status == Status::Closing { + self.fail(io::ErrorKind::BrokenPipe, "host storage transport is closed"); + } else { + self.fail( + io::ErrorKind::Other, + &format!("host storage callback rejected an admitted request: {status:?}"), + ); + } } if let Some(result) = response.wait(deadline) { return result; } - self.release(request_id); - let _ = response.take(); + self.abandoned_waiters.fetch_add(1, Ordering::Relaxed); Err(io::Error::new( io::ErrorKind::TimedOut, - "host storage request timed out", + "host storage response timed out", )) } @@ -158,6 +182,7 @@ impl HostTransport { response_bytes: usize, response: &Arc, deadline: Option, + class: AdmissionClass, ) -> io::Result<()> { let retained_bytes = request_bytes .checked_add(response_bytes) @@ -173,16 +198,22 @@ impl HostTransport { 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 { + if self.has_capacity(&state, retained_bytes, class)? { break; } + if class == AdmissionClass::Cleanup { + return Err(io::Error::new( + io::ErrorKind::WouldBlock, + "low-priority host storage capacity is unavailable", + )); + } 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", + "host storage admission timed out", )); } let (state, wait) = self @@ -192,7 +223,7 @@ impl HostTransport { if wait.timed_out() { return Err(io::Error::new( io::ErrorKind::TimedOut, - "host storage request timed out", + "host storage admission timed out", )); } state @@ -205,26 +236,102 @@ impl HostTransport { } state.operations += 1; state.bytes += retained_bytes; + if class == AdmissionClass::Cleanup { + state.cleanup_operations += 1; + state.cleanup_bytes += retained_bytes; + } state.pending.insert( request_id, PendingRequest { retained_bytes, + class, response: Arc::downgrade(response), }, ); Ok(()) } + 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(|| { + io::Error::new( + io::ErrorKind::InvalidInput, + "low-priority host storage admission is not configured", + ) + })?; + if retained_bytes > capacity.max_bytes { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "low-priority host storage reservation exceeds its byte limit", + )); + } + Some(capacity) + } else { + None + }; + if state.operations >= self.max_operations || state.bytes.saturating_add(retained_bytes) > self.max_bytes { + return Ok(false); + } + let Some(capacity) = cleanup_capacity else { + return Ok(true); + }; + Ok(state.cleanup_operations == 0 + && state.cleanup_bytes.saturating_add(retained_bytes) <= capacity.max_bytes + && state + .bytes + .saturating_add(retained_bytes) + .saturating_add(capacity.foreground_reserved_bytes) + <= self.max_bytes) + } + + 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( + io::ErrorKind::InvalidInput, + "low-priority host storage requires two operation slots and positive byte limits", + )); + } + if foreground_reserved_bytes + .checked_add(max_cleanup_bytes) + .is_none_or(|required| required > self.max_bytes) + { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "host storage byte capacity cannot hold foreground headroom and one cleanup request", + )); + } + let capacity = CleanupCapacity { + foreground_reserved_bytes, + max_bytes: max_cleanup_bytes, + }; + let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + match state.cleanup_capacity { + Some(existing) if existing != capacity => Err(io::Error::new( + io::ErrorKind::InvalidInput, + "host storage cleanup capacity is already configured differently", + )), + Some(_) => Ok(()), + None => { + state.cleanup_capacity = Some(capacity); + 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.remove(&request_id) else { return; }; - if state.operations == 0 || state.bytes < pending.retained_bytes { + let cleanup_underflow = pending.class == AdmissionClass::Cleanup + && (state.cleanup_operations == 0 || state.cleanup_bytes < pending.retained_bytes); + if state.operations == 0 || state.bytes < pending.retained_bytes || cleanup_underflow { state.closed = Some("host storage transport accounting failed".to_owned()); state.operations = 0; state.bytes = 0; + state.cleanup_operations = 0; + state.cleanup_bytes = 0; let remaining = std::mem::take(&mut state.pending); drop(state); self.capacity.notify_all(); @@ -242,6 +349,10 @@ impl HostTransport { let response = pending.response.upgrade(); state.operations -= 1; state.bytes -= pending.retained_bytes; + if pending.class == AdmissionClass::Cleanup { + state.cleanup_operations -= 1; + state.cleanup_bytes -= pending.retained_bytes; + } self.capacity.notify_all(); response }; @@ -250,20 +361,6 @@ 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 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(); - } - } - fn fail(&self, kind: io::ErrorKind, message: &str) { let pending = { let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); @@ -272,6 +369,8 @@ impl HostTransport { } state.operations = 0; state.bytes = 0; + state.cleanup_operations = 0; + state.cleanup_bytes = 0; std::mem::take(&mut state.pending) }; self.capacity.notify_all(); @@ -326,13 +425,6 @@ impl ResponseSlot { } 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> { @@ -365,14 +457,50 @@ const VALUE_MISSING: u8 = 0; const VALUE_PRESENT: u8 = 1; const MUTATION_PUT: u8 = 1; const MUTATION_DELETE: u8 = 2; +const HOST_PROTOCOL_HEADER_BYTES: usize = 2; +const HOST_LENGTH_PREFIX_BYTES: usize = 4; const READ_RESPONSE_OVERHEAD: usize = 7; +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")) +} + +fn minimum_cleanup_reservation( + max_read_response_bytes: usize, + max_control_response_bytes: usize, + max_read_request_bytes: usize, + max_mutation_request_bytes: usize, +) -> io::Result { + let read = max_read_response_bytes + .checked_add(max_read_request_bytes) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "cleanup read reservation overflow"))?; + let write = max_control_response_bytes + .checked_add(max_mutation_request_bytes.min(RECLAIM_MAX_BATCH_REQUEST_BYTES)) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "cleanup write reservation overflow"))?; + Ok(read.max(write)) +} + +fn validate_cleanup_request(request_bytes: usize, limit: Option, operation: &str) -> io::Result<()> { + if limit.is_some_and(|limit| request_bytes > limit) { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("low-priority host storage {operation} request exceeds its byte limit"), + )); + } + Ok(()) +} + #[derive(Clone)] struct HostKvStore { transport: Arc, identity: KvStoreIdentity, max_read_response_bytes: usize, max_control_response_bytes: usize, + max_cleanup_read_request_bytes: Option, + max_cleanup_mutation_request_bytes: Option, + class: AdmissionClass, } impl HostKvStore { @@ -394,10 +522,61 @@ impl HostKvStore { identity, max_read_response_bytes, max_control_response_bytes, + max_cleanup_read_request_bytes: None, + max_cleanup_mutation_request_bytes: None, + class: AdmissionClass::Foreground, }) } + fn minimum_cleanup_bytes(&self, namespace: &[u8], budget: ReclaimBudget) -> io::Result { + minimum_cleanup_reservation( + self.max_read_response_bytes, + self.max_control_response_bytes, + host_read_request_bytes(reclaim_read_key_bytes(namespace))?, + budget.max_request_bytes, + ) + } + + fn cleanup_directory( + &self, + namespace: &[u8], + foreground_reserved_bytes: usize, + max_cleanup_bytes: usize, + budget: ReclaimBudget, + ) -> io::Result> { + let max_read_request_bytes = host_read_request_bytes(reclaim_read_key_bytes(namespace))?; + let max_mutation_request_bytes = budget.max_request_bytes.min(RECLAIM_MAX_BATCH_REQUEST_BYTES); + if max_read_request_bytes == 0 || max_mutation_request_bytes == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "low-priority host storage request limits must be positive", + )); + } + let minimum_reservation = minimum_cleanup_reservation( + self.max_read_response_bytes, + self.max_control_response_bytes, + max_read_request_bytes, + max_mutation_request_bytes, + )?; + if max_cleanup_bytes < minimum_reservation { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("low-priority host storage byte limit must be at least {minimum_reservation} bytes"), + )); + } + self.transport + .configure_cleanup(foreground_reserved_bytes, max_cleanup_bytes)?; + let store = Self { + class: AdmissionClass::Cleanup, + max_cleanup_read_request_bytes: Some(max_read_request_bytes), + max_cleanup_mutation_request_bytes: Some(max_mutation_request_bytes), + ..self.clone() + }; + Ok(KvDirectory::with_namespace(store, namespace)) + } + fn request(&self, request: Vec, response_budget: usize) -> io::Result { + validate_cleanup_request(request.len(), self.max_cleanup_read_request_bytes, "read")?; // Read deadlines cover admission and host execution; a timed-out read has no storage side effect. let deadline = Instant::now() .checked_add(self.transport.read_timeout) @@ -406,6 +585,7 @@ impl HostKvStore { } fn request_mutation(&self, request: Vec, response_budget: usize) -> io::Result { + validate_cleanup_request(request.len(), self.max_cleanup_mutation_request_bytes, "mutation")?; // A dispatched JavaScript mutation cannot be canceled, so wait for its definitive result. self.request_with_deadline(request, response_budget, None) } @@ -416,7 +596,9 @@ impl HostKvStore { response_budget: usize, deadline: Option, ) -> io::Result { - let response = self.transport.round_trip(request, response_budget, deadline)?; + let response = self + .transport + .round_trip(request, response_budget, deadline, self.class)?; let mut decoder = ResponseDecoder::new(response); if decoder.u8()? != HOST_PROTOCOL_VERSION { return Err(io::Error::new( @@ -704,6 +886,7 @@ pub fn test_host_round_trip( request: Buffer, response_budget: u32, use_timeout: bool, + low_priority: bool, callback: JsFunction, ) -> boundary::Result<()> { boundary::run_stateless(|| { @@ -718,7 +901,16 @@ pub fn test_host_round_trip( .spawn(move || { let result = test_thread_result(|| { let deadline = use_timeout.then(|| Instant::now() + transport.read_timeout); - transport.round_trip(request, response_budget as usize, deadline) + transport.round_trip( + request, + response_budget as usize, + deadline, + if low_priority { + AdmissionClass::Cleanup + } else { + AdmissionClass::Foreground + }, + ) }); let _ = completion.call(result, ThreadsafeFunctionCallMode::NonBlocking); }) @@ -727,6 +919,105 @@ pub fn test_host_round_trip( })? } +#[cfg(feature = "test-panic")] +#[napi(catch_unwind, skip_typescript, js_name = "__testConfigureHostTransportCleanup")] +pub fn test_configure_host_transport_cleanup( + handle: u32, + foreground_reserved_bytes: u32, + max_cleanup_bytes: u32, +) -> boundary::Result<()> { + boundary::run_stateless(|| { + registry() + .get(&handle) + .cloned() + .ok_or_else(|| napi::Error::new("E_CLOSED", "unknown or closed host storage transport"))? + .configure_cleanup(foreground_reserved_bytes as usize, max_cleanup_bytes as usize) + .map_err(|error| napi::Error::new("E_INVALID_ARGUMENT", error.to_string())) + })? +} + +#[cfg(feature = "test-panic")] +#[napi(catch_unwind, skip_typescript, js_name = "__testHostTransportStats")] +pub fn test_host_transport_stats(handle: u32) -> 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 state = transport.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + Ok(vec![ + state.operations.to_string(), + state.bytes.to_string(), + state.cleanup_operations.to_string(), + state.cleanup_bytes.to_string(), + transport.abandoned_waiters.load(Ordering::Relaxed).to_string(), + ]) + })? +} + +#[cfg(feature = "test-panic")] +#[napi(catch_unwind, skip_typescript, js_name = "__testHoldHostTransportCapacity")] +pub fn test_hold_host_transport_capacity( + handle: u32, + request_bytes: u32, + response_bytes: u32, + low_priority: bool, +) -> 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 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 response = Arc::new(ResponseSlot::new()); + transport + .admit( + request_id, + request_bytes as usize, + response_bytes as usize, + &response, + Some(Instant::now()), + if low_priority { + AdmissionClass::Cleanup + } else { + AdmissionClass::Foreground + }, + ) + .map_err(|error| napi::Error::new("E_NATIVE_FAILURE", error.to_string()))?; + Ok(request_id.to_string()) + })? +} + +#[cfg(feature = "test-panic")] +#[napi(catch_unwind, skip_typescript, js_name = "__testReleaseHostTransportCapacity")] +pub fn test_release_host_transport_capacity(handle: u32, request_id: String) -> 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 request_id = request_id + .parse() + .map_err(|_| napi::Error::new("E_INVALID_ARGUMENT", "invalid host storage request id"))?; + let exists = transport + .state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .pending + .contains_key(&request_id); + if exists { + transport.complete(request_id, Ok(Vec::new())); + } + Ok(exists) + })? +} + #[cfg(feature = "test-panic")] #[napi(catch_unwind, skip_typescript, js_name = "__testVerifyTantivyOnHostTransport")] pub fn test_verify_tantivy_on_host_transport( @@ -781,6 +1072,106 @@ pub fn test_verify_tantivy_on_host_transport( })? } +#[cfg(feature = "test-panic")] +#[napi(catch_unwind, skip_typescript, js_name = "__testReclaimOnHostTransport")] +pub fn test_reclaim_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) + .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-reclaim-test-{handle}")) + .spawn(move || { + 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 foreground_reserved_bytes = (max_read_response_bytes as usize) + .checked_add(max_control_response_bytes as usize) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "foreground reserve overflow"))?; + let budget = ReclaimBudget { + max_point_reads: 1_024, + max_mutations: 4_096, + max_request_bytes: 1024 * 1024, + max_elapsed: Duration::from_secs(5), + }; + let run = NEXT_TRANSPORT_HANDLE.fetch_add(1, Ordering::Relaxed); + let namespace = format!("host-reclaim/{run}"); + let max_cleanup_bytes = store.minimum_cleanup_bytes(namespace.as_bytes(), budget)?; + if store + .cleanup_directory( + namespace.as_bytes(), + foreground_reserved_bytes, + max_cleanup_bytes - 1, + budget, + ) + .is_ok() + { + return Err(io::Error::other("undersized cleanup storage view was accepted")); + } + let cleanup = store.cleanup_directory( + namespace.as_bytes(), + foreground_reserved_bytes, + max_cleanup_bytes, + budget, + )?; + let foreground = KvDirectory::with_namespace(store, namespace.as_bytes()); + let payload = vec![7_u8; CHUNK_SIZE + 17]; + for path in ["pinned", "unpinned"] { + let path = Path::new(path); + let mut writer = foreground + .open_write(path) + .map_err(|error| io::Error::other(error.to_string()))?; + writer.write_all(&payload)?; + writer.terminate()?; + } + let pinned = foreground + .open_read(Path::new("pinned")) + .map_err(|error| io::Error::other(error.to_string()))?; + for path in ["pinned", "unpinned"] { + foreground + .delete(Path::new(path)) + .map_err(|error| io::Error::other(error.to_string()))?; + } + let first = cleanup.reclaim(budget)?; + if first.entries_reclaimed == 0 || first.pinned_skips == 0 { + return Err(io::Error::other( + "host reclamation did not reclaim an unpinned object while preserving a pinned object", + )); + } + drop(pinned); + let second = cleanup.reclaim(budget)?; + if second.entries_reclaimed == 0 || second.has_more { + return Err(io::Error::other( + "host reclamation did not drain after the retained reader closed", + )); + } + Ok(format!( + "{},{},{},{}", + first.entries_reclaimed + second.entries_reclaimed, + first.pinned_skips, + first.payload_delete_mutations + second.payload_delete_mutations, + payload.len() * 2 + ) + .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 = "__testCloseHostTransport")] pub fn test_close_host_transport(handle: u32) -> boundary::Result { @@ -798,7 +1189,7 @@ mod tests { use super::*; #[test] - fn closing_wakes_every_pending_request() { + fn response_slot_returns_a_close_error() { let response = Arc::new(ResponseSlot::new()); response.complete(Err(io::Error::new(io::ErrorKind::BrokenPipe, "closed"))); assert_eq!( @@ -814,4 +1205,27 @@ mod tests { response.complete(Ok(vec![2])); assert_eq!(response.wait(Some(Instant::now())).unwrap().unwrap(), vec![1]); } + + #[test] + fn cleanup_reservation_covers_reads_and_bounded_writes() { + assert_eq!(minimum_cleanup_reservation(100, 10, 6, 20).unwrap(), 106); + assert_eq!(minimum_cleanup_reservation(10, 100, 6, 20).unwrap(), 120); + assert_eq!( + minimum_cleanup_reservation(10, 100, 6, usize::MAX).unwrap(), + 100 + RECLAIM_MAX_BATCH_REQUEST_BYTES + ); + assert_eq!(reclaim_read_key_bytes(b""), 30); + assert_eq!(reclaim_read_key_bytes(b"index"), 35); + assert_eq!(host_read_request_bytes(reclaim_read_key_bytes(b"")).unwrap(), 36); + assert_eq!(host_read_request_bytes(reclaim_read_key_bytes(b"index")).unwrap(), 41); + assert!(minimum_cleanup_reservation(usize::MAX, 10, 6, 20).is_err()); + assert!(minimum_cleanup_reservation(10, usize::MAX, 6, 20).is_err()); + } + + #[test] + fn cleanup_request_limits_are_enforced() { + assert!(validate_cleanup_request(6, Some(6), "read").is_ok()); + assert!(validate_cleanup_request(7, Some(6), "read").is_err()); + assert!(validate_cleanup_request(usize::MAX, None, "read").is_ok()); + } } diff --git a/src/phase0.rs b/src/phase0.rs index e64cc21..a17fee9 100644 --- a/src/phase0.rs +++ b/src/phase0.rs @@ -93,6 +93,11 @@ pub trait KvStore: Clone + Send + Sync + 'static { pub(crate) const CHUNK_SIZE: usize = 256 * 1024; +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() +} + impl Mutation { fn key(&self) -> &[u8] { match self { @@ -2157,7 +2162,7 @@ const RECLAIM_ENTRY_WHOLE_OBJECT: u8 = 1; const RECLAIM_SHARD_COUNT: usize = 64; const _: () = assert!(RECLAIM_SHARD_COUNT <= u64::BITS as usize); const RECLAIM_MAX_BATCH_MUTATIONS: usize = 512; -const RECLAIM_MAX_BATCH_REQUEST_BYTES: usize = 64 * 1024; +pub(crate) const RECLAIM_MAX_BATCH_REQUEST_BYTES: usize = 64 * 1024; const READER_REGISTRATION_SHARD_COUNT: usize = 256; const KEY_KIND_COUNTER: u8 = 1; const KEY_KIND_BINDING: u8 = 2; diff --git a/test/fixtures/host-storage-transport-worker.mjs b/test/fixtures/host-storage-transport-worker.mjs new file mode 100644 index 0000000..fe1b084 --- /dev/null +++ b/test/fixtures/host-storage-transport-worker.mjs @@ -0,0 +1,11 @@ +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); +addon.__testConfigureHostTransportCleanup(handle, 256, 512); +addon.__testHoldHostTransportCapacity(handle, 8, 128, true); +addon.__testHoldHostTransportCapacity(handle, 10, 128, false); +parentPort.postMessage({ handle, stats: addon.__testHostTransportStats(handle) }); +setInterval(() => {}, 1_000); diff --git a/test/host-storage-transport.test.mjs b/test/host-storage-transport.test.mjs index 8cfda23..a9242a6 100644 --- a/test/host-storage-transport.test.mjs +++ b/test/host-storage-transport.test.mjs @@ -1,5 +1,6 @@ import assert from 'node:assert'; import test from 'node:test'; +import { Worker } from 'node:worker_threads'; import { createHostStorageHandler } from '../dist/host-storage.js'; import { loadAddon } from '../dist/load-addon.js'; @@ -92,7 +93,7 @@ test('a host request can wait for a definitive result without a deadline', async assert.deepStrictEqual(await roundTrip(handle, Buffer.from('slow'), 128, false), Buffer.from('slow')); }); -test('a read timeout fences only that request', async (context) => { +test('a timed-out callback stays charged until its response completes', async (context) => { let calls = 0; const handle = addon.__testOpenHostTransport( (request) => { @@ -105,10 +106,88 @@ test('a read timeout fences only that request', async (context) => { ); context.after(() => addon.__testCloseHostTransport(handle)); - await assert.rejects(roundTrip(handle, Buffer.from('slow')), /timed out/); + const first = roundTrip(handle, Buffer.from('slow')); + const second = roundTrip(handle, Buffer.from('blocked')); + const results = await Promise.allSettled([first, second]); + assert.deepStrictEqual( + results.map((result) => result.status), + ['rejected', 'rejected'], + ); + assert.deepStrictEqual(results.map((result) => result.reason.message).sort(), [ + 'host storage admission timed out', + 'host storage response timed out', + ]); + assert.deepStrictEqual(addon.__testHostTransportStats(handle), ['0', '0', '0', '0', '1']); assert.deepStrictEqual(await roundTrip(handle, Buffer.from('recovered')), Buffer.from('recovered')); }); +test('cleanup uses one operation while preserving foreground admission', (context) => { + const handle = addon.__testOpenHostTransport((request) => request, 2, 1_024, 1_000); + context.after(() => addon.__testCloseHostTransport(handle)); + addon.__testConfigureHostTransportCleanup(handle, 256, 512); + + const cleanup = addon.__testHoldHostTransportCapacity(handle, 8, 128, true); + assert.throws( + () => addon.__testHoldHostTransportCapacity(handle, 8, 128, true), + /low-priority.*capacity is unavailable/, + ); + const foreground = addon.__testHoldHostTransportCapacity(handle, 10, 128, false); + assert.deepStrictEqual(addon.__testHostTransportStats(handle), ['2', '274', '1', '136', '0']); + assert.strictEqual(addon.__testReleaseHostTransportCapacity(handle, cleanup), true); + assert.strictEqual(addon.__testReleaseHostTransportCapacity(handle, foreground), true); +}); + +test('cleanup preserves foreground byte headroom and rejects impossible reservations', (context) => { + const handle = addon.__testOpenHostTransport((request) => request, 3, 300, 1_000); + context.after(() => addon.__testCloseHostTransport(handle)); + addon.__testConfigureHostTransportCleanup(handle, 128, 172); + + const foreground = addon.__testHoldHostTransportCapacity(handle, 20, 80, false); + assert.throws(() => addon.__testHoldHostTransportCapacity(handle, 20, 80, true), /capacity is unavailable/); + assert.throws( + () => addon.__testHoldHostTransportCapacity(handle, 45, 128, true), + /reservation exceeds its byte limit/, + ); + assert.strictEqual(addon.__testReleaseHostTransportCapacity(handle, foreground), true); +}); + +test('cleanup configuration is opt-in and must leave foreground capacity', async (context) => { + const handle = addon.__testOpenHostTransport((request) => request, 1, 128, 1_000); + context.after(() => addon.__testCloseHostTransport(handle)); + + assert.throws(() => addon.__testHoldHostTransportCapacity(handle, 7, 16, true), /is not configured/); + assert.throws(() => addon.__testConfigureHostTransportCleanup(handle, 64, 64), /requires two operation slots/); +}); + +test('cleanup configuration is idempotent but cannot change for a live transport', (context) => { + const handle = addon.__testOpenHostTransport((request) => request, 2, 1_024, 1_000); + context.after(() => addon.__testCloseHostTransport(handle)); + + addon.__testConfigureHostTransportCleanup(handle, 256, 512); + addon.__testConfigureHostTransportCleanup(handle, 256, 512); + assert.throws(() => addon.__testConfigureHostTransportCleanup(handle, 128, 512), /already configured differently/); +}); + +test('worker teardown closes a transport with charged foreground and cleanup capacity', async () => { + const worker = new Worker(new URL('./fixtures/host-storage-transport-worker.mjs', import.meta.url)); + const { handle, stats } = await new Promise((resolve, reject) => { + worker.once('error', reject); + worker.once('message', resolve); + }); + assert.deepStrictEqual(stats, ['2', '274', '1', '136', '0']); + await worker.terminate(); + assert.throws(() => addon.__testHostTransportStats(handle), /unknown or closed/); +}); + +test('close wakes a request waiting indefinitely for foreground admission', async () => { + const handle = addon.__testOpenHostTransport((request) => request, 1, 1_024, 1_000); + addon.__testHoldHostTransportCapacity(handle, 8, 128, false); + const waiting = roundTrip(handle, Buffer.from('waiting'), 128, false); + await new Promise((resolve) => setImmediate(resolve)); + assert.strictEqual(addon.__testCloseHostTransport(handle), true); + await assert.rejects(waiting, /host storage transport is closed/); +}); + 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); @@ -162,6 +241,49 @@ test('KvDirectory and Tantivy operate through the host storage transport', async assert.ok(entries.size > 0, 'Tantivy state remains in host storage for reopen'); }); +test('reclamation shares reader pins across foreground and cleanup storage views', async (context) => { + const entries = new Map(); + let payloadDeletes = 0; + let physicallyDeletedBytes = 0; + const handler = createHostStorageHandler( + { + read(key) { + return entries.get(key.toString('hex')); + }, + write(mutations) { + for (const mutation of mutations) { + const key = mutation.key.toString('hex'); + if (mutation.type === 'put') entries.set(key, Buffer.from(mutation.value)); + else { + const value = entries.get(key); + if (value) { + physicallyDeletedBytes += value.length; + payloadDeletes++; + } + entries.delete(key); + } + } + }, + sync() {}, + }, + { + 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)); + + const result = (await reclaimOnHostTransport(handle)).toString().split(',').map(Number); + assert.strictEqual(result[0], 2); + assert.ok(result[1] > 0, 'the cleanup view observed the foreground reader pin'); + assert.ok(result[2] > 0, 'the cleanup view deleted retired payload keys'); + assert.ok(payloadDeletes >= result[2]); + assert.ok(physicallyDeletedBytes >= result[3], 'both retired file payloads were physically removed'); +}); + test('host directory rejects a read budget that cannot carry one full chunk', async (context) => { const maxReadResponseBytes = 256 * 1024; const storage = { @@ -281,9 +403,9 @@ test('host storage handler preserves no-WAL policy and rejects malformed frames' assert.match(decodeHandlerError(handler(Buffer.from([1, 1, 4, 0, 0]))), /truncated/); }); -function roundTrip(handle, request, responseBudget = 128, useTimeout = true) { +function roundTrip(handle, request, responseBudget = 128, useTimeout = true, lowPriority = false) { return new Promise((resolve, reject) => { - addon.__testHostRoundTrip(handle, request, responseBudget, useTimeout, (encoded) => { + addon.__testHostRoundTrip(handle, request, responseBudget, useTimeout, lowPriority, (encoded) => { if (encoded[0] === 0) { resolve(encoded.subarray(1)); } else { @@ -305,6 +427,18 @@ function verifyTantivy(handle, maxReadResponseBytes = readResponseBytes) { }); } +function reclaimOnHostTransport(handle) { + return new Promise((resolve, reject) => { + addon.__testReclaimOnHostTransport(handle, readResponseBytes, controlResponseBytes, (encoded) => { + if (encoded[0] === 0) { + resolve(encoded.subarray(1)); + } else { + reject(new Error(encoded.subarray(1).toString())); + } + }); + }); +} + function decodeHandlerError(response) { assert.strictEqual(response[0], 1); assert.strictEqual(response[1], 1); diff --git a/ts/load-addon.ts b/ts/load-addon.ts index fdc4ef9..96f9786 100644 --- a/ts/load-addon.ts +++ b/ts/load-addon.ts @@ -35,14 +35,30 @@ interface NativeAddonApi { request: Buffer, responseBudget: number, useTimeout: boolean, + lowPriority: boolean, callback: NativeCallback, ): void; + __testConfigureHostTransportCleanup?(handle: number, foregroundReservedBytes: number, maxCleanupBytes: number): void; + __testHostTransportStats?(handle: number): string[]; + __testHoldHostTransportCapacity?( + handle: number, + requestBytes: number, + responseBytes: number, + lowPriority: boolean, + ): string; + __testReleaseHostTransportCapacity?(handle: number, requestId: string): boolean; __testVerifyTantivyOnHostTransport?( handle: number, maxReadResponseBytes: number, maxControlResponseBytes: number, callback: NativeCallback, ): void; + __testReclaimOnHostTransport?( + handle: number, + maxReadResponseBytes: number, + maxControlResponseBytes: number, + callback: NativeCallback, + ): void; __testCloseHostTransport?(handle: number): boolean; }