Add bounded reclamation for Harper-backed Tantivy directories

benches/kv_directory.rs

@@ -9,10 +9,10 @@ use std::process::Command;
 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use std::sync::mpsc;
 use std::sync::{Arc, Condvar, Mutex, OnceLock, RwLock};
-use std::time::{Instant, SystemTime, UNIX_EPOCH};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use harper_fulltext::phase0::{
-	FaultingDirectory, FaultingKv, KvDirectory, KvStore, KvStoreIdentity, Mutation, WritePolicy,
+	FaultingDirectory, FaultingKv, KvDirectory, KvStore, KvStoreIdentity, Mutation, ReclaimBudget, WritePolicy,
 };
 use harper_fulltext::TANTIVY_VERSION;
 use tantivy::directory::{Directory, OwnedBytes, TerminatingWrite, WritePtr};
@@ -241,6 +241,13 @@ fn main() -> io::Result<()> {
 		&arguments,
 		delete_case(true, arguments.smoke),
 	)?);
+	results.push(measure_case(
+		"reclaim-retired-4k".to_owned(),
+		"retired-object",
+		1,
+		&arguments,
+		reclaim_case(arguments.smoke),
+	)?);
 	results.push(measure_case(
 		"open-read-retained".to_owned(),
 		"open-read",
@@ -544,6 +551,41 @@ fn delete_case(active_writer: bool, smoke: bool) -> impl FnMut(usize) -> io::Res
 	}
 }
 
+fn reclaim_case(smoke: bool) -> impl FnMut(usize) -> io::Result<Sample> {
+	let payload = vec![19u8; 4 * 1024];
+	move |sample| {
+		let directory = FaultingDirectory::new(FaultingKv::default());
+		let operations = if smoke { 4 } else { STORAGE_OPERATIONS_PER_SAMPLE };
+		for operation in 0..operations {
+			let path = format!("reclaim-{sample}-{operation}");
+			let mut writer = open_writer(&directory, Path::new(&path))?;
+			writer.write_all(&payload)?;
+			writer.terminate()?;
+			delete_path(&directory, Path::new(&path))?;
+		}
+		let budget = ReclaimBudget {
+			max_point_reads: operations * 8 + 128,
+			max_mutations: operations * 8 + 512,
+			max_request_bytes: 1024 * 1024,
+			max_elapsed: Duration::from_secs(1),
+		};
+		let started = Instant::now();
+		let outcome = directory.reclaim(budget)?;
+		let elapsed_nanoseconds = started.elapsed().as_nanos();
+		if outcome.entries_reclaimed != operations {
+			return Err(io::Error::other(format!(
+				"reclamation completed {} of {operations} retired objects",
+				outcome.entries_reclaimed
+			)));
+		}
+		Ok(Sample {
+			elapsed_nanoseconds,
+			operations: operations as u64,
+			bytes: (operations * payload.len()) as u64,
+		})
+	}
+}
+
 fn open_read_case(retain_handles: bool, smoke: bool) -> impl FnMut(usize) -> io::Result<Sample> {
 	move |sample| {
 		let directory = FaultingDirectory::new(FaultingKv::default());

docs/phase-0-rocks-bridge-plan.md

@@ -165,6 +165,9 @@ namespace / index generation
   tail kind: object-id, revision       -> immutable final partial chunk
   binding kind: logical path           -> v3 object-id, published state and physical-key high-waters
   atomic kind: logical path            -> complete small-file bytes
+  reclaim tail/head kinds: shard       -> next enqueue sequence / oldest retained sequence
+  reclaim entry kind: shard, sequence  -> v2 retired object identity, published chunks and high-waters
+  reclaim progress: shard, sequence    -> next chunk ordinal and tail revision
 ```
 
 The keyspace uses a fixed magic, format version, length-prefixed namespace, and one-byte key-kind
@@ -181,7 +184,9 @@ 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
+The consumer advances the unreleased directory key format to v3 because reclaim entry v2 adds the
+published chunk count and introduces sequence-keyed progress. The former v2 prototype is rejected
+rather than partially interpreted. New logical key kinds receive distinct kind tags under that
 version. A storage provider must also change `KvStoreIdentity` whenever close, restore, or column-
 family replacement can change the bytes behind an identity.