[bug] Indexing split cache eviction blocks merge fetches while deleting splits #7004

Describe the bug

IndexingSplitCache is one cache per indexing node, shared by every pipeline, and it is guarded by a single Mutex<InnerSplitCache>. Inserting a split holds that mutex while the cache frees room by deleting cached split directories one at a time, then renames the new directory into place. The number of deletions is not constant: it is the byte deficit divided by the average size of a cached split, capped by split_store_max_num_splits, which defaults to 1000. A large incoming split, such as a merge output, that enters a cache filled with smaller splits therefore deletes hundreds of directories inside one critical section.

Merge fetches take the same mutex. IndexingSplitStore::fetch_and_open_split calls get_cached_split, which locks the cache before it can report a hit or a miss, so every merge on the node waits for the deletion loop to finish. The eviction side issues one remove_dir_all per split, and the fetch side performs a blocking std::fs::rename under the same async mutex, so both directions put file system work inside the critical section.

Root cause

The excerpts follow one insertion from the node-level construction of the cache to the merge fetch that waits on it. Line numbers refer to commit 7b256dd3.

1. One cache per node, constructed when the node merges its own splits.

// quickwit-serve/src/lib.rs:332-349
async fn indexing_split_cache_for_config(
    node_config: &NodeConfig,
) -> anyhow::Result<Arc<IndexingSplitCache>> {
    let runs_indexer = node_config.is_service_enabled(QuickwitService::Indexer);
    let runs_local_compactor = node_config.is_service_enabled(QuickwitService::Compactor);
    let merges_own_splits =
        runs_indexer && (!node_config.enable_standalone_compactors || runs_local_compactor);
    if merges_own_splits {
        let cache = IndexingSplitCache::from_config(
            &node_config.indexer_config,
            &node_config.data_dir_path,
        )
        .await?;
        Ok(Arc::new(cache))
    } else {
        Ok(Arc::new(IndexingSplitCache::no_caching()))
    }
}

2. The cache has a single mutex.

// quickwit-indexing/src/split_store/indexing_split_cache.rs:171-173
pub struct IndexingSplitCache {
    inner: Mutex<InnerSplitCache>,
}

There is no sharding by index or by pipeline.

3. The insertion holds the mutex across the walk, the evictions and the rename.

// quickwit-indexing/src/split_store/indexing_split_cache.rs:350-364
    async fn move_into_cache(&mut self, split_id: SplitId, split_path: &Path) -> io::Result<bool> {
        let split_folder = SplitFolder::create(split_id, split_path).await?;
        let split_id = split_folder.split_id.clone();
        let should_move_split = self.make_room_and_record_split(split_folder).await?;
        if !should_move_split {
            return Ok(false);
        }
        let to_full_path = self.split_path(&split_id);
        if let Err(io_err) = tokio::fs::rename(split_path, &to_full_path).await {
            // keep the registry stats accurate
            self.split_registry.remove(&split_id);
            return Err(io_err);
        }
        Ok(true)
    }

The public entry point takes the guard first (quickwit-indexing/src/split_store/indexing_split_cache.rs:549-556).

4. Room is freed by a loop that deletes one split directory per iteration.

// quickwit-indexing/src/split_store/indexing_split_cache.rs:380-399
    async fn make_room_and_record_split(&mut self, split_folder: SplitFolder) -> io::Result<bool> {
        // We don't accept splits that are too large.
        if split_folder.num_bytes > self.split_registry.quota().max_num_bytes() {
            return Ok(false);
        }

        while !self
            .split_registry
            .quota()
            .can_fit_split(split_folder.num_bytes)
        {
            self.evict_one_split().await?;
        }

        if let Some(creation_time_limit) = split_folder.created_at.checked_sub(SPLIT_MAX_AGE) {
            self.remove_splits_older_than_limit(creation_time_limit)
                .await?;
        };
        Ok(self.split_registry.insert(split_folder))
    }

