diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index fd22a20..630c440 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -27,8 +27,8 @@ of `npm pack` into a temporary consumer project. - Keep search, indexing, scheduling, and lifecycle behavior shared across storage backends. - Keep backend choice explicit; do not introduce automatic fallback. -- Do not link RocksDB into this addon. The future Rocks backend must use the versioned lease owned - by rocksdb-js. +- Do not link RocksDB into this addon. Harper storage uses the bounded host transport and + Harper-owned RocksDB operations; the historical rocksdb-js lease remains test-only. - Do not expose generated Node-API declarations as the public TypeScript API. - Keep CPU and sustained I/O work off the Node.js event loop. The promise-shaped capability call may synchronously load the addon once; no search, indexing, commit, or storage operation gets that diff --git a/docs/phase-0-rocks-bridge-plan.md b/docs/phase-0-rocks-bridge-plan.md index ec1b90e..e56d91e 100644 --- a/docs/phase-0-rocks-bridge-plan.md +++ b/docs/phase-0-rocks-bridge-plan.md @@ -160,12 +160,31 @@ before the persisted format is declared stable. ```text namespace / index generation - 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 + 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 + atomic kind: logical path -> complete small-file bytes ``` +The keyspace uses a fixed magic, format version, length-prefixed namespace, and one-byte key-kind +tag before kind-specific bytes. Namespace and path delimiters therefore cannot alias another +index's keys. An unversioned, length-prefixed format-marker key records the active key version so a +future implementation detects an unsupported format instead of opening an empty parallel keyspace. +The unreleased delimiter-based prototype is rejected when its counter or known atomic metadata +sentinels exist; it is rebuilt rather than migrated. Read-only access validates but does not create +the marker, while the first write creates it durably before any payload. The sentinel check is a +prototype guard, not a general mixed-keyspace detector: backup restore replaces a closed generation +and its storage incarnation rather than merging bytes into a live namespace. After one clean +marker-less probe, read-only calls continue checking the marker but do not repeat the prototype +sentinel reads. + +A key-format version change requires dropping the old namespace storage and rebuilding the derived +generation from Harper source data; changing the version prefix alone would strand old payload. +New logical key kinds, including reclamation metadata, receive new kind tags under the existing key +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 diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md new file mode 100644 index 0000000..d0f3b81 --- /dev/null +++ b/docs/reclamation-plan.md @@ -0,0 +1,166 @@ +# Bounded Directory reclamation + +Issue: [Implement Harper-backed Tantivy Directory mapping and atomic publication #11](https://github.com/HarperFast/fulltext/issues/11). + +Reclamation may delete a chunk or tail only when no current logical binding, open file handle, or +active writer can reference it. It must also discover objects abandoned before their first +publication, remain bounded at hundreds of millions of source records, and never make search +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. + +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 +owns source replay and exclusive index-writer lifecycle; it does not infer Tantivy object +reachability. + +## Design + +### 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. + +`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: +it persists a managed path before calling `open_write()`, and garbage collection calls the +underlying `delete()` before removing that path from `.managed.json`. A crash-abandoned writer +therefore remains named by its binding and managed path until it is placed on the reclaim FIFO. + +When `flush()` supersedes a non-empty tail, its publication batch appends a tail-only reclaim entry +for the old revision. This avoids deleting bytes needed by an open handle and prevents an active, +long-lived logical file from accumulating tail revisions until final deletion. Queue entries are +the durable consequence of the transition that made the bytes unreachable, not a second source-data +journal. + +The FIFO is split into a fixed number of shards selected by object id. Each shard has its own head, +tail, sequence-addressed entries, and enqueue mutex; unrelated publishers do not wait on one global +host mutation. Enqueue and tail advance share the binding publication or deletion batch. Dequeue +progress and payload deletes share one WAL-only batch; final entry deletion and head advance share +one WAL-only batch. Missing payload keys are normal, and all operations are idempotent after crash. +Reclamation uses only existing point reads and atomic batch writes, so `KvStore`, the host protocol, +the TypeScript handler, and rocksdb-js gain no new primitive. + +### Protect handles and writers + +Replace the directory-wide mutation mutex with per-path shared/exclusive state plus a dedicated +allocator mutex for the read-modify-write object counter and per-shard FIFO-enqueue mutexes. The +path registry performs an O(1) lookup and never runs a whole-map retain on the indexing or reload +path. File-handle opens take a shared path gate while reading and registering the immutable binding; +opens on the same file remain concurrent. Chunk staging, flush, delete, and replacement take the +exclusive side, preventing delete from capturing a high-water while the old writer creates another +chunk. A deleted writer is marked retired so a later write cannot stage new payload. + +Pins are indexed by object id, not logical path, so delete/recreate cannot hide handles to the prior +object. A handle owns an `Arc` pin, and drop only decrements the `Arc`. Pin insertion prunes dead weak +entries above a small fixed threshold, and FIFO visits prune them again. This bounds control-block +retention during repeated open/drop churn. Writer state is likewise retained through failed and +unterminated writes. Gates recover poisoned state rather than panicking the writer actor. + +Cleanup never holds a path gate or pin-registry lock across storage I/O. A retired object cannot gain +a new pin because its binding is gone; a tail-only entry can gain no new pin for its old revision +after the newer binding is published. Whole-object entries match any pin for that object, while +tail-only entries match only the exact object id and tail revision. If a matching pin remains, the +entry is rotated to its shard's tail with bounded backoff rather than blocking later garbage. + +This process-local pin model is valid only while Harper's derived-index lifecycle guarantees one +active generation owner. Cleanup requires an explicit owner lease supplied by that lifecycle; the +production host cannot invoke it without one. Admission checks the lease before work, and worker +loss or close revokes admission, fails the transport before joining, waits for any definitive +mutation result, and drains the cleanup task. The Directory does not invent a second cross-process +lock or compare-and-set protocol that the supported storage contract cannot enforce. + +### Reclaim with bounded point operations + +The worker point-reads only the FIFO head, its current entry, and bounded progress. A whole-object +entry deletes derived chunk and tail ranges over as many batches as necessary. A tail-only entry +deletes one derived key. A pinned entry rotates behind other work; repeated rotations are rate-limited +and surface a blocked-reclamation health state rather than consuming the JavaScript service thread. +Cleanup cost is proportional to garbage queued, never to object ids or records ever created. + +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 rotations, and no-progress state 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 +guessed from point misses: this is a derived index, so the generation is marked corrupt and rebuilt +from Harper source data. A caught cleanup panic enters the same terminal health state. Neither case +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. | + +## Persistence format and delivery sequence + +Namespace keys require a versioned, length-prefixed encoding so delimiter-containing namespaces +cannot alias. Because that changes the prototype format, land it as a focused prerequisite change +with an explicit format marker and rejection of old prototype data; no migration is promised for +the unreleased prototype. Do not silently create an empty new-format index beside old keys. + +Then deliver reclamation in reviewable slices: + +1. versioned namespace encoding and format validation; +2. binding high-waters and atomic updates at chunk staging and flush; update the writer replacement + guard and in-memory binding together so high-water-only changes cannot look like path replacement; +3. per-path shared/exclusive state, object-id/revision pins, and atomic sharded-FIFO enqueue at tail + supersession and object deletion; key writer retirement by object id so path reuse cannot retire + the replacement writer; +4. bounded FIFO draining, low-priority admission, failure observability, close fencing, crash + recovery, and Harper's + exclusive-owner integration. + +The production Harper package remains disabled until slice 4's lifecycle fence and Harper host- +storage integration both pass. No native RocksDB storage provider is added to the public library. + +## 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 +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, +payload deletion, progress updates, rotation, and head advance; every crash/reopen result must expose +the old complete binding, the new complete binding, or logical absence—never a reference to missing +bytes. + +Efficacy is measured as well as safety: after repeated real Tantivy merge/delete cycles, physical +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. + +The FIFO tests include parallel enqueue on different shards, sequence/batch failure, pinned-entry +rotation cost, corrupt binding and entry handling, independent namespaces on one store, and a +numeric post-drain bound. Host tests drain to quiescence, assert at least one entry was reclaimed, +and run foreground reads with cleanup occupying every cleanup-eligible transport slot. + +Run formatting, lint, all-feature Rust tests, Node tests, packed-package checks, and local Claude and +Gemini reviews before opening each implementation PR. diff --git a/src/phase0.rs b/src/phase0.rs index c16dc60..61cf8c7 100644 --- a/src/phase0.rs +++ b/src/phase0.rs @@ -4,7 +4,7 @@ use std::io; use std::io::Write; use std::ops::Range; use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicU64, AtomicU8, Ordering}; use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak}; use std::time::{Duration, Instant}; @@ -43,6 +43,7 @@ pub enum Mutation { Delete(Vec), } +/// Identifies one storage incarnation; providers must mint a new value after close, restore, or replacement. #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub struct KvStoreIdentity(pub u64, pub u64, pub u64); @@ -237,6 +238,7 @@ fn apply_if_newer(entries: &mut BTreeMap, VersionedValue>, key: Vec, pub struct KvDirectory { store: S, namespace: Arc<[u8]>, + format: Arc, state: Arc, } @@ -264,6 +266,24 @@ struct DirectoryIdentity { static DIRECTORY_STATES: OnceLock>>> = OnceLock::new(); +struct FormatValidation { + state: AtomicU8, +} + +impl Default for FormatValidation { + fn default() -> Self { + Self { + state: AtomicU8::new(FORMAT_UNKNOWN), + } + } +} + +impl FormatValidation { + fn record(&self, state: u8) { + self.state.fetch_max(state, Ordering::AcqRel); + } +} + impl fmt::Debug for KvDirectory { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter.write_str("KvDirectory") @@ -298,14 +318,33 @@ impl KvDirectory { Self { store, namespace: Arc::from(namespace), + format: Arc::new(FormatValidation::default()), state, } } + fn ensure_format(&self, create: bool) -> io::Result<()> { + let state = self.format.state.load(Ordering::Acquire); + if state == FORMAT_PRESENT { + return Ok(()); + } + if state == FORMAT_ABSENT && !create { + if let Some(state) = read_format_marker(&self.store, &format_marker_key(&self.namespace))? { + self.format.record(state); + } + return Ok(()); + } + let state = validate_format(&self.store, &self.namespace, create)?; + self.format.record(state); + Ok(()) + } + fn read_binding(&self, path: &Path) -> Result where S: KvStore, { + self.ensure_format(false) + .map_err(|error| OpenReadError::wrap_io_error(error, path.to_path_buf()))?; let bytes = self .store .read(&binding_key(&self.namespace, path)) @@ -456,6 +495,10 @@ impl Directory for KvDirectory { } fn delete(&self, path: &Path) -> Result<(), DeleteError> { + self.ensure_format(false).map_err(|error| DeleteError::IoError { + io_error: Arc::new(error), + filepath: path.to_path_buf(), + })?; let _mutation = self.state.mutation.lock().unwrap(); let binding = binding_key(&self.namespace, path); let atomic = atomic_key(&self.namespace, path); @@ -487,6 +530,8 @@ impl Directory for KvDirectory { } fn exists(&self, path: &Path) -> Result { + self.ensure_format(false) + .map_err(|error| OpenReadError::wrap_io_error(error, path.to_path_buf()))?; Ok(self .store .read(&binding_key(&self.namespace, path)) @@ -501,6 +546,8 @@ impl Directory for KvDirectory { fn open_write(&self, path: &Path) -> Result { let _mutation = self.state.mutation.lock().unwrap(); + self.ensure_format(true) + .map_err(|error| OpenWriteError::wrap_io_error(error, path.to_path_buf()))?; let key = binding_key(&self.namespace, path); if self .store @@ -555,6 +602,8 @@ impl Directory for KvDirectory { } fn atomic_read(&self, path: &Path) -> Result, OpenReadError> { + self.ensure_format(false) + .map_err(|error| OpenReadError::wrap_io_error(error, path.to_path_buf()))?; self.store .read(&atomic_key(&self.namespace, path)) .map_err(|error| OpenReadError::wrap_io_error(error, path.to_path_buf()))? @@ -565,6 +614,7 @@ impl Directory for KvDirectory { fn atomic_write(&self, path: &Path, data: &[u8]) -> io::Result<()> { { let _mutation = self.state.mutation.lock().unwrap(); + self.ensure_format(true)?; self.store.write( &[Mutation::Put(atomic_key(&self.namespace, path), data.to_vec())], WritePolicy::WAL_SYNC, @@ -577,6 +627,7 @@ impl Directory for KvDirectory { } fn sync_directory(&self) -> io::Result<()> { + self.ensure_format(false)?; self.store.sync() } @@ -804,22 +855,104 @@ struct Binding { } fn counter_key(namespace: &[u8]) -> Vec { - namespaced_prefix(namespace, b"counter") + namespaced_prefix(namespace, KEY_KIND_COUNTER) } fn binding_key(namespace: &[u8], path: &Path) -> Vec { - prefixed_path(&namespaced_prefix(namespace, b"binding/"), path) + prefixed_path(&namespaced_prefix(namespace, KEY_KIND_BINDING), path) } fn atomic_key(namespace: &[u8], path: &Path) -> Vec { - prefixed_path(&namespaced_prefix(namespace, b"atomic/"), path) + prefixed_path(&namespaced_prefix(namespace, KEY_KIND_ATOMIC), path) +} + +const KEY_FORMAT_VERSION: u8 = 1; +const KEY_KIND_COUNTER: u8 = 1; +const KEY_KIND_BINDING: u8 = 2; +const KEY_KIND_ATOMIC: u8 = 3; +const KEY_KIND_CHUNK: u8 = 4; +const KEY_KIND_TAIL: u8 = 5; +const KEY_PREFIX: &[u8; 4] = b"HFTK"; +const FORMAT_MARKER_PREFIX: &[u8; 4] = b"HFTM"; +const FORMAT_UNKNOWN: u8 = 0; +const FORMAT_ABSENT: u8 = 1; +const FORMAT_PRESENT: u8 = 2; + +fn namespaced_prefix(namespace: &[u8], kind: u8) -> Vec { + namespaced_prefix_with_capacity(namespace, kind, 0) +} + +fn format_marker_key(namespace: &[u8]) -> Vec { + let mut key = Vec::with_capacity(FORMAT_MARKER_PREFIX.len() + 8 + namespace.len()); + key.extend_from_slice(FORMAT_MARKER_PREFIX); + key.extend_from_slice(&(namespace.len() as u64).to_be_bytes()); + key.extend_from_slice(namespace); + key +} + +fn validate_format(store: &S, namespace: &[u8], create: bool) -> io::Result { + let marker = format_marker_key(namespace); + if let Some(state) = read_format_marker(store, &marker)? { + return Ok(state); + } + for legacy_key in legacy_sentinel_keys(namespace) { + if store.read(&legacy_key)?.is_some() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "the namespace contains the unsupported prototype directory format", + )); + } + } + if !create { + return Ok(FORMAT_ABSENT); + } + store.write( + &[Mutation::Put(marker, vec![KEY_FORMAT_VERSION])], + WritePolicy::WAL_SYNC, + )?; + Ok(FORMAT_PRESENT) +} + +fn read_format_marker(store: &S, marker: &[u8]) -> io::Result> { + let Some(value) = store.read(marker)? else { + return Ok(None); + }; + match value.as_slice() { + [KEY_FORMAT_VERSION] => Ok(Some(FORMAT_PRESENT)), + [version] => Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("unsupported directory format version {version}"), + )), + _ => Err(io::Error::new( + io::ErrorKind::InvalidData, + "malformed directory format marker", + )), + } +} + +fn legacy_sentinel_keys(namespace: &[u8]) -> [Vec; 3] { + [ + legacy_namespaced_key(namespace, b"counter"), + legacy_namespaced_key(namespace, b"atomic/.managed.json"), + legacy_namespaced_key(namespace, b"atomic/meta.json"), + ] +} + +fn legacy_namespaced_key(namespace: &[u8], suffix: &[u8]) -> Vec { + let mut key = Vec::with_capacity(namespace.len() + suffix.len() + 1); + key.extend_from_slice(namespace); + key.push(b'/'); + key.extend_from_slice(suffix); + key } -fn namespaced_prefix(namespace: &[u8], suffix: &[u8]) -> Vec { - let mut prefix = Vec::with_capacity(namespace.len() + suffix.len() + 1); +fn namespaced_prefix_with_capacity(namespace: &[u8], kind: u8, additional: usize) -> Vec { + let mut prefix = Vec::with_capacity(KEY_PREFIX.len() + 10 + namespace.len() + additional); + prefix.extend_from_slice(KEY_PREFIX); + prefix.push(KEY_FORMAT_VERSION); + prefix.extend_from_slice(&(namespace.len() as u64).to_be_bytes()); prefix.extend_from_slice(namespace); - prefix.push(b'/'); - prefix.extend_from_slice(suffix); + prefix.push(kind); prefix } @@ -831,27 +964,19 @@ fn prefixed_path(prefix: &[u8], path: &Path) -> Vec { } fn chunk_key(namespace: &[u8], object_id: u64, chunk: u32) -> Vec { - let mut key = namespaced_prefix_with_capacity(namespace, b"chunk/", 12); + let mut key = namespaced_prefix_with_capacity(namespace, KEY_KIND_CHUNK, 12); key.extend_from_slice(&object_id.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); + let mut key = namespaced_prefix_with_capacity(namespace, KEY_KIND_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(33); bytes.push(2); @@ -922,6 +1047,12 @@ mod tests { reads: Arc, } + #[derive(Clone)] + struct FixedIdentityKv { + inner: FaultingKv, + identity: KvStoreIdentity, + } + impl CountingKv { fn new() -> Self { Self { @@ -954,10 +1085,42 @@ mod tests { } } + impl KvStore for FixedIdentityKv { + fn identity(&self) -> KvStoreIdentity { + self.identity + } + + fn read(&self, key: &[u8]) -> io::Result> { + 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()) } + fn prototype_directory() -> FaultingDirectory { + let store = FaultingKv::default(); + store + .write( + &[Mutation::Put( + legacy_namespaced_key(b"catalog", b"counter"), + 1_u64.to_be_bytes().to_vec(), + )], + WritePolicy::WAL_SYNC, + ) + .unwrap(); + FaultingDirectory::with_namespace(store, b"catalog") + } + #[test] fn synced_metadata_makes_earlier_wal_objects_durable() { let store = FaultingKv::default(); @@ -1129,6 +1292,197 @@ mod tests { assert_eq!(second.atomic_read(Path::new("meta.json")).unwrap(), b"two"); } + #[test] + fn namespace_and_path_delimiters_cannot_alias() { + let first = binding_key(b"a", Path::new("b/binding/x")); + let second = binding_key(b"a/binding/b", Path::new("x")); + assert_ne!(first, second); + assert_ne!(binding_key(b"a", Path::new("x")), atomic_key(b"a", Path::new("x"))); + } + + #[test] + fn format_marker_is_durable_and_reusable() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::with_namespace(store.clone(), b"catalog"); + assert!(!directory.exists(Path::new("missing")).unwrap()); + assert_eq!(store.get(&format_marker_key(b"catalog")), None); + directory.atomic_write(Path::new("meta.json"), b"metadata").unwrap(); + assert_eq!( + store.get_durable(&format_marker_key(b"catalog")), + Some(vec![KEY_FORMAT_VERSION]) + ); + let reopened = FaultingDirectory::with_namespace(store.crash(), b"catalog"); + assert_eq!(reopened.atomic_read(Path::new("meta.json")).unwrap(), b"metadata"); + } + + #[test] + fn format_initialization_failure_is_returned_and_retryable() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::with_namespace(store.clone(), b"catalog"); + store.fail_next_write(); + assert!(directory.atomic_write(Path::new("meta.json"), b"metadata").is_err()); + assert_eq!(store.get(&format_marker_key(b"catalog")), None); + directory.atomic_write(Path::new("meta.json"), b"metadata").unwrap(); + assert_eq!(directory.atomic_read(Path::new("meta.json")).unwrap(), b"metadata"); + } + + #[test] + fn absent_marker_is_revalidated_on_later_reads() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::with_namespace(store.clone(), b"catalog"); + assert!(!directory.exists(Path::new("meta.json")).unwrap()); + store + .write( + &[Mutation::Put( + format_marker_key(b"catalog"), + vec![KEY_FORMAT_VERSION + 1], + )], + WritePolicy::WAL_SYNC, + ) + .unwrap(); + assert!(directory.exists(Path::new("meta.json")).is_err()); + } + + #[test] + fn clean_absent_format_rechecks_only_the_marker() { + let store = CountingKv::new(); + let directory = KvDirectory::with_namespace(store.clone(), b"catalog"); + assert!(!directory.exists(Path::new("meta.json")).unwrap()); + assert_eq!(store.take_reads(), 6); + assert!(!directory.exists(Path::new("meta.json")).unwrap()); + assert_eq!(store.take_reads(), 3); + } + + #[test] + fn format_state_never_moves_backward() { + let format = FormatValidation::default(); + format.record(FORMAT_PRESENT); + format.record(FORMAT_ABSENT); + assert_eq!(format.state.load(Ordering::Acquire), FORMAT_PRESENT); + } + + #[test] + fn reconstructed_store_revalidates_a_reused_identity() { + let identity = KvStoreIdentity(7, 8, 9); + let first_store = FixedIdentityKv { + inner: FaultingKv::default(), + identity, + }; + let first = KvDirectory::with_namespace(first_store, b"catalog"); + first.atomic_write(Path::new("meta.json"), b"metadata").unwrap(); + + let replacement = FaultingKv::default(); + replacement + .write( + &[Mutation::Put( + legacy_namespaced_key(b"catalog", b"counter"), + 1_u64.to_be_bytes().to_vec(), + )], + WritePolicy::WAL_SYNC, + ) + .unwrap(); + let second = KvDirectory::with_namespace( + FixedIdentityKv { + inner: replacement, + identity, + }, + b"catalog", + ); + assert!(second.exists(Path::new("meta.json")).is_err()); + } + + #[test] + fn rejects_the_prototype_key_format() { + for sentinel in [b"counter".as_slice(), b"atomic/.managed.json", b"atomic/meta.json"] { + let store = FaultingKv::default(); + store + .write( + &[Mutation::Put( + legacy_namespaced_key(b"catalog", sentinel), + b"prototype".to_vec(), + )], + WritePolicy::WAL_SYNC, + ) + .unwrap(); + let directory = FaultingDirectory::with_namespace(store, b"catalog"); + let error = directory.exists(Path::new("meta.json")).unwrap_err(); + let OpenReadError::IoError { io_error, .. } = error else { + panic!("prototype format did not return an I/O error"); + }; + assert_eq!(io_error.kind(), io::ErrorKind::InvalidData); + assert!(io_error.to_string().contains("prototype directory format")); + } + } + + #[test] + fn every_storage_entry_point_rejects_the_prototype_format() { + fn assert_prototype_error(result: Result) { + let error = result.err().expect("prototype format should be rejected"); + assert!( + error.to_string().contains("prototype directory format"), + "unexpected error: {error}" + ); + } + + let path = Path::new("meta.json"); + assert_prototype_error(prototype_directory().exists(path)); + assert_prototype_error(prototype_directory().atomic_read(path)); + assert_prototype_error(prototype_directory().atomic_write(path, b"metadata")); + assert_prototype_error(prototype_directory().open_write(path)); + assert_prototype_error(prototype_directory().delete(path)); + assert_prototype_error(prototype_directory().sync_directory()); + assert_prototype_error(prototype_directory().open_read(path)); + } + + #[test] + fn rejects_an_unknown_directory_format() { + let store = FaultingKv::default(); + store + .write( + &[Mutation::Put( + format_marker_key(b"catalog"), + vec![KEY_FORMAT_VERSION + 1], + )], + WritePolicy::WAL_SYNC, + ) + .unwrap(); + let error = FaultingDirectory::with_namespace(store, b"catalog") + .exists(Path::new("meta.json")) + .unwrap_err(); + assert!(error.to_string().contains("version 2")); + } + + #[test] + fn rejects_malformed_directory_format_markers() { + for value in [Vec::new(), vec![KEY_FORMAT_VERSION, 0]] { + let store = FaultingKv::default(); + store + .write( + &[Mutation::Put(format_marker_key(b"catalog"), value)], + WritePolicy::WAL_SYNC, + ) + .unwrap(); + let error = FaultingDirectory::with_namespace(store, b"catalog") + .exists(Path::new("meta.json")) + .unwrap_err(); + assert!(error.to_string().contains("malformed directory format marker")); + } + } + + #[test] + fn independent_directories_converge_on_one_format() { + let store = FaultingKv::default(); + let first = FaultingDirectory::with_namespace(store.clone(), b"catalog"); + let second = FaultingDirectory::with_namespace(store.clone(), b"catalog"); + let first_write = std::thread::spawn(move || first.atomic_write(Path::new("one"), b"one")); + let second_write = std::thread::spawn(move || second.atomic_write(Path::new("two"), b"two")); + first_write.join().unwrap().unwrap(); + second_write.join().unwrap().unwrap(); + let directory = FaultingDirectory::with_namespace(store, b"catalog"); + assert_eq!(directory.atomic_read(Path::new("one")).unwrap(), b"one"); + assert_eq!(directory.atomic_read(Path::new("two")).unwrap(), b"two"); + } + #[test] fn file_handle_reads_across_flush_boundaries() { let directory = FaultingDirectory::new(FaultingKv::default());