diff --git a/README.md b/README.md
index 5e78487..e23e3cd 100644
--- a/README.md
+++ b/README.md
@@ -98,14 +98,18 @@ upserts include delete terms. CI runs only the correctness smoke profile; timing
 require controlled hardware.
 
 The `kv-directory` benchmark measures the caller-visible buffered write path, empty and dirty
-flushes, 256 KiB chunk publication, closed- and active-writer deletion, and distinct-file
-concurrency at one, two, four, and eight threads. It reports percentiles across per-sample mean
-latencies and uses the deterministic in-memory Phase 0 store to isolate directory coordination from
-RocksDB and Node transport costs. That store serializes access, so the concurrency cases detect
-coordination regressions but do not predict RocksDB scaling. Compare two optimized builds on the
-same quiet host; records include the Git revision and dirty state, while `--revision` can add a run
-label and `--samples` and `--warmup` control the run. CI executes only `--smoke`, whose timings are
-not comparable to a full run, and applies no timing threshold.
+flushes, 256 KiB chunk publication, closed- and active-writer deletion, retained and churned read
+handle opens, distinct-file read-open and write concurrency at one, two, four, and eight threads,
+and shared-file read-open concurrency at two, four, and eight threads. It reports percentiles across
+per-sample mean latencies and uses a deterministic in-memory store to isolate directory coordination
+from RocksDB and Node transport costs. The store permits concurrent point reads but serializes
+writes, so read-open cases isolate registration contention while write cases detect coordination
+regressions; neither predicts RocksDB scaling. The `-rw-` read-open case names establish a new
+comparison series and must not be compared with results from the earlier serialized-store cases.
+Compare two optimized builds on the same quiet host; records include the Git revision and dirty
+state, while `--revision` can add a run label and `--samples` and `--warmup` control the run. CI
+executes only `--smoke`, whose timings are not comparable to a full run, and applies no timing
+threshold.
 
 Generated Node-API declarations in `ts/addon.d.ts` are private implementation types. Consumers use
 only the types exported from a package entry point.
diff --git a/benches/kv_directory.rs b/benches/kv_directory.rs
index 82faac5..f94cddc 100644
--- a/benches/kv_directory.rs
+++ b/benches/kv_directory.rs
@@ -1,3 +1,4 @@
+use std::collections::BTreeMap;
 use std::env;
 use std::fmt::Write as FmtWrite;
 use std::hint::black_box;
@@ -5,23 +6,78 @@ use std::io;
 use std::io::Write as IoWrite;
 use std::path::Path;
 use std::process::Command;
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use std::sync::mpsc;
-use std::sync::{Arc, Condvar, Mutex, OnceLock};
+use std::sync::{Arc, Condvar, Mutex, OnceLock, RwLock};
 use std::time::{Instant, SystemTime, UNIX_EPOCH};
 
-use harper_fulltext::phase0::{FaultingDirectory, FaultingKv};
+use harper_fulltext::phase0::{
+	FaultingDirectory, FaultingKv, KvDirectory, KvStore, KvStoreIdentity, Mutation, WritePolicy,
+};
 use harper_fulltext::TANTIVY_VERSION;
-use tantivy::directory::{Directory, TerminatingWrite, WritePtr};
+use tantivy::directory::{Directory, OwnedBytes, TerminatingWrite, WritePtr};
 
 const WRITE_BYTES_PER_SAMPLE: usize = 128 * 1024;
 const MIN_WRITE_OPERATIONS_PER_SAMPLE: usize = 256;
 const CHUNK_BYTES: usize = 256 * 1024;
 const STORAGE_OPERATIONS_PER_SAMPLE: usize = 64;
 const EMPTY_FLUSHES_PER_SAMPLE: usize = 10_000;
+const RETAINED_OPEN_READS_PER_SAMPLE: usize = 1_024;
+const OPEN_READS_PER_SAMPLE: usize = 10_000;
 const CONCURRENT_BYTES_PER_FILE: usize = 4 * 1024;
 const CONCURRENT_FILES_PER_THREAD: usize = 64;
 
+static NEXT_BENCHMARK_KV_IDENTITY: AtomicU64 = AtomicU64::new(1);
+
+#[derive(Clone)]
+struct BenchmarkKv {
+	identity: u64,
+	values: Arc<RwLock<BTreeMap<Vec<u8>, Vec<u8>>>>,
+}
+
+type BenchmarkDirectory = KvDirectory<BenchmarkKv>;
+
+impl Default for BenchmarkKv {
+	fn default() -> Self {
+		let identity = NEXT_BENCHMARK_KV_IDENTITY.fetch_add(1, Ordering::Relaxed);
+		assert_ne!(identity, 0, "benchmark store identity space exhausted");
+		Self {
+			identity,
+			values: Arc::new(RwLock::new(BTreeMap::new())),
+		}
+	}
+}
+
+impl KvStore for BenchmarkKv {
+	fn identity(&self) -> KvStoreIdentity {
+		KvStoreIdentity(1, self.identity, 0)
+	}
+
+	fn read(&self, key: &[u8]) -> io::Result<Option<OwnedBytes>> {
+		let values = self.values.read().unwrap_or_else(|poisoned| poisoned.into_inner());
+		Ok(values.get(key).cloned().map(OwnedBytes::new))
+	}
+
+	fn write(&self, mutations: &[Mutation], _policy: WritePolicy) -> io::Result<()> {
+		let mut values = self.values.write().unwrap_or_else(|poisoned| poisoned.into_inner());
+		for mutation in mutations {
+			match mutation {
+				Mutation::Put(key, value) => {
+					values.insert(key.clone(), value.clone());
+				}
+				Mutation::Delete(key) => {
+					values.remove(key);
+				}
+			}
+		}
+		Ok(())
+	}
+
+	fn sync(&self) -> io::Result<()> {
+		Ok(())
+	}
+}
+
 struct Arguments {
 	samples: usize,
 	warmup_samples: usize,
@@ -185,6 +241,38 @@ fn main() -> io::Result<()> {
 		&arguments,
 		delete_case(true, arguments.smoke),
 	)?);