// quickwit-indexing/src/split_store/indexing_split_cache.rs:324-341
    async fn evict_one_split(&mut self) -> io::Result<()> {
        let evicted_split: SplitFolder = self
            .split_registry
            .pop_oldest()
            .expect("split cache should not be empty");
        let result = tokio::fs::remove_dir_all(&self.split_path(&evicted_split.split_id)).await;
        if let Err(io_err) = result {
            if io_err.kind() == io::ErrorKind::NotFound {
                // This could happen if some files have been manually deleted
                // from the FS for instance.
                warn!(split_id=%evicted_split.split_id, "cached split missing from local split directory");
                return Ok(());
            } else {
                return Err(io_err);
            }
        }
        Ok(())
    }

Each iteration unlinks a complete split directory, so the loop cost is the number of deletions times the cost of removing one split.

5. The quota decides how many iterations run.

// quickwit-indexing/src/split_store/split_store_quota.rs:59-69
    pub fn can_fit_split(&self, split_size_in_bytes: ByteSize) -> bool {
        if self.num_splits_in_cache >= self.max_num_splits {
            return false;
        }
        if self.size_in_bytes_in_cache.as_u64() + split_size_in_bytes.as_u64()
            > self.max_num_bytes.as_u64()
        {
            return false;
        }
        true
    }

The defaults are 100 GiB and 1000 splits (quickwit-config/src/node_config/mod.rs:240-246). When the byte limit binds first, the loop count is the byte deficit divided by the average cached split size, and the incoming split may evict up to the whole cache.

6. Every immature split takes this path.

// quickwit-indexing/src/split_store/indexing_split_store.rs:148-159
        if !is_mature {
            debug!("store-in-cache");
            if self
                .inner
                .split_cache
                .move_into_cache(split.split_id().clone(), split_folder_path)
                .await?
            {
                return Ok(());
            }
        }
        tokio::fs::remove_dir_all(split_folder_path).await?;

The merge policy keeps splits immature for 48 hours by default (quickwit-config/src/merge_policy_config.rs:107-109), and merge operation outputs carry that maturity (quickwit-indexing/src/merge_policy/stable_log_merge_policy.rs:120), so a large merge output passes through this insertion path instead of bypassing the cache.

7. The merge side waits on the same mutex.

// quickwit-indexing/src/split_store/indexing_split_cache.rs:527-539
    pub(super) async fn get_cached_split(
        &self,
        split_id: &SplitId,
        output_dir_path: &Path,
    ) -> StorageResult<Option<PathBuf>> {
        let mut split_store_lock = self.inner.lock().await;
        let split_file_opt: Option<PathBuf> =
            split_store_lock.move_out(split_id, output_dir_path).await?;
        if split_file_opt.is_none() {
            debug!(split_id=%split_id, "split folder not in cache");
        }
        Ok(split_file_opt)
    }

fetch_and_open_split calls it first (quickwit-indexing/src/split_store/indexing_split_store.rs:187-192), so a merge cannot even learn whether the split is cached while an eviction loop is running. move_out also performs a blocking std::fs::rename under the same async mutex (quickwit-indexing/src/split_store/indexing_split_cache.rs:284).

Impact

A single insertion can delete up to 1000 split directories while the node-wide cache mutex is held, and every merge fetch on that node waits for the whole deletion sequence. The stall grows exactly when the cache is under pressure, because the deficit that triggers the deletions is the same pressure that keeps merges busy, and it grows with the size of the incoming split relative to the cached ones.

The path has no deadline. Each deletion is a local file system operation, so a slow or hung data directory extends the hold without bound, and the same mutex then blocks every merge on the node. The measured rate below is about 0.31 ms per small split directory, which places a full 1000 directory eviction near 0.3 s on the same storage and in the seconds range once split directories carry real tantivy files.

Steps to reproduce

