diff --git a/benches/kv_directory.rs b/benches/kv_directory.rs index f94cddc..8eae39b 100644 --- a/benches/kv_directory.rs +++ b/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 { + 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 { move |sample| { let directory = FaultingDirectory::new(FaultingKv::default()); diff --git a/docs/phase-0-rocks-bridge-plan.md b/docs/phase-0-rocks-bridge-plan.md index 4f04689..656bfee 100644 --- a/docs/phase-0-rocks-bridge-plan.md +++ b/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. diff --git a/docs/reclamation-plan.md b/docs/reclamation-plan.md index 7c36a8e..714193c 100644 --- a/docs/reclamation-plan.md +++ b/docs/reclamation-plan.md @@ -9,13 +9,14 @@ readiness wait for a complete sweep. ## Grounding -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). +This plan is written against fulltext main at `15045c9`, the squash-merged result of +[Protect Harper-backed Tantivy readers during reclamation #27](https://github.com/HarperFast/fulltext/pull/27). `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. +logical binding and enqueues the object's derivable physical extent, and open handles retain +process-local object/revision pins, 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 @@ -79,7 +80,7 @@ underlying `delete()` before removing that path from `.managed.json`. A crash-ab therefore remains named by its binding and managed path until it is placed on the reclaim FIFO. Whole-object deletion already covers every tail revision through `tail_high_water`. Tail-only -enqueue is deferred to slice 4 and retained only if measurement shows that repeated non-empty tail +enqueue is deferred to slice 6 and retained only if measurement shows that repeated non-empty tail publication materially increases live-index storage before final file deletion. If retained, the publication batch appends the exact old revision so bytes cannot be deleted while an earlier handle still references them. Queue entries are the durable consequence of the transition that made bytes @@ -87,9 +88,11 @@ 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. +host mutation. Enqueue and tail advance share the binding publication or deletion batch. Progress +is keyed by the entry sequence, and progress advancement shares one WAL-only batch with the payload +deletes it represents. Final payload deletion, progress deletion, entry deletion, and any head +advance likewise 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. @@ -148,11 +151,13 @@ gain a new pin because its binding is gone. Before tail-only reclamation is enab 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. +match only the exact object id and tail revision. If a matching pin remains, cleanup leaves the +durable entry in place and advances a process-local scan cursor to later sequence-addressed entries. +Completed later entries become holes in the FIFO. The durable head advances only across missing +entries and stops at the oldest retained or partially processed entry. Foreground publishers mutate +only the tail, so a low-priority cleanup host operation never owns the mutex or counter needed by a +foreground enqueue. Restart discards the scan cursor together with the process-local pins; cleanup +then resumes from the durable head without persisting a stale reason for deferral. 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 @@ -202,7 +207,7 @@ A definitively failed delete rereads the binding while it still owns the path li same binding remains, the writer fence is reopened and the error is returned; if the binding is gone, the atomic enqueue also committed and delete succeeds. An unreadable or unexpected binding keeps the writer retired and fails closed. Invalid persisted directory or queue data is a -generation-corruption signal; slice 4 must surface that terminal health state to Harper before +generation-corruption signal; slice 6 must surface that terminal health state to Harper before production is enabled rather than relying on Tantivy's repeated GC attempt. Queue writers for one namespace must share one `DirectoryState`. This follows the existing @@ -216,16 +221,72 @@ when that handle is its sole access path. 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. +deletes one derived key. A pinned entry is skipped in memory for the current admission, while later +sequence-addressed entries in the same shard remain eligible. Each starting sequence is examined at +most once per admission, and a pass that finds only pinned work reports a blocked-reclamation state +rather than consuming the JavaScript service thread. Cleanup cost is proportional to garbage queued, never to object ids or records ever created. +The first consumer-core unit remains synchronous and available only through the experimental Rust +`phase0` surface. It adds one persisted head per shard and keys progress by shard and entry sequence. +Progress contains the next chunk ordinal and next tail revision for that exact entry. Every +payload-deletion batch also persists the resulting progress; the final batch removes the entry and +its progress and advances the head when it completes the current head. A crash can therefore repeat +deletes, which are idempotent, but cannot recover a cursor beyond bytes that may still exist or +apply one entry's cursor to another entry. + +Because the unreleased reclaim entry gains the published chunk count and the consumer adds head and +progress key kinds, this unit advances the directory key-format marker to v3. A namespace marked +with the former v2 prototype is rejected and rebuilt; the implementation does not strand v2 keys +under a parallel prefix or attempt a migration for data that has never shipped. + +The consumer snapshots each visited shard's tail and never scans beyond that bound during an +admission. A pinned entry remains durable while a process-local cursor moves to the next sequence. +An unpinned later entry may complete out of order, leaving a missing slot. When the oldest +entry completes, bounded point reads advance the durable head across consecutive holes. This +prevents one retained handle from blocking later garbage without adding another durable queue, +copying entries, or sharing foreground tail state. The cursor never advances beyond the tail +snapshot and remains on an unpinned entry until all of that entry's bounded batches complete. It is +process-local; after restart all pins are gone and scanning safely resumes at the durable head. The +result reports pinned skips and whether the remaining queue made no deletion progress; the later +background-task unit owns retry timing and health thresholds. + +Reclaim entries add the published full-chunk count alongside the existing high-waters. Published +chunks are a known contiguous prefix and are deleted directly. Any staged chunks between the +published count and the reserved high-water are probed in order and deletion stops at the first +missing chunk; writer staging is contiguous, so no later chunk can exist. This avoids writing up to +63 unnecessary RocksDB tombstones for every retired object while preserving discovery of a writer's +unpublished staged prefix. Tail revisions remain bounded blind deletes because an empty-tail flush +can make their physical presence sparse. + +One process-local consumer mutex, round-robin shard cursor, and lazy per-shard head/tail hints live +in `DirectoryState`. They serialize cleanup for one storage identity and namespace without coordinating different indexes, +prevent a small budget from always starting at shard zero, and make repeated empty admissions avoid +host reads after a shard has been observed empty. Enqueue advances the corresponding tail hint after +a definitive commit and invalidates it after an unknown result. The call accepts hard limits for +point reads, batch mutations, encoded mutation bytes, and elapsed work. It starts an operation only +when that operation fits the remaining count and byte limits, and checks elapsed time between host +operations. A synchronous operation already admitted to the host remains uncancellable and may +finish after the elapsed target. Limits that cannot admit the smallest legal cleanup step are +rejected rather than reported as no progress. Invalid persisted queue data latches the affected bit +and first error text in the outcome while the admission finishes healthy shards. Terminal shards +are excluded from `has_more`, so callers do not hot-loop work that requires a generation rebuild. +This unit does not create a thread, reserve host-transport capacity, +accept an owner lease, expose Node.js API, or enable Harper cleanup; those lifecycle and admission +rules remain the next unit. + +An enqueue with an unknown commit result invalidates the cached shard tail before returning the +error. The next admission rereads durable state, so an applied batch cannot be stranded behind a +stale empty hint. A namespace proven absent after bounded format validation returns immediately; +the default read budget covers format validation plus one cold 64-shard sweep for a present +namespace. + 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 +boundary; terminal failure, queue depth, pinned skips, 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 @@ -234,26 +295,31 @@ guessed from point misses: this is a derived index, so the generation is marked from Harper source data. A caught cleanup panic enters the same terminal health state. Neither case silently retries forever or advances past unknown data. +A progress record without its sequence-matched entry is terminal corruption, not disposable +garbage. The entry is the only durable copy of the retired object id and physical extent; removing +the progress record would allow unknown payload to leak permanently. Harper rebuilds that derived +generation from source data instead of masking the broken atomicity invariant. + ## 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. `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 | Slice 3 enqueues only whole-object deletion. It leaves superseded tails within the object's high-water until final deletion; tail-only enqueue moves to slice 4 and ships only if measured tail accumulation justifies publication-path cost. | -| Higher-layer rotation | Rebuild into a fresh Harper generation and drop the old column family. This remains the corruption-recovery path, but routine reclamation would require replaying hundreds of millions of records and would move cleanup outside the standalone library. | -| Reader lifetime | Process-wide object/revision `Arc` pins are chosen because they exactly model Tantivy handle lifetime without a read-time storage call. A Harper two-worker shared-state test is an enablement gate. A searcher epoch would require a new cross-layer reader API, and a wall-clock grace cannot prove that a reader released the bytes. | -| Enqueue naming | An object-id-addressed record requires scanning every allocated id because `KvStore` has no enumeration. Rewriting a tombstone binding under the logical path loses the old object on path reuse. A persisted sharded sequence FIFO is chosen because its point-read cost is proportional to garbage transitions. | -| Writer fence | Holding a path lock across each chunk write would add lock convoying around an uncancellable host round trip. A per-writer atomic retired/in-flight fence is chosen: delete closes admission and waits before its final binding read, while chunk staging adds no mutex or allocation. | -| Object-id allocation | Updating the counter with every binding creation serializes file creation across a host round trip. The allocator reserves fixed durable strides, reducing that shared operation to one per stride; unused ids after failure or restart are harmless. | -| Queue-tail reads | Cache the next sequence and reread only after errors, or point-read it before every enqueue. The durable point read is chosen because slice 3 enqueues only file deletion rather than indexed documents, removes cache recovery state, and is measured explicitly. | -| Cleanup priority | Cleanup does not rotate through the foreground ingress tail. A cleanup-only deferred queue lets the consumer advance past pinned entries without holding a foreground enqueue mutex or counter across low-priority host I/O. | -| 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 | Payload-prefix enumeration or a dense object-id sweep. The Directory's frozen `KvStore` and host transport expose no enumeration primitive. `RocksLease::scan_page` sits outside that contract and returns values, so using it for 256 KiB payload discovery would transfer the object being reclaimed and would not support the standalone host transport. Either scan would also 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 | Slice 3 enqueues only whole-object deletion. It leaves superseded tails within the object's high-water until final deletion; tail-only enqueue moves to slice 6 and ships only if measured tail accumulation justifies publication-path cost. | +| Higher-layer rotation | Rebuild into a fresh Harper generation and drop the old column family. This remains the corruption-recovery path, but routine reclamation would require replaying hundreds of millions of records and would move cleanup outside the standalone library. | +| Reader lifetime | Process-wide object/revision `Arc` pins are chosen because they exactly model Tantivy handle lifetime without a read-time storage call. A Harper two-worker shared-state test is an enablement gate. A searcher epoch would require a new cross-layer reader API, and a wall-clock grace cannot prove that a reader released the bytes. | +| Enqueue naming | An object-id-addressed record requires scanning every allocated id because `KvStore` has no enumeration. Rewriting a tombstone binding under the logical path loses the old object on path reuse. A persisted sharded sequence FIFO is chosen because its point-read cost is proportional to garbage transitions. | +| Writer fence | Holding a path lock across each chunk write would add lock convoying around an uncancellable host round trip. A per-writer atomic retired/in-flight fence is chosen: delete closes admission and waits before its final binding read, while chunk staging adds no mutex or allocation. | +| Object-id allocation | Updating the counter with every binding creation serializes file creation across a host round trip. The allocator reserves fixed durable strides, reducing that shared operation to one per stride; unused ids after failure or restart are harmless. | +| Queue-tail reads | Cache the next sequence and reread only after errors, or point-read it before every enqueue. The durable point read is chosen because slice 3 enqueues only file deletion rather than indexed documents, removes cache recovery state, and is measured explicitly. | +| Cleanup priority | Cleanup leaves pinned entries durable and skips them with a process-local sequence cursor. Later entries may complete out of order, while the durable head advances only across observed holes. This lets the consumer advance without a second persistent queue or holding a foreground enqueue mutex or counter across low-priority host I/O. | +| Chosen | Binding high-waters plus one transition-fed durable FIFO per shard, sequence-keyed progress, object-id pins, and bounded low-priority draining. | ## Persistence format and delivery sequence @@ -272,9 +338,10 @@ Then deliver reclamation in reviewable slices: replacement writer; 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. +5. add bounded sparse FIFO draining and its synchronous admission API; +6. add low-priority scheduling, failure observability, close fencing, crash recovery, and Harper's + exclusive-owner integration. Measure tail accumulation and add tail-supersession entries only if + that data justifies publication-path cost. 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. @@ -310,10 +377,20 @@ writer retirement. The release benchmark adds retained and churned opens plus sh distinct-file concurrency. Distinct-file cases run at one, two, four, and eight threads; shared-file cases start at two threads. +Slice 5 tests durable per-sequence progress, crash/reopen resumption, out-of-order completion behind +a real retained handle, later head compaction across holes, multi-batch entry completion, +tail-bounded cursor resumption, applied-but-reported-failed final batches, ambiguous enqueue-result +recovery, a cold-tail-read race with concurrent enqueue, terminal corruption latching without losing +healthy-shard accounting, concurrent admission, staged-prefix probing, and exact read and mutation +budgets. Format validation is charged to the same point-read and elapsed-work budget as queue +processing. A real Tantivy merge and garbage-collection cycle is reopened after the queue drains to +prove that physical cleanup preserves the index. The release benchmark times reclamation of deleted +4 KiB directory objects with creation and logical deletion outside the measured region. + 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 +and active writers, physical reclamation, 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 @@ -330,7 +407,7 @@ abandoned writers, partial object cleanup, restart during a range and between FI 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 +payload deletion, progress updates, out-of-order completion, 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. @@ -345,7 +422,8 @@ owner loss with a second worker, close drain, cleanup health reporting, backup/r 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 +skip cost, out-of-order completion and bounded head compaction, 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. diff --git a/src/phase0.rs b/src/phase0.rs index ff6656d..e64cc21 100644 --- a/src/phase0.rs +++ b/src/phase0.rs @@ -6,7 +6,7 @@ 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, RwLock, Weak}; +use std::sync::{Arc, Condvar, Mutex, OnceLock, RwLock, TryLockError, Weak}; use std::time::{Duration, Instant}; use async_trait::async_trait; @@ -38,6 +38,42 @@ impl WritePolicy { }; } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ReclaimBudget { + pub max_point_reads: usize, + pub max_mutations: usize, + pub max_request_bytes: usize, + pub max_elapsed: Duration, +} + +impl Default for ReclaimBudget { + fn default() -> Self { + Self { + max_point_reads: RECLAIM_SHARD_COUNT * 2 + 4, + max_mutations: RECLAIM_MAX_BATCH_MUTATIONS, + max_request_bytes: RECLAIM_MAX_BATCH_REQUEST_BYTES, + max_elapsed: Duration::from_millis(10), + } + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct ReclaimOutcome { + pub busy: bool, + pub budget_exhausted: bool, + pub has_more: bool, + pub no_progress: bool, + pub terminal_shard_mask: u64, + pub terminal_error: Option, + pub point_reads: usize, + pub write_batches: usize, + pub mutations: usize, + pub request_bytes: usize, + pub payload_delete_mutations: usize, + pub entries_reclaimed: usize, + pub pinned_skips: usize, +} + #[derive(Clone, Debug, PartialEq, Eq)] pub enum Mutation { Put(Vec, Vec), @@ -70,6 +106,28 @@ impl Mutation { Self::Delete(_) => None, } } + + fn request_bytes(&self) -> io::Result { + let bytes = 1_usize + .checked_add(4) + .and_then(|bytes| bytes.checked_add(self.key().len())) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "mutation request size overflow"))?; + match self { + Self::Put(_, value) => bytes + .checked_add(4) + .and_then(|bytes| bytes.checked_add(value.len())) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "mutation request size overflow")), + Self::Delete(_) => Ok(bytes), + } + } +} + +fn mutation_batch_request_bytes(mutations: &[Mutation]) -> io::Result { + mutations.iter().try_fold(7_usize, |bytes, mutation| { + bytes + .checked_add(mutation.request_bytes()?) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "mutation batch request size overflow")) + }) } #[derive(Clone, Debug)] @@ -86,6 +144,8 @@ struct State { pending_wal: Vec, VersionedValue)>>, fail_next_write: bool, fail_after_next_write: bool, + fail_read_after_next_write: bool, + fail_next_read: bool, fail_next_flush: bool, } @@ -142,6 +202,7 @@ impl FaultingKv { return Err(io::Error::other("injected write failure")); } let fail_after_write = std::mem::take(&mut state.fail_after_next_write); + let fail_read_after_write = std::mem::take(&mut state.fail_read_after_next_write); let mut wal_batch = Vec::with_capacity(mutations.len()); for mutation in mutations { state.next_sequence += 1; @@ -166,6 +227,9 @@ impl FaultingKv { } } } + if fail_read_after_write { + state.fail_next_read = true; + } if fail_after_write { Err(io::Error::other("injected post-commit write failure")) } else { @@ -201,6 +265,8 @@ impl FaultingKv { pending_wal: Vec::new(), fail_next_write: false, fail_after_next_write: false, + fail_read_after_next_write: false, + fail_next_read: false, fail_next_flush: false, })), } @@ -214,6 +280,12 @@ impl FaultingKv { self.state.lock().unwrap().fail_after_next_write = true; } + pub fn fail_after_next_write_and_next_read(&self) { + let mut state = self.state.lock().unwrap(); + state.fail_after_next_write = true; + state.fail_read_after_next_write = true; + } + pub fn fail_next_flush(&self) { self.state.lock().unwrap().fail_next_flush = true; } @@ -225,7 +297,15 @@ impl KvStore for FaultingKv { } fn read(&self, key: &[u8]) -> io::Result> { - Ok(self.get(key).map(OwnedBytes::new)) + let mut state = self.state.lock().unwrap(); + if std::mem::take(&mut state.fail_next_read) { + return Err(io::Error::other("injected read failure")); + } + Ok(state + .visible + .get(key) + .and_then(|entry| entry.value.clone()) + .map(OwnedBytes::new)) } fn write(&self, mutations: &[Mutation], policy: WritePolicy) -> io::Result<()> { @@ -268,11 +348,25 @@ struct DirectoryState { reader_pins: Arc, reader_registration_shards: [RwLock<()>; READER_REGISTRATION_SHARD_COUNT], reclaim_shards: [Mutex<()>; RECLAIM_SHARD_COUNT], + reclaim_hints: [ReclaimHint; RECLAIM_SHARD_COUNT], + reclaim_consumer: Mutex<()>, + reclaim_next_shard: AtomicUsize, + reclaim_terminal_shard_mask: AtomicU64, locks: Mutex, locks_changed: Condvar, watches: WatchCallbackList, } +#[derive(Default)] +struct ReclaimHint { + head: AtomicU64, + head_known: AtomicBool, + tail: AtomicU64, + tail_known: AtomicBool, + tail_invalidations: AtomicU64, + next_sequence: AtomicU64, +} + #[derive(Default)] struct ObjectIdAllocator { next: u64, @@ -359,6 +453,64 @@ struct DirectoryIdentity { static DIRECTORY_STATES: OnceLock>>> = OnceLock::new(); +enum Budgeted { + Performed(T), + Exhausted, +} + +struct ReclaimAdmission { + budget: ReclaimBudget, + started: Instant, + outcome: ReclaimOutcome, +} + +impl ReclaimAdmission { + fn new(budget: ReclaimBudget) -> Self { + Self { + budget, + started: Instant::now(), + outcome: ReclaimOutcome::default(), + } + } + + fn elapsed(&self) -> bool { + self.started.elapsed() >= self.budget.max_elapsed + } + + fn read(&mut self, store: &S, key: &[u8]) -> io::Result>> { + if self.outcome.point_reads >= self.budget.max_point_reads || self.elapsed() { + self.outcome.budget_exhausted = true; + return Ok(Budgeted::Exhausted); + } + self.outcome.point_reads += 1; + store.read(key).map(Budgeted::Performed) + } + + fn batch_fits(&self, mutations: usize, request_bytes: usize) -> bool { + mutations <= RECLAIM_MAX_BATCH_MUTATIONS + && request_bytes <= RECLAIM_MAX_BATCH_REQUEST_BYTES + && self.admission_fits(mutations, request_bytes) + } + + fn admission_fits(&self, mutations: usize, request_bytes: usize) -> bool { + self.outcome.mutations.saturating_add(mutations) <= self.budget.max_mutations + && self.outcome.request_bytes.saturating_add(request_bytes) <= self.budget.max_request_bytes + } + + fn write(&mut self, store: &S, mutations: &[Mutation]) -> io::Result> { + let request_bytes = mutation_batch_request_bytes(mutations)?; + if self.elapsed() || !self.batch_fits(mutations.len(), request_bytes) { + self.outcome.budget_exhausted = true; + return Ok(Budgeted::Exhausted); + } + self.outcome.write_batches += 1; + self.outcome.mutations += mutations.len(); + self.outcome.request_bytes += request_bytes; + store.write(mutations, WritePolicy::WAL)?; + Ok(Budgeted::Performed(())) + } +} + impl PathRegistry { fn state(self: &Arc, path: &Path) -> Arc { let mut states = self.states.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); @@ -482,7 +634,6 @@ impl ReaderPinRegistry { }) } - #[cfg(test)] fn is_pinned(&self, object_id: u64, tail_revision: Option) -> bool { let shard = usize::from(reclaim_shard(object_id)); let objects = self.shards[shard] @@ -525,7 +676,6 @@ impl ReaderRevisions { Ok(()) } - #[cfg(test)] fn contains(&self, tail_revision: u64) -> bool { match self { Self::None => false, @@ -672,6 +822,10 @@ impl KvDirectory { reader_pins: Arc::new(ReaderPinRegistry::default()), reader_registration_shards: std::array::from_fn(|_| RwLock::new(())), reclaim_shards: std::array::from_fn(|_| Mutex::new(())), + reclaim_hints: std::array::from_fn(|_| ReclaimHint::default()), + reclaim_consumer: Mutex::new(()), + reclaim_next_shard: AtomicUsize::new(0), + reclaim_terminal_shard_mask: AtomicU64::new(0), locks: Mutex::new(DirectoryLocks::default()), locks_changed: Condvar::new(), watches: WatchCallbackList::default(), @@ -754,7 +908,7 @@ impl KvDirectory { Ok(object_id) } - fn reclaim_enqueue_mutations(&self, binding: &Binding) -> io::Result> { + fn reclaim_enqueue_mutations(&self, binding: &Binding) -> io::Result<(Vec, u64)> { let shard = reclaim_shard(binding.object_id); let tail_key = reclaim_tail_key(&self.namespace, shard); let sequence = self @@ -773,10 +927,460 @@ impl KvDirectory { let next = sequence .checked_add(1) .ok_or_else(|| io::Error::other("reclaim queue sequence exhausted"))?; - Ok(vec![ - Mutation::Put(entry_key, encode_reclaim_entry(&ReclaimEntry::from_binding(binding))?), - Mutation::Put(tail_key, next.to_be_bytes().to_vec()), - ]) + Ok(( + vec![ + Mutation::Put(entry_key, encode_reclaim_entry(&ReclaimEntry::from_binding(binding))?), + Mutation::Put(tail_key, next.to_be_bytes().to_vec()), + ], + next, + )) + } + + fn record_reclaim_tail(&self, shard: u8, tail: u64) { + let hint = &self.state.reclaim_hints[usize::from(shard)]; + hint.tail.fetch_max(tail, Ordering::AcqRel); + hint.tail_known.store(true, Ordering::Release); + } + + fn invalidate_reclaim_tail(&self, shard: u8) { + let hint = &self.state.reclaim_hints[usize::from(shard)]; + hint.tail_invalidations.fetch_add(1, Ordering::AcqRel); + hint.tail_known.store(false, Ordering::Release); + } + + pub fn reclaim(&self, budget: ReclaimBudget) -> io::Result { + self.validate_reclaim_budget(budget)?; + let mut admission = ReclaimAdmission::new(budget); + if matches!(self.ensure_reclaim_format(&mut admission)?, Budgeted::Exhausted) { + admission.outcome.has_more = true; + admission.outcome.no_progress = true; + return Ok(admission.outcome); + } + if self.format.state.load(Ordering::Acquire) == FORMAT_ABSENT { + admission.outcome.no_progress = true; + return Ok(admission.outcome); + } + let _consumer = match self.state.reclaim_consumer.try_lock() { + Ok(consumer) => consumer, + Err(TryLockError::Poisoned(poisoned)) => poisoned.into_inner(), + Err(TryLockError::WouldBlock) => { + admission.outcome.busy = true; + admission.outcome.no_progress = true; + admission.outcome.has_more = true; + return Ok(admission.outcome); + } + }; + let start = self.state.reclaim_next_shard.fetch_add(1, Ordering::Relaxed) % RECLAIM_SHARD_COUNT; + for offset in 0..RECLAIM_SHARD_COUNT { + if admission.outcome.budget_exhausted { + break; + } + let shard = (start + offset) % RECLAIM_SHARD_COUNT; + if self.state.reclaim_terminal_shard_mask.load(Ordering::Acquire) & (1_u64 << shard) != 0 { + continue; + } + if let Err(error) = self.reclaim_shard(shard as u8, &mut admission) { + let terminal = self.state.reclaim_terminal_shard_mask.load(Ordering::Acquire); + if terminal & (1_u64 << shard) == 0 { + return Err(error); + } + admission + .outcome + .terminal_error + .get_or_insert_with(|| error.to_string()); + } + } + admission.outcome.terminal_shard_mask = self.state.reclaim_terminal_shard_mask.load(Ordering::Acquire); + admission.outcome.has_more = self.state.reclaim_hints.iter().enumerate().any(|(shard, hint)| { + admission.outcome.terminal_shard_mask & (1_u64 << shard) == 0 + && (!hint.head_known.load(Ordering::Acquire) + || !hint.tail_known.load(Ordering::Acquire) + || hint.head.load(Ordering::Acquire) < hint.tail.load(Ordering::Acquire)) + }); + admission.outcome.no_progress = admission.outcome.mutations == 0; + Ok(admission.outcome) + } + + fn ensure_reclaim_format(&self, admission: &mut ReclaimAdmission) -> io::Result> { + let state = self.format.state.load(Ordering::Acquire); + if state == FORMAT_PRESENT { + return Ok(Budgeted::Performed(())); + } + let marker = match admission.read(&self.store, &format_marker_key(&self.namespace))? { + Budgeted::Performed(marker) => marker, + Budgeted::Exhausted => return Ok(Budgeted::Exhausted), + }; + if let Some(state) = decode_format_marker(marker)? { + self.format.record(state); + return Ok(Budgeted::Performed(())); + } + if state == FORMAT_ABSENT { + return Ok(Budgeted::Performed(())); + } + for legacy_key in legacy_sentinel_keys(&self.namespace) { + let value = match admission.read(&self.store, &legacy_key)? { + Budgeted::Performed(value) => value, + Budgeted::Exhausted => return Ok(Budgeted::Exhausted), + }; + if value.is_some() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "the namespace contains the unsupported prototype directory format", + )); + } + } + self.format.record(FORMAT_ABSENT); + Ok(Budgeted::Performed(())) + } + + fn validate_reclaim_budget(&self, budget: ReclaimBudget) -> io::Result<()> { + if budget.max_point_reads < 4 || budget.max_mutations < 4 || budget.max_elapsed.is_zero() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "reclaim budget must admit four point reads, four mutations, and elapsed work", + )); + } + let minimum_batch = vec![ + Mutation::Delete(tail_key(&self.namespace, u64::MAX, u64::MAX)), + Mutation::Delete(reclaim_progress_key(&self.namespace, u8::MAX, u64::MAX)), + Mutation::Delete(reclaim_entry_key(&self.namespace, u8::MAX, u64::MAX)), + Mutation::Put( + reclaim_head_key(&self.namespace, u8::MAX), + u64::MAX.to_be_bytes().to_vec(), + ), + ]; + let minimum_bytes = mutation_batch_request_bytes(&minimum_batch)?; + if budget.max_request_bytes < minimum_bytes { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("reclaim request-byte budget must be at least {minimum_bytes}"), + )); + } + Ok(()) + } + + fn reclaim_shard(&self, shard: u8, admission: &mut ReclaimAdmission) -> io::Result<()> { + let Some((mut head, tail)) = self.reclaim_bounds(shard, admission)? else { + return Ok(()); + }; + if head == tail { + return Ok(()); + } + let hint = &self.state.reclaim_hints[usize::from(shard)]; + let mut sequence = hint.next_sequence.load(Ordering::Relaxed); + if sequence < head || sequence >= tail { + sequence = head; + } + let depth = tail - head; + let mut examined = 0_u64; + while sequence < tail && examined < depth && !admission.outcome.budget_exhausted { + let entry_key = reclaim_entry_key(&self.namespace, shard, sequence); + let entry_bytes = match admission.read(&self.store, &entry_key)? { + Budgeted::Performed(entry) => entry, + Budgeted::Exhausted => break, + }; + let progress_key = reclaim_progress_key(&self.namespace, shard, sequence); + let Some(entry_bytes) = entry_bytes else { + let progress = match admission.read(&self.store, &progress_key)? { + Budgeted::Performed(progress) => progress, + Budgeted::Exhausted => break, + }; + if progress.is_some() { + return Err(self.terminal_reclaim_error( + shard, + io::Error::new( + io::ErrorKind::InvalidData, + "missing reclaim entry has persisted progress", + ), + )); + } + if sequence == head { + let next = head.checked_add(1).ok_or_else(|| { + self.terminal_reclaim_error(shard, io::Error::other("reclaim head exhausted")) + })?; + let mutation = Mutation::Put(reclaim_head_key(&self.namespace, shard), next.to_be_bytes().to_vec()); + match admission.write(&self.store, std::slice::from_ref(&mutation))? { + Budgeted::Performed(()) => { + head = next; + hint.head.store(head, Ordering::Release); + hint.head_known.store(true, Ordering::Release); + } + Budgeted::Exhausted => break, + } + } + sequence += 1; + examined += 1; + continue; + }; + let entry = + decode_reclaim_entry(&entry_bytes).map_err(|error| self.terminal_reclaim_error(shard, error))?; + if reclaim_shard(entry.object_id) != shard { + return Err(self.terminal_reclaim_error( + shard, + io::Error::new(io::ErrorKind::InvalidData, "reclaim entry is stored in the wrong shard"), + )); + } + if self.state.reader_pins.is_pinned(entry.object_id, None) { + admission.outcome.pinned_skips += 1; + sequence += 1; + examined += 1; + continue; + } + let progress = match admission.read(&self.store, &progress_key)? { + Budgeted::Performed(Some(progress)) => decode_reclaim_progress(&progress, &entry) + .map_err(|error| self.terminal_reclaim_error(shard, error))?, + Budgeted::Performed(None) => ReclaimProgress::default(), + Budgeted::Exhausted => break, + }; + match self.reclaim_entry(shard, sequence, head, &entry, progress, admission)? { + Budgeted::Performed(true) => { + if sequence == head { + head += 1; + } + } + Budgeted::Performed(false) => continue, + Budgeted::Exhausted => break, + } + sequence += 1; + examined += 1; + } + hint.next_sequence.store( + if sequence >= tail || examined >= depth { + head + } else { + sequence + }, + Ordering::Relaxed, + ); + Ok(()) + } + + fn reclaim_bounds(&self, shard: u8, admission: &mut ReclaimAdmission) -> io::Result> { + let hint = &self.state.reclaim_hints[usize::from(shard)]; + let head = if hint.head_known.load(Ordering::Acquire) { + hint.head.load(Ordering::Acquire) + } else { + let key = reclaim_head_key(&self.namespace, shard); + let value = match admission.read(&self.store, &key)? { + Budgeted::Performed(value) => value, + Budgeted::Exhausted => return Ok(None), + }; + let head = value + .map(|value| decode_u64(&value)) + .transpose() + .map_err(|error| self.terminal_reclaim_error(shard, error))? + .unwrap_or(0); + hint.head.store(head, Ordering::Release); + hint.head_known.store(true, Ordering::Release); + head + }; + let tail = if hint.tail_known.load(Ordering::Acquire) { + hint.tail.load(Ordering::Acquire) + } else { + let invalidations = hint.tail_invalidations.load(Ordering::Acquire); + let key = reclaim_tail_key(&self.namespace, shard); + let value = match admission.read(&self.store, &key)? { + Budgeted::Performed(value) => value, + Budgeted::Exhausted => return Ok(None), + }; + let tail = value + .map(|value| decode_u64(&value)) + .transpose() + .map_err(|error| self.terminal_reclaim_error(shard, error))? + .unwrap_or(0); + hint.tail.fetch_max(tail, Ordering::AcqRel); + hint.tail_known.store(true, Ordering::Release); + if hint.tail_invalidations.load(Ordering::Acquire) != invalidations { + hint.tail_known.store(false, Ordering::Release); + } + hint.tail.load(Ordering::Acquire) + }; + if head > tail { + return Err(self.terminal_reclaim_error( + shard, + io::Error::new(io::ErrorKind::InvalidData, "reclaim head exceeds its tail"), + )); + } + Ok(Some((head, tail))) + } + + fn reclaim_entry( + &self, + shard: u8, + sequence: u64, + head: u64, + entry: &ReclaimEntry, + mut progress: ReclaimProgress, + admission: &mut ReclaimAdmission, + ) -> io::Result> { + let progress_key = reclaim_progress_key(&self.namespace, shard, sequence); + let entry_key = reclaim_entry_key(&self.namespace, shard, sequence); + let next_head = sequence + .checked_add(1) + .ok_or_else(|| self.terminal_reclaim_error(shard, io::Error::other("reclaim sequence exhausted")))?; + let mut final_mutations = vec![Mutation::Delete(progress_key.clone()), Mutation::Delete(entry_key)]; + if sequence == head { + final_mutations.push(Mutation::Put( + reclaim_head_key(&self.namespace, shard), + next_head.to_be_bytes().to_vec(), + )); + } + let progress_mutation = Mutation::Put(progress_key, encode_reclaim_progress(progress)); + let final_bytes = mutation_batch_request_bytes(&final_mutations)?; + let progress_bytes = mutation_batch_request_bytes(std::slice::from_ref(&progress_mutation))?; + let reserved_mutations = final_mutations.len().max(1); + let reserved_payload_bytes = final_bytes.max(progress_bytes) - 7; + if !admission.batch_fits(reserved_mutations, reserved_payload_bytes + 7) { + admission.outcome.budget_exhausted = true; + return Ok(Budgeted::Exhausted); + } + + let remaining_chunks = u64::from(entry.chunk_high_water.saturating_sub(progress.next_chunk)); + let remaining_tails = entry + .tail_high_water + .saturating_add(1) + .saturating_sub(progress.next_tail); + let remaining_work = usize::try_from(remaining_chunks.saturating_add(remaining_tails)).unwrap_or(usize::MAX); + let capacity = RECLAIM_MAX_BATCH_MUTATIONS + .min( + admission + .budget + .max_mutations + .saturating_sub(admission.outcome.mutations), + ) + .min(remaining_work.saturating_add(reserved_mutations)); + let mut mutations = Vec::with_capacity(capacity); + let mut payload_bytes = 0_usize; + let mut progress_changed = false; + while progress.next_chunk < entry.full_chunks { + let mutation = Mutation::Delete(chunk_key(&self.namespace, entry.object_id, progress.next_chunk)); + if !self.reclaim_payload_fits( + admission, + &mutations, + payload_bytes, + &mutation, + reserved_mutations, + reserved_payload_bytes, + )? { + break; + } + payload_bytes += mutation.request_bytes()?; + mutations.push(mutation); + progress.next_chunk += 1; + progress_changed = true; + } + while progress.next_chunk < entry.chunk_high_water && !admission.outcome.budget_exhausted { + let key = chunk_key(&self.namespace, entry.object_id, progress.next_chunk); + let exists = match admission.read(&self.store, &key)? { + Budgeted::Performed(value) => value.is_some(), + Budgeted::Exhausted => break, + }; + if !exists { + progress.next_chunk = entry.chunk_high_water; + progress_changed = true; + break; + } + let mutation = Mutation::Delete(key); + if !self.reclaim_payload_fits( + admission, + &mutations, + payload_bytes, + &mutation, + reserved_mutations, + reserved_payload_bytes, + )? { + break; + } + payload_bytes += mutation.request_bytes()?; + mutations.push(mutation); + progress.next_chunk += 1; + progress_changed = true; + } + while progress.next_chunk == entry.chunk_high_water && progress.next_tail <= entry.tail_high_water { + let mutation = Mutation::Delete(tail_key(&self.namespace, entry.object_id, progress.next_tail)); + if !self.reclaim_payload_fits( + admission, + &mutations, + payload_bytes, + &mutation, + reserved_mutations, + reserved_payload_bytes, + )? { + break; + } + payload_bytes += mutation.request_bytes()?; + mutations.push(mutation); + progress.next_tail += 1; + progress_changed = true; + } + let completed = progress.next_chunk == entry.chunk_high_water && progress.next_tail > entry.tail_high_water; + if completed { + mutations.extend(final_mutations); + } else if progress_changed { + mutations.push(Mutation::Put( + reclaim_progress_key(&self.namespace, shard, sequence), + encode_reclaim_progress(progress), + )); + } else { + admission.outcome.budget_exhausted = true; + return Ok(Budgeted::Exhausted); + } + let payload_deletes = mutations.len() - if completed { reserved_mutations } else { 1 }; + match admission.write(&self.store, &mutations)? { + Budgeted::Performed(()) => { + admission.outcome.payload_delete_mutations += payload_deletes; + if completed { + admission.outcome.entries_reclaimed += 1; + if sequence == head { + let hint = &self.state.reclaim_hints[usize::from(shard)]; + hint.head.store(next_head, Ordering::Release); + hint.head_known.store(true, Ordering::Release); + } + } + Ok(Budgeted::Performed(completed)) + } + Budgeted::Exhausted => Ok(Budgeted::Exhausted), + } + } + + fn reclaim_payload_fits( + &self, + admission: &mut ReclaimAdmission, + mutations: &[Mutation], + payload_bytes: usize, + candidate: &Mutation, + reserved_mutations: usize, + reserved_payload_bytes: usize, + ) -> io::Result { + let candidate_bytes = candidate.request_bytes()?; + let mutation_count = mutations + .len() + .checked_add(1) + .and_then(|count| count.checked_add(reserved_mutations)) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "reclaim mutation count overflow"))?; + let request_bytes = 7_usize + .checked_add(payload_bytes) + .and_then(|bytes| bytes.checked_add(candidate_bytes)) + .and_then(|bytes| bytes.checked_add(reserved_payload_bytes)) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "reclaim request size overflow"))?; + if admission.batch_fits(mutation_count, request_bytes) { + Ok(true) + } else { + if !admission.admission_fits(mutation_count, request_bytes) { + admission.outcome.budget_exhausted = true; + } + Ok(false) + } + } + + fn terminal_reclaim_error(&self, shard: u8, error: io::Error) -> io::Error { + self.state + .reclaim_terminal_shard_mask + .fetch_or(1_u64 << shard, Ordering::AcqRel); + io::Error::new( + io::ErrorKind::InvalidData, + format!("reclaim shard {shard} is corrupt: {error}"), + ) } fn pinned_binding( @@ -1010,8 +1614,8 @@ impl Directory for KvDirectory { let _queue = self.state.reclaim_shards[usize::from(shard)] .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); - let mut mutations = match self.reclaim_enqueue_mutations(&final_binding) { - Ok(mutations) => mutations, + let (mut mutations, next_reclaim_tail) = match self.reclaim_enqueue_mutations(&final_binding) { + Ok(enqueue) => enqueue, Err(error) => { if let Some(fence) = &fence { fence.reactivate(); @@ -1029,11 +1633,13 @@ impl Directory for KvDirectory { }; match write_result { Ok(()) => { + self.record_reclaim_tail(shard, next_reclaim_tail); lifecycle.writer = None; Ok(()) } Err(error) => match self.store.read(&binding_key) { Ok(None) => { + self.record_reclaim_tail(shard, next_reclaim_tail); lifecycle.writer = None; Ok(()) } @@ -1043,7 +1649,10 @@ impl Directory for KvDirectory { } Err(delete_io_error(path, error)) } - Ok(Some(_)) | Err(_) => Err(delete_io_error(path, error)), + Ok(Some(_)) | Err(_) => { + self.invalidate_reclaim_tail(shard); + Err(delete_io_error(path, error)) + } }, } } @@ -1494,6 +2103,7 @@ impl Binding { #[derive(Clone, Debug, PartialEq, Eq)] struct ReclaimEntry { object_id: u64, + full_chunks: u32, chunk_high_water: u32, tail_high_water: u64, } @@ -1502,12 +2112,28 @@ impl ReclaimEntry { fn from_binding(binding: &Binding) -> Self { Self { object_id: binding.object_id, + full_chunks: binding.full_chunks, chunk_high_water: binding.chunk_high_water, tail_high_water: binding.tail_high_water, } } } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct ReclaimProgress { + next_chunk: u32, + next_tail: u64, +} + +impl Default for ReclaimProgress { + fn default() -> Self { + Self { + next_chunk: 0, + next_tail: 1, + } + } +} + fn counter_key(namespace: &[u8]) -> Vec { namespaced_prefix(namespace, KEY_KIND_COUNTER) } @@ -1520,14 +2146,18 @@ fn atomic_key(namespace: &[u8], path: &Path) -> Vec { prefixed_path(&namespaced_prefix(namespace, KEY_KIND_ATOMIC), path) } -const KEY_FORMAT_VERSION: u8 = 2; +const KEY_FORMAT_VERSION: u8 = 3; const BINDING_FORMAT_VERSION: u8 = 3; const CHUNK_RESERVATION_STRIDE: u32 = 64; const OBJECT_ID_RESERVATION_STRIDE: u64 = 1_024; const MAX_OBJECT_EXTENT_BYTES: u128 = 1 << 40; -const RECLAIM_ENTRY_FORMAT_VERSION: u8 = 1; +const RECLAIM_ENTRY_FORMAT_VERSION: u8 = 2; +const RECLAIM_PROGRESS_FORMAT_VERSION: u8 = 1; const RECLAIM_ENTRY_WHOLE_OBJECT: u8 = 1; const RECLAIM_SHARD_COUNT: usize = 64; +const _: () = assert!(RECLAIM_SHARD_COUNT <= u64::BITS as usize); +const RECLAIM_MAX_BATCH_MUTATIONS: usize = 512; +const RECLAIM_MAX_BATCH_REQUEST_BYTES: usize = 64 * 1024; const READER_REGISTRATION_SHARD_COUNT: usize = 256; const KEY_KIND_COUNTER: u8 = 1; const KEY_KIND_BINDING: u8 = 2; @@ -1536,6 +2166,8 @@ const KEY_KIND_CHUNK: u8 = 4; const KEY_KIND_TAIL: u8 = 5; const KEY_KIND_RECLAIM_TAIL: u8 = 7; const KEY_KIND_RECLAIM_ENTRY: u8 = 8; +const KEY_KIND_RECLAIM_HEAD: u8 = 9; +const KEY_KIND_RECLAIM_PROGRESS: u8 = 10; const KEY_PREFIX: &[u8; 4] = b"HFTK"; const FORMAT_MARKER_PREFIX: &[u8; 4] = b"HFTM"; const FORMAT_UNKNOWN: u8 = 0; @@ -1578,7 +2210,11 @@ fn validate_format(store: &S, namespace: &[u8], create: bool) -> io: } fn read_format_marker(store: &S, marker: &[u8]) -> io::Result> { - let Some(value) = store.read(marker)? else { + decode_format_marker(store.read(marker)?) +} + +fn decode_format_marker(value: Option) -> io::Result> { + let Some(value) = value else { return Ok(None); }; match value.as_slice() { @@ -1654,6 +2290,19 @@ fn reclaim_entry_key(namespace: &[u8], shard: u8, sequence: u64) -> Vec { key } +fn reclaim_head_key(namespace: &[u8], shard: u8) -> Vec { + let mut key = namespaced_prefix_with_capacity(namespace, KEY_KIND_RECLAIM_HEAD, 1); + key.push(shard); + key +} + +fn reclaim_progress_key(namespace: &[u8], shard: u8, sequence: u64) -> Vec { + let mut key = namespaced_prefix_with_capacity(namespace, KEY_KIND_RECLAIM_PROGRESS, 9); + key.push(shard); + key.extend_from_slice(&sequence.to_be_bytes()); + key +} + fn reclaim_shard(object_id: u64) -> u8 { (object_id % RECLAIM_SHARD_COUNT as u64) as u8 } @@ -1678,23 +2327,23 @@ fn encode_binding(binding: &Binding) -> Vec { } fn encode_reclaim_entry(entry: &ReclaimEntry) -> io::Result> { - validate_physical_extent(entry.chunk_high_water, entry.tail_high_water, "reclaim entry")?; + validate_reclaim_entry(entry)?; if entry.object_id == 0 { return Err(io::Error::new( io::ErrorKind::InvalidData, "reclaim entry has an invalid object id", )); } - let mut bytes = Vec::with_capacity(22); + let mut bytes = Vec::with_capacity(26); bytes.push(RECLAIM_ENTRY_FORMAT_VERSION); bytes.push(RECLAIM_ENTRY_WHOLE_OBJECT); bytes.extend_from_slice(&entry.object_id.to_be_bytes()); + bytes.extend_from_slice(&entry.full_chunks.to_be_bytes()); bytes.extend_from_slice(&entry.chunk_high_water.to_be_bytes()); bytes.extend_from_slice(&entry.tail_high_water.to_be_bytes()); Ok(bytes) } -#[cfg(test)] fn decode_reclaim_entry(bytes: &[u8]) -> io::Result { if bytes.first().copied() != Some(RECLAIM_ENTRY_FORMAT_VERSION) { let version = bytes.first().copied().unwrap_or(0); @@ -1712,19 +2361,24 @@ fn decode_reclaim_entry(bytes: &[u8]) -> io::Result { format!("unsupported reclaim entry kind {}", bytes[1]), )); } - if bytes.len() != 22 { + if bytes.len() != 26 { return Err(io::Error::new(io::ErrorKind::InvalidData, "malformed reclaim entry")); } let entry = ReclaimEntry { object_id: decode_u64(&bytes[2..10])?, - chunk_high_water: u32::from_be_bytes( + full_chunks: u32::from_be_bytes( bytes[10..14] + .try_into() + .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid reclaim published chunk count"))?, + ), + chunk_high_water: u32::from_be_bytes( + bytes[14..18] .try_into() .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid reclaim chunk high-water"))?, ), - tail_high_water: decode_u64(&bytes[14..22])?, + tail_high_water: decode_u64(&bytes[18..26])?, }; - validate_physical_extent(entry.chunk_high_water, entry.tail_high_water, "reclaim entry")?; + validate_reclaim_entry(&entry)?; if entry.object_id == 0 { return Err(io::Error::new( io::ErrorKind::InvalidData, @@ -1734,6 +2388,56 @@ fn decode_reclaim_entry(bytes: &[u8]) -> io::Result { Ok(entry) } +fn validate_reclaim_entry(entry: &ReclaimEntry) -> io::Result<()> { + if entry.full_chunks > entry.chunk_high_water { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "reclaim published chunk count exceeds its high-water", + )); + } + validate_physical_extent(entry.chunk_high_water, entry.tail_high_water, "reclaim entry") +} + +fn encode_reclaim_progress(progress: ReclaimProgress) -> Vec { + let mut bytes = Vec::with_capacity(13); + bytes.push(RECLAIM_PROGRESS_FORMAT_VERSION); + bytes.extend_from_slice(&progress.next_chunk.to_be_bytes()); + bytes.extend_from_slice(&progress.next_tail.to_be_bytes()); + bytes +} + +fn decode_reclaim_progress(bytes: &[u8], entry: &ReclaimEntry) -> io::Result { + if bytes.first().copied() != Some(RECLAIM_PROGRESS_FORMAT_VERSION) { + let version = bytes.first().copied().unwrap_or(0); + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("unsupported reclaim progress format version {version}"), + )); + } + if bytes.len() != 13 { + return Err(io::Error::new(io::ErrorKind::InvalidData, "malformed reclaim progress")); + } + let progress = ReclaimProgress { + next_chunk: u32::from_be_bytes( + bytes[1..5] + .try_into() + .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid reclaim chunk progress"))?, + ), + next_tail: decode_u64(&bytes[5..13])?, + }; + let tail_end = entry + .tail_high_water + .checked_add(1) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "reclaim tail progress exceeds its format"))?; + if progress.next_chunk > entry.chunk_high_water || !(1..=tail_end).contains(&progress.next_tail) { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "reclaim progress is outside its entry", + )); + } + Ok(progress) +} + fn decode_binding(bytes: &[u8]) -> io::Result { if bytes.first().copied() != Some(BINDING_FORMAT_VERSION) { let version = bytes.first().copied().unwrap_or(0); @@ -1833,7 +2537,8 @@ mod tests { #[derive(Clone)] struct BlockingKv { inner: FaultingKv, - block: Arc<(Mutex, Condvar)>, + write_block: Arc<(Mutex, Condvar)>, + read_block: Arc<(Mutex, Condvar)>, } #[derive(Default)] @@ -1909,12 +2614,13 @@ mod tests { fn new() -> Self { Self { inner: FaultingKv::default(), - block: Arc::new((Mutex::new(BlockState::default()), Condvar::new())), + write_block: Arc::new((Mutex::new(BlockState::default()), Condvar::new())), + read_block: Arc::new((Mutex::new(BlockState::default()), Condvar::new())), } } fn arm_next_write(&self) { - let (state, _) = &*self.block; + let (state, _) = &*self.write_block; let mut state = state.lock().unwrap(); state.armed = true; state.entered = false; @@ -1922,7 +2628,23 @@ mod tests { } fn wait_until_blocked(&self) -> bool { - let (state, changed) = &*self.block; + Self::wait_until(&self.write_block) + } + + fn arm_next_read(&self) { + let (state, _) = &*self.read_block; + let mut state = state.lock().unwrap(); + state.armed = true; + state.entered = false; + state.released = false; + } + + fn wait_until_read_blocked(&self) -> bool { + Self::wait_until(&self.read_block) + } + + fn wait_until(block: &Arc<(Mutex, Condvar)>) -> bool { + let (state, changed) = &**block; let deadline = Instant::now() + Duration::from_secs(5); let mut state = state.lock().unwrap(); while !state.entered { @@ -1944,7 +2666,15 @@ mod tests { } fn release_write(&self) { - let (state, changed) = &*self.block; + Self::release(&self.write_block); + } + + fn release_read(&self) { + Self::release(&self.read_block); + } + + fn release(block: &Arc<(Mutex, Condvar)>) { + let (state, changed) = &**block; let mut state = state.lock().unwrap(); state.released = true; changed.notify_all(); @@ -1957,11 +2687,22 @@ mod tests { } fn read(&self, key: &[u8]) -> io::Result> { - KvStore::read(&self.inner, key) + let value = KvStore::read(&self.inner, key)?; + let (state, changed) = &*self.read_block; + let mut state = state.lock().unwrap(); + if state.armed { + state.armed = false; + state.entered = true; + changed.notify_all(); + while !state.released { + state = changed.wait(state).unwrap(); + } + } + Ok(value) } fn write(&self, mutations: &[Mutation], policy: WritePolicy) -> io::Result<()> { - let (state, changed) = &*self.block; + let (state, changed) = &*self.write_block; let mut state = state.lock().unwrap(); if state.armed { state.armed = false; @@ -2002,7 +2743,10 @@ mod tests { let store = FaultingKv::default(); store .write( - &[Mutation::Put(format_marker_key(b"catalog"), vec![1])], + &[Mutation::Put( + format_marker_key(b"catalog"), + vec![KEY_FORMAT_VERSION - 1], + )], WritePolicy::WAL_SYNC, ) .unwrap(); @@ -2188,6 +2932,45 @@ mod tests { .count() } + fn generous_reclaim_budget() -> ReclaimBudget { + ReclaimBudget { + max_point_reads: 4_096, + max_mutations: 4_096, + max_request_bytes: 1024 * 1024, + max_elapsed: Duration::from_secs(1), + } + } + + fn drain_reclamation(directory: &KvDirectory) { + for _ in 0..128 { + let outcome = directory.reclaim(generous_reclaim_budget()).unwrap(); + assert_eq!(outcome.terminal_shard_mask, 0); + if !outcome.has_more { + return; + } + assert!(!outcome.no_progress, "reclamation stopped with reachable work"); + } + panic!("reclamation did not reach quiescence"); + } + + fn install_reclaim_entries(store: &FaultingKv, namespace: &[u8], entries: &[ReclaimEntry]) { + assert!(!entries.is_empty()); + let shard = reclaim_shard(entries[0].object_id); + let mut mutations = Vec::with_capacity(entries.len() + 1); + for (sequence, entry) in entries.iter().enumerate() { + assert_eq!(reclaim_shard(entry.object_id), shard); + mutations.push(Mutation::Put( + reclaim_entry_key(namespace, shard, sequence as u64), + encode_reclaim_entry(entry).unwrap(), + )); + } + mutations.push(Mutation::Put( + reclaim_tail_key(namespace, shard), + (entries.len() as u64).to_be_bytes().to_vec(), + )); + store.write(&mutations, WritePolicy::WAL_SYNC).unwrap(); + } + #[test] fn synced_metadata_makes_earlier_wal_objects_durable() { let store = FaultingKv::default(); @@ -2577,6 +3360,69 @@ mod tests { assert!(store.get(&reclaim_entry_key(b"phase0", shard, 1)).is_none()); } + #[test] + fn ambiguous_delete_invalidates_the_cached_reclaim_tail() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + directory.atomic_write(Path::new("marker"), b"ready").unwrap(); + assert!(!directory.reclaim(generous_reclaim_budget()).unwrap().has_more); + let path = Path::new("segment"); + let mut writer = directory.open_write(path).unwrap(); + writer.write_all(b"contents").unwrap(); + writer.terminate().unwrap(); + let binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); + + store.fail_after_next_write_and_next_read(); + assert!(directory.delete(path).is_err()); + let outcome = directory.reclaim(generous_reclaim_budget()).unwrap(); + + assert_eq!(outcome.entries_reclaimed, 1); + assert!(!outcome.has_more); + assert_eq!( + store.get(&tail_key(b"phase0", binding.object_id, binding.tail_revision)), + None + ); + } + + #[test] + fn cold_tail_read_cannot_overwrite_a_concurrent_enqueue() { + let store = BlockingKv::new(); + let directory = KvDirectory::new(store.clone()); + directory.atomic_write(Path::new("marker"), b"ready").unwrap(); + let path = Path::new("segment"); + let mut writer = directory.open_write(path).unwrap(); + writer.write_all(b"contents").unwrap(); + writer.terminate().unwrap(); + let binding = decode_binding(&store.inner.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); + let shard = usize::from(reclaim_shard(binding.object_id)); + for hint in &directory.state.reclaim_hints { + hint.head.store(0, Ordering::Release); + hint.head_known.store(true, Ordering::Release); + hint.tail.store(0, Ordering::Release); + hint.tail_known.store(true, Ordering::Release); + } + directory.state.reclaim_hints[shard] + .tail_known + .store(false, Ordering::Release); + store.arm_next_read(); + let reclaiming = directory.clone(); + let thread = std::thread::spawn(move || reclaiming.reclaim(generous_reclaim_budget())); + assert!(store.wait_until_read_blocked()); + + directory.delete(path).unwrap(); + store.release_read(); + let outcome = thread.join().unwrap().unwrap(); + + assert_eq!(outcome.entries_reclaimed, 1); + assert!(!outcome.has_more); + assert_eq!( + store + .inner + .get(&tail_key(b"phase0", binding.object_id, binding.tail_revision)), + None + ); + } + #[test] fn failed_delete_reactivates_its_writer() { let store = FaultingKv::default(); @@ -3148,7 +3994,9 @@ mod tests { 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")); + assert!(error + .to_string() + .contains(&format!("directory format version {}", KEY_FORMAT_VERSION - 1))); } let path = Path::new("meta.json"); @@ -3176,7 +4024,9 @@ mod tests { let error = FaultingDirectory::with_namespace(store, b"catalog") .exists(Path::new("meta.json")) .unwrap_err(); - assert!(error.to_string().contains("version 3")); + assert!(error + .to_string() + .contains(&format!("version {}", KEY_FORMAT_VERSION + 1))); } #[test] @@ -3466,6 +4316,7 @@ mod tests { fn reclaim_entries_round_trip_and_enforce_the_extent_limit() { let entry = ReclaimEntry { object_id: 17, + full_chunks: 5, chunk_high_water: 7, tail_high_water: 11, }; @@ -3475,6 +4326,7 @@ mod tests { ); let error = encode_reclaim_entry(&ReclaimEntry { object_id: 17, + full_chunks: 0, chunk_high_water: u32::MAX, tail_high_water: u64::MAX, }) @@ -3499,6 +4351,33 @@ mod tests { .unwrap_err() .to_string() .contains("unsupported reclaim entry kind")); + let mut invalid_chunks = entry.clone(); + invalid_chunks.full_chunks = invalid_chunks.chunk_high_water + 1; + assert!(encode_reclaim_entry(&invalid_chunks) + .unwrap_err() + .to_string() + .contains("published chunk count")); + + let progress = ReclaimProgress { + next_chunk: 3, + next_tail: 4, + }; + assert_eq!( + decode_reclaim_progress(&encode_reclaim_progress(progress), &entry).unwrap(), + progress + ); + let mut invalid_progress = encode_reclaim_progress(progress); + invalid_progress[1..5].copy_from_slice(&(entry.chunk_high_water + 1).to_be_bytes()); + assert_eq!( + decode_reclaim_progress(&invalid_progress, &entry).unwrap_err().kind(), + io::ErrorKind::InvalidData + ); + let mut unknown_progress = encode_reclaim_progress(progress); + unknown_progress[0] += 1; + assert!(decode_reclaim_progress(&unknown_progress, &entry) + .unwrap_err() + .to_string() + .contains("progress format version")); } #[test] @@ -3570,6 +4449,407 @@ mod tests { let recovered = store.crash(); assert!(reclaim_entry_count(&recovered, b"phase0") > 0); assert_reclamation_inventory(&recovered, b"phase0"); + drop(writer); + drop(index); + let recovered_directory = FaultingDirectory::new(recovered.clone()); + drain_reclamation(&recovered_directory); + assert_eq!(reclaim_entry_count(&recovered, b"phase0"), 0); + assert_reclamation_inventory(&recovered, b"phase0"); + let reopened = tantivy::Index::open(recovered_directory).unwrap(); + assert_eq!(reopened.searchable_segment_ids().unwrap().len(), 1); + } + + #[test] + fn reclaim_deletes_a_retired_object_and_advances_its_head() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + directory.atomic_write(Path::new("marker"), b"ready").unwrap(); + assert!(!directory.reclaim(generous_reclaim_budget()).unwrap().has_more); + let path = Path::new("segment"); + let mut writer = directory.open_write(path).unwrap(); + writer.write_all(&vec![7; CHUNK_SIZE * 2 + 17]).unwrap(); + writer.terminate().unwrap(); + let binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); + directory.delete(path).unwrap(); + + drain_reclamation(&directory); + + let shard = reclaim_shard(binding.object_id); + assert_eq!(store.get(&chunk_key(b"phase0", binding.object_id, 0)), None); + assert_eq!(store.get(&chunk_key(b"phase0", binding.object_id, 1)), None); + assert_eq!( + store.get(&tail_key(b"phase0", binding.object_id, binding.tail_revision)), + None + ); + assert_eq!(store.get(&reclaim_entry_key(b"phase0", shard, 0)), None); + assert_eq!(store.get(&reclaim_progress_key(b"phase0", shard, 0)), None); + assert_eq!( + decode_u64(&store.get(&reclaim_head_key(b"phase0", shard)).unwrap()).unwrap(), + 1 + ); + } + + #[test] + fn pinned_head_does_not_block_a_later_entry() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + directory.atomic_write(Path::new("marker"), b"ready").unwrap(); + let path = Path::new("pinned-segment"); + let mut writer = directory.open_write(path).unwrap(); + writer.write_all(b"first").unwrap(); + writer.terminate().unwrap(); + let first_binding = decode_binding(&store.get(&binding_key(b"phase0", path)).unwrap()).unwrap(); + let first_handle = directory.open_read(path).unwrap(); + directory.delete(path).unwrap(); + let first = ReclaimEntry::from_binding(&first_binding); + let second = ReclaimEntry { + object_id: first.object_id + RECLAIM_SHARD_COUNT as u64, + full_chunks: 0, + chunk_high_water: 0, + tail_high_water: 1, + }; + let shard = reclaim_shard(first.object_id); + store + .write( + &[ + Mutation::Put( + reclaim_entry_key(b"phase0", shard, 1), + encode_reclaim_entry(&second).unwrap(), + ), + Mutation::Put(reclaim_tail_key(b"phase0", shard), 2_u64.to_be_bytes().to_vec()), + Mutation::Put(tail_key(b"phase0", second.object_id, 1), vec![2]), + ], + WritePolicy::WAL_SYNC, + ) + .unwrap(); + directory.record_reclaim_tail(shard, 2); + + let first_pass = directory.reclaim(generous_reclaim_budget()).unwrap(); + assert_eq!(first_pass.entries_reclaimed, 1); + assert_eq!(first_pass.pinned_skips, 1); + assert!(first_pass.has_more); + assert!(store + .get(&tail_key(b"phase0", first.object_id, first_binding.tail_revision)) + .is_some()); + assert_eq!(first_handle.read_bytes().unwrap().as_slice(), b"first"); + assert_eq!(store.get(&tail_key(b"phase0", second.object_id, 1)), None); + assert!(store.get(&reclaim_entry_key(b"phase0", shard, 0)).is_some()); + assert_eq!(store.get(&reclaim_entry_key(b"phase0", shard, 1)), None); + + drop(first_handle); + drain_reclamation(&directory); + assert_eq!( + store.get(&tail_key(b"phase0", first.object_id, first_binding.tail_revision)), + None + ); + assert_eq!( + decode_u64(&store.get(&reclaim_head_key(b"phase0", shard)).unwrap()).unwrap(), + 2 + ); + } + + #[test] + fn persisted_progress_resumes_after_restart() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + directory.atomic_write(Path::new("marker"), b"ready").unwrap(); + let entry = ReclaimEntry { + object_id: 1, + full_chunks: 10, + chunk_high_water: 10, + tail_high_water: 0, + }; + install_reclaim_entries(&store, b"phase0", std::slice::from_ref(&entry)); + store + .write( + &(0..entry.full_chunks) + .map(|chunk| Mutation::Put(chunk_key(b"phase0", entry.object_id, chunk), vec![chunk as u8])) + .collect::>(), + WritePolicy::WAL_SYNC, + ) + .unwrap(); + let partial = directory + .reclaim(ReclaimBudget { + max_point_reads: 8, + max_mutations: 4, + max_request_bytes: 4_096, + max_elapsed: Duration::from_secs(1), + }) + .unwrap(); + assert!(partial.budget_exhausted); + assert_eq!(partial.payload_delete_mutations, 1); + let progress = + decode_reclaim_progress(&store.get(&reclaim_progress_key(b"phase0", 1, 0)).unwrap(), &entry).unwrap(); + assert_eq!(progress.next_chunk, 1); + directory.atomic_write(Path::new("barrier"), b"durable").unwrap(); + + let recovered = store.crash(); + let reopened = FaultingDirectory::new(recovered.clone()); + drain_reclamation(&reopened); + for chunk in 0..entry.full_chunks { + assert_eq!(recovered.get(&chunk_key(b"phase0", entry.object_id, chunk)), None); + } + assert_eq!(recovered.get(&reclaim_progress_key(b"phase0", 1, 0)), None); + } + + #[test] + fn reclaim_finishes_multiple_batches_before_moving_its_cursor() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + directory.atomic_write(Path::new("marker"), b"ready").unwrap(); + let entry = ReclaimEntry { + object_id: 1, + full_chunks: 600, + chunk_high_water: 600, + tail_high_water: 0, + }; + install_reclaim_entries(&store, b"phase0", std::slice::from_ref(&entry)); + store + .write( + &(0..entry.full_chunks) + .map(|chunk| Mutation::Put(chunk_key(b"phase0", entry.object_id, chunk), vec![chunk as u8])) + .collect::>(), + WritePolicy::WAL_SYNC, + ) + .unwrap(); + + let outcome = directory.reclaim(generous_reclaim_budget()).unwrap(); + + assert_eq!(outcome.entries_reclaimed, 1); + assert_eq!(outcome.write_batches, 2); + assert_eq!(outcome.payload_delete_mutations, entry.full_chunks as usize); + assert!(!outcome.has_more); + } + + #[test] + fn resumed_cursor_stops_at_the_snapshotted_tail() { + let store = CountingKv::new(); + let directory = KvDirectory::new(store.clone()); + directory.atomic_write(Path::new("marker"), b"ready").unwrap(); + let entry = ReclaimEntry { + object_id: 1, + full_chunks: 0, + chunk_high_water: 0, + tail_high_water: 0, + }; + install_reclaim_entries(&store.inner, b"phase0", std::slice::from_ref(&entry)); + store + .inner + .write( + &[ + Mutation::Put(reclaim_tail_key(b"phase0", 1), 3_u64.to_be_bytes().to_vec()), + Mutation::Put(reclaim_head_key(b"phase0", 1), 0_u64.to_be_bytes().to_vec()), + ], + WritePolicy::WAL_SYNC, + ) + .unwrap(); + let pin = directory + .state + .reader_pins + .register(&Binding { + object_id: entry.object_id, + full_chunks: 0, + tail_revision: 0, + tail_length: 0, + visible_length: 0, + chunk_high_water: 0, + tail_high_water: 0, + }) + .unwrap(); + let hint = &directory.state.reclaim_hints[1]; + for empty in &directory.state.reclaim_hints { + empty.head.store(0, Ordering::Release); + empty.head_known.store(true, Ordering::Release); + empty.tail.store(0, Ordering::Release); + empty.tail_known.store(true, Ordering::Release); + } + hint.head.store(0, Ordering::Release); + hint.head_known.store(true, Ordering::Release); + hint.tail.store(3, Ordering::Release); + hint.tail_known.store(true, Ordering::Release); + hint.next_sequence.store(2, Ordering::Release); + store.take_io_counts(); + + let outcome = directory.reclaim(generous_reclaim_budget()).unwrap(); + + assert_eq!(outcome.point_reads, 2); + assert_eq!(store.take_reads(), 2); + assert_eq!(hint.next_sequence.load(Ordering::Acquire), 0); + drop(pin); + } + + #[test] + fn reclaim_probes_only_the_staged_chunk_prefix() { + let store = CountingKv::new(); + let directory = KvDirectory::new(store.clone()); + directory.atomic_write(Path::new("marker"), b"ready").unwrap(); + let entry = ReclaimEntry { + object_id: 1, + full_chunks: 2, + chunk_high_water: CHUNK_RESERVATION_STRIDE, + tail_high_water: 0, + }; + install_reclaim_entries(&store.inner, b"phase0", std::slice::from_ref(&entry)); + store + .inner + .write( + &[ + Mutation::Put(chunk_key(b"phase0", entry.object_id, 0), vec![0]), + Mutation::Put(chunk_key(b"phase0", entry.object_id, 1), vec![1]), + ], + WritePolicy::WAL_SYNC, + ) + .unwrap(); + store.take_io_counts(); + + let outcome = directory.reclaim(generous_reclaim_budget()).unwrap(); + assert_eq!(outcome.entries_reclaimed, 1); + assert_eq!(outcome.payload_delete_mutations, 2); + assert_eq!(store.take_io_counts().2, outcome.mutations); + } + + #[test] + fn empty_reclaim_hints_avoid_repeated_storage_reads() { + let store = CountingKv::new(); + let directory = KvDirectory::new(store.clone()); + directory.atomic_write(Path::new("marker"), b"ready").unwrap(); + store.take_io_counts(); + + let first = directory.reclaim(generous_reclaim_budget()).unwrap(); + assert_eq!(first.point_reads, RECLAIM_SHARD_COUNT * 2); + assert!(!first.has_more); + store.take_io_counts(); + let second = directory.reclaim(generous_reclaim_budget()).unwrap(); + assert_eq!(second.point_reads, 0); + assert_eq!(store.take_reads(), 0); + } + + #[test] + fn reclaim_rejects_a_budget_that_cannot_make_progress() { + let directory = FaultingDirectory::new(FaultingKv::default()); + let error = directory + .reclaim(ReclaimBudget { + max_point_reads: 3, + ..generous_reclaim_budget() + }) + .unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput); + + let error = directory + .reclaim(ReclaimBudget { + max_request_bytes: 1, + ..generous_reclaim_budget() + }) + .unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput); + } + + #[test] + fn reclaim_format_validation_obeys_the_point_read_budget() { + let store = CountingKv::new(); + let directory = KvDirectory::new(store.clone()); + let outcome = directory + .reclaim(ReclaimBudget { + max_point_reads: 4, + ..generous_reclaim_budget() + }) + .unwrap(); + + assert_eq!(outcome.point_reads, 4); + assert_eq!(store.take_reads(), 4); + assert!(!outcome.budget_exhausted); + assert!(outcome.no_progress); + assert!(!outcome.has_more); + } + + #[test] + fn missing_entry_progress_latches_a_terminal_shard() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + directory.atomic_write(Path::new("marker"), b"ready").unwrap(); + store + .write( + &[ + Mutation::Put(reclaim_tail_key(b"phase0", 0), 1_u64.to_be_bytes().to_vec()), + Mutation::Put( + reclaim_entry_key(b"phase0", 0, 0), + encode_reclaim_entry(&ReclaimEntry { + object_id: 64, + full_chunks: 0, + chunk_high_water: 0, + tail_high_water: 0, + }) + .unwrap(), + ), + Mutation::Put(reclaim_tail_key(b"phase0", 1), 1_u64.to_be_bytes().to_vec()), + Mutation::Put( + reclaim_progress_key(b"phase0", 1, 0), + encode_reclaim_progress(ReclaimProgress::default()), + ), + ], + WritePolicy::WAL_SYNC, + ) + .unwrap(); + + let first = directory.reclaim(generous_reclaim_budget()).unwrap(); + assert_eq!(first.entries_reclaimed, 1); + assert_eq!(first.terminal_shard_mask, 1_u64 << 1); + assert!(first.terminal_error.unwrap().contains("shard 1")); + assert!(!first.has_more); + let next = directory.reclaim(generous_reclaim_budget()).unwrap(); + assert_eq!(next.terminal_shard_mask, 1_u64 << 1); + assert!(!next.has_more); + } + + #[test] + fn applied_but_reported_failed_reclaim_is_retryable() { + let store = FaultingKv::default(); + let directory = FaultingDirectory::new(store.clone()); + directory.atomic_write(Path::new("marker"), b"ready").unwrap(); + let entry = ReclaimEntry { + object_id: 1, + full_chunks: 1, + chunk_high_water: 1, + tail_high_water: 0, + }; + install_reclaim_entries(&store, b"phase0", std::slice::from_ref(&entry)); + store + .write( + &[Mutation::Put(chunk_key(b"phase0", entry.object_id, 0), vec![1])], + WritePolicy::WAL_SYNC, + ) + .unwrap(); + store.fail_after_next_write(); + + assert!(directory.reclaim(generous_reclaim_budget()).is_err()); + drain_reclamation(&directory); + assert_eq!(store.get(&chunk_key(b"phase0", entry.object_id, 0)), None); + assert_eq!(store.get(&reclaim_entry_key(b"phase0", 1, 0)), None); + } + + #[test] + fn concurrent_reclaim_returns_busy() { + let store = BlockingKv::new(); + let directory = KvDirectory::new(store.clone()); + directory.atomic_write(Path::new("marker"), b"ready").unwrap(); + let entry = ReclaimEntry { + object_id: 1, + full_chunks: 0, + chunk_high_water: 0, + tail_high_water: 0, + }; + install_reclaim_entries(&store.inner, b"phase0", std::slice::from_ref(&entry)); + store.arm_next_write(); + let reclaiming = directory.clone(); + let thread = std::thread::spawn(move || reclaiming.reclaim(generous_reclaim_budget())); + assert!(store.wait_until_blocked()); + + let busy = directory.reclaim(generous_reclaim_budget()).unwrap(); + assert!(busy.busy); + assert!(busy.no_progress); + assert!(busy.has_more); + store.release_write(); + thread.join().unwrap().unwrap(); } #[test]