+	results.push(measure_case(
+		"open-read-retained".to_owned(),
+		"open-read",
+		1,
+		&arguments,
+		open_read_case(true, arguments.smoke),
+	)?);
+	results.push(measure_case(
+		"open-read-churn".to_owned(),
+		"open-read",
+		1,
+		&arguments,
+		open_read_case(false, arguments.smoke),
+	)?);
+	for threads in [1, 2, 4, 8] {
+		results.push(measure_case(
+			format!("concurrent-open-read-rw-{threads}t"),
+			"open-read",
+			threads,
+			&arguments,
+			concurrent_open_read_case(threads, arguments.smoke, false),
+		)?);
+	}
+	for threads in [2, 4, 8] {
+		results.push(measure_case(
+			format!("concurrent-shared-open-read-rw-{threads}t"),
+			"open-read",
+			threads,
+			&arguments,
+			concurrent_open_read_case(threads, arguments.smoke, true),
+		)?);
+	}
 
 	for threads in [1, 2, 4, 8] {
 		results.push(measure_case(
@@ -456,6 +544,143 @@ fn delete_case(active_writer: bool, smoke: bool) -> impl FnMut(usize) -> io::Res
 	}
 }
 
+fn open_read_case(retain_handles: bool, smoke: bool) -> impl FnMut(usize) -> io::Result<Sample> {
+	move |sample| {
+		let directory = FaultingDirectory::new(FaultingKv::default());
+		let operations = if smoke {
+			4
+		} else if retain_handles {
+			RETAINED_OPEN_READS_PER_SAMPLE
+		} else {
+			OPEN_READS_PER_SAMPLE
+		};
+		let file_count = if retain_handles { operations } else { 1 };
+		let mut paths = Vec::with_capacity(file_count);
+		for file in 0..file_count {
+			let path = format!("open-read-{retain_handles}-{sample}-{file}");
+			let mut writer = open_writer(&directory, Path::new(&path))?;
+			writer.write_all(b"contents")?;
+			writer.terminate()?;
+			paths.push(path);
+		}
+		let mut handles = Vec::with_capacity(if retain_handles { operations } else { 1 });
+		let started = Instant::now();
+		for operation in 0..operations {
+			let path = &paths[operation % file_count];
+			let handle = directory
+				.open_read(Path::new(path))
+				.map_err(|error| io::Error::other(error.to_string()))?;
+			if retain_handles {
+				handles.push(handle);
+			} else {
+				black_box(handle);
+			}
+		}
+		let elapsed_nanoseconds = started.elapsed().as_nanos();
+		black_box(handles);
+		Ok(Sample {
+			elapsed_nanoseconds,
+			operations: operations as u64,
+			bytes: 0,
+		})
+	}
+}
+
+fn concurrent_open_read_case(
+	threads: usize,
+	smoke: bool,
+	shared_paths: bool,
+) -> impl FnMut(usize) -> io::Result<Sample> {
+	let files_per_thread = if smoke { 2 } else { CONCURRENT_FILES_PER_THREAD };
+	move |sample| {
+		let directory = BenchmarkDirectory::new(BenchmarkKv::default());
+		let mut paths = Vec::with_capacity(threads);
+		let path_groups = if shared_paths { 1 } else { threads };
+		for thread in 0..path_groups {
+			let mut thread_paths = Vec::with_capacity(files_per_thread);
+			for file in 0..files_per_thread {
+				let path = format!("concurrent-open-read-{shared_paths}-{sample}-{thread}-{file}");
+				let mut writer = open_writer(&directory, Path::new(&path))?;
+				writer.write_all(b"contents")?;
+				writer.terminate()?;
+				thread_paths.push(path);
+			}
+			paths.push(thread_paths);
+		}
+		if shared_paths {
+			let shared = paths[0].clone();
+			paths.resize_with(threads, || shared.clone());
+		}
+		let start_gate = Arc::new(StartGate::new());
+		let remaining = Arc::new(AtomicUsize::new(threads));
+		let (completion, completed) = mpsc::sync_channel(1);
+		let elapsed_nanoseconds = std::thread::scope(|scope| -> io::Result<u128> {
+			let mut handles = Vec::with_capacity(threads);
+			for (thread, thread_paths) in paths.into_iter().enumerate() {
+				let opened = Vec::with_capacity(thread_paths.len());
+				let directory = directory.clone();
+				let worker_start_gate = start_gate.clone();
+				let remaining = remaining.clone();
+				let completion = completion.clone();
+				let handle = std::thread::Builder::new()
+					.name(format!("kv-directory-open-read-benchmark-{thread}"))
+					.spawn_scoped(scope, move || {
+						let Some(started) = worker_start_gate.wait() else {
+							return Ok(Vec::new());
+						};
+						let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || {
+							let mut opened = opened;
+							for path in thread_paths {
+								opened.push(
+									directory
+										.open_read(Path::new(&path))
+										.map_err(|error| io::Error::other(error.to_string()))?,
+								);
+							}
+							black_box(&opened);
+							Ok::<_, io::Error>(opened)
+						}))
+						.unwrap_or_else(|_| Err(io::Error::other("benchmark worker panicked")));
+						if remaining.fetch_sub(1, Ordering::AcqRel) == 1 {
+							let _ = completion.send(started.elapsed().as_nanos());
+						}
+						result
+					});
+				match handle {
+					Ok(handle) => handles.push(handle),
+					Err(error) => {
+						start_gate.cancel();
+						for handle in handles {
+							handle
+								.join()
+								.map_err(|_| io::Error::other("benchmark worker panicked"))??;
+						}
+						return Err(error);
+					}
+				}
+			}
+			start_gate.start(threads);
+			drop(completion);
+			let elapsed_nanoseconds = completed
+				.recv()
+				.map_err(|_| io::Error::other("benchmark workers did not report completion"))?;
+			for handle in handles {
+				let opened = handle
+					.join()
+					.map_err(|_| io::Error::other("benchmark worker panicked"))??;
+				black_box(opened);
+			}
+			Ok(elapsed_nanoseconds)
+		})?;
+		let operations = threads * files_per_thread;
+		Ok(Sample {
+			elapsed_nanoseconds,
+			operations: operations as u64,
+			bytes: 0,
+		})
+	}
+}
+
 fn concurrent_case(threads: usize, smoke: bool) -> impl FnMut(usize) -> io::Result<Sample> {
 	let files_per_thread = if smoke { 2 } else { CONCURRENT_FILES_PER_THREAD };
 	move |sample| {
@@ -528,7 +753,7 @@ fn concurrent_case(threads: usize, smoke: bool) -> impl FnMut(usize) -> io::Resu
 	}
 }
 