The harness below builds a real IndexingSplitCache and a real IndexingSplitStore over a temporary directory, fills the cache to a scaled quota of 1000 splits and 8 MiB, then inserts one split that is large relative to the cached ones. It records the same insertion into an empty cache as a control, and it probes a merge fetch while the eviction loop is running. The probe starts 20 ms into the insertion and the assertions compare its wait with the remaining hold, so the test does not depend on the speed of the machine. Place the file at quickwit-indexing/tests/split_cache_eviction_blocks_merges.rs and run cargo test -p quickwit-indexing --test split_cache_eviction_blocks_merges -- --nocapture from the Cargo workspace root.

use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant};

use bytesize::ByteSize;
use quickwit_common::io::IoControls;
use quickwit_indexing::{IndexingSplitCache, IndexingSplitStore, SplitStoreQuota};
use quickwit_metastore::{SplitMetadata, SplitMaturity};
use quickwit_proto::types::SplitId;
use quickwit_storage::RamStorage;
use tempfile::tempdir;
use time::OffsetDateTime;
use tokio::time::sleep;

const QUOTA_SPLITS: usize = 1_000;
const QUOTA_BYTES: u64 = 8 * 1024 * 1024;
const CACHED_FILES_PER_SPLIT: usize = 4;
const CACHED_BYTES_PER_FILE: usize = 2 * 1024;
const INCOMING_SPLIT_BYTES: usize = 4 * 1024 * 1024;
const PROBE_DELAY: Duration = Duration::from_millis(20);

fn immature_split(split_id: &SplitId) -> SplitMetadata {
    SplitMetadata {
        split_id: split_id.clone(),
        create_timestamp: OffsetDateTime::now_utc().unix_timestamp(),
        maturity: SplitMaturity::Immature {
            maturation_period: Duration::from_hours(48),
        },
        ..Default::default()
    }
}

async fn build_store(work_dir: &Path) -> IndexingSplitStore {
    let cache = IndexingSplitCache::open(
        work_dir.join("splits"),
        SplitStoreQuota::try_new(QUOTA_SPLITS, ByteSize::b(QUOTA_BYTES)).unwrap(),
    )
    .await
    .unwrap();
    IndexingSplitStore::new(Arc::new(RamStorage::default()), Arc::new(cache))
}

