This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 8d08e48a [core] Reduce shared disk cache lock contention (#1186)
8d08e48a is described below
commit 8d08e48a09d9119b701e993cf09bf2e83d08943c
Author: XiaoHongbo <[email protected]>
AuthorDate: Wed Oct 21 20:03:19 2026 +0800
[core] Reduce shared disk cache lock contention (#1186)
---
crates/paimon/src/io/cache/disk.rs | 44 +++-
crates/paimon/src/io/cache/mod.rs | 20 ++
crates/paimon/src/io/cache/reader.rs | 117 +++++++++-
crates/paimon/src/io/cache/shared.rs | 426 +++++++++++++++++++++++++++++++----
docs/src/getting-started.md | 6 +
5 files changed, 554 insertions(+), 59 deletions(-)
diff --git a/crates/paimon/src/io/cache/disk.rs
b/crates/paimon/src/io/cache/disk.rs
index 3b217e04..baf25235 100644
--- a/crates/paimon/src/io/cache/disk.rs
+++ b/crates/paimon/src/io/cache/disk.rs
@@ -693,14 +693,22 @@ fn encode_block(key: &BlockKey, payload: &Bytes) ->
Vec<u8> {
}
fn decode_block(key: &BlockKey, encoded: &[u8]) -> Result<Bytes,
BlockDecodeError> {
- let (decoded_key, payload) = decode_block_any(encoded)?;
+ let (decoded_key, payload) = validate_block(encoded)?;
if &decoded_key != key {
return Err(BlockDecodeError("cache block key does not match"));
}
- Ok(payload)
+ Ok(Bytes::copy_from_slice(&encoded[payload]))
}
-fn decode_block_any(encoded: &[u8]) -> Result<(BlockKey, Bytes),
BlockDecodeError> {
+fn decode_block_owned(key: &BlockKey, encoded: Vec<u8>) -> Result<Bytes,
BlockDecodeError> {
+ let (decoded_key, payload) = validate_block(&encoded)?;
+ if &decoded_key != key {
+ return Err(BlockDecodeError("cache block key does not match"));
+ }
+ Ok(Bytes::from(encoded).slice(payload))
+}
+
+fn validate_block(encoded: &[u8]) -> Result<(BlockKey,
std::ops::Range<usize>), BlockDecodeError> {
let (key, layout) = decode_block_header(encoded, encoded.len() as u64)?;
let payload_len = usize::try_from(layout.payload_len)
.map_err(|_| BlockDecodeError("cache block payload is too large"))?;
@@ -717,10 +725,7 @@ fn decode_block_any(encoded: &[u8]) -> Result<(BlockKey,
Bytes), BlockDecodeErro
return Err(BlockDecodeError("cache block checksum does not match"));
}
- Ok((
- key,
- Bytes::copy_from_slice(&encoded[layout.header_len..payload_end]),
- ))
+ Ok((key, layout.header_len..payload_end))
}
#[derive(Clone, Copy, Debug)]
@@ -846,6 +851,31 @@ mod tests {
assert_eq!(decoded, payload);
}
+ #[test]
+ fn test_owned_block_decode_reuses_allocation_and_validates() {
+ let key = BlockKey::new("s3://bucket/table/data.parquet", 1024, 3);
+ let payload = Bytes::from_static(b"cached block");
+ let encoded = encode_block(&key, &payload);
+ let (_, range) = validate_block(&encoded).unwrap();
+ let expected_address = encoded.as_ptr() as usize + range.start;
+ let decoded = decode_block_owned(&key, encoded).unwrap();
+ assert_eq!(decoded, payload);
+ assert_eq!(decoded.as_ptr() as usize, expected_address);
+
+ let wrong_key = BlockKey::new("s3://bucket/table/data.parquet", 1024,
4);
+ assert!(decode_block_owned(&wrong_key, encode_block(&key,
&payload)).is_err());
+ let encoded = encode_block(&key, &payload);
+ for end in 0..encoded.len() {
+ assert!(decode_block_owned(&key,
encoded[..end].to_vec()).is_err());
+ }
+ let mut corrupt = encoded;
+ corrupt[range.start] ^= 1;
+ assert!(decode_block_owned(&key, corrupt).is_err());
+ assert!(decode_block_owned(&key, encode_block(&key, &Bytes::new()))
+ .unwrap()
+ .is_empty());
+ }
+
#[test]
fn test_disk_block_codec_rejects_crc_corruption() {
let key = BlockKey::new("s3://bucket/table/snapshot/snapshot-1", 1024,
3);
diff --git a/crates/paimon/src/io/cache/mod.rs
b/crates/paimon/src/io/cache/mod.rs
index 701f305b..5891f283 100644
--- a/crates/paimon/src/io/cache/mod.rs
+++ b/crates/paimon/src/io/cache/mod.rs
@@ -48,6 +48,8 @@ pub(crate) struct LocalCache {
block_size: u64,
whitelist: HashSet<FileType>,
file_size_capacity: usize,
+ // Bound queued and active background fills.
+ fill_slots: Arc<tokio::sync::Semaphore>,
}
#[derive(Debug)]
@@ -76,6 +78,7 @@ impl LocalCache {
block_size,
whitelist: FileType::parse_whitelist(whitelist),
file_size_capacity: DEFAULT_FILE_SIZE_CAPACITY,
+ fill_slots: Arc::new(tokio::sync::Semaphore::new(8)),
})
}
@@ -124,6 +127,7 @@ impl LocalCache {
block_size: config.block_size,
whitelist: config.whitelist,
file_size_capacity,
+ fill_slots: Arc::new(tokio::sync::Semaphore::new(8)),
})
}
@@ -131,6 +135,22 @@ impl LocalCache {
self.block_size
}
+ fn background_fill_slot(&self) ->
Option<tokio::sync::OwnedSemaphorePermit> {
+ // Current-thread runtimes stop driving spawned fills when block_on
returns.
+ // Keep those fills synchronous so their block locks cannot outlive
the read.
+ if !matches!(
+ tokio::runtime::Handle::try_current().map(|handle|
handle.runtime_flavor()),
+ Ok(tokio::runtime::RuntimeFlavor::MultiThread)
+ ) {
+ return None;
+ }
+ // Limit background payloads to 8 MiB per cache; otherwise fill
synchronously.
+ if !matches!(self.backend, CacheBackend::SharedDisk(_)) ||
self.block_size > 1024 * 1024 {
+ return None;
+ }
+ self.fill_slots.clone().try_acquire_owned().ok()
+ }
+
fn block_key(&self, path: &str, block_index: u64) -> BlockKey {
BlockKey::with_namespace(&self.namespace, path, self.block_size,
block_index)
}
diff --git a/crates/paimon/src/io/cache/reader.rs
b/crates/paimon/src/io/cache/reader.rs
index bcb83ddd..35781cc5 100644
--- a/crates/paimon/src/io/cache/reader.rs
+++ b/crates/paimon/src/io/cache/reader.rs
@@ -82,8 +82,8 @@ impl CachedFileReader {
}
let load_lock = self.cache.block_load_lock(&key).await;
- let load_guard = load_lock.lock().await;
- let _shared_guard = self.cache.shared_load_lock(&key).await;
+ let load_guard = load_lock.clone().lock_owned().await;
+ let shared_guard = self.cache.shared_load_lock(&key).await;
let result = if let Some(payload) = self
.cache
.get_block(&key, expected_len, &self.read_token)
@@ -107,6 +107,20 @@ impl CachedFileReader {
} else {
match self.delegate.read(start..end).await {
Ok(payload) if payload.len() == expected_len => {
+ if let Some(slot) = self.cache.background_fill_slot() {
+ let cache = self.cache.clone();
+ let token = self.read_token.clone();
+ let fill_payload = payload.clone();
+ // Hold download guards until publication completes.
+ tokio::spawn(async move {
+ let _slot = slot;
+ let _shared_guard = shared_guard;
+ cache.put_block(&key, fill_payload, &token).await;
+ drop(load_guard);
+ cache.release_block_load_lock(&key,
&load_lock).await;
+ });
+ return Ok(payload);
+ }
self.cache
.put_block(&key, payload.clone(), &self.read_token)
.await;
@@ -125,6 +139,7 @@ impl CachedFileReader {
Err(error) => Err(error),
}
};
+ drop(shared_guard);
drop(load_guard);
self.cache.release_block_load_lock(&key, &load_lock).await;
result
@@ -356,6 +371,104 @@ mod tests {
}
}
+ #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+ async fn
shared_fill_returns_before_publication_and_preserves_single_flight() {
+ let directory = tempfile::tempdir().unwrap();
+ let delegate = Arc::new(CountingReader {
+ data: Bytes::from_static(b"abcdefgh"),
+ reads: AtomicUsize::new(0),
+ });
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ shared: true,
+ dir: Some(directory.path().to_path_buf()),
+ namespace: "test".into(),
+ max_size: Some(4096),
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ let reader = CachedFileReader::new(delegate.clone(), "data.parquet",
8, cache.clone());
+ let gate = std::fs::OpenOptions::new()
+ .read(true)
+ .write(true)
+ .open(
+ directory
+ .path()
+ .join("paimon-shared-local-cache-v1/directory.lock"),
+ )
+ .unwrap();
+ gate.lock().unwrap();
+ assert_eq!(
+ tokio::time::timeout(std::time::Duration::from_millis(500),
reader.read(0..4))
+ .await
+ .unwrap()
+ .unwrap(),
+ "abcd"
+ );
+ assert_eq!(cache.fill_slots.available_permits(), 7);
+ assert!(
+ tokio::time::timeout(std::time::Duration::from_millis(50),
reader.read(0..4))
+ .await
+ .is_err()
+ );
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+ drop(gate);
+ assert_eq!(reader.read(0..4).await.unwrap(), "abcd");
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+ let permits = cache
+ .fill_slots
+ .clone()
+ .acquire_many_owned(8)
+ .await
+ .unwrap();
+ assert!(cache.background_fill_slot().is_none());
+ drop(permits);
+ }
+
+ #[test]
+ fn shared_fill_does_not_block_reads_on_another_runtime() {
+ let directory = tempfile::tempdir().unwrap();
+ let delegate = Arc::new(CountingReader {
+ data: Bytes::from_static(b"abcdefgh"),
+ reads: AtomicUsize::new(0),
+ });
+ let cache = Arc::new(
+ LocalCache::new(LocalCacheConfig {
+ shared: true,
+ dir: Some(directory.path().to_path_buf()),
+ namespace: "test".into(),
+ max_size: Some(4096),
+ block_size: 4,
+ whitelist: std::collections::HashSet::from([FileType::Meta]),
+ })
+ .unwrap(),
+ );
+ let reader = CachedFileReader::new(delegate.clone(), "data.parquet",
8, cache);
+ let first_runtime = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ assert_eq!(first_runtime.block_on(reader.read(0..4)).unwrap(), "abcd");
+
+ // Keep the first runtime alive without driving its background tasks.
+ let second_runtime = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ let second = second_runtime.block_on(async {
+ tokio::time::timeout(std::time::Duration::from_millis(500),
reader.read(0..4)).await
+ });
+ assert_eq!(
+ second
+ .expect("read must not wait for an idle runtime")
+ .unwrap(),
+ "abcd"
+ );
+ assert_eq!(delegate.reads.load(Ordering::SeqCst), 1);
+ }
+
#[tokio::test]
async fn test_cached_range_single_flight_reads_cold_block_once() {
let delegate = Arc::new(SlowCountingReader {
diff --git a/crates/paimon/src/io/cache/shared.rs
b/crates/paimon/src/io/cache/shared.rs
index 0ace0c0e..f5b3dcdc 100644
--- a/crates/paimon/src/io/cache/shared.rs
+++ b/crates/paimon/src/io/cache/shared.rs
@@ -16,12 +16,12 @@
// under the License.
//! Opt-in node-local cache. Block files, not a process-local index, are
authoritative.
-//! A shared directory lock protects cache hits, while an exclusive lock
serializes
-//! publication, eviction, and invalidation. File-striped locks coalesce
downloads.
+//! Epoch validation protects immutable hits; an exclusive directory lock
serializes
+//! publication, eviction, and invalidation. Block-striped locks coalesce
downloads.
use super::*;
use std::fs::{File, OpenOptions};
-use std::io::{self, Write};
+use std::io::{self, Read, Write};
use std::time::Duration;
const MUTATION_MARKER: &str = "mutation-pending";
@@ -31,6 +31,88 @@ pub(crate) struct SharedDiskCache {
root: PathBuf,
max_size: u64,
high_priority_ratio: f64,
+ staging_slots: Arc<tokio::sync::Semaphore>,
+}
+
+/// Unpublished inode, locked against cleanup and isolated from legacy
temporaries.
+struct StagedBlock {
+ path: PathBuf,
+ _file: File,
+ len: u64,
+}
+
+impl StagedBlock {
+ fn create(root: &Path, encoded: &[u8]) -> io::Result<Self> {
+ let directory = root.join("staging");
+ std::fs::create_dir_all(&directory)?;
+ let path = directory.join(uuid::Uuid::new_v4().to_string());
+ let mut options = OpenOptions::new();
+ options.read(true).write(true).create_new(true);
+ #[cfg(unix)]
+ {
+ use std::os::unix::fs::OpenOptionsExt;
+ options.mode(0o600);
+ }
+ let file = options.open(&path)?;
+ let mut staged = Self {
+ path,
+ _file: file,
+ len: encoded.len() as u64,
+ };
+ // Cleanup may reclaim the inode before we lock it.
+ staged._file.try_lock().map_err(io::Error::other)?;
+ if !staged.path.try_exists()? {
+ return Err(io::Error::new(
+ io::ErrorKind::NotFound,
+ "staging inode was reclaimed",
+ ));
+ }
+ staged._file.write_all(encoded)?;
+ Ok(staged)
+ }
+}
+
+impl Drop for StagedBlock {
+ fn drop(&mut self) {
+ // Remove unpublished staging while still holding its lock.
+ let _ = std::fs::remove_file(&self.path);
+ }
+}
+
+fn cleanup_staging(root: &Path) -> io::Result<()> {
+ let entries = match std::fs::read_dir(root.join("staging")) {
+ Ok(entries) => entries,
+ Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),
+ Err(e) => return Err(e),
+ };
+ for entry in entries {
+ let entry = entry?;
+ if
uuid::Uuid::parse_str(&entry.file_name().to_string_lossy()).is_err() {
+ continue;
+ }
+ match entry.file_type() {
+ Ok(kind) if kind.is_file() => (),
+ Ok(_) => continue,
+ Err(e) if e.kind() == io::ErrorKind::NotFound => continue,
+ Err(e) => return Err(e),
+ }
+ let file = match
OpenOptions::new().read(true).write(true).open(entry.path()) {
+ Ok(file) => file,
+ Err(e) if e.kind() == io::ErrorKind::NotFound => continue,
+ Err(e) => return Err(e),
+ };
+ // Do not wait: the publisher may be waiting for our directory lock.
+ match file.try_lock() {
+ Ok(()) => match std::fs::remove_file(entry.path()) {
+ Ok(()) => (),
+ Err(e) if e.kind() == io::ErrorKind::NotFound => (),
+ Err(e) => return Err(e),
+ },
+ Err(std::fs::TryLockError::WouldBlock) => (),
+ Err(std::fs::TryLockError::Error(e)) => return Err(e),
+ }
+ }
+ Ok(())
}
fn open_lock(path: &Path) -> io::Result<File> {
@@ -69,6 +151,7 @@ async fn lock(path: PathBuf) -> Option<File> {
}
}
+#[cfg(test)]
async fn read_lock(path: PathBuf) -> Option<File> {
let file = tokio::task::spawn_blocking(move || open_lock(&path))
.await
@@ -200,6 +283,7 @@ impl SharedDiskCache {
let used = recover_usage(&root, max_size,
high_priority_ratio)?;
finish_usage(&root, used)?;
}
+ cleanup_staging(&root)?;
Ok(())
};
initialize().map_err(|e| crate::Error::ConfigInvalid {
@@ -209,6 +293,7 @@ impl SharedDiskCache {
root,
max_size,
high_priority_ratio,
+ staging_slots: Arc::new(tokio::sync::Semaphore::new(8)),
})
}
@@ -218,9 +303,9 @@ impl SharedDiskCache {
}
pub(crate) async fn load_lock(&self, key: &BlockKey) -> Option<File> {
- // All blocks of one file use the same stripe, also coordinating
read_full.
- let stripe =
- BlockKey::with_namespace(&key.namespace, &key.path, 0,
0).cache_relative_path();
+ // Best-effort block coalescing; read_full locks only the first block.
+ // Publication safety relies on the directory lock and generation
checks.
+ let stripe = key.cache_relative_path();
lock(self.root.join(format!(
"load-{}.lock",
stripe.components().next()?.as_os_str().to_string_lossy()
@@ -232,23 +317,39 @@ impl SharedDiskCache {
if generation.is_none() {
return None;
}
- let gate = read_lock(self.root.join("directory.lock")).await?;
let root = self.root.clone();
let key = generation_key(key, generation.as_ref()?);
let generation = generation.clone();
tokio::task::spawn_blocking(move || {
- let _gate = gate;
if !mutation_cleared(&root)
|| std::fs::read_to_string(root.join("generation")).ok() !=
generation
{
return None;
}
let path = root.join(key.cache_relative_path());
- let data = std::fs::read(&path).ok()?;
- let payload = decode_block(&key, &data).ok()?;
- // Persist recency so eviction observes hits from every process.
- if let Ok(file) = File::open(path) {
- let _ = file.set_modified(std::time::SystemTime::now());
+ // Immutable inodes survive unlink; recheck the epoch after reading
+ // to reject concurrent invalidation.
+ let mut file = File::open(&path).ok()?;
+ let metadata = file.metadata().ok()?;
+ // Throttle recency writes for approximate cross-process LRU.
+ if metadata
+ .modified()
+ .ok()
+ .and_then(|t| t.elapsed().ok())
+ .is_some_and(|age| age >= Duration::from_secs(10))
+ {
+ // Windows requires a write-capable handle to update mtime.
+ // A failed touch must not turn a valid cached block into a
miss.
+ let _ = OpenOptions::new()
+ .write(true)
+ .open(&path)
+ .and_then(|file|
file.set_modified(std::time::SystemTime::now()));
+ }
+ let mut data =
Vec::with_capacity(usize::try_from(metadata.len()).ok()?);
+ file.read_to_end(&mut data).ok()?;
+ let payload = decode_block_owned(&key, data).ok()?;
+ if !mutation_cleared(&root) || read_generation(&root) !=
generation {
+ return None;
}
Some(payload)
})
@@ -261,25 +362,57 @@ impl SharedDiskCache {
if generation.is_none() {
return;
}
- let Some(gate) = lock(self.root.join("directory.lock")).await else {
+ let Some(staged) = self.stage_block(key, payload, generation).await
else {
return;
};
+ self.publish_staged(staged, generation).await;
+ }
+
+ async fn stage_block(
+ &self,
+ key: &BlockKey,
+ payload: Bytes,
+ generation: &Option<String>,
+ ) -> Option<(BlockKey, StagedBlock, tokio::sync::OwnedSemaphorePermit)> {
+ let root = self.root.clone();
+ let key = generation_key(key, generation.as_ref()?);
+ let max_size = self.max_size;
+ // Stage outside the directory lock; recheck the generation before
publishing.
+ let slot = self.staging_slots.clone().acquire_owned().await.ok()?;
+ let staging_root = root.clone();
+ tokio::task::spawn_blocking(move || -> io::Result<_> {
+ let encoded = encode_block(&key, &payload);
+ if encoded.len() as u64 > max_size {
+ return Err(io::Error::other("cache block exceeds capacity"));
+ }
+ let staged = StagedBlock::create(&staging_root, &encoded)?;
+ Ok((key, staged, slot))
+ })
+ .await
+ .ok()?
+ .ok()
+ }
+
+ async fn publish_staged(
+ &self,
+ (key, staged, slot): (BlockKey, StagedBlock,
tokio::sync::OwnedSemaphorePermit),
+ generation: &Option<String>,
+ ) {
let root = self.root.clone();
- let key = generation_key(key, generation.as_ref().unwrap());
let generation = generation.clone();
let max_size = self.max_size;
let high_priority_ratio = self.high_priority_ratio;
+ let Some(gate) = lock(self.root.join("directory.lock")).await else {
+ return;
+ };
let result = tokio::task::spawn_blocking(move || -> io::Result<()> {
let _gate = gate;
+ let _slot = slot;
if !mutation_cleared(&root)
|| std::fs::read_to_string(root.join("generation")).ok() !=
generation
{
return Ok(());
}
- let encoded = encode_block(&key, &payload);
- if encoded.len() as u64 > max_size {
- return Ok(());
- }
let path = root.join(key.cache_relative_path());
let mut used = if root.join("dirty").exists() {
recover_usage(&root, max_size, high_priority_ratio)?
@@ -296,47 +429,23 @@ impl SharedDiskCache {
// A killed process leaves it for the next writer to recover.
std::fs::write(root.join("dirty"), b"")?;
let mut previous = std::fs::metadata(&path).map(|m|
m.len()).unwrap_or(0);
- if used
- .saturating_sub(previous)
- .saturating_add(encoded.len() as u64)
- > max_size
- {
+ if used.saturating_sub(previous).saturating_add(staged.len) >
max_size {
// Evict in batches, not one full directory scan per new block.
used = recover_usage(
&root,
- (max_size - encoded.len() as u64).min(max_size - max_size
/ 10),
+ (max_size - staged.len).min(max_size - max_size / 10),
high_priority_ratio,
)?;
previous = std::fs::metadata(&path).map(|m|
m.len()).unwrap_or(0);
}
let parent = path.parent().expect("cache shard");
std::fs::create_dir_all(parent)?;
- let temporary = parent.join(format!(
- ".{}.tmp.{}",
- path.file_name().unwrap().to_string_lossy(),
- uuid::Uuid::new_v4()
- ));
- let mut options = OpenOptions::new();
- options.write(true).create_new(true);
- #[cfg(unix)]
{
- use std::os::unix::fs::OpenOptionsExt;
- options.mode(0o600);
+ std::fs::rename(&staged.path, &path)?;
}
- let publish = (|| {
- let mut file = options.open(&temporary)?;
- file.write_all(&encoded)?;
- drop(file);
- std::fs::rename(&temporary, &path)
- })();
- if publish.is_err() {
- let _ = std::fs::remove_file(&temporary);
- }
- publish?;
finish_usage(
&root,
- used.saturating_sub(previous)
- .saturating_add(encoded.len() as u64),
+ used.saturating_sub(previous).saturating_add(staged.len),
)
})
.await;
@@ -422,6 +531,7 @@ fn finish_invalidation(root: &Path) -> io::Result<()> {
let _ = std::fs::remove_file(root.join(key.cache_relative_path()));
}
cleanup_temporary_files(root)?;
+ cleanup_staging(root)?;
finish_usage(root, existing_usage(root))
}
@@ -445,13 +555,14 @@ fn clean_usage(root: &Path, max_size: u64) -> Option<u64>
{
}
fn recover_usage(root: &Path, limit: u64, high_priority_ratio: f64) ->
io::Result<u64> {
- // Publication holds the directory lock, so every temporary block found
- // here belongs to an interrupted publisher. Remove it before clearing
dirty.
+ // Legacy temporaries are orphaned; staged files may still have live
writers.
cleanup_temporary_files(root)?;
+ cleanup_staging(root)?;
#[cfg(test)]
RECOVERY_SCANS.with(|count| count.set(count.get() + 1));
let mut state = CacheState::new(Some(limit), high_priority_ratio);
- for (key, encoded_size) in scan_existing_blocks(root) {
+ let blocks = scan_existing_blocks(root);
+ for (key, encoded_size) in blocks {
insert_state_entry(&mut state, key, encoded_size);
}
for key in collect_evictions(&mut state) {
@@ -539,6 +650,11 @@ mod tests {
let cache = SharedDiskCache::new(root.clone(), 4096).unwrap();
let token = cache.generation();
match std::env::var("PAIMON_SHARED_TEST_MODE").unwrap().as_str() {
+ "stage-hold" => {
+ let _staged = StagedBlock::create(&root, b"staged").unwrap();
+ std::fs::write(root.join("locked"), b"ready").unwrap();
+ std::future::pending::<()>().await;
+ }
"populate" => {
let _guard = cache.load_lock(&key()).await.unwrap();
if cache.get(&key(), &token).await.is_none() {
@@ -666,6 +782,100 @@ mod tests {
assert!(reopened.generation().is_none());
}
+ #[tokio::test]
+ async fn staged_file_survives_recovery_and_publishes_after_gate_release() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(),
4096).unwrap();
+ let generation = cache.generation();
+ let staged = cache
+ .stage_block(&key(), Bytes::from_static(b"value"), &generation)
+ .await
+ .unwrap();
+ let staged_path = staged.1.path.clone();
+ let gate =
open_lock(&directory.path().join("directory.lock")).unwrap();
+ gate.lock().unwrap();
+ assert_eq!(
+ recover_usage(directory.path(), 4096,
DEFAULT_HIGH_PRIORITY_POOL_RATIO).unwrap(),
+ 0
+ );
+ assert!(staged_path.exists());
+ drop(gate);
+ cache.publish_staged(staged, &generation).await;
+ assert!(!staged_path.exists());
+ assert_eq!(cache.get(&key(), &generation).await.unwrap(), "value");
+ assert_eq!(
+ std::fs::read_to_string(directory.path().join("usage"))
+ .unwrap()
+ .parse::<u64>()
+ .unwrap(),
+ existing_usage(directory.path())
+ );
+ }
+
+ #[test]
+ fn process_exit_releases_staging_for_cleanup() {
+ let directory = tempfile::tempdir().unwrap();
+ let mut process = child(directory.path(),
"stage-hold").spawn().unwrap();
+ let deadline = std::time::Instant::now() + Duration::from_secs(5);
+ while !directory.path().join("locked").exists() &&
std::time::Instant::now() < deadline {
+ std::thread::sleep(Duration::from_millis(5));
+ }
+ let ready = directory.path().join("locked").exists();
+ cleanup_staging(directory.path()).unwrap();
+ let live = std::fs::read_dir(directory.path().join("staging"))
+ .unwrap()
+ .count();
+ process.kill().unwrap();
+ process.wait().unwrap();
+ assert!(ready);
+ assert_eq!(live, 1);
+ SharedDiskCache::new(directory.path().to_path_buf(), 4096).unwrap();
+ assert_eq!(
+ std::fs::read_dir(directory.path().join("staging"))
+ .unwrap()
+ .count(),
+ 0
+ );
+ }
+
+ #[test]
+ fn dropping_unpublished_stage_removes_it() {
+ let directory = tempfile::tempdir().unwrap();
+ let staged = StagedBlock::create(directory.path(), b"value").unwrap();
+ let path = staged.path.clone();
+ assert!(path.exists());
+ drop(staged);
+ assert!(!path.exists());
+ }
+
+ #[tokio::test]
+ async fn encoded_publication_cannot_cross_invalidation() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = std::sync::Arc::new(
+ SharedDiskCache::new(directory.path().to_path_buf(), 4 * 1024 *
1024).unwrap(),
+ );
+ let old = cache.generation();
+ let mutation = cache.begin_invalidation().await.unwrap();
+ let writer = {
+ let cache = cache.clone();
+ tokio::spawn(async move {
+ cache
+ .put(&key(), Bytes::from(vec![42; 1024 * 1024]), &old)
+ .await;
+ })
+ };
+ // Reject the old generation regardless of when encoding finishes.
+ tokio::task::yield_now().await;
+ mutation.finish().await;
+ writer.await.unwrap();
+ let current = cache.generation();
+ assert!(cache.get(&key(), ¤t).await.is_none());
+ cache
+ .put(&key(), Bytes::from_static(b"current"), ¤t)
+ .await;
+ assert_eq!(cache.get(&key(), ¤t).await.unwrap(), "current");
+ }
+
#[tokio::test]
async fn cache_hits_can_share_the_directory_read_lock() {
let directory = tempfile::tempdir().unwrap();
@@ -703,6 +913,87 @@ mod tests {
assert!(guard.is_some());
}
+ #[tokio::test]
+ async fn hit_does_not_wait_for_directory_writer() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(),
4096).unwrap();
+ let token = cache.generation();
+ cache
+ .put(&key(), Bytes::from_static(b"value"), &token)
+ .await;
+ let gate =
open_lock(&directory.path().join("directory.lock")).unwrap();
+ gate.lock().unwrap();
+ assert_eq!(
+ tokio::time::timeout(Duration::from_millis(500), cache.get(&key(),
&token))
+ .await
+ .unwrap()
+ .unwrap(),
+ "value"
+ );
+ // A marker alone must fail closed, even before generation rotation.
+ std::fs::write(directory.path().join(MUTATION_MARKER), b"").unwrap();
+ assert!(cache.get(&key(), &token).await.is_none());
+ }
+
+ #[tokio::test]
+ async fn
opened_block_survives_invalidation_but_new_reads_reject_old_epoch() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(),
4096).unwrap();
+ let token = cache.generation();
+ cache.put(&key(), Bytes::from_static(b"old"), &token).await;
+ let encoded_key = generation_key(&key(), token.as_ref().unwrap());
+ let gate = read_lock(directory.path().join("directory.lock"))
+ .await
+ .unwrap();
+ let mut opened =
+
File::open(directory.path().join(encoded_key.cache_relative_path())).unwrap();
+ drop(gate);
+ cache.invalidate().await.unwrap();
+ assert!(cache.get(&key(), &token).await.is_none());
+ let current = cache.generation();
+ cache
+ .put(&key(), Bytes::from_static(b"new"), ¤t)
+ .await;
+ let mut bytes = Vec::new();
+ opened.read_to_end(&mut bytes).unwrap();
+ assert_eq!(decode_block_owned(&encoded_key, bytes).unwrap(), "old");
+ assert_eq!(cache.get(&key(), ¤t).await.unwrap(), "new");
+ }
+
+ #[tokio::test]
+ async fn hits_refresh_stale_recency_but_do_not_touch_recent_blocks() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(),
4096).unwrap();
+ let token = cache.generation();
+ cache
+ .put(&key(), Bytes::from_static(b"value"), &token)
+ .await;
+ let path = directory
+ .path()
+ .join(generation_key(&key(),
token.as_ref().unwrap()).cache_relative_path());
+ let recent = std::time::SystemTime::now() - Duration::from_secs(1);
+ OpenOptions::new()
+ .write(true)
+ .open(&path)
+ .unwrap()
+ .set_modified(recent)
+ .unwrap();
+ let before = std::fs::metadata(&path).unwrap().modified().unwrap();
+ assert!(cache.get(&key(), &token).await.is_some());
+ assert_eq!(
+ std::fs::metadata(&path).unwrap().modified().unwrap(),
+ before
+ );
+ OpenOptions::new()
+ .write(true)
+ .open(&path)
+ .unwrap()
+ .set_modified(recent - Duration::from_secs(60))
+ .unwrap();
+ assert!(cache.get(&key(), &token).await.is_some());
+ assert!(std::fs::metadata(&path).unwrap().modified().unwrap() >
before);
+ }
+
#[tokio::test]
async fn capacity_and_corruption() {
let directory = tempfile::tempdir().unwrap();
@@ -767,6 +1058,41 @@ mod tests {
assert!(used <= 512);
}
+ #[tokio::test]
+ async fn download_locks_allow_different_blocks_of_the_same_file() {
+ let directory = tempfile::tempdir().unwrap();
+ let cache = SharedDiskCache::new(directory.path().to_path_buf(), 1024
* 1024).unwrap();
+ let first = key();
+ let mut second = first.clone();
+ // Select a different stripe; collisions remain intentionally bounded.
+ loop {
+ second.block_index += 1;
+ if first.cache_relative_path().components().next()
+ != second.cache_relative_path().components().next()
+ {
+ break;
+ }
+ }
+ let guard = cache.load_lock(&first).await.unwrap();
+ let other = tokio::time::timeout(Duration::from_secs(1),
cache.load_lock(&second))
+ .await
+ .expect("another block must not wait for this file's download")
+ .unwrap();
+ assert!(
+ tokio::time::timeout(Duration::from_millis(50),
cache.load_lock(&first))
+ .await
+ .is_err()
+ );
+ drop(guard);
+ assert!(
+ tokio::time::timeout(Duration::from_secs(1),
cache.load_lock(&first))
+ .await
+ .unwrap()
+ .is_some()
+ );
+ drop(other);
+ }
+
#[tokio::test]
async fn interrupted_temporary_blocks_are_removed_before_clean_startup() {
for invalidate in [false, true] {
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index dcfa30e6..b6c8ff9b 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -436,6 +436,12 @@ do not reuse the affected path until invalidation succeeds.
A canceled or failed rename bypasses this cache; after confirming storage
state,
use a new cache directory to resume sharing.
+Shared range reads allow eight background fills per cache instance for blocks
+up to 1 MiB; larger blocks, whole-file reads, or saturated fills stay
synchronous.
+Concurrent whole-file and range reads may download overlapping data.
+Allow disk headroom beyond `local-cache.max-size` for up to eight staged writes
+per cache instance; this staging limit is not node-wide.
+
### BLOB Index Cache
Decoded `.blob` indexes are cached per catalog. Set
