diff --git a/docs/phase-0-rocks-bridge-plan.md b/docs/phase-0-rocks-bridge-plan.md index e56d91e..4f04689 100644 --- a/docs/phase-0-rocks-bridge-plan.md +++ b/docs/phase-0-rocks-bridge-plan.md @@ -163,7 +163,7 @@ namespace / index generation format marker kind -> directory key-format version chunk kind: object-id, ordinal -> immutable 256 KiB data chunk tail kind: object-id, revision -> immutable final partial chunk - binding kind: logical path -> v2 object-id, chunk count, tail revision and visible length + binding kind: logical path -> v3 object-id, published state and physical-key high-waters atomic kind: logical path -> complete small-file bytes ``` @@ -185,14 +185,22 @@ New logical key kinds, including reclamation metadata, receive new kind tags und version. A storage provider must also change `KvStoreIdentity` whenever close, restore, or column- family replacement can change the bytes behind an identity. -`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. +`open_write()` creates a new object identity and a zero-length binding. Before writing the first +chunk outside the binding's exclusive chunk high-water, the writer reserves 64 ordinals with one +WAL binding update. Covered chunks keep the one-put path, so a 1 GiB file adds 64 reservation reads +and writes rather than one binding round trip per 256 KiB chunk. The reservation is visible before +its payloads, and missing keys inside it are valid. The writer retains at most one partial chunk. +`flush()` atomically publishes a new binding and an immutable, revisioned tail while advancing the +tail high-water, including for an empty 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. +Applied-but-reported-failed reservations, chunk puts, and publications are retried idempotently. + +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. Binding +decode enforces that the published chunk count and tail revision stay within the high-waters and +that their maximum possible physical extent is no more than the 1 TiB format safety bound. 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 diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md index d0f3b81..ccf5a3b 100644 --- a/docs/reclamation-plan.md +++ b/docs/reclamation-plan.md @@ -9,11 +9,13 @@ readiness wait for a complete sweep. ## Grounding -The merged fulltext baseline is `2e853d4`. `KvDirectory` stores immutable 256 KiB chunks and -revisioned tails. `delete()` currently removes only logical bindings, each successful `flush()` can -leave the previous tail revision unreachable, and a crashed writer can leave staged chunks that no -binding ever named. `KvStore` supplies point read, atomic batch write, and sync, but no key -enumeration. Harper PR #2535 supplies the same narrow storage shape from Harper-owned RocksDB. +This plan is written against fulltext main at `b9964b0`, the rebase-merged result of +[Prevent Harper full-text index key collisions #24](https://github.com/HarperFast/fulltext/pull/24). +`KvDirectory` stores immutable 256 KiB chunks and revisioned tails. `delete()` currently removes +only logical bindings, each successful `flush()` can leave the previous tail revision unreachable, +and a crashed writer can leave staged chunks that no binding ever named. `KvStore` supplies point +read, atomic batch write, and sync, but no key enumeration. Harper PR #2535 supplies the same narrow +storage shape from Harper-owned RocksDB. Tantivy's `ManagedDirectory` decides when a logical file name is retired. `KvDirectory` owns the physical object behind that name and the lifetime of opened handles. Harper's derived-index runtime @@ -24,12 +26,51 @@ reachability. ### Make every physical key derivable -Extend the binding with monotonic chunk and tail high-waters. Every payload write advances its bound -in the same atomic batch: staged chunk `n` records at least `n + 1`, and published tail revision `r` -records at least `r`. `flush()` updates the published fields and binding in that batch as well. The -complete physical key set is therefore arithmetic—chunk ordinals `0..chunk_high_water` and tail -revisions `1..=tail_high_water`—even when a writer crashes before publication. A payload key can -never become durable without a durable record that bounds its existence. +Extend the binding with monotonic chunk and tail high-waters. Before staging a chunk outside the +current bound, the writer durably reserves a fixed stride of chunk ordinals in the binding. The +reservation write completes before any covered payload write is issued; RocksDB WAL prefix ordering +and the later publication barrier therefore cannot recover or publish a chunk without its earlier +bound. Tail payload and tail high-water advance together in the existing publication batch. The +complete possible physical key set is arithmetic—chunk ordinals `0..chunk_high_water` and tail +revisions `1..=tail_high_water`—even when a writer crashes before publication. Bounds may include +keys that were never written; missing payload keys are normal during reclamation. + +Slice 2 encodes these fields in binding format v3 and advances the directory key-format marker so +every storage entry point rejects a slice-1 namespace at one choke point. The chunk high-water is an +exclusive ordinal and must be at least the published full-chunk count. The tail high-water is the +greatest possible revision, including revisions advanced by an empty-tail flush, so the published +tail revision must never exceed it. Decoding rejects malformed lengths, inconsistent published +lengths, either violated bound, and a possible physical extent above the format's 1 TiB hard limit +before any cleanup work can use the values. Harper may impose a lower operational limit, but the +library does not expose a customer setting for the on-disk safety bound. The prior unreleased key and +binding formats are rebuilt rather than migrated. + +The first chunk in each 64-ordinal reservation takes the existing directory mutation gate, rereads +the binding, and verifies that the object id and published file state still match the writer. A +stored reservation ahead of the writer is adopted as the result of an applied-but-reported-failed +attempt; a lower stored bound is corruption. Otherwise one binding mutation reserves the next +stride. Covered chunks keep the current single-put path without a binding read or directory gate. +An applied-but-reported-failed chunk put is retried idempotently with the same bytes and ordinal. +Slice 3 installs per-path writer retirement before enabling cleanup, so delete cannot drain a +captured reservation and then allow its old writer to stage into it. Slice 2 alone does not claim to +reclaim a deleted binding and remains unavailable to production. + +`flush()` treats object identity and published fields—not bookkeeping-only high-waters—as the file +replacement guard. It preserves the greatest stored bounds and allocates every revision above both +the published revision and tail high-water, even when the new tail is empty; this makes the decode +invariant unconditional and prevents later payload reuse. Publication writes any tail payload, +updated bounds, and published binding in one WAL batch. A retry recognizes an already-published +intended binding as success before accepting later bytes; a definitely unapplied attempt is cleared +and recomputed from the preserved writer buffer. Any other published-state change is replacement. +All mutation gates touched by this slice recover poisoned mutex state and return storage errors +rather than creating a repeatable actor panic. + +The 64-chunk stride bounds overstatement to less than 16 MiB per reservation while reducing a 1 GiB +file from 4,096 binding reads and updates to 64. A `CountingKv` test pins reservation reads, write +calls, mutation count, and overstatement so the stride remains an explicit performance choice rather +than an accidental constant. The existing host-transport large-file test exercises reservation, +chunk staging, and publication through the Node boundary; host operation counts remain a production +integration measurement. `delete()` atomically removes the binding and appends a reclaim entry containing the object id and its high-waters to a durable FIFO. Tantivy 0.26.1's `ManagedDirectory` makes this transition complete: @@ -104,15 +145,17 @@ silently retries forever or advances past unknown data. ## Alternatives -| Axis | Candidate and disposition | -| ------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Different layer | RocksDB compaction and Harper log retention cannot see Tantivy handles. `ManagedDirectory` owns logical retirement, while `KvDirectory` owns physical retirement. | -| Discovery | Prefix enumeration or a dense object-id sweep. Both do work proportional to stored history rather than garbage and add machinery that transition records avoid. | -| Different timing | Delete up to a fixed number of chunks synchronously in `delete()` and enqueue only the remainder. This may help small objects, but it lengthens Tantivy metadata GC and is deferred until measurement shows a net win. | -| Lower-layer range delete | Add a range-tombstone primitive. This expands the frozen Harper storage surface and makes foreground reads pay tombstone checks until compaction in a shared column family, so it is rejected for the first release. | -| Deeper cause | Record existence in the batch that creates each key and enqueue retirement in the batch that makes it unreachable. This is the chosen foundation. | -| Do less | Reclaim only final object deletion and leave superseded tails until then. Real Tantivy flush-count measurement remains a gate, but tail-only FIFO entries cheaply bound the general Directory contract. | -| Chosen | Binding high-waters plus a transition-fed durable FIFO, with object-id pins and bounded low-priority draining. | +| Axis | Candidate and disposition | +| ------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Different layer | RocksDB compaction and Harper log retention cannot see Tantivy handles. `ManagedDirectory` owns logical retirement, while `KvDirectory` owns physical retirement. | +| Discovery | Prefix enumeration or a dense object-id sweep. `KvStore` exposes no enumeration primitive, and either scan would do work proportional to stored history rather than garbage. | +| Managed paths | Treat Tantivy's `.managed.json` as the discovery source. It names logical paths, not object ids, chunk ordinals, or tail revisions, so delete/recreate cannot recover the retired object's extent. | +| Bound placement | Update the binding with every payload, reserve bounded strides in the binding, or add a separate per-object extent key. Strided binding reservation is chosen: it keeps deletion capture atomic in slice 3 without per-chunk host reads. | +| Different timing | Delete up to a fixed number of chunks synchronously in `delete()` and enqueue only the remainder. This may help small objects, but it lengthens Tantivy metadata GC and is deferred until measurement shows a net win. | +| Lower-layer range delete | Add a range-tombstone primitive. This expands the frozen Harper storage surface and makes foreground reads pay tombstone checks until compaction in a shared column family, so it is rejected for the first release. | +| Deeper cause | Record existence in the batch that creates each key and enqueue retirement in the batch that makes it unreachable. This is the chosen foundation. | +| Do less | Reclaim only final object deletion and leave superseded tails until then. Reusing tail keys is invalid because opened handles require immutable revisions; real flush-count measurement remains a gate. | +| Chosen | Binding high-waters plus a transition-fed durable FIFO, with object-id pins and bounded low-priority draining. | ## Persistence format and delivery sequence @@ -138,8 +181,14 @@ storage integration both pass. No native RocksDB storage provider is added to th ## Verification -The mapping tests cover repeated flush, delete/recreate with a retained old reader, abandoned -writers, partial object cleanup, restart during a range and between FIFO entries, namespace +Slice 2 adds an applied-then-report-failure mode to `FaultingKv` and an invariant audit over its +physical keys. Before deletion is introduced in slice 3, every visible or crash-recovered chunk and +tail must fall within its surviving binding's bounds. Tests cover reservation failure and adoption, +idempotent chunk retry, already-published flush retry, empty-tail bounds, former-format rejection at +every entry point, decode-time extent limits, and exact reservation I/O counts. + +The completed mapping tests cover repeated flush, delete/recreate with a retained old reader, +abandoned writers, partial object cleanup, restart during a range and between FIFO entries, namespace isolation, concurrent `open_write()` on distinct paths, and cleanup concurrent with Tantivy merge completion. A deterministic hook forces the binding-read/pin-register race. Failure injection covers staging and high-water updates, publication and tail enqueue, object retirement and enqueue, diff --git a/src/phase0.rs b/src/phase0.rs index 61cf8c7..e57ed81 100644 --- a/src/phase0.rs +++ b/src/phase0.rs @@ -82,8 +82,9 @@ struct State { next_sequence: u64, visible: BTreeMap, VersionedValue>, durable: BTreeMap, VersionedValue>, - pending_wal: Vec<(Vec, VersionedValue)>, + pending_wal: Vec, VersionedValue)>>, fail_next_write: bool, + fail_after_next_write: bool, fail_next_flush: bool, } @@ -139,6 +140,8 @@ impl FaultingKv { if std::mem::take(&mut state.fail_next_write) { return Err(io::Error::other("injected write failure")); } + let fail_after_write = std::mem::take(&mut state.fail_after_next_write); + let mut wal_batch = Vec::with_capacity(mutations.len()); for mutation in mutations { state.next_sequence += 1; let entry = VersionedValue { @@ -148,16 +151,25 @@ impl FaultingKv { let key = mutation.key().to_vec(); state.visible.insert(key.clone(), entry.clone()); if policy.wal_enabled { - state.pending_wal.push((key, entry)); + wal_batch.push((key, entry)); } } + if !wal_batch.is_empty() { + state.pending_wal.push(wal_batch); + } if policy.sync { let pending = std::mem::take(&mut state.pending_wal); - for (key, entry) in pending { - apply_if_newer(&mut state.durable, key, entry); + for batch in pending { + for (key, entry) in batch { + apply_if_newer(&mut state.durable, key, entry); + } } } - Ok(()) + if fail_after_write { + Err(io::Error::other("injected post-commit write failure")) + } else { + Ok(()) + } } pub fn begin_flush(&self) -> FlushBarrier { @@ -187,6 +199,7 @@ impl FaultingKv { durable, pending_wal: Vec::new(), fail_next_write: false, + fail_after_next_write: false, fail_next_flush: false, })), } @@ -196,6 +209,10 @@ impl FaultingKv { self.state.lock().unwrap().fail_next_write = true; } + pub fn fail_after_next_write(&self) { + self.state.lock().unwrap().fail_after_next_write = true; + } + pub fn fail_next_flush(&self) { self.state.lock().unwrap().fail_next_flush = true; } @@ -499,7 +516,11 @@ impl Directory for KvDirectory { io_error: Arc::new(error), filepath: path.to_path_buf(), })?; - let _mutation = self.state.mutation.lock().unwrap(); + let _mutation = self + .state + .mutation + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); let binding = binding_key(&self.namespace, path); let atomic = atomic_key(&self.namespace, path); if self @@ -545,7 +566,11 @@ impl Directory for KvDirectory { } fn open_write(&self, path: &Path) -> Result { - let _mutation = self.state.mutation.lock().unwrap(); + let _mutation = self + .state + .mutation + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); self.ensure_format(true) .map_err(|error| OpenWriteError::wrap_io_error(error, path.to_path_buf()))?; let key = binding_key(&self.namespace, path); @@ -579,6 +604,8 @@ impl Directory for KvDirectory { tail_revision: 0, tail_length: 0, visible_length: 0, + chunk_high_water: 0, + tail_high_water: 0, }; self.store .write( @@ -598,6 +625,7 @@ impl Directory for KvDirectory { tail: Vec::new(), staged_full_chunks: 0, dirty: false, + pending_publication: None, }))) } @@ -613,7 +641,11 @@ impl Directory for KvDirectory { fn atomic_write(&self, path: &Path, data: &[u8]) -> io::Result<()> { { - let _mutation = self.state.mutation.lock().unwrap(); + let _mutation = self + .state + .mutation + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); self.ensure_format(true)?; self.store.write( &[Mutation::Put(atomic_key(&self.namespace, path), data.to_vec())], @@ -732,6 +764,7 @@ struct KvWriter { tail: Vec, staged_full_chunks: u32, dirty: bool, + pending_publication: Option, } impl Write for KvWriter { @@ -739,6 +772,7 @@ impl Write for KvWriter { if bytes.is_empty() { return Ok(0); } + self.reconcile_pending_publication()?; self.stage_full_tail()?; let accepted = bytes.len().min(CHUNK_SIZE - self.tail.len()); self.tail @@ -750,21 +784,89 @@ impl Write for KvWriter { } fn flush(&mut self) -> io::Result<()> { + self.reconcile_pending_publication()?; if !self.dirty { return Ok(()); } self.stage_full_tail()?; - let _mutation = self.state.mutation.lock().unwrap(); + let state = self.state.clone(); + let _mutation = state.mutation.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let current = self.current_binding()?; + let expected = self.next_binding()?; + if current == expected { + self.finish_flush(expected); + return Ok(()); + } + self.adopt_high_waters(¤t)?; + let next = self.next_binding()?; + let binding = Mutation::Put(binding_key(&self.namespace, &self.path), encode_binding(&next)); + self.pending_publication = Some(next.clone()); + let result = 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, next.tail_revision), + self.tail.clone(), + ), + binding, + ], + WritePolicy::WAL, + ) + }; + result?; + self.finish_flush(next); + Ok(()) + } +} + +impl KvWriter { + fn reconcile_pending_publication(&mut self) -> io::Result<()> { + let Some(pending) = self.pending_publication.clone() else { + return Ok(()); + }; + let state = self.state.clone(); + let _mutation = state.mutation.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let current = self.current_binding()?; + if current == pending { + self.finish_flush(pending); + return Ok(()); + } + self.adopt_high_waters(¤t)?; + self.pending_publication = None; + Ok(()) + } + + fn current_binding(&self) -> io::Result { 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)? != self.binding { + decode_binding(¤t) + } + + fn adopt_high_waters(&mut self, current: &Binding) -> io::Result<()> { + if !current.same_published_state(&self.binding) { return Err(io::Error::new( io::ErrorKind::NotFound, "file was replaced while its writer was open", )); } + if current.chunk_high_water < self.binding.chunk_high_water + || current.tail_high_water < self.binding.tail_high_water + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "binding high-water moved backward", + )); + } + self.binding.chunk_high_water = current.chunk_high_water; + self.binding.tail_high_water = current.tail_high_water; + Ok(()) + } + + fn next_binding(&self) -> io::Result { let full_chunks = self .binding .full_chunks @@ -777,7 +879,7 @@ impl Write for KvWriter { .ok_or_else(|| io::Error::other("visible length exhausted"))?; let tail_revision = self .binding - .tail_revision + .tail_high_water .checked_add(1) .ok_or_else(|| io::Error::other("tail revision exhausted"))?; let next = Binding { @@ -787,30 +889,20 @@ impl Write for KvWriter { tail_length: u32::try_from(self.tail.len()) .map_err(|_| io::Error::other("tail length exceeds its format"))?, visible_length, + chunk_high_water: self.binding.chunk_high_water, + tail_high_water: tail_revision, }; - 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, - )?; - } + validate_binding(&next)?; + Ok(next) + } + + fn finish_flush(&mut self, next: Binding) { self.binding = next; self.staged_full_chunks = 0; self.dirty = false; - Ok(()) + self.pending_publication = None; } -} -impl KvWriter { fn stage_full_tail(&mut self) -> io::Result<()> { if self.tail.len() != CHUNK_SIZE { return Ok(()); @@ -820,6 +912,7 @@ impl KvWriter { .full_chunks .checked_add(self.staged_full_chunks) .ok_or_else(|| io::Error::other("chunk count exhausted"))?; + self.reserve_chunk(chunk)?; let mutation = Mutation::Put( chunk_key(&self.namespace, self.binding.object_id, chunk), std::mem::take(&mut self.tail), @@ -837,6 +930,49 @@ impl KvWriter { self.staged_full_chunks += 1; Ok(()) } + + fn reserve_chunk(&mut self, chunk: u32) -> io::Result<()> { + if chunk < self.binding.chunk_high_water { + return Ok(()); + } + let state = self.state.clone(); + let _mutation = state.mutation.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let current = self.current_binding()?; + self.adopt_high_waters(¤t)?; + if chunk < self.binding.chunk_high_water { + return Ok(()); + } + if chunk != self.binding.chunk_high_water { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "chunk reservation is not contiguous", + )); + } + let tail_extent = self.binding.tail_high_water as u128 * (CHUNK_SIZE - 1) as u128; + let available = MAX_OBJECT_EXTENT_BYTES + .checked_sub(tail_extent) + .ok_or_else(|| io::Error::other("object physical extent exhausted"))?; + let max_high_water = u32::try_from(available / CHUNK_SIZE as u128) + .map_err(|_| io::Error::other("chunk reservation exceeds its format"))?; + let desired = chunk.saturating_add(CHUNK_RESERVATION_STRIDE).min(max_high_water); + if desired <= chunk { + return Err(io::Error::other("object physical extent exhausted")); + } + let mut reserved = self.binding.clone(); + reserved.chunk_high_water = desired; + validate_binding(&reserved)?; + let result = self.store.write( + &[Mutation::Put( + binding_key(&self.namespace, &self.path), + encode_binding(&reserved), + )], + WritePolicy::WAL, + ); + if result.is_ok() { + self.binding = reserved; + } + result + } } impl TerminatingWrite for KvWriter { @@ -852,6 +988,18 @@ struct Binding { tail_revision: u64, tail_length: u32, visible_length: usize, + chunk_high_water: u32, + tail_high_water: u64, +} + +impl Binding { + fn same_published_state(&self, other: &Self) -> bool { + self.object_id == other.object_id + && self.full_chunks == other.full_chunks + && self.tail_revision == other.tail_revision + && self.tail_length == other.tail_length + && self.visible_length == other.visible_length + } } fn counter_key(namespace: &[u8]) -> Vec { @@ -866,7 +1014,10 @@ fn atomic_key(namespace: &[u8], path: &Path) -> Vec { prefixed_path(&namespaced_prefix(namespace, KEY_KIND_ATOMIC), path) } -const KEY_FORMAT_VERSION: u8 = 1; +const KEY_FORMAT_VERSION: u8 = 2; +const BINDING_FORMAT_VERSION: u8 = 3; +const CHUNK_RESERVATION_STRIDE: u32 = 64; +const MAX_OBJECT_EXTENT_BYTES: u128 = 1 << 40; const KEY_KIND_COUNTER: u8 = 1; const KEY_KIND_BINDING: u8 = 2; const KEY_KIND_ATOMIC: u8 = 3; @@ -978,25 +1129,27 @@ fn tail_key(namespace: &[u8], object_id: u64, revision: u64) -> Vec { } fn encode_binding(binding: &Binding) -> Vec { - let mut bytes = Vec::with_capacity(33); - bytes.push(2); + let mut bytes = Vec::with_capacity(45); + bytes.push(BINDING_FORMAT_VERSION); bytes.extend_from_slice(&binding.object_id.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.extend_from_slice(&binding.chunk_high_water.to_be_bytes()); + bytes.extend_from_slice(&binding.tail_high_water.to_be_bytes()); bytes } fn decode_binding(bytes: &[u8]) -> io::Result { - if bytes.first().copied() != Some(2) { + if bytes.first().copied() != Some(BINDING_FORMAT_VERSION) { let version = bytes.first().copied().unwrap_or(0); return Err(io::Error::new( io::ErrorKind::InvalidData, format!("unsupported binding format version {version}"), )); } - if bytes.len() != 33 { + if bytes.len() != 45 { return Err(io::Error::new(io::ErrorKind::InvalidData, "malformed binding")); } let visible_length = usize::try_from(decode_u64(&bytes[25..33])?) @@ -1015,7 +1168,18 @@ fn decode_binding(bytes: &[u8]) -> io::Result { .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid tail length"))?, ), visible_length, + chunk_high_water: u32::from_be_bytes( + bytes[33..37] + .try_into() + .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid chunk high-water"))?, + ), + tail_high_water: decode_u64(&bytes[37..45])?, }; + validate_binding(&binding)?; + Ok(binding) +} + +fn validate_binding(binding: &Binding) -> io::Result<()> { let expected_length = usize::try_from(binding.full_chunks) .ok() .and_then(|chunks| chunks.checked_mul(CHUNK_SIZE)) @@ -1026,7 +1190,21 @@ fn decode_binding(bytes: &[u8]) -> io::Result { "binding length is inconsistent", )); } - Ok(binding) + if binding.chunk_high_water < binding.full_chunks || binding.tail_high_water < binding.tail_revision { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "binding high-water is below published state", + )); + } + let possible_extent = binding.chunk_high_water as u128 * CHUNK_SIZE as u128 + + binding.tail_high_water as u128 * (CHUNK_SIZE - 1) as u128; + if possible_extent > MAX_OBJECT_EXTENT_BYTES { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "binding physical extent exceeds the format limit", + )); + } + Ok(()) } fn decode_u64(bytes: &[u8]) -> io::Result { @@ -1045,6 +1223,8 @@ mod tests { struct CountingKv { inner: FaultingKv, reads: Arc, + writes: Arc, + mutations: Arc, } #[derive(Clone)] @@ -1058,12 +1238,22 @@ mod tests { Self { inner: FaultingKv::default(), reads: Arc::new(AtomicUsize::new(0)), + writes: Arc::new(AtomicUsize::new(0)), + mutations: Arc::new(AtomicUsize::new(0)), } } fn take_reads(&self) -> usize { self.reads.swap(0, Ordering::Relaxed) } + + fn take_io_counts(&self) -> (usize, usize, usize) { + ( + self.reads.swap(0, Ordering::Relaxed), + self.writes.swap(0, Ordering::Relaxed), + self.mutations.swap(0, Ordering::Relaxed), + ) + } } impl KvStore for CountingKv { @@ -1077,6 +1267,8 @@ mod tests { } fn write(&self, mutations: &[Mutation], policy: WritePolicy) -> io::Result<()> { + self.writes.fetch_add(1, Ordering::Relaxed); + self.mutations.fetch_add(mutations.len(), Ordering::Relaxed); KvStore::write(&self.inner, mutations, policy) } @@ -1121,6 +1313,85 @@ mod tests { FaultingDirectory::with_namespace(store, b"catalog") } + fn former_format_directory() -> FaultingDirectory { + let store = FaultingKv::default(); + store + .write( + &[Mutation::Put(format_marker_key(b"catalog"), vec![1])], + WritePolicy::WAL_SYNC, + ) + .unwrap(); + FaultingDirectory::with_namespace(store, b"catalog") + } + + fn empty_binding() -> Binding { + Binding { + object_id: 1, + full_chunks: 0, + tail_revision: 0, + tail_length: 0, + visible_length: 0, + chunk_high_water: 0, + tail_high_water: 0, + } + } + + fn assert_physical_keys_within_high_water(store: &FaultingKv, namespace: &[u8], path: &Path) { + let state = store.state.lock().unwrap(); + let binding = state + .visible + .get(&binding_key(namespace, path)) + .and_then(|entry| entry.value.as_ref()) + .map(|bytes| decode_binding(bytes).unwrap()) + .expect("object binding is missing"); + assert_object_keys_within_high_water(&state.visible, namespace, path, binding.object_id); + } + + fn assert_object_keys_within_high_water( + entries: &BTreeMap, VersionedValue>, + namespace: &[u8], + path: &Path, + object_id: u64, + ) { + let binding = entries + .get(&binding_key(namespace, path)) + .and_then(|entry| entry.value.as_ref()) + .map(|bytes| decode_binding(bytes).unwrap()); + let mut chunk_prefix = namespaced_prefix_with_capacity(namespace, KEY_KIND_CHUNK, 8); + chunk_prefix.extend_from_slice(&object_id.to_be_bytes()); + let mut tail_prefix = namespaced_prefix_with_capacity(namespace, KEY_KIND_TAIL, 8); + tail_prefix.extend_from_slice(&object_id.to_be_bytes()); + for (key, value) in entries { + if value.value.is_none() { + continue; + } + if let Some(suffix) = key.strip_prefix(chunk_prefix.as_slice()) { + let binding = binding.as_ref().expect("recovered chunk has no binding"); + assert_eq!(binding.object_id, object_id); + let chunk = u32::from_be_bytes(suffix.try_into().unwrap()); + assert!(chunk < binding.chunk_high_water); + } + if let Some(suffix) = key.strip_prefix(tail_prefix.as_slice()) { + let binding = binding.as_ref().expect("recovered tail has no binding"); + assert_eq!(binding.object_id, object_id); + let revision = u64::from_be_bytes(suffix.try_into().unwrap()); + assert!(revision <= binding.tail_high_water); + } + } + } + + fn assert_every_wal_prefix_respects_high_waters(store: &FaultingKv, namespace: &[u8], path: &Path) { + let state = store.state.lock().unwrap(); + let mut recovered = state.durable.clone(); + assert_object_keys_within_high_water(&recovered, namespace, path, 1); + for batch in &state.pending_wal { + for (key, entry) in batch { + apply_if_newer(&mut recovered, key.clone(), entry.clone()); + } + assert_object_keys_within_high_water(&recovered, namespace, path, 1); + } + } + #[test] fn synced_metadata_makes_earlier_wal_objects_durable() { let store = FaultingKv::default(); @@ -1276,9 +1547,25 @@ mod tests { writer.write_all(b"pending").unwrap(); store.fail_next_write(); assert!(writer.flush().is_err()); + writer.write_all(b" plus more").unwrap(); + writer.flush().unwrap(); + let file = directory.open_read(Path::new("segment")).unwrap(); + assert_eq!(file.read_bytes().unwrap().as_slice(), b"pending plus more"); + } + + #[test] + fn applied_but_reported_failed_publication_is_retryable() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + let mut writer = directory.open_write(Path::new("segment")).unwrap(); + writer.write_all(b"pending").unwrap(); + store.fail_after_next_write(); + assert!(writer.flush().is_err()); + writer.write_all(b" plus more").unwrap(); writer.flush().unwrap(); + let file = directory.open_read(Path::new("segment")).unwrap(); - assert_eq!(file.read_bytes().unwrap().as_slice(), b"pending"); + assert_eq!(file.read_bytes().unwrap().as_slice(), b"pending plus more"); } #[test] @@ -1434,6 +1721,23 @@ mod tests { assert_prototype_error(prototype_directory().open_read(path)); } + #[test] + fn every_storage_entry_point_rejects_the_former_directory_format() { + fn assert_version_error(result: Result) { + let error = result.err().expect("former directory format should be rejected"); + assert!(error.to_string().contains("directory format version 1")); + } + + let path = Path::new("meta.json"); + assert_version_error(former_format_directory().exists(path)); + assert_version_error(former_format_directory().atomic_read(path)); + assert_version_error(former_format_directory().atomic_write(path, b"metadata")); + assert_version_error(former_format_directory().open_write(path)); + assert_version_error(former_format_directory().delete(path)); + assert_version_error(former_format_directory().sync_directory()); + assert_version_error(former_format_directory().open_read(path)); + } + #[test] fn rejects_an_unknown_directory_format() { let store = FaultingKv::default(); @@ -1449,7 +1753,7 @@ mod tests { let error = FaultingDirectory::with_namespace(store, b"catalog") .exists(Path::new("meta.json")) .unwrap_err(); - assert!(error.to_string().contains("version 2")); + assert!(error.to_string().contains("version 3")); } #[test] @@ -1510,6 +1814,8 @@ mod tests { assert_eq!(binding.full_chunks, 3); assert_eq!(binding.tail_length, 17); assert_eq!(binding.visible_length, bytes.len()); + assert_eq!(binding.chunk_high_water, CHUNK_RESERVATION_STRIDE); + assert_eq!(binding.tail_high_water, binding.tail_revision); assert_eq!( directory .open_read(Path::new("segment")) @@ -1533,6 +1839,9 @@ mod tests { 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); + assert_eq!(binding.chunk_high_water, CHUNK_RESERVATION_STRIDE); + assert_eq!(binding.tail_high_water, binding.tail_revision); + assert_eq!(binding.tail_revision, 1); let file = directory.open_read(Path::new("segment")).unwrap(); assert_eq!( file.read_bytes_slice(0..CHUNK_SIZE).unwrap().as_slice(), @@ -1540,6 +1849,44 @@ mod tests { ); } + #[test] + fn chunk_reservations_amortize_binding_io() { + let store = CountingKv::new(); + let directory = KvDirectory::new(store.clone()); + let mut writer = directory.open_write(Path::new("segment")).unwrap(); + store.take_io_counts(); + writer.write_all(&vec![7; CHUNK_SIZE * 65 + 17]).unwrap(); + writer.flush().unwrap(); + + assert_eq!(store.take_io_counts(), (3, 68, 69)); + let binding = decode_binding(&store.inner.get(&binding_key(b"phase0", Path::new("segment"))).unwrap()).unwrap(); + assert_eq!(binding.full_chunks, 65); + assert_eq!(binding.chunk_high_water, 128); + assert_eq!(binding.tail_length, 17); + assert_every_wal_prefix_respects_high_waters(&store.inner, b"phase0", Path::new("segment")); + } + + #[test] + fn physical_keys_stay_within_binding_high_waters_across_reopen() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + let path = Path::new("segment"); + let mut writer = directory.open_write(path).unwrap(); + writer.write_all(&vec![3; CHUNK_SIZE * 2 + 17]).unwrap(); + writer.flush().unwrap(); + writer.write_all(b"next revision").unwrap(); + writer.flush().unwrap(); + assert_physical_keys_within_high_water(&store, b"phase0", path); + + directory + .atomic_write(Path::new("meta.json"), b"durability barrier") + .unwrap(); + let recovered = store.crash(); + assert_physical_keys_within_high_water(&recovered, b"phase0", path); + let reopened = FaultingDirectory::new(recovered); + assert_eq!(reopened.open_read(path).unwrap().len(), CHUNK_SIZE * 2 + 30); + } + #[test] fn range_reads_fetch_only_intersecting_chunks() { let store = CountingKv::new(); @@ -1620,16 +1967,49 @@ mod tests { } #[test] - fn staging_failure_does_not_consume_the_failing_write() { + fn applied_but_reported_failed_chunk_reservation_and_put_are_retryable() { 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, + let binding = empty_binding(); + store + .write( + &[Mutation::Put( + binding_key(b"phase0", Path::new("segment")), + encode_binding(&binding), + )], + WritePolicy::WAL, + ) + .unwrap(); + let mut writer = KvWriter { + store: store.clone(), + state: directory.state.clone(), + namespace: directory.namespace.clone(), + path: PathBuf::from("segment"), + binding, + tail: vec![7; CHUNK_SIZE], + staged_full_chunks: 0, + dirty: true, + pending_publication: None, }; + + store.fail_after_next_write(); + assert!(writer.flush().is_err()); + let reserved = decode_binding(&store.get(&binding_key(b"phase0", Path::new("segment"))).unwrap()).unwrap(); + assert_eq!(reserved.chunk_high_water, CHUNK_RESERVATION_STRIDE); + store.fail_after_next_write(); + assert!(writer.flush().is_err()); + assert_eq!(store.get(&chunk_key(b"phase0", 1, 0)), Some(vec![7; CHUNK_SIZE])); + writer.flush().unwrap(); + + let file = directory.open_read(Path::new("segment")).unwrap(); + assert_eq!(file.read_bytes().unwrap().as_slice(), vec![7; CHUNK_SIZE]); + } + + #[test] + fn staging_failure_does_not_consume_the_failing_write() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + let binding = empty_binding(); store .write( &[Mutation::Put( @@ -1648,6 +2028,7 @@ mod tests { tail: Vec::new(), staged_full_chunks: 0, dirty: false, + pending_publication: None, }; let prefix = vec![1; CHUNK_SIZE - 4_096]; let suffix = vec![2; 8_192]; @@ -1664,18 +2045,43 @@ mod tests { #[test] fn rejects_an_older_binding_format_with_its_version() { - let error = decode_binding(&[1]).unwrap_err(); + let error = decode_binding(&[2; 33]).unwrap_err(); assert_eq!(error.kind(), io::ErrorKind::InvalidData); - assert!(error.to_string().contains("version 1")); + assert!(error.to_string().contains("version 2")); } #[test] fn distinguishes_a_malformed_current_binding() { - let error = decode_binding(&[2; 20]).unwrap_err(); + let error = decode_binding(&[BINDING_FORMAT_VERSION; 20]).unwrap_err(); assert_eq!(error.kind(), io::ErrorKind::InvalidData); assert_eq!(error.to_string(), "malformed binding"); } + #[test] + fn rejects_binding_high_waters_below_published_state() { + let mut chunk = empty_binding(); + chunk.full_chunks = 1; + chunk.visible_length = CHUNK_SIZE; + let error = decode_binding(&encode_binding(&chunk)).unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); + assert!(error.to_string().contains("high-water is below")); + + let mut tail = empty_binding(); + tail.tail_revision = 1; + let error = decode_binding(&encode_binding(&tail)).unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); + assert!(error.to_string().contains("high-water is below")); + } + + #[test] + fn rejects_binding_high_waters_beyond_the_extent_limit() { + let mut binding = empty_binding(); + binding.chunk_high_water = u32::MAX; + let error = decode_binding(&encode_binding(&binding)).unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); + assert!(error.to_string().contains("physical extent exceeds")); + } + #[test] fn directory_supports_a_real_tantivy_lifecycle_and_crash_reopen() { let store = FaultingKv::default();