diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index ccdffa1..494cf30 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -75,7 +75,7 @@ The eviction policies share those bounds but spend metadata differently: | CLOCK | The default uses one visited bit and a bounded shard-local hand. It keeps policy metadata and hit work small. | | S3-FIFO | Uses small and main resident FIFOs, a metadata-only ghost FIFO, and a saturating two-bit frequency. Hits update frequency without moving queue nodes; repeated small-queue entries and ghost hits can enter the main queue. It spends additional fixed metadata to distinguish reused entries from one-hit traffic. | -Evicted values may remain alive while a caller owns them. Admission never waits for such retained values to release their memory charge; it selects another bounded victim or bypasses L1. +Evicted values may remain alive while a caller owns them. Admission never waits for such retained values to release their memory charge; it selects another bounded victim or bypasses L1. Replacement and eviction can reuse one detached, exclusively owned allocation when its combined key/value size matches the new entry. The allocation and its charge become the new resident value; there is no extra free-buffer pool, and caller-held values remain immutable. This reduces allocator churn for fixed-size turnover under both CLOCK and S3-FIFO. ### L2 index @@ -97,14 +97,14 @@ This layout converts small foreground mutations into ordered, batched Region wri ### `put` and `put_l2` -1. Compute the key hash and encoded size once. -2. Enter bounded mutation admission and reserve an append position under one short manager try-lock. +1. Compute the key hash, encoded size, and payload checksum once. +2. Enter bounded mutation admission and try the hash-routed append shard once. Reserve an append position under one short manager try-lock. 3. Encode into the selected shard's current buffer. 4. For `put`, attempt bounded L1 admission. For `put_l2`, stage through Region and remove an older exact-key L1 entry best effort. 5. Return after bounded in-memory admission. 6. The shard worker writes sealed batches and publishes their L2 mappings only after write completion. -Full staging or short-path contention returns structured `ErrorKind::Overloaded`. Staging admission is shard-local; optional adaptive fill budgets are shared across shards. Success means accepted staging; `put_l2` becomes visible when publication completes. `drain` fences all mutations accepted before its operation barrier and waits for their Region writes and L2 publication, but does not issue the recovery durability syncs. +Full staging or lock contention returns structured `ErrorKind::Overloaded`. Every fill uses the same hash route during initial growth, capacity reclaim, and recovery; there is no neighboring-shard retry or routing mode switch. Reads follow the physical address in the index, so records written to another shard by an older runtime remain readable. Lock contention does not force an early flush; capacity and rotation pressure still wake the affected shard worker urgently. A Region capacity request remains in manager authority through its final write, allowing the worker to flush and rotate without waiting for another rejected put; duplicate notifications cannot rotate an empty new Region or retain obsolete rotation pressure. Staging admission is shard-local; optional adaptive fill budgets are shared across shards. Success means accepted staging; `put_l2` becomes visible when publication completes. `drain` fences all mutations accepted before its operation barrier and waits for their Region writes and L2 publication, but does not issue the recovery durability syncs. ### `get` diff --git a/CHANGELOG.md b/CHANGELOG.md index ec75180..af1ebcc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ ## Unreleased +### Bug Fixes + +- Lock contention during foreground staging returns overload without forcing partial Region flushes. `DetailedCacheSnapshot::write_buffer_rejections` excludes lock contention; `CacheSnapshot::write_rejections` continues to count all mutation rejections. Actual capacity and rotation pressure still request urgent progress. +- A full Region retains its capacity request through the final write and rotates without requiring another rejected put. Duplicate notifications do not rotate the new empty Region or retain obsolete flush pressure. +- `CacheSnapshot::served_bytes` counts the payload actually returned when L2 promotion selects a newer L1 value of a different size. + +### Improvements + +- L1 replacement and eviction reuse detached, exclusively owned allocations of the same size under CLOCK and S3-FIFO, reducing repeated allocation during fixed-size turnover without an additional buffer pool. Caller-held values remain immutable and retain their separate memory charge. When activity statistics are enabled, mutations with no eviction also avoid a shared zero-value counter update. + ## v0.6.0 (2026-09-21) ### Breaking Changes diff --git a/CONFIGURATION.md b/CONFIGURATION.md index c4202f6..9ce2477 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -223,11 +223,11 @@ The controls operate at different stages: | `write_flush_threshold_bytes` | Controls when a partial buffer requests publication | Smaller values issue more, smaller writes; larger values favor batching | | Region size | Changes per-shard burst reservoir and rotation frequency | Multiplies staging and reclaim memory and changes eviction granularity | -Start with append shards near the number of genuinely concurrent foreground writers, then measure. More shards are not monotonic: excess shards fragment data across additional partial buffers and Active Regions, consume memory, and can reduce useful capacity. Hash skew can still overload one shard even when aggregate buffers look empty. +Start with append shards near the number of genuinely concurrent foreground writers, then measure. More shards are not monotonic: excess shards fragment data across additional partial buffers and Active Regions, consume memory, and can reduce useful capacity. Each fill tries its hash-routed shard once, with the same path during initial growth, reclaim, and recovery. Lock contention returns overload; capacity pressure requests progress on that shard. Hash skew can still overload one shard even when aggregate buffers look empty. Keep append-shard and I/O-worker counts fixed when comparing revisions, and tune them separately to the actual foreground concurrency and storage throughput. Start write execution capacity near the append-shard count. Increase it only when write I/O in-flight peaks at the configured capacity or slot-wait time grows while the device still has headroom. More execution capacity cannot fix a foreground mutation-gate collision, a full shard-local staging pair, or a saturated SSD. -The flush threshold is per append shard, 4 KiB-aligned, and ranges from 4 KiB through 4 MiB. Lower it when publication latency matters more than write amplification. Keep it high for throughput and batching. Pressure, the bounded flush delay, rotation, and `drain` can flush below the threshold, so it is not a visibility deadline. +The flush threshold is per append shard, defaults to 4 MiB, is 4 KiB-aligned, and ranges from 4 KiB through 4 MiB. Smaller thresholds start writes sooner and reduce the final batch left for `drain`; larger thresholds reduce write-request overhead but can delay publication and capacity recovery. Compare accepted puts plus `drain` as well as continuous traffic when tuning it. Pressure, the bounded flush delay, rotation, and `drain` can flush below the threshold, so it is not a visibility deadline. When a record requires a new Region, its capacity request remains pending through the final write and the worker rotates as soon as it can safely do so, without requiring another rejected put. `put` and `put_l2` intentionally do not wait for staging capacity. A larger Region or more appropriately placed shards can absorb more of a finite burst, but configuration does not turn immediate admission into an unbounded lossless queue. Applications that require higher acceptance must retry with their own deadline, bypass the cache, or split bursts with completion barriers. `put` attempts immediate L1 visibility; L2 visibility for both operations follows publication. `drain` waits for accepted Region writes and index publication. diff --git a/cache2/src/cache/runtime/mod.rs b/cache2/src/cache/runtime/mod.rs index 23f75ea..e0afefa 100644 --- a/cache2/src/cache/runtime/mod.rs +++ b/cache2/src/cache/runtime/mod.rs @@ -77,6 +77,7 @@ use crate::memory::MemoryMetricsSnapshot; use crate::memory::MemoryReadToken; use crate::memory::MemoryStore; use crate::memory::MemoryValue; +use crate::region::RegionRotation; use crate::region::RegionStageValue; use crate::region::RegionStore; use crate::region::RegionValue; @@ -903,7 +904,16 @@ impl CacheRuntime { } Ok(seqno) } - RegionStageValue::NeedsProgress => { + RegionStageValue::Contended => { + // The lock owner retains normal flush responsibility. A collision + // is not buffer pressure and must not force a partial flush. + drop(operation); + if state.activity_counters { + state.metrics.record_write_rejection(); + } + Err(write_overload_error()) + } + RegionStageValue::NeedsFlush => { reject_staged_write(state, control, WAKE_URGENT, operation) } RegionStageValue::NeedsRotation => { @@ -1220,21 +1230,25 @@ impl CacheRuntime { Ok(None) } Ok(Some(value)) => { - if let Some(activity) = activity { - RuntimeMetrics::increment(&activity.l2_hits); - RuntimeMetrics::add(&activity.served_bytes, value.value().len()); - } let promoted = state .memory .promote(read_token, hash, key, value.value(), value.seqno()); - if let Some(promoted) = promoted { + let read = if let Some(promoted) = promoted { if let Some(activity) = activity { RuntimeMetrics::increment(&activity.l1_promotions); } - return Ok(Some(CacheRead::PromotedL2(promoted))); + CacheRead::PromotedL2(promoted) + } else { + CacheRead::L2(value) + }; + // Promotion may select a newer resident value of a different + // size. Count the payload that the caller actually receives. + if let Some(activity) = activity { + RuntimeMetrics::increment(&activity.l2_hits); + RuntimeMetrics::add(&activity.served_bytes, read.value().len()); } - Ok(Some(CacheRead::L2(value))) + Ok(Some(read)) } Ok(None) => { if let Some(activity) = activity { @@ -1871,7 +1885,9 @@ fn reclaim_worker_result( staged_reinsert = true; Ok(true) } - RegionStageValue::NeedsProgress | RegionStageValue::NeedsRotation => { + RegionStageValue::Contended + | RegionStageValue::NeedsFlush + | RegionStageValue::NeedsRotation => { accepting_reinserts = false; Ok(false) } @@ -1917,10 +1933,11 @@ fn append_worker_result( control: &AppendWorkerControl, ) -> io::Result<()> { let mut deadline = None; + let mut rotate_pending = false; loop { let (flags, drain_generation, stop, timed_out) = wait_for_shard_work(control, deadline)?; let draining = drain_generation != 0; - let force_flush = flags & WAKE_URGENT != 0 || timed_out || draining; + let force_flush = flags & WAKE_URGENT != 0 || timed_out || draining || rotate_pending; let rotate = flags & WAKE_ROTATE != 0; match state.staging.shard_fill_snapshot(shard_id) { @@ -1931,7 +1948,8 @@ fn append_worker_result( )?); } if force_flush || fill.bytes >= state.write_flush_threshold_bytes { - let essential = flags & (WAKE_URGENT | WAKE_ROTATE) != 0 || draining; + let essential = + flags & (WAKE_URGENT | WAKE_ROTATE) != 0 || draining || rotate_pending; let bytes = fill.bytes as u64; let records = u32::try_from(fill.records).unwrap_or(u32::MAX); let charge = match &state.fill_control { @@ -1966,15 +1984,6 @@ fn append_worker_result( } Ok(None) => { deadline = None; - if rotate { - let rotated = state.regions.rotate_shard(shard_id)?; - if rotated && state.activity_counters { - RuntimeMetrics::increment(&state.metrics.region_rotations); - } - if rotated { - state.reclaim_control.notify()?; - } - } } Err(StagingError::WouldBlock) => { deadline = Some(Instant::now() + STAGING_RETRY_DELAY); @@ -1982,6 +1991,22 @@ fn append_worker_result( Err(error) => return Err(staging_runtime_error(error)), } + if rotate || rotate_pending { + // Manager authority distinguishes a pending capacity request from a + // duplicate wake for an already rotated, empty Region. + match state.regions.rotate_shard(shard_id)? { + RegionRotation::Idle => rotate_pending = false, + RegionRotation::Pending => rotate_pending = true, + RegionRotation::Rotated => { + rotate_pending = false; + if state.activity_counters { + RuntimeMetrics::increment(&state.metrics.region_rotations); + } + state.reclaim_control.notify()?; + } + } + } + if draining { // Owner drains fence producers. A reclaimer requests the same // completion boundary without fencing foreground mutations, so a @@ -2372,6 +2397,856 @@ mod tests { assert!(!should_wake_write(4096, 4160, 4096)); } + fn write_pressure_store( + region_size: u64, + append_shards: u32, + ) -> (crate::cache::session::CacheSession, std::path::PathBuf) { + use crate::FillControlOptions; + use crate::FillLimits; + + runtime_store( + region_size, + RuntimeOptions { + append_shards, + l1_capacity_bytes: 0, + fill_control: FillControlOptions::Adaptive(FillLimits::new(1_048_576, 10_000)), + stats: crate::StatsOptions { + activity_counters: true, + ..Default::default() + }, + ..RuntimeOptions::default() + }, + ) + } + + fn runtime_store( + region_size: u64, + options: RuntimeOptions, + ) -> (crate::cache::session::CacheSession, std::path::PathBuf) { + use crate::region::persistence::RegionPaths; + use crate::region::recovery::PersistentId; + + let id = LANE_TEST_ID.fetch_add(1, Ordering::Relaxed); + let path = + env::temp_dir().join(format!("cache2-write-pressure-{}-{id}", std::process::id())); + std::fs::create_dir(&path).unwrap(); + let files = RegionPaths::new(path.join("data"), path.join("state"), path.join("image")); + let data = DataSuperblock { + generation: 1, + cache_uuid: PersistentId::from_bytes([1; 16]).unwrap(), + data_identity: PersistentId::from_bytes([2; 16]).unwrap(), + geometry: DataGeometry { + data_file_len: DataGeometry::expected_file_len(region_size, 4).unwrap(), + region_size, + region_count: 4, + }, + hash_seed: 3, + storage_fingerprint: 4, + }; + let store = + crate::cache::session::CacheSession::for_test_with_options(files, data, 8, options) + .unwrap(); + (store, path) + } + + #[test] + fn served_bytes_count_the_value_returned_after_promotion() { + use crate::L1EvictionPolicy; + + for policy in [L1EvictionPolicy::Clock, L1EvictionPolicy::S3Fifo] { + for (l1_capacity, activity_counters, old_length, new_length) in [ + (4096, true, 64, Some(128)), + (4096, true, 128, Some(64)), + (4096, true, 64, None), + (0, true, 64, Some(128)), + (4096, false, 64, Some(128)), + ] { + let (mut store, path) = runtime_store( + 8192, + RuntimeOptions { + append_shards: 1, + l1_capacity_bytes: l1_capacity, + l1_shards: 1, + l1_eviction_policy: policy, + stats: crate::StatsOptions { + activity_counters, + ..Default::default() + }, + ..RuntimeOptions::default() + }, + ); + let plane = store.runtime().unwrap().clone(); + let old_value = vec![1; old_length]; + plane.put_l2(b"key", &old_value).unwrap(); + plane.drain().unwrap(); + let PreparedGet::Pending(pending) = plane.prepare_get(b"key", None).unwrap() else { + panic!("the initial value must be read from L2"); + }; + // Complete the disk read, then deterministically publish a + // newer value before the completion attempts L1 promotion. + let completed = pending.wait(); + let new_value = new_length.map(|length| vec![2; length]); + if let Some(value) = &new_value { + plane.put(b"key", value).unwrap(); + } + let observed = plane.finish_get(completed, b"key").unwrap().unwrap(); + let expected = if l1_capacity == 0 { + &old_value + } else { + new_value.as_ref().unwrap_or(&old_value) + }; + assert_eq!(observed.value(), expected); + assert_eq!( + matches!(&observed, CacheRead::PromotedL2(_)), + l1_capacity != 0 + ); + let snapshot = plane.snapshot().unwrap(); + let returned_bytes = observed.value().len() as u64; + drop(observed); + store.close_warm().unwrap(); + std::fs::remove_dir_all(path).unwrap(); + assert_eq!(snapshot.l1_hits, 0); + assert_eq!(snapshot.l2_hits, u64::from(activity_counters)); + assert_eq!( + snapshot.served_bytes, + if activity_counters { returned_bytes } else { 0 }, + "served bytes must follow the final value with {policy:?}" + ); + } + } + } + + #[test] + fn promoted_value_retains_its_bytes_and_charge_during_concurrent_replacement() { + use crate::L1EvictionPolicy; + + for policy in [L1EvictionPolicy::Clock, L1EvictionPolicy::S3Fifo] { + for activity_counters in [false, true] { + for (old_length, returned_length, final_length) in + [(0, 1, 0), (1, 0, 1), (64, 128, 128), (128, 64, 257)] + { + let (mut store, path) = runtime_store( + 8192, + RuntimeOptions { + append_shards: 1, + l1_capacity_bytes: 4096, + l1_shards: 1, + l1_eviction_policy: policy, + stats: crate::StatsOptions { + activity_counters, + ..Default::default() + }, + ..RuntimeOptions::default() + }, + ); + let plane = store.runtime().unwrap().clone(); + let key = b"key"; + let old_value = vec![1; old_length]; + let returned_value = vec![2; returned_length]; + let final_value = vec![3; final_length]; + plane.put_l2(key, &old_value).unwrap(); + plane.drain().unwrap(); + let PreparedGet::Pending(pending) = plane.prepare_get(key, None).unwrap() + else { + panic!("the initial value must be read from L2"); + }; + let completed = pending.wait(); + plane.put(key, &returned_value).unwrap(); + let held = plane.finish_get(completed, key).unwrap().unwrap(); + let CacheRead::PromotedL2(promoted) = &held else { + panic!("promotion must select the newer resident value"); + }; + let held_clone = promoted.clone(); + assert_eq!(held.value(), returned_value); + let before = plane.state.memory.detailed_snapshot().unwrap(); + assert_eq!(before.resident_entries, 1); + assert_eq!(before.retained_bytes, 0); + let promoted_charge = before.resident_bytes; + + // Retain both caller references while another thread + // replaces the L1 entry. Reserve a real logical sequence + // and publish directly to isolate L1 ownership from I/O. + let seqno = plane + .regions + .lock_manager_for_test() + .allocate_seqno() + .unwrap(); + let concurrent = plane.clone(); + let replacement = final_value.clone(); + std::thread::spawn(move || { + let hash = hash_key(concurrent.data.hash_seed, key); + assert!( + concurrent + .state + .memory + .publish(hash, key, &replacement, seqno) + ); + }) + .join() + .unwrap(); + assert_eq!(held.value(), returned_value); + assert_eq!(held_clone.as_ref(), returned_value); + let snapshot = plane.snapshot().unwrap(); + assert_eq!( + snapshot.served_bytes, + if activity_counters { + returned_length as u64 + } else { + 0 + } + ); + assert_eq!(snapshot.l1_hits, 0); + assert_eq!(snapshot.l2_hits, u64::from(activity_counters)); + let l1 = plane.state.memory.detailed_snapshot().unwrap(); + assert_eq!(l1.resident_entries, 1); + assert_eq!(l1.retained_bytes, promoted_charge); + assert_eq!( + l1.resident_bytes, + promoted_charge - returned_length + final_length + ); + let current = plane.get(key).unwrap().unwrap(); + assert_eq!(current.value(), final_value); + let snapshot = plane.snapshot().unwrap(); + assert_eq!( + snapshot.served_bytes, + if activity_counters { + (returned_length + final_length) as u64 + } else { + 0 + } + ); + assert_eq!(snapshot.l1_hits, u64::from(activity_counters)); + assert_eq!(snapshot.l2_hits, u64::from(activity_counters)); + drop(current); + drop(held); + assert_eq!( + plane + .state + .memory + .detailed_snapshot() + .unwrap() + .retained_bytes, + promoted_charge + ); + drop(held_clone); + let released = plane.state.memory.detailed_snapshot().unwrap(); + assert_eq!(released.resident_bytes, l1.resident_bytes); + assert_eq!(released.retained_bytes, 0); + store.close_warm().unwrap(); + std::fs::remove_dir_all(path).unwrap(); + } + } + } + } + + #[test] + fn contention_rejects_without_waking_but_full_buffers_flush_and_publish() { + use crate::region::record::RECORD_HEADER_SIZE; + + let (mut store, path) = write_pressure_store(4096, 1); + let plane = store.runtime().unwrap().clone(); + for manager_contention in [false, true] { + let shard = (!manager_contention).then(|| plane.regions.lock_append_for_test(0)); + let manager = manager_contention.then(|| plane.regions.lock_manager_for_test()); + // A contended producer must not even acquire the worker's wake + // mutex. Release it before joining so a regression fails, not hangs. + let state = plane.state.append_controls[0].lock().unwrap(); + let (send, receive) = std::sync::mpsc::channel(); + let writer_plane = plane.clone(); + let writer = std::thread::spawn(move || { + send.send(writer_plane.put(b"contended", b"value")).unwrap(); + }); + let result = receive.recv_timeout(Duration::from_secs(1)); + assert_eq!(state.wake_flags, 0); + drop(state); + drop(manager); + drop(shard); + writer.join().unwrap(); + assert!(matches!(result, Ok(Err(error)) if error.kind() == io::ErrorKind::WouldBlock)); + } + assert_eq!(plane.snapshot().unwrap().write_rejections, 2); + assert_eq!( + plane.detailed_snapshot().unwrap().write_buffer_rejections, + 0 + ); + let controller = plane.state.fill_control.as_ref().unwrap(); + assert_eq!(controller.snapshot().outstanding_operations, 0); + + // Fill without notifying the worker, then make the public path hit + // actual capacity pressure. It must still wake, flush and publish. + let key = b"pressure"; + let value = vec![7; 4096 - RECORD_HEADER_SIZE - key.len()]; + assert!(matches!( + plane + .regions + .try_stage_value( + &plane.state.staging, + 0, + hash_key(plane.data.hash_seed, key), + 4096, + key, + &value + ) + .unwrap(), + RegionStageValue::Staged { .. } + )); + assert_eq!( + plane.put(b"next", b"value").unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + assert_eq!( + plane.detailed_snapshot().unwrap().write_buffer_rejections, + 1 + ); + // A drain would wake the worker itself and hide a missing pressure + // notification. Require automatic completion and publication first. + let deadline = Instant::now() + Duration::from_secs(2); + while plane + .regions + .begin_point_read(hash_key(plane.data.hash_seed, key)) + .is_none() + { + assert!( + Instant::now() < deadline, + "capacity pressure must flush and publish without an explicit drain" + ); + std::thread::sleep(Duration::from_millis(1)); + } + assert!(plane.get(key).unwrap().is_some()); + store.close_warm().unwrap(); + std::fs::remove_dir_all(path).unwrap(); + } + + #[test] + fn full_staging_capacity_request_rotates_without_another_put_or_drain() { + use crate::region::record::RECORD_HEADER_SIZE; + + let (mut store, path) = write_pressure_store(4096, 1); + let plane = store.runtime().unwrap().clone(); + let key = b"full"; + let value = vec![7; 4096 - RECORD_HEADER_SIZE - key.len()]; + // Fill the entire Region before the worker is notified. The next + // public put must preserve rotation despite failing staging preflight. + assert!(matches!( + plane + .regions + .try_stage_value( + &plane.state.staging, + 0, + hash_key(plane.data.hash_seed, key), + 4096, + key, + &value, + ) + .unwrap(), + RegionStageValue::Staged { .. } + )); + assert_eq!( + plane.put(b"next", b"value").unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + let deadline = Instant::now() + Duration::from_secs(2); + while plane.snapshot().unwrap().region_rotations == 0 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(1)); + } + let rotations = plane.snapshot().unwrap().region_rotations; + assert_eq!(plane.get(key).unwrap().unwrap().value(), value); + plane.state.append_controls[0].notify(WAKE_ROTATE).unwrap(); + plane.drain().unwrap(); + let snapshot = plane.detailed_snapshot().unwrap(); + store.close_warm().unwrap(); + std::fs::remove_dir_all(path).unwrap(); + assert_eq!( + rotations, 1, + "one rejection must flush and rotate the full Region" + ); + assert_eq!( + snapshot.summary.region_rotations, 1, + "a duplicate wake must remain idle" + ); + assert_eq!(snapshot.summary.write_rejections, 1); + assert_eq!(snapshot.write_buffer_rejections, 1); + } + + #[test] + fn staging_record_limit_accounts_for_flush_padding_before_rotation() { + use crate::region::record::RECORD_HEADER_SIZE; + + const REGION_BYTES: usize = 1024 * 1024; + for needs_rotation in [false, true] { + let (mut store, path) = write_pressure_store(REGION_BYTES as u64, 1); + let plane = store.runtime().unwrap().clone(); + let key = b"key"; + let hash = hash_key(plane.data.hash_seed, key); + // Start with a 96-byte record so filling the record vector leaves + // an unaligned tail that must be padded before the next append. + let first = vec![1; 96 - RECORD_HEADER_SIZE - key.len()]; + assert!(matches!( + plane + .regions + .try_stage_value(&plane.state.staging, 0, hash, 96, key, &first) + .unwrap(), + RegionStageValue::Staged { .. } + )); + let mut used = 96; + loop { + match plane + .regions + .try_stage_value(&plane.state.staging, 0, hash, 64, key, b"value") + .unwrap() + { + RegionStageValue::Staged { .. } => used += 64, + RegionStageValue::NeedsFlush => break, + outcome => panic!("unexpected staging outcome: {outcome:?}"), + } + assert!( + used < REGION_BYTES, + "the record limit must leave Region capacity" + ); + } + assert_eq!(used % 4096, 32); + // A small next record still fits after flush. A record that fills + // the unpadded remainder needs rotation once padding is included. + let next_record_bytes = if needs_rotation { + REGION_BYTES - used + } else { + 64 + }; + let next = vec![2; next_record_bytes - RECORD_HEADER_SIZE - key.len()]; + assert_eq!( + plane.put(key, &next).unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + let deadline = Instant::now() + Duration::from_secs(2); + loop { + let fill = plane.state.staging.shard_fill_snapshot(0); + assert!(matches!(fill, Ok(_) | Err(StagingError::WouldBlock))); + let rotations = plane.snapshot().unwrap().region_rotations; + if matches!(fill, Ok(None)) && rotations == u64::from(needs_rotation) { + break; + } + assert!( + Instant::now() < deadline, + "record capacity must complete publication and the requested rotation" + ); + std::thread::sleep(Duration::from_millis(1)); + } + // Empty staging means every index entry has been published. An + // earlier lookup can succeed between publications, then a second + // lookup can miss on the publisher's next partition lock. + // Capture progress before drain, which must not supply a missing + // wake or rotation request on behalf of the rejected producer. + let published = plane.regions.begin_point_read(hash).is_some(); + let rotations = plane.snapshot().unwrap().region_rotations; + plane.drain().unwrap(); + let final_rotations = plane.snapshot().unwrap().region_rotations; + assert_eq!(plane.get(key).unwrap().unwrap().value(), b"value"); + store.close_warm().unwrap(); + std::fs::remove_dir_all(path).unwrap(); + assert!(published, "record capacity must wake the worker"); + assert_eq!(rotations, u64::from(needs_rotation)); + assert_eq!(final_rotations, rotations); + } + } + + #[test] + fn staging_record_limit_distinguishes_exact_fit_from_one_unit_overflow() { + use crate::region::record::RECORD_ALIGNMENT; + + const REGION_BYTES: usize = 1024 * 1024; + const ALIGNMENT: usize = crate::io::file::DIRECT_IO_ALIGNMENT; + let unit = RECORD_ALIGNMENT as usize; + for prefix in [0, ALIGNMENT] { + for extra in [0, unit, ALIGNMENT - unit] { + for overflow in [0, unit] { + let (mut store, path) = write_pressure_store(REGION_BYTES as u64, 1); + let plane = store.runtime().unwrap().clone(); + if prefix != 0 { + let value = vec![8; prefix - RECORD_HEADER_SIZE - b"prefix".len()]; + plane.put_l2(b"prefix", &value).unwrap(); + plane.drain().unwrap(); + } + let key = b"key"; + let hash = hash_key(plane.data.hash_seed, key); + // Vary the first record so the full metadata vector + // requires zero, maximal, or minimal flush padding. + let first_bytes = 64 + extra; + let first = vec![1; first_bytes - RECORD_HEADER_SIZE - key.len()]; + assert!(matches!( + plane + .regions + .try_stage_value( + &plane.state.staging, + 0, + hash, + first_bytes as u32, + key, + &first + ) + .unwrap(), + RegionStageValue::Staged { .. } + )); + let mut used = prefix + first_bytes; + loop { + match plane + .regions + .try_stage_value(&plane.state.staging, 0, hash, 64, key, b"value") + .unwrap() + { + RegionStageValue::Staged { .. } => used += 64, + RegionStageValue::NeedsFlush => break, + outcome => panic!("unexpected staging outcome: {outcome:?}"), + } + assert!( + used < REGION_BYTES, + "the record limit must leave Region capacity" + ); + } + let aligned = used.div_ceil(ALIGNMENT) * ALIGNMENT; + let record_bytes = REGION_BYTES - aligned + overflow; + let value = vec![2; record_bytes - RECORD_HEADER_SIZE - key.len()]; + assert_eq!( + plane.put_l2(key, &value).unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + let expected_rotations = u64::from(overflow != 0); + let deadline = Instant::now() + Duration::from_secs(3); + // Observe autonomous progress before a retry or drain + // could supply the original missing rotation request. + loop { + let rotations = plane.snapshot().unwrap().region_rotations; + let fill = plane.state.staging.shard_fill_snapshot(0); + assert!(matches!(fill, Ok(_) | Err(StagingError::WouldBlock))); + if matches!(fill, Ok(None)) && rotations == expected_rotations { + break; + } + assert!( + Instant::now() < deadline, + "no progress: prefix={prefix}, extra={extra}, overflow={overflow}, rotations={rotations}" + ); + std::thread::sleep(Duration::from_millis(1)); + } + { + let manager = plane.regions.lock_manager_for_test(); + let active = manager.active_regions()[0] as usize; + assert_eq!( + manager.regions()[active].completed_used, + if expected_rotations == 0 { + aligned as u64 + } else { + 0 + } + ); + } + // Admission may briefly collide with the reclaimer after + // rotation; only retry after asserting the above progress. + let deadline = Instant::now() + Duration::from_secs(3); + loop { + match plane.put_l2(key, &value) { + Ok(_) => break, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + assert!( + Instant::now() < deadline, + "write admission did not resume" + ); + std::thread::sleep(Duration::from_millis(1)); + } + Err(error) => panic!("unexpected write failure: {error}"), + } + } + plane.drain().unwrap(); + assert_eq!(plane.get(key).unwrap().unwrap().value(), value); + assert_eq!( + plane.snapshot().unwrap().region_rotations, + expected_rotations + ); + store.close_warm().unwrap(); + std::fs::remove_dir_all(path).unwrap(); + } + } + } + } + + #[test] + fn capacity_request_flushes_then_rotates_without_another_put_or_drain() { + use crate::region::record::RECORD_HEADER_SIZE; + + let (mut store, path) = write_pressure_store(8192, 1); + let plane = store.runtime().unwrap().clone(); + let first = vec![1; 4096 - RECORD_HEADER_SIZE - b"first".len()]; + plane.put(b"first", &first).unwrap(); + plane.drain().unwrap(); + + // Leave one aligned record unit at the end of the Region. This fill + // fits the staging buffer, but the next record needs a new Region. + let tail = vec![2; 4064 - RECORD_HEADER_SIZE - b"tail".len()]; + assert!(matches!( + plane + .regions + .try_stage_value( + &plane.state.staging, + 0, + hash_key(plane.data.hash_seed, b"tail"), + 4064, + b"tail", + &tail, + ) + .unwrap(), + RegionStageValue::Staged { .. } + )); + assert_eq!( + plane.put(b"next", b"value").unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + let deadline = Instant::now() + Duration::from_secs(2); + while plane.snapshot().unwrap().region_rotations == 0 { + assert!( + Instant::now() < deadline, + "one capacity request must flush and rotate without another put or drain" + ); + std::thread::sleep(Duration::from_millis(1)); + } + assert_eq!(plane.get(b"first").unwrap().unwrap().value(), first); + assert_eq!(plane.get(b"tail").unwrap().unwrap().value(), tail); + // A duplicate wake, including reclaim's free-capacity notification, + // must not rotate the newly empty Active Region. + plane.state.append_controls[0].notify(WAKE_ROTATE).unwrap(); + plane.drain().unwrap(); + assert_eq!(plane.snapshot().unwrap().region_rotations, 1); + assert_eq!(plane.snapshot().unwrap().write_rejections, 1); + store.close_warm().unwrap(); + std::fs::remove_dir_all(path).unwrap(); + } + + #[test] + fn primary_contention_rejects_without_probing_other_shards() { + let (mut store, path) = write_pressure_store(8192, 3); + let plane = store.runtime().unwrap().clone(); + let key = b"fixed-route"; + let value = [7; 64]; + let primary = plane + .regions + .append_shard(hash_key(plane.data.hash_seed, key)); + let primary_gate = plane.regions.lock_append_for_test(primary); + let primary_wake = plane.state.append_controls[primary].lock().unwrap(); + let (send, receive) = std::sync::mpsc::channel(); + let writer_plane = plane.clone(); + let writer = std::thread::spawn(move || { + send.send(writer_plane.put(key, &value)).unwrap(); + }); + let result = receive.recv_timeout(Duration::from_secs(1)); + assert_eq!(primary_wake.wake_flags, 0); + drop(primary_wake); + drop(primary_gate); + writer.join().unwrap(); + assert_eq!( + result + .expect("contention must not wait for a mutation or wake lock") + .unwrap_err() + .kind(), + io::ErrorKind::WouldBlock + ); + let controller = plane.state.fill_control.as_ref().unwrap(); + assert_eq!(controller.snapshot().outstanding_operations, 0); + let snapshot = plane.detailed_snapshot().unwrap(); + assert_eq!(snapshot.summary.puts, 0); + assert_eq!(snapshot.summary.write_rejections, 1); + assert_eq!(snapshot.write_buffer_rejections, 0); + assert_eq!(snapshot.summary.fill_control.outstanding_operations, 0); + + plane.put(key, &value).unwrap(); + plane.drain().unwrap(); + assert_eq!(plane.get(key).unwrap().unwrap().value(), value); + let snapshot = plane.detailed_snapshot().unwrap(); + assert_eq!(snapshot.summary.puts, 1); + assert_eq!(snapshot.summary.write_rejections, 1); + assert_eq!(snapshot.write_buffer_rejections, 0); + store.close_warm().unwrap(); + std::fs::remove_dir_all(path).unwrap(); + } + + #[test] + fn reclaimed_cache_rejects_primary_contention_without_trying_a_free_neighbor() { + let (mut store, path) = write_pressure_store(8192, 3); + let plane = store.runtime().unwrap().clone(); + let key = b"stable-route"; + let primary = plane + .regions + .append_shard(hash_key(plane.data.hash_seed, key)); + plane + .regions + .lock_manager_for_test() + .request_rotation_for_test(primary) + .unwrap(); + assert_eq!( + plane.regions.rotate_shard(primary).unwrap(), + RegionRotation::Rotated + ); + let receipt = plane.regions.begin_reclaim().unwrap().unwrap(); + plane.regions.complete_reclaim(receipt).unwrap(); + + let primary_gate = plane.regions.lock_append_for_test(primary); + let primary_wake = plane.state.append_controls[primary].lock().unwrap(); + let (send, receive) = std::sync::mpsc::channel(); + let writer_plane = plane.clone(); + let writer = std::thread::spawn(move || { + send.send(writer_plane.put(key, &[7; 64])).unwrap(); + }); + let result = receive.recv_timeout(Duration::from_secs(1)); + assert_eq!(primary_wake.wake_flags, 0); + drop(primary_wake); + drop(primary_gate); + writer.join().unwrap(); + assert_eq!( + result + .expect("contention must not wait for a mutation or wake lock") + .unwrap_err() + .kind(), + io::ErrorKind::WouldBlock + ); + let snapshot = plane.detailed_snapshot().unwrap(); + assert_eq!(snapshot.summary.puts, 0); + assert_eq!(snapshot.summary.write_rejections, 1); + assert_eq!(snapshot.write_buffer_rejections, 0); + assert_eq!(snapshot.summary.fill_control.outstanding_operations, 0); + assert_eq!( + plane + .regions + .begin_point_read(hash_key(plane.data.hash_seed, key)), + None + ); + plane.put(key, &[7; 64]).unwrap(); + plane.drain().unwrap(); + assert_eq!(plane.get(key).unwrap().unwrap().value(), &[7; 64]); + store.close_warm().unwrap(); + std::fs::remove_dir_all(path).unwrap(); + } + + #[test] + fn existing_alternate_records_survive_reclaim_and_warm_reopen() { + use crate::FillControlOptions; + use crate::FillLimits; + use crate::region::persistence::RegionPaths; + + let (mut store, path) = write_pressure_store(8192, 3); + let plane = store.runtime().unwrap().clone(); + let key = b"fallback-route"; + let value = [9; 64]; + let data = plane.data; + let hash = hash_key(data.hash_seed, key); + let primary = plane.regions.append_shard(hash); + let alternate = (primary + 1) % plane.state.append_controls.len(); + let alternate_region = plane.regions.lock_manager_for_test().active_regions()[alternate]; + + // Model a record written by an older runtime that used another + // append shard. Reads must follow the stored address, not today's route. + let record_bytes = required_record_bytes(key.len(), value.len()).unwrap(); + assert!(matches!( + plane + .regions + .try_stage_value( + &plane.state.staging, + alternate, + hash, + record_bytes, + key, + &value, + ) + .unwrap(), + RegionStageValue::Staged { .. } + )); + plane.drain().unwrap(); + assert_eq!( + plane + .regions + .begin_point_read(hash) + .unwrap() + .entry + .location + .region_id(), + alternate_region + ); + assert_eq!(plane.get(key).unwrap().unwrap().value(), value); + + // Reclaim an empty primary Region while the alternate record remains + // live. L1 is disabled, so all read assertions exercise the L2 address. + plane + .regions + .lock_manager_for_test() + .request_rotation_for_test(primary) + .unwrap(); + assert_eq!( + plane.regions.rotate_shard(primary).unwrap(), + RegionRotation::Rotated + ); + let receipt = plane.regions.begin_reclaim().unwrap().unwrap(); + assert_ne!(receipt.region_id, alternate_region); + assert_eq!(receipt.physical_record_count, 0); + plane.regions.complete_reclaim(receipt).unwrap(); + assert!(!plane.regions.reclaim_needed().unwrap()); + assert_eq!(plane.get(key).unwrap().unwrap().value(), value); + assert_eq!( + plane + .regions + .begin_point_read(hash) + .unwrap() + .entry + .location + .region_id(), + alternate_region + ); + store.close_warm().unwrap(); + drop(plane); + drop(store); + + let files = RegionPaths::new(path.join("data"), path.join("state"), path.join("image")); + let options = RuntimeOptions { + append_shards: 3, + l1_capacity_bytes: 0, + fill_control: FillControlOptions::Adaptive(FillLimits::new(1_048_576, 10_000)), + stats: crate::StatsOptions { + activity_counters: true, + ..Default::default() + }, + ..RuntimeOptions::default() + }; + let mut reopened = + crate::cache::session::CacheSession::for_test_with_options(files, data, 8, options) + .unwrap(); + let plane = reopened.runtime().unwrap().clone(); + // Recovery preserves the old physical address. New foreground writes + // use the same hash route as they did before reclaim and close. + assert_eq!(plane.get(key).unwrap().unwrap().value(), value); + assert_eq!( + plane + .regions + .begin_point_read(hash) + .unwrap() + .entry + .location + .region_id(), + alternate_region + ); + plane.put_l2(key, &[11; 64]).unwrap(); + plane.drain().unwrap(); + assert_eq!(plane.get(key).unwrap().unwrap().value(), &[11; 64]); + let primary_region = plane.regions.lock_manager_for_test().active_regions()[primary]; + assert_eq!( + plane + .regions + .begin_point_read(hash) + .unwrap() + .entry + .location + .region_id(), + primary_region + ); + reopened.close_warm().unwrap(); + drop(plane); + drop(reopened); + std::fs::remove_dir_all(path).unwrap(); + } + #[test] fn mutation_gate_fences_existing_and_new_mutations() { let gate = MutationGate::new(); diff --git a/cache2/src/memory/mod.rs b/cache2/src/memory/mod.rs index 037e868..8132f15 100644 --- a/cache2/src/memory/mod.rs +++ b/cache2/src/memory/mod.rs @@ -43,6 +43,9 @@ use crate::snapshot::CacheL1Snapshot; mod eviction; +#[cfg(test)] +mod reuse_tests; + /// Charged retained-value ownership. Fixed entry, policy, and directory /// storage is planned and allocated separately during open. const MEMORY_ENTRY_OVERHEAD_BYTES: usize = 64; @@ -179,6 +182,7 @@ impl MemoryValue { value: &[u8], seqno: u64, charge: MemoryCharge, + recycled: Option, ) -> Result { let Some(length) = key .len() @@ -187,6 +191,19 @@ impl MemoryValue { else { return Err(charge); }; + // Reuse only a detached, exclusively owned allocation of the exact + // size. Caller-held values remain immutable and keep their own charge. + if let Some(mut recycled) = recycled + && let Some(inner) = Arc::get_mut(&mut recycled.0) + && inner.bytes.len() == length + { + inner.bytes[..key.len()].copy_from_slice(key); + inner.bytes[key.len()..key.len() + value.len()].copy_from_slice(value); + inner.bytes[length - MEMORY_VALUE_SEQNO_BYTES..].copy_from_slice(&seqno.to_le_bytes()); + inner.key_length = key.len(); + inner._charge = charge; + return Ok(recycled); + } let mut bytes = Vec::new(); if bytes.try_reserve_exact(length).is_err() { return Err(charge); @@ -426,20 +443,23 @@ impl MemoryShard { } fn remove_slot(&mut self, index: usize) { - if self.slots.get(index).and_then(Option::as_ref).is_none() { - return; - } - let Some(hash) = self.policy_slots.get(index).map(PolicySlot::hash) else { - return; - }; + drop(self.take_slot(index)); + } + + fn take_slot(&mut self, index: usize) -> Option { + self.slots.get(index).and_then(Option::as_ref)?; + let hash = self.policy_slots.get(index).map(PolicySlot::hash)?; self.eviction.remove(&mut self.policy_slots, index); - self.remove_slot_after_policy(index, hash, false); + self.remove_slot_after_policy(index, hash, false) } - fn remove_slot_after_policy(&mut self, index: usize, hash: u64, transfer_charge: bool) { - let Some(mut cursor) = self.directory_head(hash) else { - return; - }; + fn remove_slot_after_policy( + &mut self, + index: usize, + hash: u64, + transfer_charge: bool, + ) -> Option { + let mut cursor = self.directory_head(hash)?; let mut predecessor = None; let mut found = false; for _ in 0..MAX_SAME_HASH_ENTRIES { @@ -461,7 +481,7 @@ impl MemoryShard { } debug_assert!(found, "resident slot must remain in its hash chain"); if !found { - return; + return None; } let entry_next = self.slots[index] @@ -501,13 +521,14 @@ impl MemoryShard { .charged_bytes() ); } - let _entry = self.slots[index].take().expect("resident slot disappeared"); + let entry = self.slots[index].take().expect("resident slot disappeared"); self.resident_bytes = self.resident_bytes.saturating_sub(resident_bytes); debug_assert!(self.free_count < self.slots.len()); let packed = u32::try_from(index).expect("memory slot index exceeds u32"); self.policy_slots[index] = PolicySlot::new_free(self.free_head); self.free_head = packed; self.free_count += 1; + Some(entry.value) } fn detach_for_admission(&mut self, index: usize) -> DetachedAdmissionEntry { @@ -533,11 +554,17 @@ impl MemoryShard { .restore_for_admission(&mut self.policy_slots, entry.index, entry.policy); } - fn commit_detached(&mut self, entry: DetachedAdmissionEntry, record_eviction: bool) { - self.remove_slot_after_policy(entry.index, entry.policy.hash(), entry.transfers_charge); + fn commit_detached( + &mut self, + entry: DetachedAdmissionEntry, + record_eviction: bool, + ) -> Option { + let value = + self.remove_slot_after_policy(entry.index, entry.policy.hash(), entry.transfers_charge); if record_eviction { self.eviction.commit_eviction(entry.policy); } + value } fn restore_admission( @@ -564,14 +591,14 @@ impl MemoryShard { let needs_slot = replacement.is_none() && self.free_head == NO_SLOT_INDEX; match self.budget.try_charge(required) { MemoryChargeAttempt::Charged(charge) => { - if let Some(replacement) = replacement { - self.remove_slot(replacement); - } else if needs_slot { + if needs_slot { return self.charge_with_slot_eviction(charge); } + let recycled = replacement.and_then(|index| self.take_slot(index)); return ChargeResult::Charged { charge, evictions: 0, + recycled, }; } MemoryChargeAttempt::Contended => { @@ -624,15 +651,19 @@ impl MemoryShard { } }; - if let Some(replacement) = replacement { - self.commit_detached(replacement, false); - } + let mut recycled = replacement.and_then(|entry| self.commit_detached(entry, false)); for victim in victims.iter() { - self.commit_detached(victim, true); + let value = self.commit_detached(victim, true); + if recycled.as_ref().is_none_or(|value| { + value.charged_bytes() != required || !value.releases_charge_on_remove() + }) { + recycled = value; + } } ChargeResult::Charged { charge, evictions: victims.len, + recycled, } } @@ -649,10 +680,11 @@ impl MemoryShard { // The candidate was charged independently because byte capacity was // available. Let an exclusive victim release its own charge normally. victim.transfers_charge = false; - self.commit_detached(victim, true); + let recycled = self.commit_detached(victim, true); ChargeResult::Charged { charge, evictions: 1, + recycled, } } @@ -671,12 +703,17 @@ impl MemoryShard { return MemoryInsertResult::bypassed(); } - let (charge, evictions) = match self.charge_with_eviction(charged_bytes, replacement) { - ChargeResult::Charged { charge, evictions } => (charge, evictions), - ChargeResult::Rejected { evictions } => { - return MemoryInsertResult::rejected(evictions); - } - }; + let (charge, evictions, recycled) = + match self.charge_with_eviction(charged_bytes, replacement) { + ChargeResult::Charged { + charge, + evictions, + recycled, + } => (charge, evictions, recycled), + ChargeResult::Rejected { evictions } => { + return MemoryInsertResult::rejected(evictions); + } + }; let packed_index = self.free_head; if packed_index == NO_SLOT_INDEX { return MemoryInsertResult::rejected(evictions); @@ -684,7 +721,7 @@ impl MemoryShard { let Ok(index) = usize::try_from(packed_index) else { return MemoryInsertResult::rejected(evictions); }; - let Ok(memory_value) = MemoryValue::try_new(key, value, seqno, charge) else { + let Ok(memory_value) = MemoryValue::try_new(key, value, seqno, charge, recycled) else { return MemoryInsertResult::rejected(evictions); }; let previous_head = self.directory_head(hash); @@ -712,6 +749,7 @@ enum ChargeResult { Charged { charge: MemoryCharge, evictions: usize, + recycled: Option, }, Rejected { evictions: usize, @@ -1049,10 +1087,12 @@ impl MemoryStore { if !self.metrics.enabled { return; } - self.metrics.evictions.fetch_add( - u64::try_from(result.evictions).unwrap_or(u64::MAX), - Ordering::Relaxed, - ); + if result.evictions != 0 { + self.metrics.evictions.fetch_add( + u64::try_from(result.evictions).unwrap_or(u64::MAX), + Ordering::Relaxed, + ); + } if !result.inserted() { self.metrics.bypasses.fetch_add(1, Ordering::Relaxed); } @@ -1119,6 +1159,77 @@ mod tests { assert_eq!(size_of::>(), expected); } + #[test] + fn recycled_eviction_rewrites_key_and_sequence_with_exact_accounting() { + for policy in [L1EvictionPolicy::Clock, L1EvictionPolicy::S3Fifo] { + let charge = memory_entry_bytes(b"a", &[1; 128]).unwrap(); + // Exercise both byte pressure (transferred charge) and slot + // pressure with spare byte capacity (independent new charge). + for capacity in [charge, 2 * charge] { + let store = MemoryStore::new(capacity, 1, 1, policy, true).unwrap(); + assert!(store.publish(7, b"a", &[1; 128], 1)); + let old_address = match store.lookup(7, b"a") { + MemoryLookup::Hit(value) => Arc::as_ptr(&value.0), + MemoryLookup::Miss(_) => panic!("initial value must be resident"), + }; + assert!(store.publish(8, b"bb", &[2; 127], 2)); + assert_miss(&store, 7, b"a"); + let MemoryLookup::Hit(value) = store.lookup(8, b"bb") else { + panic!("replacement must be resident"); + }; + assert_eq!(Arc::as_ptr(&value.0), old_address); + assert_eq!(value.key(), b"bb"); + assert_eq!(value.as_ref(), &[2; 127]); + assert_eq!(value.seqno(), 2); + let snapshot = store.detailed_snapshot().unwrap(); + assert_eq!(snapshot.resident_entries, 1); + assert_eq!(snapshot.resident_bytes, charge); + assert_eq!(snapshot.retained_bytes, 0); + assert_eq!(store.shards[0].lock().unwrap().budget.used_bytes(), charge); + assert_eq!(store.metrics_snapshot().evictions, 1); + } + } + } + + #[test] + fn recycled_replacement_preserves_caller_owned_values() { + for policy in [L1EvictionPolicy::Clock, L1EvictionPolicy::S3Fifo] { + let charge = memory_entry_bytes(b"key", &[1; 128]).unwrap(); + let store = MemoryStore::new(2 * charge, 2, 1, policy, true).unwrap(); + assert!(store.publish(7, b"key", &[1; 128], 1)); + let MemoryLookup::Hit(held) = store.lookup(7, b"key") else { + panic!("initial value must be resident"); + }; + assert!(store.publish(7, b"key", &[2; 128], 2)); + let address = match store.lookup(7, b"key") { + MemoryLookup::Hit(value) => { + assert!(!Arc::ptr_eq(&value.0, &held.0)); + assert_eq!(value.as_ref(), &[2; 128]); + Arc::as_ptr(&value.0) + } + MemoryLookup::Miss(_) => panic!("newer value must be resident"), + }; + assert_eq!(held.as_ref(), &[1; 128]); + assert_eq!(held.seqno(), 1); + assert_eq!(store.detailed_snapshot().unwrap().retained_bytes, charge); + + // The newer resident allocation is exclusive, while the caller's + // old value still owns its separate charge. Reuse must not exceed + // the full budget or alter that old value. + assert!(store.publish(7, b"key", &[3; 128], 3)); + let MemoryLookup::Hit(value) = store.lookup(7, b"key") else { + panic!("latest value must be resident"); + }; + assert_eq!(Arc::as_ptr(&value.0), address); + assert_eq!(value.as_ref(), &[3; 128]); + assert_eq!(value.seqno(), 3); + assert_eq!(held.as_ref(), &[1; 128]); + drop(held); + assert_eq!(store.detailed_snapshot().unwrap().retained_bytes, 0); + assert_eq!(store.shards[0].lock().unwrap().budget.used_bytes(), charge); + } + } + #[test] fn published_value_is_visible_and_immediately_evictable() { let store = store(512, 1); diff --git a/cache2/src/memory/reuse_tests.rs b/cache2/src/memory/reuse_tests.rs new file mode 100644 index 0000000..22b8382 --- /dev/null +++ b/cache2/src/memory/reuse_tests.rs @@ -0,0 +1,327 @@ +// Copyright 2026 ScopeDB, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::HashMap; + +use super::*; + +const MIXED_OPERATIONS: u64 = 4_000; +const CONCURRENT_OPERATIONS: usize = 12_000; +const CONCURRENT_WORKERS: usize = 8; +const KEY_COUNT: usize = 64; +const VALUE_LENGTHS: [usize; 11] = [0, 1, 63, 64, 127, 128, 193, 257, 511, 1024, 2048]; + +#[derive(Clone)] +struct RetainedValueSnapshot { + value: MemoryValue, + key: Vec, + bytes: Vec, + seqno: u64, +} + +impl RetainedValueSnapshot { + fn new(value: MemoryValue) -> Self { + Self { + key: value.key().to_vec(), + bytes: value.as_ref().to_vec(), + seqno: value.seqno(), + value, + } + } + + fn assert_unchanged(&self) { + assert_eq!(self.value.key(), self.key); + assert_eq!(self.value.as_ref(), self.bytes); + assert_eq!(self.value.seqno(), self.seqno); + } +} + +fn next_random(state: &mut u64) -> u64 { + // Fixed nonzero seeds keep failures reproducible without a new dependency. + *state ^= *state << 13; + *state ^= *state >> 7; + *state ^= *state << 17; + *state +} + +fn random_index(state: &mut u64, length: usize) -> usize { + (next_random(state) % length as u64) as usize +} + +fn key_and_hash(id: usize) -> (Vec, u64) { + let key = format!("key:{id:02}{}", "x".repeat(id % 9)).into_bytes(); + // Deliberate full-hash collisions also exercise removal from hash chains. + let hash = (id as u64 % 12).wrapping_mul(0x9e37_79b9_7f4a_7c15); + (key, hash) +} + +/// Compare charges with unique live allocations after all writers have stopped. +/// Clones shared by residents and callers must contribute only one charge. +fn assert_accounting(store: &MemoryStore, retained: &[RetainedValueSnapshot]) { + let mut allocations = HashMap::new(); + let mut budgets = HashMap::new(); + let mut total_resident = 0; + let mut total_entries = 0; + let mut record = |value: &MemoryValue| { + let inner = &value.0; + assert_eq!(inner._charge.bytes, value.charged_bytes()); + let budget = inner + ._charge + .budget + .as_ref() + .expect("live allocation must keep its charge armed"); + allocations.insert( + Arc::as_ptr(inner), + (Arc::as_ptr(budget), inner._charge.bytes), + ); + }; + for shard in &store.shards { + let shard = shard.lock().unwrap(); + let mut resident = 0; + let mut entries = 0; + for (index, entry) in shard.slots.iter().enumerate() { + if let Some(entry) = entry { + let hash = shard.policy_slots[index].hash(); + assert_eq!(shard.find(hash, entry.value.key()), Some(index)); + assert!(Arc::ptr_eq( + entry.value.0._charge.budget.as_ref().unwrap(), + &shard.budget, + )); + record(&entry.value); + resident += entry.value.charged_bytes(); + entries += 1; + } + } + assert_eq!(shard.resident_bytes, resident); + assert_eq!(shard.free_count + entries, shard.slots.len()); + assert!(shard.budget.used_bytes() <= shard.budget.capacity_bytes); + budgets.insert(Arc::as_ptr(&shard.budget), shard.budget.used_bytes()); + total_resident += resident; + total_entries += entries; + } + for value in retained { + value.assert_unchanged(); + record(&value.value); + } + let mut expected = HashMap::<_, usize>::new(); + for (budget, bytes) in allocations.values() { + assert!(budgets.contains_key(budget)); + *expected.entry(*budget).or_default() += bytes; + } + for (budget, used) in budgets { + assert_eq!(used, expected.get(&budget).copied().unwrap_or(0)); + } + let snapshot = store.detailed_snapshot().unwrap(); + assert_eq!(snapshot.resident_entries, total_entries); + assert_eq!(snapshot.resident_bytes, total_resident); + assert_eq!( + snapshot.retained_bytes + total_resident, + expected.values().sum() + ); +} + +fn budgets(store: &MemoryStore) -> Vec> { + store + .shards + .iter() + .map(|shard| Arc::clone(&shard.lock().unwrap().budget)) + .collect() +} + +#[test] +fn mixed_sizes_and_retained_owners_have_exact_charges() { + for policy in [L1EvictionPolicy::Clock, L1EvictionPolicy::S3Fifo] { + let mut saw_multi_victim = false; + // Cover byte pressure, slot pressure, and uneven per-shard budgets. + for (capacity, slots, shards) in [(4096, 40, 1), (8192, 7, 3), (8192, 80, 3)] { + for seed in [1, 42, 777] { + let store = MemoryStore::new(capacity, slots, shards, policy, true).unwrap(); + let mut retained: Vec = Vec::new(); + let mut random = seed; + for seqno in 1..=MIXED_OPERATIONS { + let (key, hash) = key_and_hash(random_index(&mut random, KEY_COUNT)); + let value_len = VALUE_LENGTHS[random_index(&mut random, VALUE_LENGTHS.len())]; + let value = vec![(seqno % 251) as u8; value_len]; + let previous_evictions = store.metrics_snapshot().evictions; + match next_random(&mut random) % 10 { + 0..=4 => { + store.publish(hash, &key, &value, seqno); + } + 5 => { + let result = match store.lookup(hash, &key) { + MemoryLookup::Hit(value) => Some(value), + MemoryLookup::Miss(token) => { + store.promote(token, hash, &key, &value, seqno) + } + }; + if let Some(value) = result { + assert_eq!(value.key(), key); + retained.push(RetainedValueSnapshot::new(value)); + } + } + 6 => { + store.delete(hash, &key, seqno); + } + 7 => { + if !retained.is_empty() { + let index = random_index(&mut random, retained.len()); + retained.push(retained[index].clone()); + } + } + 8 => { + if !retained.is_empty() { + let index = random_index(&mut random, retained.len()); + retained.swap_remove(index); + } + } + _ => { + if let MemoryLookup::Hit(value) = store.lookup(hash, &key) { + assert_eq!(value.key(), key); + retained.push(RetainedValueSnapshot::new(value)); + } + } + } + saw_multi_victim |= store.metrics_snapshot().evictions - previous_evictions > 1; + if retained.len() > 24 { + retained.remove(0); + } + assert_accounting(&store, &retained); + } + retained.clear(); + assert_accounting(&store, &retained); + let budgets = budgets(&store); + drop(store); + for budget in budgets { + assert_eq!(budget.used_bytes(), 0); + } + } + } + assert!( + saw_multi_victim, + "each policy must transfer charges from multiple victims" + ); + } +} + +#[test] +fn concurrent_publication_clone_and_drop_preserve_old_values() { + for policy in [L1EvictionPolicy::Clock, L1EvictionPolicy::S3Fifo] { + for shards in [1, 4] { + let store = Arc::new(MemoryStore::new(32768, 128, shards, policy, true).unwrap()); + for id in 0..KEY_COUNT { + let (key, hash) = key_and_hash(id); + assert!(store.publish(hash, &key, &[1; 64], id as u64 + 1)); + } + let sequence = Arc::new(AtomicU64::new(KEY_COUNT as u64 + 1)); + let start = Arc::new(Barrier::new(CONCURRENT_WORKERS)); + let mut workers = Vec::new(); + for worker in 0..CONCURRENT_WORKERS { + let store = Arc::clone(&store); + let sequence = Arc::clone(&sequence); + let start = Arc::clone(&start); + // Every worker owns a known old value before mutation starts, + // even if scheduling makes its later lookups miss. + let (key, hash) = key_and_hash(worker); + let MemoryLookup::Hit(value) = store.lookup(hash, &key) else { + panic!("preloaded value must be resident before writers start"); + }; + let initial = RetainedValueSnapshot::new(value); + workers.push(std::thread::spawn(move || { + let mut random = worker as u64 + 1; + let mut retained: Vec = Vec::new(); + start.wait(); + for iteration in 0..CONCURRENT_OPERATIONS { + let (key, hash) = key_and_hash(random_index(&mut random, KEY_COUNT)); + let seqno = sequence.fetch_add(1, Ordering::Relaxed); + if worker < CONCURRENT_WORKERS / 2 { + let value = + vec![(seqno % 251) as u8; [64, 128, 512, 2048][iteration % 4]]; + if iteration % 23 == 0 { + store.delete(hash, &key, seqno); + } else { + store.publish(hash, &key, &value, seqno); + } + } else if let MemoryLookup::Hit(value) = store.lookup(hash, &key) { + assert_eq!(value.key(), key); + retained.push(RetainedValueSnapshot::new(value)); + } + if !retained.is_empty() && iteration % 3 == 0 { + let index = random_index(&mut random, retained.len()); + retained.push(retained[index].clone()); + } + initial.assert_unchanged(); + for value in &retained { + value.assert_unchanged(); + } + while retained.len() > 12 { + let index = random_index(&mut random, retained.len()); + retained.swap_remove(index); + } + if iteration % 31 == 0 { + std::thread::yield_now(); + } + } + retained.push(initial); + retained + })); + } + let mut retained = Vec::new(); + for worker in workers { + retained.extend(worker.join().unwrap()); + } + assert_accounting(&store, &retained); + retained.clear(); + assert_accounting(&store, &retained); + let budgets = budgets(&store); + drop(store); + for budget in budgets { + assert_eq!(budget.used_bytes(), 0); + } + } + } +} + +#[test] +fn last_external_drop_before_reuse_releases_original_charge_once() { + for policy in [L1EvictionPolicy::Clock, L1EvictionPolicy::S3Fifo] { + let required = memory_entry_bytes(b"old", &[1; 128]).unwrap(); + let mut shard = MemoryShard::new(2 * required, 2, policy).unwrap(); + let result = shard.insert(1, b"old", &[1; 128], 1, required, None); + let slot = result.slot.unwrap() as usize; + let held = shard.slots[slot].as_ref().unwrap().value.clone(); + let address = Arc::as_ptr(&held.0); + let ChargeResult::Charged { + charge, recycled, .. + } = shard.charge_with_eviction(required, Some(slot)) + else { + panic!("independent new charge should fit"); + }; + assert_eq!(shard.budget.used_bytes(), 2 * required); + assert!(!recycled.as_ref().unwrap().releases_charge_on_remove()); + + // Force the last caller to drop after detachment but before the + // factory rechecks exclusivity. The original charge is still armed. + std::thread::spawn(move || drop(held)).join().unwrap(); + let replacement = MemoryValue::try_new(b"new", &[2; 128], 2, charge, recycled) + .ok() + .unwrap(); + assert_eq!(Arc::as_ptr(&replacement.0), address); + assert_eq!(shard.budget.used_bytes(), required); + assert_eq!(replacement.key(), b"new"); + assert_eq!(replacement.as_ref(), &[2; 128]); + assert_eq!(replacement.seqno(), 2); + drop(replacement); + assert_eq!(shard.budget.used_bytes(), 0); + } +} diff --git a/cache2/src/region/manager.rs b/cache2/src/region/manager.rs index 9f1f566..e87c7c3 100644 --- a/cache2/src/region/manager.rs +++ b/cache2/src/region/manager.rs @@ -388,6 +388,53 @@ impl RegionManager { !self.free_regions.is_empty() } + pub fn rotation_requested(&self, shard_id: usize) -> Result { + self.append_states + .get(shard_id) + .map(|shard| shard.rotation_requested) + .ok_or(RegionMutationError::InvalidShard) + } + + /// Retains rotation pressure for an append rejected by staging preflight. + /// A record-count limit may require only a flush, so request rotation only + /// when the next record cannot fit after the required aligned flush. No + /// position or sequence number is reserved. + pub fn request_rotation_for_append( + &mut self, + shard_id: usize, + record_bytes: u32, + ) -> Result { + let region_id = *self + .active_regions + .get(shard_id) + .ok_or(RegionMutationError::InvalidShard)?; + let region = self + .regions + .get(region_id as usize) + .ok_or(RegionMutationError::Invariant( + "active Region id is out of bounds", + ))?; + if region.state != RegionState::Active { + return Err(RegionMutationError::Invariant( + "data shard does not own an Active Region", + )); + } + let padding = region.reserved_used.wrapping_neg() & (DIRECT_IO_ALIGNMENT as u64 - 1); + let end = region + .reserved_used + .checked_add(padding) + .and_then(|end| end.checked_add(u64::from(record_bytes))) + .ok_or(RegionMutationError::ArithmeticOverflow)?; + let needs_rotation = end > self.region_size; + if needs_rotation { + self.append_states + .get_mut(shard_id) + .ok_or(RegionMutationError::InvalidShard)? + .rotation_requested = true; + } + Ok(needs_rotation) + } + /// Allocates one process-local ordering version. Sequence exhaustion is a /// terminal condition for the current cache identity; `u64::MAX` is never /// issued because clean metadata reserves it as invalid. @@ -441,16 +488,14 @@ impl RegionManager { .checked_add(u64::from(record_bytes)) .ok_or(RegionMutationError::ArithmeticOverflow)?; if end > self.region_size { - let error = if shard.open_span.is_some() { + let error = if shard.open_span.is_some() || shard.submitted_span.is_some() { RegionMutationError::FlushBeforeRotation - } else if shard.submitted_span.is_some() { - RegionMutationError::WouldBlock } else { RegionMutationError::RegionFull }; - if error == RegionMutationError::RegionFull { - self.append_states[shard_id].rotation_requested = true; - } + // Keep the capacity request through the final write so completion + // can rotate without requiring a second rejected foreground fill. + self.append_states[shard_id].rotation_requested = true; return Err(error); } let offset = u32::try_from(region.reserved_used) @@ -1549,29 +1594,166 @@ mod tests { } #[test] - fn full_region_flushes_its_open_tail_before_requesting_rotation() { - let mut manager = RegionManager::from_metadata(sample()).unwrap(); - let active_region = manager.active_regions()[0] as usize; - let record_bytes = u32::try_from(manager.region_size() - RECORD_ALIGNMENT as u64).unwrap(); - let reservation = manager.reserve_append(0, record_bytes).unwrap(); - manager.stage_reservation(reservation).unwrap(); + fn rotation_request_respects_every_flush_alignment_boundary() { + #[derive(Clone, Copy, Debug)] + enum PreviousSpan { + None, + Completed, + Submitted, + } - assert_eq!( - manager.reserve_append(0, 64), - Err(RegionMutationError::FlushBeforeRotation) - ); - let padding = manager.reserve_write_padding(0).unwrap().unwrap(); - assert_eq!(padding.padding_bytes(), Some(RECORD_ALIGNMENT)); - let span = manager.seal_write_span_with_padding(padding).unwrap(); - manager.complete_write_span(span).unwrap(); - assert_eq!( - manager.regions()[active_region].completed_used, - manager.region_size() - ); - assert_eq!( - manager.reserve_append(0, RECORD_ALIGNMENT), - Err(RegionMutationError::RegionFull) - ); + const REGION_BYTES: u64 = 16 * 1024; + let alignment = DIRECT_IO_ALIGNMENT as u32; + let unit = i64::from(RECORD_ALIGNMENT); + let mut checked = 0; + // Cover every possible record-aligned tail in one I/O block, both + // before and after an earlier flush, including an in-flight span. + for previous in [ + PreviousSpan::None, + PreviousSpan::Completed, + PreviousSpan::Submitted, + ] { + for tail in (RECORD_ALIGNMENT..=alignment).step_by(RECORD_ALIGNMENT as usize) { + for delta in [-unit, 0, unit] { + let mut metadata = sample(); + metadata.root.region_size = REGION_BYTES; + let mut manager = RegionManager::from_metadata(metadata).unwrap(); + let active = manager.active_regions()[0]; + let mut submitted = None; + if !matches!(previous, PreviousSpan::None) { + let reservation = manager.reserve_append(0, alignment).unwrap(); + manager.stage_reservation(reservation).unwrap(); + let span = manager.seal_write_span(0).unwrap(); + if matches!(previous, PreviousSpan::Completed) { + manager.complete_write_span(span).unwrap(); + } else { + submitted = Some(span); + } + } + let reservation = manager.reserve_append(0, tail).unwrap(); + manager.stage_reservation(reservation).unwrap(); + let used = manager.regions()[active as usize].reserved_used; + let padded = used.div_ceil(u64::from(alignment)) * u64::from(alignment); + let remaining = REGION_BYTES - padded; + let next = u32::try_from(remaining.checked_add_signed(delta).unwrap()).unwrap(); + let sequence = manager.next_seqno(); + let completed = manager.regions()[active as usize].completed_used; + let needs_rotation = delta > 0; + // Repeated preflight failures retain intent without + // consuming another position or logical sequence number. + for _ in 0..2 { + assert_eq!( + manager.request_rotation_for_append(0, next).unwrap(), + needs_rotation, + "previous={previous:?}, tail={tail}, delta={delta}" + ); + assert_eq!(manager.rotation_requested(0).unwrap(), needs_rotation); + assert_eq!(manager.next_seqno(), sequence); + assert_eq!(manager.regions()[active as usize].reserved_used, used); + assert_eq!(manager.regions()[active as usize].completed_used, completed); + assert_eq!( + manager.rotation_candidate(0), + Err(RegionMutationError::WouldBlock) + ); + } + if let Some(submitted) = submitted { + assert_eq!( + manager.reserve_write_padding(0), + Err(RegionMutationError::WouldBlock) + ); + manager.complete_write_span(submitted).unwrap(); + assert_eq!(manager.regions()[active as usize].reserved_used, used); + assert_eq!(manager.rotation_requested(0).unwrap(), needs_rotation); + } + // Compare the preflight decision with actual padding and + // completion, using the absolute Region offset throughout. + let span = match manager.reserve_write_padding(0).unwrap() { + Some(padding) => { + assert_eq!(padding.padding_bytes(), Some((padded - used) as u32)); + manager.seal_write_span_with_padding(padding).unwrap() + } + None => manager.seal_write_span(0).unwrap(), + }; + assert_eq!(span.end_offset, padded); + assert_eq!( + manager.rotation_candidate(0), + Err(RegionMutationError::WouldBlock) + ); + manager.complete_write_span(span).unwrap(); + assert_eq!(manager.regions()[active as usize].completed_used, padded); + if needs_rotation { + let candidate = manager.rotation_candidate(0).unwrap(); + let receipt = manager.begin_rotation(candidate).unwrap(); + manager.finish_rotation(receipt).unwrap(); + assert!(!manager.rotation_requested(0).unwrap()); + assert_ne!(manager.active_regions()[0], active); + } + assert_eq!( + manager.rotation_candidate(0), + Err(RegionMutationError::WouldBlock) + ); + let next_reservation = manager.reserve_append(0, next).unwrap(); + assert_eq!( + u64::from(next_reservation.offset), + if needs_rotation { 0 } else { padded } + ); + checked += 1; + } + } + } + assert_eq!(checked, 3 * (alignment / RECORD_ALIGNMENT) * 3); + } + + #[test] + fn full_region_retains_rotation_request_until_its_final_write_completes() { + for submitted in [false, true] { + let mut manager = RegionManager::from_metadata(sample()).unwrap(); + let active_region = manager.active_regions()[0] as usize; + let record_bytes = + u32::try_from(manager.region_size() - RECORD_ALIGNMENT as u64).unwrap(); + let reservation = manager.reserve_append(0, record_bytes).unwrap(); + manager.stage_reservation(reservation).unwrap(); + + if !submitted { + assert_eq!( + manager.reserve_append(0, 64), + Err(RegionMutationError::FlushBeforeRotation) + ); + assert_eq!( + manager.rotation_candidate(0), + Err(RegionMutationError::WouldBlock) + ); + } + let padding = manager.reserve_write_padding(0).unwrap().unwrap(); + assert_eq!(padding.padding_bytes(), Some(RECORD_ALIGNMENT)); + let span = manager.seal_write_span_with_padding(padding).unwrap(); + if submitted { + assert_eq!( + manager.reserve_append(0, 64), + Err(RegionMutationError::FlushBeforeRotation) + ); + } + assert_eq!( + manager.rotation_candidate(0), + Err(RegionMutationError::WouldBlock) + ); + manager.complete_write_span(span).unwrap(); + assert_eq!( + manager.regions()[active_region].completed_used, + manager.region_size() + ); + // No new failed reservation is needed to make rotation eligible. + let rotation = manager + .begin_rotation(manager.rotation_candidate(0).unwrap()) + .unwrap(); + manager.finish_rotation(rotation).unwrap(); + assert_ne!(manager.active_regions()[0] as usize, active_region); + assert_eq!( + manager.rotation_candidate(0), + Err(RegionMutationError::WouldBlock), + "the retained request must rotate exactly once" + ); + } } #[test] diff --git a/cache2/src/region/mod.rs b/cache2/src/region/mod.rs index 2bb9c1c..0022748 100644 --- a/cache2/src/region/mod.rs +++ b/cache2/src/region/mod.rs @@ -265,10 +265,18 @@ pub enum RegionStageValue { previous_bytes: usize, current_bytes: usize, }, - NeedsProgress, + Contended, + NeedsFlush, NeedsRotation, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum RegionRotation { + Idle, + Pending, + Rotated, +} + pub struct RegionValue { buffer: BufferLease, buffer_len: usize, @@ -295,6 +303,16 @@ impl RegionValue { } impl RegionStore { + #[cfg(test)] + pub fn lock_manager_for_test(&self) -> MutexGuard<'_, RegionManager> { + self.manager.lock().unwrap() + } + + #[cfg(test)] + pub fn lock_append_for_test(&self, shard_id: usize) -> MutexGuard<'_, ()> { + self.append_gates[shard_id].mutation.lock().unwrap() + } + /// Consume the index and Region metadata into one live store. pub fn from_recovery( index: PartitionedIndexStorage, @@ -883,7 +901,7 @@ impl RegionStore { let payload = RecordPayload::new(key, value); let _shard_mutation = match self.append_gate(shard_id)?.mutation.try_lock() { Ok(guard) => guard, - Err(TryLockError::WouldBlock) => return Ok(RegionStageValue::NeedsProgress), + Err(TryLockError::WouldBlock) => return Ok(RegionStageValue::Contended), Err(TryLockError::Poisoned(_)) => { self.health.enter_miss_only(); return Err(io::Error::new( @@ -892,29 +910,42 @@ impl RegionStore { )); } }; - match staging.preflight_append(shard_id, record_bytes) { - Ok(StageAppend::Appended { .. }) => {} - Ok(StageAppend::NeedsSeal) => return Ok(RegionStageValue::NeedsProgress), + let needs_seal = match staging.preflight_append(shard_id, record_bytes) { + Ok(StageAppend::Appended { .. }) => false, + Ok(StageAppend::NeedsSeal) => true, Err(StagingError::WouldBlock) => { - return Ok(RegionStageValue::NeedsProgress); + return Ok(RegionStageValue::Contended); } Err(error) => { self.health.enter_miss_only(); staging.close(); return Err(staging_io_error(error)); } - } + }; let receipt = { let Some(mut manager) = self.manager.try_lock()? else { - return Ok(RegionStageValue::NeedsProgress); + return Ok(RegionStageValue::Contended); }; + if needs_seal { + let needs_rotation = manager + .request_rotation_for_append(shard_id, record_bytes) + .map_err(|error| { + self.health.enter_miss_only(); + region_mutation_io_error(error) + })?; + return Ok(if needs_rotation { + RegionStageValue::NeedsRotation + } else { + RegionStageValue::NeedsFlush + }); + } let receipt = match manager.reserve_append(shard_id, record_bytes) { Ok(receipt) => receipt, Err(RegionMutationError::WouldBlock) => { - return Ok(RegionStageValue::NeedsProgress); + return Ok(RegionStageValue::NeedsFlush); } Err(RegionMutationError::FlushBeforeRotation) => { - return Ok(RegionStageValue::NeedsProgress); + return Ok(RegionStageValue::NeedsRotation); } Err(RegionMutationError::RegionFull) => { return Ok(RegionStageValue::NeedsRotation); @@ -1182,7 +1213,7 @@ impl RegionStore { /// Rotates one empty data shard. Concurrent reads validate the returned /// record identity and may observe either valid generation. - pub fn rotate_shard(&self, shard_id: usize) -> io::Result { + pub fn rotate_shard(&self, shard_id: usize) -> io::Result { self.health.require_healthy()?; let shard_mutation = self.lock_shard_mutation(shard_id)?; let rotation = self.rotation.lock().map_err(|_| { @@ -1192,10 +1223,19 @@ impl RegionStore { "Region rotation gate is poisoned", ) })?; - let candidate = match self.manager.lock()?.rotation_candidate(shard_id) { - Ok(candidate) => candidate, - Err(RegionMutationError::WouldBlock) => return Ok(false), - Err(error) => return Err(region_mutation_context("rotation selection", error)), + let candidate = { + let manager = self.manager.lock()?; + if !manager + .rotation_requested(shard_id) + .map_err(|error| region_mutation_context("rotation selection", error))? + { + return Ok(RegionRotation::Idle); + } + match manager.rotation_candidate(shard_id) { + Ok(candidate) => candidate, + Err(RegionMutationError::WouldBlock) => return Ok(RegionRotation::Pending), + Err(error) => return Err(region_mutation_context("rotation selection", error)), + } }; let receipt = self @@ -1224,7 +1264,7 @@ impl RegionStore { self.health.enter_miss_only(); return Err(region_mutation_context("rotation completion", error)); } - Ok(true) + Ok(RegionRotation::Rotated) } /// Publishes a completed batch without entering global Region authority. diff --git a/cache2/src/region/persistence/tests.rs b/cache2/src/region/persistence/tests.rs index 3836ad6..0f7ab2d 100644 --- a/cache2/src/region/persistence/tests.rs +++ b/cache2/src/region/persistence/tests.rs @@ -793,7 +793,7 @@ fn foreground_stage_bypasses_busy_manager_without_consuming_a_sequence() { regions .try_stage_value(&staging, 0, hash, record_bytes, b"key", b"value") .unwrap(), - RegionStageValue::NeedsProgress + RegionStageValue::Contended ); assert_eq!(manager.next_seqno(), next_seqno); } @@ -817,7 +817,7 @@ fn foreground_stage_rejects_busy_shard_without_reserving_then_stages_once() { drop(mutation); writer.join().unwrap(); let (result, staging) = early.expect("a busy shard must reject without waiting for its lock"); - assert!(matches!(result.unwrap(), RegionStageValue::NeedsProgress)); + assert!(matches!(result.unwrap(), RegionStageValue::Contended)); assert_eq!( regions.manager.state.lock().unwrap().next_seqno(), next_seqno @@ -909,7 +909,7 @@ fn completed_owned_span_publishes_index_without_a_steady_state_sync() { last = Some((key, hash, seqno)); staged_records += 1; } - RegionStageValue::NeedsProgress => break, + RegionStageValue::NeedsFlush => break, outcome => panic!("unexpected staging outcome: {outcome:?}"), } } @@ -1258,7 +1258,14 @@ fn rotation_is_committed_without_a_metadata_io_boundary() { .request_rotation_for_test(0) .unwrap(); - assert!(regions.rotate_shard(0).unwrap()); + assert_eq!( + regions.rotate_shard(0).unwrap(), + crate::region::RegionRotation::Rotated + ); + assert_eq!( + regions.rotate_shard(0).unwrap(), + crate::region::RegionRotation::Idle + ); assert!(regions.health.is_healthy()); let manager = regions.manager.lock().unwrap(); assert_eq!(manager.active_regions()[0], REGION_SHARDS); diff --git a/cache2/src/snapshot.rs b/cache2/src/snapshot.rs index 8b5e8eb..3c0aa13 100644 --- a/cache2/src/snapshot.rs +++ b/cache2/src/snapshot.rs @@ -290,7 +290,8 @@ pub struct RegionSnapshot { pub struct DetailedCacheSnapshot { /// Operational summary sampled for this diagnostic. pub summary: CacheSnapshot, - /// Mutations rejected specifically because an append buffer needed progress. + /// Mutations rejected because append capacity or Region rotation needed progress. + /// Excludes lock contention; total mutation rejections remain in `summary.write_rejections`. pub write_buffer_rejections: u64, /// L1 occupancy and metadata detail. pub l1: CacheL1Snapshot,