-fn open_writer(directory: &FaultingDirectory, path: &Path) -> io::Result<WritePtr> {
+fn open_writer(directory: &impl Directory, path: &Path) -> io::Result<WritePtr> {
 	directory
 		.open_write(path)
 		.map_err(|error| io::Error::other(error.to_string()))
diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md
index 5397af9..7c36a8e 100644
--- a/docs/reclamation-plan.md
+++ b/docs/reclamation-plan.md
@@ -9,13 +9,13 @@ readiness wait for a complete sweep.
 
 ## Grounding
 
-This plan is written against fulltext main at `a982e8a`, the rebase-merged result of
-[Track directory object high-water marks #25](https://github.com/HarperFast/fulltext/pull/25).
-`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 `36d96e3`, the rebase-merged result of
+[Enqueue deleted directory objects for reclamation #26](https://github.com/HarperFast/fulltext/pull/26).
+`KvDirectory` stores immutable 256 KiB chunks and revisioned tails. `delete()` atomically removes a
+logical binding and enqueues the object's derivable physical extent, but no consumer deletes those
+payloads yet. Each successful `flush()` can leave the previous tail revision unreachable. `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
@@ -114,23 +114,45 @@ an unrelated timeout. Closing the host wakes pending operations and releases the
 gate is held by chunk staging or publication across a host round trip.
 
 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. Pins land
-with the FIFO consumer in slice 4, where a test can prove that they prevent payload deletion. Harper
-may enable cleanup only after a two-worker test proves that every local search handle for one
-RocksDB-backed generation reaches the same native `DirectoryState`. A process that cannot establish
-that invariant cannot obtain the cleanup owner lease.
-
-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,
-cleanup moves the entry to a cleanup-only deferred queue with bounded backoff rather than blocking
-later garbage. Foreground publishers only mutate the ingress tail; cleanup only mutates the ingress
-head and deferred queue. A low-priority cleanup host operation therefore never owns the mutex or
-counter needed by a foreground enqueue.
+object. Each object has counted pins by tail revision. A handle owns one pin and retains the shared
+directory state; dropping it decrements the revision and object counts, and dropping the last pin
+removes the object's weak registry entry. Registry shards use the same object-id partition as
+reclamation, while revision counts use a per-object mutex and keep the common single-revision case
+inline. This bounds retained control state by live handles during repeated object, revision, and
+open/drop churn. Writer state is likewise retained through failed and unterminated writes. Gates
+recover poisoned state rather than panicking the writer actor. Pins land in the first slice 4 unit
+before the FIFO consumer. A handle reads its
+binding and registers the pin under the shared side of a fixed hash-sharded registration fence.
+Deletion takes the exclusive side of the same shard only for its final binding-removal batch, so
+concurrent opens remain parallel except for hash collisions and writer retirement does not block
+them. This preserves one storage read per handle open and adds no directory-wide coordination. The
+consumer follows only after these lifetime rules are independently covered. Harper may enable
+cleanup only after a two-worker test proves that every local search handle for one RocksDB-backed
+generation reaches the same native `DirectoryState`. A process that cannot establish that invariant
+cannot obtain the cleanup owner lease.
+
+The blocking fence deliberately spans a reader's binding read and pin registration and deletion's
+single atomic host write. Ending either side earlier reopens the registration race. A stalled delete
+can therefore delay an unrelated open whose path collides in the same one of 256 shards. Replacing
+this bounded collision risk requires a quiescence or epoch protocol that prevents cleanup until all
+readers that started before publication have either registered or exited; it is not a safe lock
+substitution. Conversely, a stalled colliding reader can hold deletion behind its binding read;
+because deletion already owns that path's lifecycle gate, this can also delay an `open_write()` for
+the path being deleted. The host transport's definitive-completion contract bounds this only when
+the host operation completes or the transport closes. The initial 256-shard count bounds memory and
+keeps collision probability low; it is not yet validated against Harper host-read p99 and must be
+measured before cleanup is enabled.
+
+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. Before tail-only reclamation is enabled, publication of
+the newer binding and its superseded-tail entry must take the exclusive side of the same
+registration fence used by deletion. That prevents a reader from registering the old revision
+after publication. 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, cleanup moves the entry
+to a cleanup-only deferred queue with bounded backoff rather than blocking later garbage.
+Foreground publishers only mutate the ingress tail; cleanup only mutates the ingress head and
+deferred queue. A low-priority cleanup host operation therefore never owns the mutex or counter
+needed by a foreground enqueue.
 
 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
@@ -248,11 +270,13 @@ Then deliver reclamation in reviewable slices:
 3. per-path lifecycle state, a strided object-id allocator, an atomic writer fence, and sharded-FIFO
    enqueue at object deletion; key writer retirement by object id so path reuse cannot retire the
    replacement writer;
-4. measure and, if justified, add tail-supersession entries; then add object-id/revision pins,
-   bounded FIFO draining and deferred queues, low-priority admission, failure observability, close
-   fencing, crash recovery, and Harper's exclusive-owner integration.
+4. add object-id/revision reader pins and coordinate registration with deletion through the
+   hash-sharded registration fence;
+5. measure and, if justified, add tail-supersession entries; then add bounded FIFO draining and
+   deferred queues, 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-
+The production Harper package remains disabled until the cleanup lifecycle fence and Harper host-
 storage integration both pass. No native RocksDB storage provider is added to the public library.
 
 ## Verification
@@ -279,16 +303,27 @@ real Tantivy merge and garbage-collection cycles and after crash/reopen. `Faulti
 recovery at atomic batch boundaries; the Harper integration separately proves that one host write
 request commits as one RocksDB batch.
 
-The dependency-free `kv_directory` release benchmark compares the merged slice-2 baseline with this
-slice through Tantivy's buffered `WritePtr`, rather than calling the fence directly. It reports
-per-sample-mean p50/p95/p99 and aggregate throughput for caller write sizes, empty and dirty flushes, chunk
-publication, deletion with closed and active writers, and distinct-file concurrency. The
-empty-flush case isolates the per-call retirement-fence cost; the buffered cases show how Tantivy's
-writer amortizes it in practice. Results are versioned JSON labeled by revision. Shared CI runs a
-correctness smoke with no timing threshold; performance decisions use alternating runs on one fixed
-host. The deterministic Phase 0 store removes RocksDB and Node transport variance but serializes
-access, so its concurrency results detect directory-coordination regressions rather than predicting
-RocksDB scaling.
+Slice 4 tests object and exact-revision pin counts, delete/recreate identity, the binding-read and
+pin-registration race, shared state across independently constructed directories, handle-owned
+state lifetime, one storage read per open, bounded open/drop churn, and reader independence from
+writer retirement. The release benchmark adds retained and churned opens plus shared- and
+distinct-file concurrency. Distinct-file cases run at one, two, four, and eight threads; shared-file
+cases start at two threads.
+
+The dependency-free `kv_directory` release benchmark compares adjacent merged slices through
+Tantivy's public directory interfaces. It reports per-sample-mean p50/p95/p99 and aggregate
+throughput for caller write sizes, empty and dirty flushes, chunk publication, deletion with closed
+and active writers, retained and churned read-handle opens, shared- and distinct-file read-open
+concurrency, and distinct-file write concurrency. The empty-flush case isolates the per-call
+retirement-fence cost; the buffered cases show how Tantivy's writer amortizes it in practice.
+Retained handles measure registration growth, churned handles exercise drop-time pin removal, and
+shared-file concurrency exercises per-object pin accounting. Results are versioned JSON labeled by
+revision. Shared CI runs a correctness smoke with no timing threshold; performance decisions use
+alternating runs on one fixed host. The deterministic benchmark store removes RocksDB and Node
+transport variance, permits concurrent point reads, and serializes writes. Read-open concurrency
+therefore isolates registration contention, while write concurrency detects directory-coordination
+regressions; neither predicts RocksDB scaling. The `-rw-` read-open case names begin a new comparison
+series because the earlier cases used the serialized fault-injection store.
 
 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
diff --git a/src/phase0.rs b/src/phase0.rs
index c4a7619..ff6656d 100644
--- a/src/phase0.rs
+++ b/src/phase0.rs
@@ -1,11 +1,12 @@
 use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
 use std::fmt;
+use std::hash::{DefaultHasher, Hash, Hasher};
 use std::io;
 use std::io::Write;
 use std::ops::Range;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, AtomicUsize, Ordering};
-use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak};
+use std::sync::{Arc, Condvar, Mutex, OnceLock, RwLock, Weak};
 use std::time::{Duration, Instant};
 
 use async_trait::async_trait;
@@ -264,6 +265,8 @@ pub type FaultingDirectory = KvDirectory<FaultingKv>;
 struct DirectoryState {
 	allocator: Mutex<ObjectIdAllocator>,
 	paths: Arc<PathRegistry>,
+	reader_pins: Arc<ReaderPinRegistry>,
+	reader_registration_shards: [RwLock<()>; READER_REGISTRATION_SHARD_COUNT],
 	reclaim_shards: [Mutex<()>; RECLAIM_SHARD_COUNT],
 	locks: Mutex<DirectoryLocks>,
 	locks_changed: Condvar,
@@ -304,6 +307,43 @@ struct WriterClaim<'a> {
 	fence: &'a WriterFence,
 }
 
+struct ReaderPinRegistry {
+	shards: [Mutex<HashMap<u64, Weak<ObjectReaderPins>>>; RECLAIM_SHARD_COUNT],
+}
+
+struct ObjectReaderPins {
+	object_id: u64,
+	registry: Weak<ReaderPinRegistry>,
+	state: Mutex<ObjectReaderPinState>,
+}
+
+#[derive(Default)]
+struct ObjectReaderPinState {
+	total: usize,
+	revisions: ReaderRevisions,
+}
+
+#[derive(Default)]
+enum ReaderRevisions {
+	#[default]
+	None,
+	One(u64, usize),
+	Many(HashMap<u64, usize>),
+}
+
+struct ReaderPin {
+	object: Arc<ObjectReaderPins>,
+	tail_revision: u64,
+}
+
+impl Default for ReaderPinRegistry {
+	fn default() -> Self {
+		Self {
+			shards: std::array::from_fn(|_| Mutex::new(HashMap::new())),
+		}
+	}
+}
+
 #[derive(Default)]
 struct DirectoryLocks {
 	held: HashSet<PathBuf>,
@@ -408,6 +448,173 @@ impl Drop for WriterClaim<'_> {
 	}
 }
 
+impl ReaderPinRegistry {
+	fn register(self: &Arc<Self>, binding: &Binding) -> io::Result<ReaderPin> {
+		let shard = usize::from(reclaim_shard(binding.object_id));
+		let object = {
+			let mut objects = self.shards[shard]
+				.lock()
+				.unwrap_or_else(|poisoned| poisoned.into_inner());
+			objects
+				.get(&binding.object_id)
+				.and_then(Weak::upgrade)
+				.unwrap_or_else(|| {
+					let object = Arc::new(ObjectReaderPins {
+						object_id: binding.object_id,
+						registry: Arc::downgrade(self),
+						state: Mutex::new(ObjectReaderPinState::default()),
+					});
+					objects.insert(binding.object_id, Arc::downgrade(&object));
+					object
+				})
+		};
+		let mut state = object.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+		let total = state
+			.total
+			.checked_add(1)
+			.ok_or_else(|| io::Error::other("reader pin count exhausted"))?;
+		state.revisions.increment(binding.tail_revision)?;
+		state.total = total;
+		drop(state);
+		Ok(ReaderPin {
+			object,
+			tail_revision: binding.tail_revision,
+		})
+	}
+
+	#[cfg(test)]
+	fn is_pinned(&self, object_id: u64, tail_revision: Option<u64>) -> bool {
+		let shard = usize::from(reclaim_shard(object_id));
+		let objects = self.shards[shard]
+			.lock()
+			.unwrap_or_else(|poisoned| poisoned.into_inner());
+		let Some(object) = objects.get(&object_id).and_then(Weak::upgrade) else {
+			return false;
+		};
+		drop(objects);
+		let state = object.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+		tail_revision.map_or(state.total != 0, |revision| state.revisions.contains(revision))
+	}
+}
+
+impl ReaderRevisions {
+	fn increment(&mut self, tail_revision: u64) -> io::Result<()> {
+		match self {
+			Self::None => *self = Self::One(tail_revision, 1),
+			Self::One(revision, count) if *revision == tail_revision => {
+				*count = count
+					.checked_add(1)
+					.ok_or_else(|| io::Error::other("reader revision pin count exhausted"))?;
+			}
+			Self::One(revision, count) => {
+				let mut revisions = HashMap::with_capacity(2);
+				revisions.insert(*revision, *count);
+				revisions.insert(tail_revision, 1);
+				*self = Self::Many(revisions);
+			}
+			Self::Many(revisions) => {
+				let count = revisions.get(&tail_revision).copied().unwrap_or(0);
+				revisions.insert(
+					tail_revision,
+					count
+						.checked_add(1)
+						.ok_or_else(|| io::Error::other("reader revision pin count exhausted"))?,
+				);
+			}
+		}
+		Ok(())
+	}
+
+	#[cfg(test)]
+	fn contains(&self, tail_revision: u64) -> bool {
+		match self {
+			Self::None => false,
+			Self::One(revision, count) => *revision == tail_revision && *count != 0,
+			Self::Many(revisions) => revisions.get(&tail_revision).is_some_and(|count| *count != 0),
+		}
+	}
+
+	fn decrement(&mut self, tail_revision: u64) -> bool {
+		match self {
+			Self::None => false,
+			Self::One(revision, count) if *revision == tail_revision && *count > 1 => {
+				*count -= 1;
+				true
+			}
+			Self::One(revision, count) if *revision == tail_revision && *count == 1 => {
+				*self = Self::None;
+				true
+			}
+			Self::One(_, _) => false,
+			Self::Many(revisions) => {
+				let remove = match revisions.get_mut(&tail_revision) {
+					Some(count) if *count > 1 => {
+						*count -= 1;
+						false
+					}
+					Some(count) if *count == 1 => true,
+					_ => return false,
+				};
+				if remove {
+					revisions.remove(&tail_revision);
+				}
+				let remaining = if revisions.len() == 1 {
+					revisions.iter().next().map(|(revision, count)| (*revision, *count))
+				} else {
+					None
+				};
+				if let Some((revision, count)) = remaining {
+					*self = Self::One(revision, count);
+				}
+				true
+			}
+		}
+	}
+}
+
+impl Drop for ReaderPin {
+	fn drop(&mut self) {
+		let mut state = self
+			.object
+			.state
+			.lock()
+			.unwrap_or_else(|poisoned| poisoned.into_inner());
+		debug_assert!(
+			state.total != 0 || std::thread::panicking(),
+			"reader object pin count is missing"
+		);
+		if state.total == 0 {
+			return;
+		}
+		let revision_found = state.revisions.decrement(self.tail_revision);
+		debug_assert!(
+			revision_found || std::thread::panicking(),
+			"reader revision pin count is missing"
+		);
+		if revision_found {
+			state.total -= 1;
+		}
+	}
+}
+
+impl Drop for ObjectReaderPins {
+	fn drop(&mut self) {
+		let Some(registry) = self.registry.upgrade() else {
+			return;
+		};
+		let shard = usize::from(reclaim_shard(self.object_id));
+		let mut objects = registry.shards[shard]
+			.lock()
+			.unwrap_or_else(|poisoned| poisoned.into_inner());
+		if objects
+			.get(&self.object_id)
+			.is_some_and(|object| std::ptr::eq(object.as_ptr(), self))
+		{
+			objects.remove(&self.object_id);
+		}
+	}
+}
+
 fn writer_retired_error() -> io::Error {
 	io::Error::new(io::ErrorKind::NotFound, "file was deleted while its writer was open")
 }
@@ -462,6 +669,8 @@ impl<S: KvStore> KvDirectory<S> {
 			let state = Arc::new(DirectoryState {
 				allocator: Mutex::new(ObjectIdAllocator::default()),
 				paths: Arc::new(PathRegistry::default()),
+				reader_pins: Arc::new(ReaderPinRegistry::default()),
+				reader_registration_shards: std::array::from_fn(|_| RwLock::new(())),
 				reclaim_shards: std::array::from_fn(|_| Mutex::new(())),
 				locks: Mutex::new(DirectoryLocks::default()),
 				locks_changed: Condvar::new(),
@@ -569,6 +778,25 @@ impl<S: KvStore> KvDirectory<S> {
 			Mutation::Put(tail_key, next.to_be_bytes().to_vec()),
 		])
 	}
+
+	fn pinned_binding(
+		&self,
+		path: &Path,
+		after_binding_read: impl FnOnce(),
+	) -> Result<(Binding, ReaderPin), OpenReadError> {
+		let registration = self.state.reader_registration_shards[reader_registration_shard(path)]
+			.read()
+			.unwrap_or_else(|poisoned| poisoned.into_inner());
+		let binding = self.read_binding(path)?;
+		after_binding_read();
+		let pin = self
+			.state
+			.reader_pins
+			.register(&binding)
+			.map_err(|error| OpenReadError::wrap_io_error(error, path.to_path_buf()))?;
+		drop(registration);
+		Ok((binding, pin))
+	}
 }
 
 struct KvFileHandle<S> {
@@ -576,6 +804,8 @@ struct KvFileHandle<S> {
 	namespace: Arc<[u8]>,
 	path: PathBuf,
 	binding: Binding,
+	_state: Arc<DirectoryState>,
+	_pin: ReaderPin,
 }
 
 impl<S> fmt::Debug for KvFileHandle<S> {
@@ -703,11 +933,14 @@ impl Drop for KvDirectoryLock {
 
 impl<S: KvStore> Directory for KvDirectory<S> {
 	fn get_file_handle(&self, path: &Path) -> Result<Arc<dyn FileHandle>, OpenReadError> {
+		let (binding, pin) = self.pinned_binding(path, || {})?;
 		Ok(Arc::new(KvFileHandle {
 			store: self.store.clone(),
 			namespace: self.namespace.clone(),
 			path: path.to_path_buf(),
-			binding: self.read_binding(path)?,
+			binding,
+			_state: self.state.clone(),
+			_pin: pin,
 		}))
 	}
 
@@ -788,7 +1021,13 @@ impl<S: KvStore> Directory for KvDirectory<S> {
 		};
 		mutations.push(Mutation::Delete(binding_key.clone()));
 		mutations.push(Mutation::Delete(atomic_key));
-		match self.store.write(&mutations, WritePolicy::WAL) {
+		let write_result = {
+			let _reader_registration = self.state.reader_registration_shards[reader_registration_shard(path)]
+				.write()
+				.unwrap_or_else(|poisoned| poisoned.into_inner());
+			self.store.write(&mutations, WritePolicy::WAL)
+		};
+		match write_result {
 			Ok(()) => {
 				lifecycle.writer = None;
 				Ok(())
@@ -1289,6 +1528,7 @@ const MAX_OBJECT_EXTENT_BYTES: u128 = 1 << 40;
 const RECLAIM_ENTRY_FORMAT_VERSION: u8 = 1;
 const RECLAIM_ENTRY_WHOLE_OBJECT: u8 = 1;
 const RECLAIM_SHARD_COUNT: usize = 64;
+const READER_REGISTRATION_SHARD_COUNT: usize = 256;
 const KEY_KIND_COUNTER: u8 = 1;
 const KEY_KIND_BINDING: u8 = 2;
 const KEY_KIND_ATOMIC: u8 = 3;
@@ -1418,6 +1658,12 @@ fn reclaim_shard(object_id: u64) -> u8 {
 	(object_id % RECLAIM_SHARD_COUNT as u64) as u8
 }
 
+fn reader_registration_shard(path: &Path) -> usize {
+	let mut hasher = DefaultHasher::new();
+	path.hash(&mut hasher);
+	(hasher.finish() % READER_REGISTRATION_SHARD_COUNT as u64) as usize
+}
+
 fn encode_binding(binding: &Binding) -> Vec<u8> {
 	let mut bytes = Vec::with_capacity(45);
 	bytes.push(BINDING_FORMAT_VERSION);
@@ -2065,6 +2311,193 @@ mod tests {
 		assert_eq!(second.read_bytes().unwrap().as_slice(), b"first second");
 	}
 
+	#[test]
+	fn open_handles_pin_their_object_and_tail_revision() {
+		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(b"first").unwrap();
+		writer.flush().unwrap();
+		let first_binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap();
+		let first = directory.open_read(path).unwrap();
+		assert!(directory.state.reader_pins.is_pinned(first_binding.object_id, None));
+		assert!(directory
+			.state
+			.reader_pins
+			.is_pinned(first_binding.object_id, Some(first_binding.tail_revision)));
+
+		directory.delete(path).unwrap();
+		let mut replacement = directory.open_write(path).unwrap();
+		replacement.write_all(b"second").unwrap();
+		replacement.flush().unwrap();
+		let second_binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap();
+		let second = directory.open_read(path).unwrap();
+		assert_ne!(first_binding.object_id, second_binding.object_id);
+		assert!(directory.state.reader_pins.is_pinned(first_binding.object_id, None));
+		assert!(directory.state.reader_pins.is_pinned(second_binding.object_id, None));
+
+		drop(first);
+		assert!(!directory.state.reader_pins.is_pinned(first_binding.object_id, None));
+		assert!(directory.state.reader_pins.is_pinned(second_binding.object_id, None));
+		drop(second);
+		assert!(!directory.state.reader_pins.is_pinned(second_binding.object_id, None));
+	}
+
+	#[test]
+	fn revision_pins_are_independent_within_one_object() {
+		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(b"first").unwrap();
+		writer.flush().unwrap();
+		let first_binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap();
+		let first = directory.open_read(path).unwrap();
+
+		writer.write_all(b" second").unwrap();
+		writer.flush().unwrap();
+		let second_binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap();
+		let second = directory.open_read(path).unwrap();
+		assert_eq!(first_binding.object_id, second_binding.object_id);
+		assert_ne!(first_binding.tail_revision, second_binding.tail_revision);
+		assert!(directory
+			.state
+			.reader_pins
+			.is_pinned(first_binding.object_id, Some(first_binding.tail_revision)));
+		assert!(directory
+			.state
+			.reader_pins
+			.is_pinned(second_binding.object_id, Some(second_binding.tail_revision)));
+
+		drop(second);
+		assert!(directory.state.reader_pins.is_pinned(first_binding.object_id, None));
+		assert!(directory
+			.state
+			.reader_pins
+			.is_pinned(first_binding.object_id, Some(first_binding.tail_revision)));
+		assert!(!directory
+			.state
+			.reader_pins
+			.is_pinned(second_binding.object_id, Some(second_binding.tail_revision)));
+		drop(first);
+		assert!(!directory.state.reader_pins.is_pinned(first_binding.object_id, None));
+	}
+
+	#[test]
+	fn binding_read_and_pin_registration_complete_before_delete() {
+		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(b"contents").unwrap();
+		writer.flush().unwrap();
+		drop(writer);
+		let binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap();
+		let (binding_read, allow_delete) = std::sync::mpsc::sync_channel(0);
+		let (delete_started, deletion_entered) = std::sync::mpsc::sync_channel(0);
+		let (deletion_completed, deleted) = std::sync::mpsc::channel();
+		let deleting_directory = directory.clone();
+		let deletion = std::thread::spawn(move || {
+			allow_delete.recv().unwrap();
+			delete_started.send(()).unwrap();
+			let result = deleting_directory.delete(path);
+			deletion_completed.send(()).unwrap();
+			result
+		});
+
+		let (opened, pin) = directory
+			.pinned_binding(path, || {
+				binding_read.send(()).unwrap();
+				deletion_entered.recv().unwrap();
+				assert!(deleted.recv_timeout(Duration::from_millis(50)).is_err());
+			})
+			.unwrap();
+		deletion.join().unwrap().unwrap();
+
+		assert_eq!(opened, binding);
+		assert!(directory.state.reader_pins.is_pinned(binding.object_id, None));
+		assert!(!directory.exists(path).unwrap());
+		drop(pin);
+		assert!(!directory.state.reader_pins.is_pinned(binding.object_id, None));
+	}
+
+	#[test]
+	fn independently_constructed_directories_share_reader_pins() {
+		let store = FaultingKv::default();
+		let first = FaultingDirectory::new(store.clone());
+		let second = FaultingDirectory::new(store.clone());
+		let path = Path::new("segment");
+		let mut writer = first.open_write(path).unwrap();
+		writer.write_all(b"contents").unwrap();
+		writer.flush().unwrap();
+		let binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap();
+		let handle = first.open_read(path).unwrap();
+
+		assert!(second.state.reader_pins.is_pinned(binding.object_id, None));
+		drop(handle);
+		assert!(!second.state.reader_pins.is_pinned(binding.object_id, None));
+	}
+
+	#[test]
+	fn open_handle_keeps_the_canonical_directory_state_alive() {
+		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(b"contents").unwrap();
+		writer.flush().unwrap();
+		drop(writer);
+		let binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap();
+		let state = Arc::downgrade(&directory.state);
+		let handle = directory.open_read(path).unwrap();
+		drop(directory);
+
+		let retained = state.upgrade().unwrap();
+		let reopened = FaultingDirectory::new(store);
+		assert!(Arc::ptr_eq(&retained, &reopened.state));
+		assert!(reopened.state.reader_pins.is_pinned(binding.object_id, None));
+		drop(handle);
+		assert!(!reopened.state.reader_pins.is_pinned(binding.object_id, None));
+	}
+
+	#[test]
+	fn open_read_adds_no_storage_operation_for_pin_registration() {
+		let store = CountingKv::new();
+		let directory = KvDirectory::new(store.clone());
+		let path = Path::new("segment");
+		let mut writer = directory.open_write(path).unwrap();
+		writer.write_all(b"contents").unwrap();
+		writer.flush().unwrap();
+		store.take_io_counts();
+
+		let handle = directory.open_read(path).unwrap();
+
+		assert_eq!(store.take_io_counts(), (1, 0, 0));
+		drop(handle);
+	}
+
+	#[test]
+	fn open_drop_churn_bounds_retained_weak_pins() {
+		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(b"contents").unwrap();
+		writer.flush().unwrap();
+		let binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap();
+
+		for _ in 0..256 {
+			drop(directory.open_read(path).unwrap());
+		}
+
+		let shard = usize::from(reclaim_shard(binding.object_id));
+		let objects = directory.state.reader_pins.shards[shard]
+			.lock()
+			.unwrap_or_else(|poisoned| poisoned.into_inner());
+		assert!(!objects.contains_key(&binding.object_id));
+	}
+
 	#[test]
 	fn independently_constructed_directories_share_writer_locks() {
 		let store = FaultingKv::default();
@@ -2376,6 +2809,52 @@ mod tests {
 		}
 	}
 
+	#[test]
+	fn open_read_does_not_wait_for_writer_retirement() {
+		let store = BlockingKv::new();
+		let directory = KvDirectory::new(store.clone());
+		let path = Path::new("segment");
+		let mut writer = directory.open_write(path).unwrap();
+		writer.write_all(b"first").unwrap();
+		writer.flush().unwrap();
+		writer.write_all(b" second").unwrap();
+		let fence = directory
+			.state
+			.paths
+			.state(path)
+			.lifecycle
+			.lock()
+			.unwrap_or_else(|poisoned| poisoned.into_inner())
+			.writer
+			.as_ref()
+			.and_then(Weak::upgrade)
+			.unwrap();
+		store.arm_next_write();
+		let helper_store = store.clone();
+		let deleting_directory = directory.clone();
+		let reading_directory = directory.clone();
+		let helper = std::thread::spawn(move || {
+			assert!(
+				helper_store.wait_until_blocked(),
+				"write did not reach the test barrier"
+			);
+			let deletion = std::thread::spawn(move || deleting_directory.delete(path));
+			assert!(wait_for_retirement(&fence), "delete did not retire the writer");
+			let (opened, received) = std::sync::mpsc::channel();
+			let reading = std::thread::spawn(move || opened.send(reading_directory.open_read(path)).unwrap());
+			let handle = received.recv_timeout(Duration::from_secs(5));
+			helper_store.release_write();
+			let handle = handle.expect("read waited for writer retirement").unwrap();
+			deletion.join().unwrap().unwrap();
+			reading.join().unwrap();
+			handle
+		});
+
+		writer.flush().unwrap();
+		let handle = helper.join().unwrap();
+		assert_eq!(handle.read_bytes().unwrap().as_slice(), b"first");
+	}
+
 	#[test]
 	fn delete_waits_for_in_flight_chunk_storage() {
 		let store = BlockingKv::new();
@@ -2435,6 +2914,10 @@ mod tests {
 			reclaim_shard(first_binding.object_id),
 			reclaim_shard(second_binding.object_id)
 		);
+		assert_ne!(
+			reader_registration_shard(Path::new("first")),
+			reader_registration_shard(Path::new("second"))
+		);
 		store.arm_next_write();
 
 		let first_directory = directory.clone();