async fn insert_split(
    store: &IndexingSplitStore,
    root: &Path,
    split_id: &str,
    file_count: usize,
    bytes_per_file: usize,
) {
    let split_id = SplitId::from(split_id);
    let folder: PathBuf = root.join(format!("{split_id}.split"));
    tokio::fs::create_dir_all(&folder).await.unwrap();
    for file_index in 0..file_count {
        let path = folder.join(format!("f{file_index}"));
        tokio::fs::write(&path, vec![0u8; bytes_per_file]).await.unwrap();
    }
    store
        .store_split(
            &immature_split(&split_id),
            &folder,
            Box::new(vec![0u8; 16]),
        )
        .await
        .unwrap();
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn cache_eviction_parks_merge_fetches() {
    let work_dir = tempdir().unwrap();
    let split_root = work_dir.path().join("incoming");
    tokio::fs::create_dir_all(&split_root).await.unwrap();
    let fetch_output = work_dir.path().join("merge");
    tokio::fs::create_dir_all(&fetch_output).await.unwrap();

    // Control: the same insertion into an empty cache, where no eviction is needed.
    let control_store = build_store(&work_dir.path().join("control")).await;
    let control_started = Instant::now();
    insert_split(
        &control_store,
        &split_root,
        "01-control-split",
        1,
        INCOMING_SPLIT_BYTES,
    )
    .await;
    let control_elapsed = control_started.elapsed();
    println!("[control] insert into an empty cache: {control_elapsed:?}");

    // Control: a merge fetch on an idle cache, with no insertion in flight.
    let idle_started = Instant::now();
    let _ = control_store
        .fetch_and_open_split(
            SplitId::from("01-idle-split"),
            &fetch_output,
            &IoControls::default(),
        )
        .await;
    println!("[control] merge fetch on an idle cache: {:?}", idle_started.elapsed());

    // Fill the cache up to the byte quota with small cached splits.
    let store = build_store(&work_dir.path().join("trigger")).await;
    for split_index in 0..QUOTA_SPLITS {
        let split_id = format!("01-cached-{split_index:04}");
        insert_split(
            &store,
            &split_root,
            &split_id,
            CACHED_FILES_PER_SPLIT,
            CACHED_BYTES_PER_FILE,
        )
        .await;
    }
    println!("[trigger] cache filled with {QUOTA_SPLITS} splits");

    // Trigger: one insertion of an incoming split that is much larger than the cached ones.
    // The quota can only be met by deleting many cached folders, one at a time.
    let insert = {
        let store = store.clone();
        let split_root = split_root.clone();
        tokio::spawn(async move {
            let started = Instant::now();
            insert_split(
                &store,
                &split_root,
                "01-incoming-split",
                1,
                INCOMING_SPLIT_BYTES,
            )
            .await;
            started.elapsed()
        })
    };
    sleep(PROBE_DELAY).await;
    assert!(
        !insert.is_finished(),
        "the insertion finished before the probe: raise INCOMING_SPLIT_BYTES or the cache fill"
    );

    // The merge path asks for an unrelated split while the eviction loop owns the mutex.
    let probe_started = Instant::now();
    let _ = store
        .fetch_and_open_split(
            SplitId::from("01-merge-split"),
            &fetch_output,
            &IoControls::default(),
        )
        .await;
    let probe_elapsed = probe_started.elapsed();
    let insert_elapsed = insert.await.unwrap();
    let remaining = insert_elapsed.saturating_sub(PROBE_DELAY);
    println!("[trigger] insert with evictions: {insert_elapsed:?}");
    println!("[trigger] merge fetch waited: {probe_elapsed:?} (remaining lock hold {remaining:?})");

    assert!(
        insert_elapsed >= PROBE_DELAY + Duration::from_millis(20),
        "the insertion was too short to measure the convoy: {insert_elapsed:?}"
    );
    assert!(
        probe_elapsed >= remaining / 2,
        "the merge fetch should have waited for the eviction loop, took {probe_elapsed:?}"
    );
    assert!(
        control_elapsed * 2 < insert_elapsed,
        "the same insertion should be cheaper without evictions: control {control_elapsed:?}, trigger {insert_elapsed:?}"
    );
}

[control] insert into an empty cache: 8.046929ms
[control] merge fetch on an idle cache: 168.927碌s
[trigger] cache filled with 1000 splits
[trigger] insert with evictions: 157.683456ms
[trigger] merge fetch waited: 137.37866ms (remaining lock hold 137.683456ms)

Expected behavior

An insertion into the local split cache does not delay merge fetches on the same node, and the cost of making room is not paid inside the cache mutex.

Actual behavior

An insertion deletes evicted split directories one by one while holding the node-wide cache mutex, and every merge fetch waits for the entire deletion sequence.

Workaround

Raising split_store_max_num_bytes or split_store_max_num_splits postpones the eviction but does not remove it. Enabling standalone compactors takes the cache out of the indexer path, because indexing_split_cache_for_config then returns IndexingSplitCache::no_caching() on nodes that only index, and the merge work moves to the compactor deployment with its own cache.

Proposed fix

Detach the deletion from the critical section: pop the evicted entries and charge the quota under the lock, drop the guard, delete the directories, then re-acquire the guard to finish the accounting. The deletions can also run in spawn_blocking so the async mutex is not held across blocking unlink work, and move_out should move its std::fs::rename out of the async critical section for the same reason.

Environment

Quickwit 0.9.1, main at commit 7b256dd3, Linux x86_64, built and tested with the workspace default toolchain. The reproduction uses a temporary directory and RAM storage for the remote side; no running node and no object store are required.