diff --git a/docs/phase-0-rocks-bridge-plan.md b/docs/phase-0-rocks-bridge-plan.md index be0696f..ec1b90e 100644 --- a/docs/phase-0-rocks-bridge-plan.md +++ b/docs/phase-0-rocks-bridge-plan.md @@ -63,9 +63,10 @@ Pinned `MmapDirectory` source also exposes a reference limitation: its path cach same older mmap to a new `open_read()` while a handle to that mmap remains alive, even after the writer flushes more bytes. The shared reference test therefore proves writer continuation, old-handle immutability, and later visibility after the old handle is released. The Rocks prototype separately -proves simultaneous old and new binding revisions because its write-once fragments support that -stronger behavior. The wrapper does not claim Mmap and Rocks have identical raw Directory cache -semantics; it requires identical observable index, commit, reopen, and search behavior. +proves simultaneous old and new binding revisions because its write-once chunks and tail revisions +support that stronger behavior. The wrapper does not claim Mmap and Rocks have identical raw +Directory cache semantics; it requires identical observable index, commit, reopen, and search +behavior. ### rocksdb-js 2.8.0 @@ -153,32 +154,48 @@ and package-isolation cases still run. ## Logical object prototype -Phase 0 uses the smallest mapping capable of exercising the Directory semantics. It is not yet the -production format. +The first Harper mapping uses fixed-size data chunks and a versioned tail. The chunk size is an +internal format choice, not a schema or factory option. It remains subject to the benchmark sweep +before the persisted format is declared stable. ```text namespace / index generation - working//fragment/ -> immutable flushed bytes - object/ -> sealed ordered fragments and total length - binding/ -> object-id and visible length - atomic/ -> complete small-file bytes and revision - pending/ -> recovery marker + chunk// -> immutable 256 KiB data chunk + tail// -> immutable final partial chunk + binding/ -> v2 object-id, chunk count, tail revision and visible length + atomic/ -> complete small-file bytes ``` -`open_write()` creates a new object identity and a logical binding with visible length zero. Bytes -may be buffered only until the next writer flush. A successful flush writes one or more new fragment -keys and atomically replaces the binding with the complete ordered fragment list and visible length. -Fragment keys are write-once: an append after flush starts a new fragment even when the preceding -fragment is smaller than the target chunk size. This is the isolation source for an open handle; a -handle captures one binding revision, object identity, fragment list, and visible length, so later -flushes never change any key it may read. Termination flushes and seals the object. Deleting removes -the binding; object bytes remain available while native handles retain them and become reclaimable -afterward. - -`atomic_write()` stores the complete small value and replaces its logical revision in one RocksDB -write batch. It is used for metadata such as `meta.json` and `.managed.json`, not large segment -output. Phase 0 verifies the semantic boundary; issue #11 defines the production encoding, -versioning, garbage collection, and bounded chunk policy. +`open_write()` creates a new object identity and a zero-length binding. The writer stages each full +chunk under its final ordinal with a WAL write while retaining at most one partial chunk. `flush()` +atomically publishes a new binding and an immutable, revisioned tail. Filling a previously published +tail creates a full chunk and a later binding revision; it never overwrites bytes visible to an +existing file handle. Every flush that replaces a partial tail leaves its previous immutable tail +revision unreachable. A file appended across K such flushes can therefore leave K-1 tails of up to +`CHUNK_SIZE - 1` bytes, while a failed publication may additionally leave an unpublished tail or +full chunks. These values remain until the derived-index reclaimer is implemented. + +A file handle captures one binding value. It computes a full-chunk key directly from the requested +offset and reads the versioned tail only when the range intersects it. A range contained in one +chunk therefore performs one payload lookup regardless of file size; ranges spanning boundaries +perform one lookup per intersecting chunk. No read walks or fetches preceding payloads. + +A full-chunk read response requires at least `CHUNK_SIZE + 7` bytes for the protocol envelope, so +the host store rejects a smaller response budget at construction. Transport admission separately +accounts for the exact encoded request and reserved response before dispatch. The prototype does +not claim a universal construction-time transport minimum: namespace and path lengths vary, and +`atomic_write()` does not yet enforce a metadata size bound. The production factory must bound +those inputs and validate its aggregate byte budget before this backend is exposed. The Phase 0 +sweep records small-read amplification and retained `OwnedBytes` because a slice keeps its complete +chunk allocation alive; a bounded chunk cache is considered only if those measurements justify it. + +Termination flushes the remaining tail. Deleting removes the binding but leaves every immutable +chunk and tail behind. They remain readable through already-open handles and become eligible for +reclamation after those handles drain. + +`atomic_write()` stores the complete small value in one RocksDB write batch. It is used for metadata +such as `meta.json` and `.managed.json`, not large segment output. Issue #11 still owns format +qualification, bounded orphan reclamation and the chunk-size performance sweep before release. ## Experiment matrix diff --git a/src/directory_harness.rs b/src/directory_harness.rs index 83baa1e..8bfa0a4 100644 --- a/src/directory_harness.rs +++ b/src/directory_harness.rs @@ -212,6 +212,50 @@ where verify_query_count(&reopened, body, "shoes", 1) } +pub fn verify_large_file(directory: D, chunk_size: usize) -> Result<(), String> +where + D: Directory, +{ + let length = chunk_size + .checked_mul(2) + .and_then(|length| length.checked_add(17)) + .ok_or_else(|| "large-file test length overflowed".to_owned())?; + let bytes = (0..length).map(|offset| (offset % 251) as u8).collect::>(); + let path = Path::new("large-file"); + let mut writer = directory.open_write(path).map_err(|error| error.to_string())?; + writer.write_all(&bytes).map_err(|error| error.to_string())?; + writer.terminate().map_err(|error| error.to_string())?; + + let file = directory.open_read(path).map_err(|error| error.to_string())?; + let full_chunk = file + .slice(0..chunk_size) + .read_bytes() + .map_err(|error| error.to_string())?; + if full_chunk.as_slice() != &bytes[..chunk_size] { + return Err("large-file full chunk returned unexpected bytes".to_owned()); + } + let range_start = chunk_size + .checked_sub(13) + .ok_or_else(|| "large-file test chunk size is too small".to_owned())?; + let range = range_start..chunk_size + 19; + let crossing = file + .slice(range.clone()) + .read_bytes() + .map_err(|error| error.to_string())?; + if crossing.as_slice() != &bytes[range] { + return Err("large-file boundary range returned unexpected bytes".to_owned()); + } + let range = length - 30..length; + let tail_crossing = file + .slice(range.clone()) + .read_bytes() + .map_err(|error| error.to_string())?; + if tail_crossing.as_slice() != &bytes[range] { + return Err("large-file tail boundary range returned unexpected bytes".to_owned()); + } + Ok(()) +} + fn verify_query_count( index: &Index, field: tantivy::schema::Field, @@ -742,6 +786,11 @@ mod tests { verify_tantivy_lifecycle(MmapDirectory::create_from_tempdir().unwrap()).unwrap(); } + #[test] + fn mmap_directory_supports_large_file_ranges() { + verify_large_file(MmapDirectory::create_from_tempdir().unwrap(), 256 * 1024).unwrap(); + } + #[test] fn harness_rejects_non_exclusive_locks() { let directory = BrokenDirectory::new(true); diff --git a/src/host_storage.rs b/src/host_storage.rs index de5c388..cdd8e9b 100644 --- a/src/host_storage.rs +++ b/src/host_storage.rs @@ -13,7 +13,7 @@ use napi_derive::napi; use tantivy::directory::OwnedBytes; use crate::boundary; -use crate::phase0::{KvDirectory, KvStore, KvStoreIdentity, Mutation, WritePolicy}; +use crate::phase0::{KvDirectory, KvStore, KvStoreIdentity, Mutation, WritePolicy, CHUNK_SIZE}; type HostCallback = ThreadsafeFunction, ErrorStrategy::Fatal>; type CompletionCallback = ThreadsafeFunction, ErrorStrategy::Fatal>; @@ -365,6 +365,7 @@ const VALUE_MISSING: u8 = 0; const VALUE_PRESENT: u8 = 1; const MUTATION_PUT: u8 = 1; const MUTATION_DELETE: u8 = 2; +const READ_RESPONSE_OVERHEAD: usize = 7; #[derive(Clone)] struct HostKvStore { @@ -380,13 +381,20 @@ impl HostKvStore { identity: KvStoreIdentity, max_read_response_bytes: usize, max_control_response_bytes: usize, - ) -> Self { - Self { + ) -> io::Result { + let minimum_read_response = CHUNK_SIZE + READ_RESPONSE_OVERHEAD; + if max_read_response_bytes < minimum_read_response { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("host read response budget must be at least {minimum_read_response} bytes"), + )); + } + Ok(Self { transport, identity, max_read_response_bytes, max_control_response_bytes, - } + }) } fn request(&self, request: Vec, response_budget: usize) -> io::Result { @@ -742,13 +750,20 @@ pub fn test_verify_tantivy_on_host_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-large-file/{run}"); + crate::directory_harness::verify_large_file( + KvDirectory::with_namespace(store.clone(), namespace.as_bytes()), + CHUNK_SIZE, + ) + }) .and_then(|_| { let namespace = format!("host-lifecycle/{run}"); crate::directory_harness::verify_tantivy_lifecycle(KvDirectory::with_namespace( diff --git a/src/lib.rs b/src/lib.rs index e751c89..c092cb4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -257,6 +257,12 @@ pub fn phase0_verify_tantivy_on_storage_lease(id: u32) -> boundary::Result<()> { phase0::KvDirectory::with_namespace(lease.clone(), namespace.as_bytes()) }) .map_err(|error| napi::Error::new("E_STORAGE", error))?; + let namespace = format!("phase0-large-file/{run}"); + directory_harness::verify_large_file( + phase0::KvDirectory::with_namespace(lease.clone(), namespace.as_bytes()), + phase0::CHUNK_SIZE, + ) + .map_err(|error| napi::Error::new("E_STORAGE", error))?; let namespace = format!("phase0-lifecycle/{run}"); directory_harness::verify_tantivy_lifecycle(phase0::KvDirectory::with_namespace(lease, namespace.as_bytes())) .map_err(|error| napi::Error::new("E_STORAGE", error)) diff --git a/src/phase0.rs b/src/phase0.rs index 61b28ba..c16dc60 100644 --- a/src/phase0.rs +++ b/src/phase0.rs @@ -53,6 +53,8 @@ pub trait KvStore: Clone + Send + Sync + 'static { fn sync(&self) -> io::Result<()>; } +pub(crate) const CHUNK_SIZE: usize = 256 * 1024; + impl Mutation { fn key(&self) -> &[u8] { match self { @@ -345,38 +347,45 @@ impl FileHandle for KvFileHandle { return Ok(OwnedBytes::empty()); } - let mut logical_offset = 0usize; + let full_length = usize::try_from(self.binding.full_chunks) + .ok() + .and_then(|chunks| chunks.checked_mul(CHUNK_SIZE)) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "binding chunk length overflow"))?; + let first_chunk = range.start / CHUNK_SIZE; + let last_chunk = (range.end - 1) / CHUNK_SIZE; + if first_chunk == last_chunk && range.end <= full_length { + let value = self.read_chunk(first_chunk)?; + let end = match range.end % CHUNK_SIZE { + 0 => CHUNK_SIZE, + end => end, + }; + return Ok(value.slice(range.start % CHUNK_SIZE..end)); + } + if range.start >= full_length { + let tail = self.read_tail()?; + return Ok(tail.slice(range.start - full_length..range.end - full_length)); + } + let mut copied = Vec::new(); - for fragment in 0..self.binding.fragments { - let value = self - .store - .read(&fragment_key(&self.namespace, self.binding.object_id, fragment))? - .ok_or_else(|| io::Error::other("binding references a missing fragment"))?; - let fragment_end = logical_offset - .checked_add(value.len()) - .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "fragment length overflow"))?; - if range.start < fragment_end && range.end > logical_offset { - let start = range.start.saturating_sub(logical_offset); - let end = value.len().min(range.end - logical_offset); - if copied.is_empty() && range.start >= logical_offset && range.end <= fragment_end { - return Ok(value.slice(start..end)); - } - if copied.is_empty() { - copied - .try_reserve(range.len()) - .map_err(|_| io::Error::other("requested file range cannot be allocated"))?; - } - copied.extend_from_slice(&value[start..end]); - } - logical_offset = fragment_end; - if logical_offset >= range.end { - break; - } + copied + .try_reserve(range.len()) + .map_err(|_| io::Error::other("requested file range cannot be allocated"))?; + let full_chunk_limit = last_chunk.saturating_add(1).min(self.binding.full_chunks as usize); + for chunk in first_chunk..full_chunk_limit { + let value = self.read_chunk(chunk)?; + let chunk_start = chunk * CHUNK_SIZE; + let start = range.start.saturating_sub(chunk_start); + let end = value.len().min(range.end - chunk_start); + copied.extend_from_slice(&value[start..end]); + } + if range.end > full_length { + let tail = self.read_tail()?; + copied.extend_from_slice(&tail[..range.end - full_length]); } if copied.len() != range.len() { return Err(io::Error::new( io::ErrorKind::InvalidData, - "binding length does not match its fragments", + "binding length does not match its chunks", )); } Ok(OwnedBytes::new(copied)) @@ -387,6 +396,42 @@ impl FileHandle for KvFileHandle { } } +impl KvFileHandle { + fn read_chunk(&self, chunk: usize) -> io::Result { + let chunk = u32::try_from(chunk) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "chunk index exceeds its format"))?; + let value = self + .store + .read(&chunk_key(&self.namespace, self.binding.object_id, chunk))? + .ok_or_else(|| io::Error::other("binding references a missing chunk"))?; + if value.len() != CHUNK_SIZE { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "full chunk has an invalid length", + )); + } + Ok(value) + } + + fn read_tail(&self) -> io::Result { + if self.binding.tail_length == 0 { + return Err(io::Error::new(io::ErrorKind::InvalidData, "binding has no tail")); + } + let value = self + .store + .read(&tail_key( + &self.namespace, + self.binding.object_id, + self.binding.tail_revision, + ))? + .ok_or_else(|| io::Error::other("binding references a missing tail"))?; + if value.len() != self.binding.tail_length as usize { + return Err(io::Error::new(io::ErrorKind::InvalidData, "tail has an invalid length")); + } + Ok(value) + } +} + struct KvDirectoryLock { state: Arc, path: PathBuf, @@ -483,7 +528,9 @@ impl Directory for KvDirectory { })?; let binding = Binding { object_id, - fragments: 0, + full_chunks: 0, + tail_revision: 0, + tail_length: 0, visible_length: 0, }; self.store @@ -501,7 +548,9 @@ impl Directory for KvDirectory { namespace: self.namespace.clone(), path: path.to_path_buf(), binding, - pending: Vec::new(), + tail: Vec::new(), + staged_full_chunks: 0, + dirty: false, }))) } @@ -629,58 +678,112 @@ struct KvWriter { namespace: Arc<[u8]>, path: PathBuf, binding: Binding, - pending: Vec, + tail: Vec, + staged_full_chunks: u32, + dirty: bool, } impl Write for KvWriter { fn write(&mut self, bytes: &[u8]) -> io::Result { - self.pending - .try_reserve(bytes.len()) + if bytes.is_empty() { + return Ok(0); + } + self.stage_full_tail()?; + let accepted = bytes.len().min(CHUNK_SIZE - self.tail.len()); + self.tail + .try_reserve(accepted) .map_err(|_| io::Error::other("writer buffer cannot be allocated"))?; - self.pending.extend_from_slice(bytes); - Ok(bytes.len()) + self.tail.extend_from_slice(&bytes[..accepted]); + self.dirty = true; + Ok(accepted) } fn flush(&mut self) -> io::Result<()> { - if self.pending.is_empty() { + if !self.dirty { return Ok(()); } + self.stage_full_tail()?; let _mutation = self.state.mutation.lock().unwrap(); let current = self .store .read(&binding_key(&self.namespace, &self.path))? .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "file was deleted while its writer was open"))?; - if decode_binding(¤t)?.object_id != self.binding.object_id { + if decode_binding(¤t)? != self.binding { return Err(io::Error::new( io::ErrorKind::NotFound, "file was replaced while its writer was open", )); } - let fragment = self.binding.fragments; - let visible_length = self + let full_chunks = self .binding - .visible_length - .checked_add(self.pending.len()) + .full_chunks + .checked_add(self.staged_full_chunks) + .ok_or_else(|| io::Error::other("chunk count exhausted"))?; + let visible_length = usize::try_from(full_chunks) + .ok() + .and_then(|chunks| chunks.checked_mul(CHUNK_SIZE)) + .and_then(|length| length.checked_add(self.tail.len())) .ok_or_else(|| io::Error::other("visible length exhausted"))?; + let tail_revision = self + .binding + .tail_revision + .checked_add(1) + .ok_or_else(|| io::Error::other("tail revision exhausted"))?; let next = Binding { object_id: self.binding.object_id, - fragments: fragment - .checked_add(1) - .ok_or_else(|| io::Error::other("fragment count exhausted"))?, + full_chunks, + tail_revision, + tail_length: u32::try_from(self.tail.len()) + .map_err(|_| io::Error::other("tail length exceeds its format"))?, visible_length, }; - let pending = std::mem::take(&mut self.pending); - let mutations = [ - Mutation::Put(fragment_key(&self.namespace, self.binding.object_id, fragment), pending), - Mutation::Put(binding_key(&self.namespace, &self.path), encode_binding(&next)), - ]; - if let Err(error) = self.store.write(&mutations, WritePolicy::WAL) { - if let Mutation::Put(_, pending) = mutations.into_iter().next().unwrap() { - self.pending = pending; - } - return Err(error); + let binding = Mutation::Put(binding_key(&self.namespace, &self.path), encode_binding(&next)); + if self.tail.is_empty() { + self.store.write(&[binding], WritePolicy::WAL)?; + } else { + self.store.write( + &[ + Mutation::Put( + tail_key(&self.namespace, self.binding.object_id, tail_revision), + self.tail.clone(), + ), + binding, + ], + WritePolicy::WAL, + )?; } self.binding = next; + self.staged_full_chunks = 0; + self.dirty = false; + Ok(()) + } +} + +impl KvWriter { + fn stage_full_tail(&mut self) -> io::Result<()> { + if self.tail.len() != CHUNK_SIZE { + return Ok(()); + } + let chunk = self + .binding + .full_chunks + .checked_add(self.staged_full_chunks) + .ok_or_else(|| io::Error::other("chunk count exhausted"))?; + let mutation = Mutation::Put( + chunk_key(&self.namespace, self.binding.object_id, chunk), + std::mem::take(&mut self.tail), + ); + let result = self.store.write(std::slice::from_ref(&mutation), WritePolicy::WAL); + let Mutation::Put(_, mut value) = mutation else { + unreachable!(); + }; + if let Err(error) = result { + self.tail = value; + return Err(error); + } + value.clear(); + self.tail = value; + self.staged_full_chunks += 1; Ok(()) } } @@ -691,10 +794,12 @@ impl TerminatingWrite for KvWriter { } } -#[derive(Clone, Debug)] +#[derive(Clone, Debug, PartialEq, Eq)] struct Binding { object_id: u64, - fragments: u32, + full_chunks: u32, + tail_revision: u64, + tail_length: u32, visible_length: usize, } @@ -725,37 +830,78 @@ fn prefixed_path(prefix: &[u8], path: &Path) -> Vec { key } -fn fragment_key(namespace: &[u8], object_id: u64, fragment: u32) -> Vec { - let mut key = namespaced_prefix(namespace, b"fragment/"); +fn chunk_key(namespace: &[u8], object_id: u64, chunk: u32) -> Vec { + let mut key = namespaced_prefix_with_capacity(namespace, b"chunk/", 12); key.extend_from_slice(&object_id.to_be_bytes()); - key.extend_from_slice(&fragment.to_be_bytes()); + key.extend_from_slice(&chunk.to_be_bytes()); key } +fn tail_key(namespace: &[u8], object_id: u64, revision: u64) -> Vec { + let mut key = namespaced_prefix_with_capacity(namespace, b"tail/", 16); + key.extend_from_slice(&object_id.to_be_bytes()); + key.extend_from_slice(&revision.to_be_bytes()); + key +} + +fn namespaced_prefix_with_capacity(namespace: &[u8], suffix: &[u8], additional: usize) -> Vec { + let mut prefix = Vec::with_capacity(namespace.len() + suffix.len() + 1 + additional); + prefix.extend_from_slice(namespace); + prefix.push(b'/'); + prefix.extend_from_slice(suffix); + prefix +} + fn encode_binding(binding: &Binding) -> Vec { - let mut bytes = Vec::with_capacity(21); - bytes.push(1); + let mut bytes = Vec::with_capacity(33); + bytes.push(2); bytes.extend_from_slice(&binding.object_id.to_be_bytes()); - bytes.extend_from_slice(&binding.fragments.to_be_bytes()); + bytes.extend_from_slice(&binding.full_chunks.to_be_bytes()); + bytes.extend_from_slice(&binding.tail_revision.to_be_bytes()); + bytes.extend_from_slice(&binding.tail_length.to_be_bytes()); bytes.extend_from_slice(&(binding.visible_length as u64).to_be_bytes()); bytes } fn decode_binding(bytes: &[u8]) -> io::Result { - if bytes.len() != 21 || bytes[0] != 1 { - return Err(io::Error::new(io::ErrorKind::InvalidData, "invalid binding")); + if bytes.first().copied() != Some(2) { + let version = bytes.first().copied().unwrap_or(0); + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("unsupported binding format version {version}"), + )); } - let visible_length = usize::try_from(decode_u64(&bytes[13..21])?) + if bytes.len() != 33 { + return Err(io::Error::new(io::ErrorKind::InvalidData, "malformed binding")); + } + let visible_length = usize::try_from(decode_u64(&bytes[25..33])?) .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "binding length exceeds this platform"))?; - Ok(Binding { + let binding = Binding { object_id: decode_u64(&bytes[1..9])?, - fragments: u32::from_be_bytes( + full_chunks: u32::from_be_bytes( bytes[9..13] .try_into() - .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid fragment count"))?, + .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid chunk count"))?, + ), + tail_revision: decode_u64(&bytes[13..21])?, + tail_length: u32::from_be_bytes( + bytes[21..25] + .try_into() + .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid tail length"))?, ), visible_length, - }) + }; + let expected_length = usize::try_from(binding.full_chunks) + .ok() + .and_then(|chunks| chunks.checked_mul(CHUNK_SIZE)) + .and_then(|length| length.checked_add(binding.tail_length as usize)); + if binding.tail_length as usize >= CHUNK_SIZE || expected_length != Some(binding.visible_length) { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "binding length is inconsistent", + )); + } + Ok(binding) } fn decode_u64(bytes: &[u8]) -> io::Result { @@ -768,6 +914,45 @@ fn decode_u64(bytes: &[u8]) -> io::Result { mod tests { use super::*; use crate::directory_harness::{verify_directory_contract, verify_tantivy_lifecycle}; + use std::sync::atomic::AtomicUsize; + + #[derive(Clone)] + struct CountingKv { + inner: FaultingKv, + reads: Arc, + } + + impl CountingKv { + fn new() -> Self { + Self { + inner: FaultingKv::default(), + reads: Arc::new(AtomicUsize::new(0)), + } + } + + fn take_reads(&self) -> usize { + self.reads.swap(0, Ordering::Relaxed) + } + } + + impl KvStore for CountingKv { + fn identity(&self) -> KvStoreIdentity { + self.inner.identity() + } + + fn read(&self, key: &[u8]) -> io::Result> { + self.reads.fetch_add(1, Ordering::Relaxed); + KvStore::read(&self.inner, key) + } + + fn write(&self, mutations: &[Mutation], policy: WritePolicy) -> io::Result<()> { + KvStore::write(&self.inner, mutations, policy) + } + + fn sync(&self) -> io::Result<()> { + KvStore::sync(&self.inner) + } + } fn put(key: &[u8], value: &[u8]) -> Mutation { Mutation::Put(key.to_vec(), value.to_vec()) @@ -945,7 +1130,7 @@ mod tests { } #[test] - fn file_handle_reads_across_fragment_boundaries() { + fn file_handle_reads_across_flush_boundaries() { let directory = FaultingDirectory::new(FaultingKv::default()); let mut writer = directory.open_write(Path::new("segment")).unwrap(); writer.write_all(b"first").unwrap(); @@ -956,6 +1141,187 @@ mod tests { assert_eq!(file.read_bytes_slice(3..9).unwrap().as_slice(), b"stseco"); } + #[test] + fn large_files_use_fixed_chunks_and_a_versioned_tail() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + let bytes = (0..CHUNK_SIZE * 3 + 17) + .map(|offset| (offset % 251) as u8) + .collect::>(); + let mut writer = directory.open_write(Path::new("segment")).unwrap(); + writer.write_all(&bytes).unwrap(); + writer.flush().unwrap(); + + let binding = decode_binding(&store.get(&binding_key(b"phase0", Path::new("segment"))).unwrap()).unwrap(); + assert_eq!(binding.full_chunks, 3); + assert_eq!(binding.tail_length, 17); + assert_eq!(binding.visible_length, bytes.len()); + assert_eq!( + directory + .open_read(Path::new("segment")) + .unwrap() + .read_bytes() + .unwrap() + .as_slice(), + bytes + ); + } + + #[test] + fn exact_chunk_multiple_has_no_tail_and_reads_its_boundary() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + let bytes = vec![7; CHUNK_SIZE * 2]; + let mut writer = directory.open_write(Path::new("segment")).unwrap(); + writer.write_all(&bytes).unwrap(); + writer.flush().unwrap(); + + let binding = decode_binding(&store.get(&binding_key(b"phase0", Path::new("segment"))).unwrap()).unwrap(); + assert_eq!(binding.full_chunks, 2); + assert_eq!(binding.tail_length, 0); + let file = directory.open_read(Path::new("segment")).unwrap(); + assert_eq!( + file.read_bytes_slice(0..CHUNK_SIZE).unwrap().as_slice(), + &bytes[..CHUNK_SIZE] + ); + } + + #[test] + fn range_reads_fetch_only_intersecting_chunks() { + let store = CountingKv::new(); + let directory = KvDirectory::new(store.clone()); + let bytes = (0..CHUNK_SIZE * 4 + 29) + .map(|offset| (offset % 251) as u8) + .collect::>(); + let mut writer = directory.open_write(Path::new("segment")).unwrap(); + writer.write_all(&bytes).unwrap(); + writer.flush().unwrap(); + let file = directory.open_read(Path::new("segment")).unwrap(); + store.take_reads(); + + let start = CHUNK_SIZE * 3 + 11; + let end = start + 97; + assert_eq!( + file.read_bytes_slice(start..end).unwrap().as_slice(), + &bytes[start..end] + ); + assert_eq!(store.take_reads(), 1); + + let start = CHUNK_SIZE / 2; + let end = CHUNK_SIZE * 2 + 100; + assert_eq!( + file.read_bytes_slice(start..end).unwrap().as_slice(), + &bytes[start..end] + ); + assert_eq!(store.take_reads(), 3); + + let start = CHUNK_SIZE * 4 - 11; + let end = CHUNK_SIZE * 4 + 17; + assert_eq!( + file.read_bytes_slice(start..end).unwrap().as_slice(), + &bytes[start..end] + ); + assert_eq!(store.take_reads(), 2); + } + + #[test] + fn filling_a_published_tail_does_not_change_an_open_handle() { + let directory = FaultingDirectory::new(FaultingKv::default()); + let first_bytes = vec![3; CHUNK_SIZE - 17]; + let appended = vec![5; 34]; + let mut writer = directory.open_write(Path::new("segment")).unwrap(); + writer.write_all(&first_bytes).unwrap(); + writer.flush().unwrap(); + let first = directory.open_read(Path::new("segment")).unwrap(); + + writer.write_all(&appended).unwrap(); + writer.flush().unwrap(); + let second = directory.open_read(Path::new("segment")).unwrap(); + assert_eq!(first.read_bytes().unwrap().as_slice(), first_bytes); + assert_eq!( + second.read_bytes().unwrap().as_slice(), + [first_bytes, appended].concat() + ); + } + + #[test] + fn failed_full_chunk_staging_can_be_retried_during_flush() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + let bytes = vec![7; CHUNK_SIZE]; + let mut writer = directory.open_write(Path::new("segment")).unwrap(); + writer.write_all(&bytes).unwrap(); + store.fail_next_write(); + assert!(writer.flush().is_err()); + writer.flush().unwrap(); + assert_eq!( + directory + .open_read(Path::new("segment")) + .unwrap() + .read_bytes() + .unwrap() + .as_slice(), + bytes + ); + } + + #[test] + fn staging_failure_does_not_consume_the_failing_write() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + let binding = Binding { + object_id: 1, + full_chunks: 0, + tail_revision: 0, + tail_length: 0, + visible_length: 0, + }; + store + .write( + &[Mutation::Put( + binding_key(b"phase0", Path::new("segment")), + encode_binding(&binding), + )], + WritePolicy::WAL, + ) + .unwrap(); + let mut writer = KvWriter { + store, + state: directory.state.clone(), + namespace: directory.namespace.clone(), + path: PathBuf::from("segment"), + binding, + tail: Vec::new(), + staged_full_chunks: 0, + dirty: false, + }; + let prefix = vec![1; CHUNK_SIZE - 4_096]; + let suffix = vec![2; 8_192]; + assert_eq!(writer.write(&prefix).unwrap(), prefix.len()); + writer.store.fail_next_write(); + assert_eq!(writer.write(&suffix).unwrap(), 4_096); + assert!(writer.write(&suffix[4_096..]).is_err()); + assert_eq!(writer.write(&suffix[4_096..]).unwrap(), 4_096); + writer.flush().unwrap(); + + let file = directory.open_read(Path::new("segment")).unwrap(); + assert_eq!(file.read_bytes().unwrap().as_slice(), [prefix, suffix].concat()); + } + + #[test] + fn rejects_an_older_binding_format_with_its_version() { + let error = decode_binding(&[1]).unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); + assert!(error.to_string().contains("version 1")); + } + + #[test] + fn distinguishes_a_malformed_current_binding() { + let error = decode_binding(&[2; 20]).unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); + assert_eq!(error.to_string(), "malformed binding"); + } + #[test] fn directory_supports_a_real_tantivy_lifecycle_and_crash_reopen() { let store = FaultingKv::default(); diff --git a/test/host-storage-transport.test.mjs b/test/host-storage-transport.test.mjs index ee965b1..8cfda23 100644 --- a/test/host-storage-transport.test.mjs +++ b/test/host-storage-transport.test.mjs @@ -162,6 +162,25 @@ 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('host directory rejects a read budget that cannot carry one full chunk', async (context) => { + const maxReadResponseBytes = 256 * 1024; + const storage = { + read() {}, + write() {}, + sync() {}, + }; + const handler = createHostStorageHandler(storage, { + maxMutations: 2, + maxReadResponseBytes, + maxControlResponseBytes: controlResponseBytes, + maxErrorBytes: controlResponseBytes, + }); + const handle = addon.__testOpenHostTransport(handler, 4, 2 * 1024 * 1024, 1_000); + context.after(() => addon.__testCloseHostTransport(handle)); + + await assert.rejects(verifyTantivy(handle, maxReadResponseBytes), /must be at least 262151 bytes/); +}); + test('a failed durability barrier does not pretend the preceding atomic write rolled back', async (context) => { const entries = new Map(); let writes = 0; @@ -274,9 +293,9 @@ function roundTrip(handle, request, responseBudget = 128, useTimeout = true) { }); } -function verifyTantivy(handle) { +function verifyTantivy(handle, maxReadResponseBytes = readResponseBytes) { return new Promise((resolve, reject) => { - addon.__testVerifyTantivyOnHostTransport(handle, readResponseBytes, controlResponseBytes, (encoded) => { + addon.__testVerifyTantivyOnHostTransport(handle, maxReadResponseBytes, controlResponseBytes, (encoded) => { if (encoded[0] === 0) { resolve(); } else {