[bug] Raft snapshot application holds the collections write lock across collection construction, blocking every collection on the node

Author: YZL0v3ZZCreated Sep 29, 2026Updated Sep 29, 2026

Labelsbugacceptedarea/storagearea/consensus

Current Behavior

When a peer installs a Raft snapshot, TableOfContent::apply_collections_snapshot acquires a single write guard on the node wide collections map and keeps it while every collection in the snapshot is constructed or reconciled. Constructing a collection builds each of its local replicas from scratch, which includes spawning segment build threads and joining them, creating the WAL, starting the update, optimizer, and flush workers, and writing an initial WAL record. Reconciling an existing collection may additionally create or remove shards and delete shard directories.

Every collection scoped request resolves its target through TableOfContent::get_collection, which takes the read side of the same map. While the write guard is held, no collection on the node can be looked up, so queries, upserts, scrolls, retrievals, collection snapshots, the public REST and gRPC APIs, and the internal peer to peer shard APIs all block until the application finishes. Because the lock is write preferring, a writer that queues during the application also fences every later reader.

The path is reachable through ordinary operations: a peer that rejoins after being offline, a peer that has fallen behind the leader's compacted Raft log, a peer that is added while the cluster is running, and a peer on which POST /cluster/recover is issued.

A PoC that applies a snapshot of 64 collections with two local shards each shows that reads time out for the whole 17.2 second application window and recover in tens of microseconds once the guard is released.

Environment

Reproduced on a checkout from the v1.17 to v1.18 line, using a pinned nightly build (rustc 1.94).
The same code path is present on the current dev branch at e857c5d9 (1.19.2-dev), from which the excerpts below are taken.
The trigger requires cluster mode. The PoC itself runs in a single process and calls the snapshot application function directly.

Steps to Reproduce

Place the PoC below at lib/storage/tests/raft_snapshot_apply_stall.rs.
Run the test. On the v1.17 to v1.18 line:
```
cargo +nightly-2026-03-21 test -p storage --test raft_snapshot_apply_stall -- --nocapture
```
On dev, use a nightly that satisfies rust-version 1.98 and apply the constructor change noted in the PoC header.
The test builds a TableOfContent with a consensus proposal sender, seeds one collection, constructs a CollectionsSnapshot that contains the seed plus 64 further collections with two local shards each, and applies the snapshot on a background thread. While the application runs, all_collections and get_collection are probed in a loop.

PoC: lib/storage/tests/raft_snapshot_apply_stall.rs

rust

//! PoC: applying a Raft collections snapshot holds the ToC `collections` write guard across
//! `Collection::new` and `Collection::apply_state`, so every collection lookup on the node
//! blocks until the whole snapshot has been applied.
//!
//! Integration location: lib/storage/tests/raft_snapshot_apply_stall.rs
//! Run: cargo +nightly-2026-03-21 test -p storage --test raft_snapshot_apply_stall -- --nocapture
//!
//! Version note: matches the validated v1.17 to v1.18 line (TableOfContent::new takes three
//! runtimes and returns Self). On dev replace the call with TableOfContent::new(&config,
//! ResourceBudget::default(), ChannelService::default(), 0,
//! Some(OperationSender::new(propose_sender))).unwrap(), and use a nightly that satisfies
//! rust-version 1.98.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};

use collection::shards::channel_service::ChannelService;
use common::budget::ResourceBudget;
use storage::content_manager::CollectionContainer;
use storage::content_manager::collection_meta_ops::{
    CollectionMetaOperations, CreateCollection, CreateCollectionOperation,
};
use storage::content_manager::consensus::operation_sender::OperationSender;
use storage::content_manager::consensus_manager::CollectionsSnapshot;
use storage::content_manager::toc::TableOfContent;
use storage::rbac::{Access, AccessRequirements};
use storage::types::StorageConfig;
use tempfile::Builder;
use tokio::runtime::Runtime;

/// Collections in the applied snapshot. Raise it if the apply ends too fast to observe.
const SNAPSHOT_COLLECTIONS: usize = 64;
const SHARDS_PER_COLLECTION: u32 = 2;
const PROBE_TIMEOUT: Duration = Duration::from_millis(50);

/// Built from JSON so field additions do not break the PoC across versions.
fn test_storage_config(storage_dir: &std::path::Path) -> StorageConfig {
    let json = serde_json::json!({
        "storage_path": storage_dir.to_string_lossy().into_owned(),
        "snapshots_path": storage_dir.join("snapshots").to_string_lossy().into_owned(),
        "temp_path": null,
        "on_disk_payload": false,
        "optimizers": {
            "deleted_threshold": 0.5,
            "vacuum_min_vector_number": 100,
            "default_segment_number": 2,
            "max_segment_size": null,
            "memmap_threshold": 100,
            "indexing_threshold": 100,
            "flush_interval_sec": 2,
            "max_optimization_threads": 2,
        },
        "wal": { "wal_capacity_mb": 32, "wal_segments_ahead": 0 },
        "performance": {
            "max_search_threads": 1,
            "max_optimization_runtime_threads": 2,
            "optimizer_cpu_budget": 0,
            "optimizer_io_budget": 0,
            "incoming_shard_transfers_limit": 1,
            "outgoing_shard_transfers_limit": 1,
        },
        "hnsw_index": { "m": 16, "ef_construct": 100, "full_scan_threshold": 10000 },
        "update_concurrency": 2,
    });
    serde_json::from_value(json).expect("valid storage config")
}

#[test]
fn raft_snapshot_apply_holds_collections_write_guard() {
    let storage_dir = Builder::new().prefix("storage").tempdir().unwrap();
    let config = test_storage_config(storage_dir.path());

    let (propose_sender, _propose_receiver) = std::sync::mpsc::channel();
    let toc = Arc::new(TableOfContent::new(
        &config,
        Runtime::new().unwrap(),
        Runtime::new().unwrap(),
        Runtime::new().unwrap(),
        ResourceBudget::default(),
        ChannelService::default(),
        0,
        Some(OperationSender::new(propose_sender)),
    ));

    // Seed one collection to derive a valid state for the collections in the snapshot.
    let create_collection: CreateCollection = serde_json::from_value(serde_json::json!({
        "vectors": { "size": 4, "distance": "Dot" },
        "shard_number": SHARDS_PER_COLLECTION,
        "hnsw_config": null,
        "wal_config": null,
        "optimizers_config": null,
        "sparse_vectors": null,
        "strict_mode_config": null,
    }))
    .expect("valid create collection request");
    CollectionContainer::perform_collection_meta_op(
        toc.as_ref(),
        CollectionMetaOperations::CreateCollection(
            CreateCollectionOperation::new("seed".to_string(), create_collection).unwrap(),
        ),
    )
    .unwrap();

    let seed_state = CollectionContainer::collections_snapshot(toc.as_ref())
        .collections
        .get("seed")
        .cloned()
        .expect("seed collection state must exist");

    // The snapshot contains the seed plus collections this node has never seen.
    let mut collections = HashMap::with_capacity(SNAPSHOT_COLLECTIONS + 1);
    collections.insert("seed".to_string(), seed_state.clone());
    for i in 0..SNAPSHOT_COLLECTIONS {
        collections.insert(format!("snapshot-{i}"), seed_state.clone());
    }
    let snapshot = CollectionsSnapshot {
        collections,
        aliases: Default::default(),
    };

    let probe_rt = Runtime::new().unwrap();
    let access = Access::full("snapshot apply stall PoC");
    let pass = access
        .check_collection_access("seed", AccessRequirements::new())
        .unwrap();

    let baseline_started = Instant::now();
    let baseline_all = probe_rt.block_on(toc.all_collections(&access));
    let baseline_get = probe_rt.block_on(toc.get_collection(&pass));
    let baseline_elapsed = baseline_started.elapsed();

    // Apply the snapshot on another thread and probe reads while it holds the write guard.
    let apply_toc = Arc::clone(&toc);
    let apply_started = Instant::now();
    let apply = std::thread::spawn(move || {
        CollectionContainer::apply_collections_snapshot(apply_toc.as_ref(), snapshot)
    });

    // A read issued while the apply runs cannot complete; both probes must time out.
    let mut blocked_all = 0usize;
    let mut blocked_get = 0usize;
    let mut longest_probe = Duration::ZERO;
    let deadline = Instant::now() + Duration::from_secs(600);
    while !apply.is_finished() && Instant::now() < deadline {
        let timed_out = probe_rt.block_on(async {
            tokio::time::timeout(PROBE_TIMEOUT, toc.all_collections(&access))
                .await
                .is_err()
        });
        blocked_all += usize::from(timed_out);

        let started = Instant::now();
        let timed_out = probe_rt.block_on(async {
            tokio::time::timeout(PROBE_TIMEOUT, toc.get_collection(&pass))
                .await
                .is_err()
        });
        longest_probe = longest_probe.max(started.elapsed());
        blocked_get += usize::from(timed_out);
    }

    let apply_elapsed = apply_started.elapsed();
    assert!(apply.is_finished(), "snapshot apply did not finish in time");
    apply.join().unwrap().unwrap();

    let final_started = Instant::now();
    let collections_after = probe_rt.block_on(toc.all_collections(&access));
    let final_elapsed = final_started.elapsed();

    println!(
        "baseline read:        {baseline_elapsed:?} ({} collections, get_collection: {})",
        baseline_all.len(),
        baseline_get.is_ok(),
    );
    println!("snapshot apply:       {apply_elapsed:?}");
    println!(
        "blocked probes:       all_collections={blocked_all}, get_collection={blocked_get} \
         (longest probe {longest_probe:?})",
    );
    println!(
        "read after apply:     {final_elapsed:?} ({} collections)",
        collections_after.len(),
    );

    assert!(
        blocked_all > 0 && blocked_get > 0,
        "reads must be blocked while the snapshot apply holds the collections write guard \
         (all_collections={blocked_all}, get_collection={blocked_get}); \
         raise SNAPSHOT_COLLECTIONS if the apply was too fast",
    );
    assert!(
        apply_elapsed >= Duration::from_millis(200),
        "snapshot apply finished too fast to observe; raise SNAPSHOT_COLLECTIONS",
    );
    assert_eq!(collections_after.len(), SNAPSHOT_COLLECTIONS + 1);
}

Observed output

$ cargo +nightly-2026-03-21 test -p storage --test raft_snapshot_apply_stall -- --nocapture
    Finished `test` profile [unoptimized + debuginfo] target(s) in 0.89s
     Running tests/raft_snapshot_apply_stall.rs (target/debug/deps/raft_snapshot_apply_stall-ab35ee18fa071ef4)

running 1 test
baseline read:        90.19µs (1 collections, get_collection: true)
snapshot apply:       17.195232385s
blocked probes:       all_collections=167, get_collection=168 (longest probe 51.955446ms)
read after apply:     67.43µs (65 collections)
test raft_snapshot_apply_holds_collections_write_guard ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 18.75s

Every read probe issued while the snapshot was being applied exceeded the 50 millisecond probe timeout: 167 all_collections probes and 168 get_collection probes, meaning that no read completed during the 17.2 second application window. Immediately after the application returned, a read completed in 67 microseconds and listed 65 collections, while the baseline read before the application took 90 microseconds.

The measured 17.2 seconds cover 64 collections with two empty shards each. In a real recovery the snapshot contains every collection of the cluster, and each local replica it assigns to the peer is built on the same code path, so the window scales with the number of collections, the number of local shards, and the amount of shard data that has to be removed or rebuilt.

Root Cause

The application holds one write guard for the entire reconciliation loop and performs all construction, destruction, and reconciliation work while it is held. Line ranges refer to the dev branch at commit e857c5d9; the same code is present in v1.17 and v1.18 with minor line shifts.

1. The Raft snapshot is applied inline on the consensus thread

rust

// src/consensus.rs:1074-1082
if !ready.snapshot().is_empty() {
    // This is a snapshot, we need to apply the snapshot at first.
    let snapshot = ready.snapshot().clone();
    log::debug!("Applying snapshot {:?}", snapshot.get_metadata());
    is_idle = false;

    if let Err(err) = store.apply_snapshot(&snapshot)? {
        log::error!("Failed to apply snapshot: {err}");
    }
}

2. The call is forwarded to the collection registry

rust

// lib/storage/src/content_manager/consensus_manager.rs:711-729 (abridged)
pub fn apply_snapshot(
    &self,
    snapshot: &raft::eraftpb::Snapshot,
) -> Result<Result<(), StorageError>, StorageError> {
    ...
    let SnapshotData {
        collections_data,
        ..
    } = snapshot.get_data().try_into()?;

    self.toc.apply_collections_snapshot(collections_data)?;
    ...
}

3. A single write guard is taken for the whole loop

rust

// lib/storage/src/content_manager/toc/collection_container.rs:176-178
fn apply_collections_snapshot(
    &self,
    data: consensus_manager::CollectionsSnapshot,
) -> Result<(), StorageError> {
    self.general_runtime.block_on(async {
        let mut existing_collections = self.collections.write().await;

4. New collections are constructed while the guard is held

rust

// lib/storage/src/content_manager/toc/collection_container.rs:219-256 (abridged)
if !collection_exists {
    ...
    let collection = Collection::new(
        ...
    )
    .await?;
    existing_collections.validate_collection_not_exists(id)?;
    existing_collections.insert(id.clone(), Arc::new(collection));
}

5. Collection construction builds every local replica

rust

// lib/collection/src/collection/mod.rs:136-173 (abridged)
for (shard_id, mut peers) in shard_distribution.shards {
    let is_local = peers.remove(&this_peer_id);
    ...
    let replica_set = ShardReplicaSet::build(
        ...
        is_local,
        ...
    )
    .await?;

    shard_holder
        .add_shard(shard_id, replica_set, shard_key)
        .await?;
}

6. Building a local shard performs blocking and long running work

rust

// lib/collection/src/shards/local_shard/mod.rs:687-748 (abridged)
let segment = thread::Builder::new()
    .name(format!("shard-build-{collection_id}-{id}"))
    .spawn(move || {
        build_segment(&path_clone, &segment_config, deferred_internal_id, true)
    })
    .unwrap();
...
let join_results = build_handlers
    .into_iter()
    .map(|handler| handler.join())
    .collect_vec();
...
let wal: SerdeWal<OperationWithClockTag> =
    SerdeWal::new(&wal_path, (&config.wal_config).into())?;
...
local_shard.insert_fake_operation().await?;

7. Existing collections are reconciled under the same guard

rust

// lib/storage/src/content_manager/toc/collection_container.rs:263-285 (abridged)
// Update collection state
if &existing_collection.state().await != state {
    ...
    existing_collection
        .apply_state(state.clone(), self.this_peer_id(), abort_transfer)
        .await?;
    ...
}

8. The guard is released only after the whole loop

rust

// lib/storage/src/content_manager/toc/collection_container.rs:306-309
let collection_names: Vec<_> = existing_collections.keys().cloned().collect();

// Drop `collections` lock
drop(existing_collections);

9. Every lookup needs the read side of the same lock

rust

// lib/storage/src/content_manager/toc/mod.rs:447-465 (abridged)
async fn get_collection_unchecked(
    &self,
    collection_name: &str,
) -> Result<Arc<Collection>, StorageError> {
    let read_collection = self.collections.read().await;
    ...
}

pub async fn get_collection(
    &self,
    collection: &CollectionPass<'_>,
) -> Result<Arc<Collection>, StorageError> {
    self.get_collection_unchecked(collection.name()).await
}

10. The same function already drops the guard for another heavy step

rust

// lib/storage/src/content_manager/toc/collection_container.rs:204-212 (abridged)
if recreate_collection {
    // Drop `collections` lock
    drop(existing_collections);

    // Delete collection
    self.delete_collection(id).await?;

    // Re-acquire `collections` lock
    existing_collections = self.collections.write().await;
}

11. The create and delete paths show the intended shape

rust

// lib/storage/src/content_manager/toc/create_collection.rs:213-249 (abridged)
let collection = Collection::new(...).await?;
...
{
    let mut write_collections = self.collections.write().await;
    write_collections.validate_collection_not_exists(collection_name)?;
    let existing_collection =
        write_collections.insert(collection_name.to_string(), Arc::new(collection));
    ...
}

rust

// lib/storage/src/content_manager/toc/collection_meta_ops.rs:248
let removed_opt = self.collections.write().await.remove(collection_name);

Expected Behavior

Installing a Raft snapshot should reconcile the local collection registry without making every collection on the node unavailable. Building or destroying a collection does not require the node wide registry lock, as the create and delete paths already demonstrate, and unrelated collections should remain serviceable while a peer catches up.

Impact

Availability only. While the snapshot is applied, the peer cannot serve any collection scoped request, including reads and writes. In deployments where a shard has a single replica, the affected peer is the only replica for those collections, so the outcome extends beyond the recovering peer itself. The window scales with the number of collections and local shards in the snapshot and with the amount of shard data that has to be removed or rebuilt. Nothing on the server side bounds the hold; request level timeouts only turn waiting requests into failures.

Possible Solution

Build and destroy collections on owned handles outside the lock, and reconcile the registry under a short exclusive section, matching the pattern already used by the create and delete paths. In outline:

Determine which collections must be created, recreated, or removed, and which existing collections need a state reconciliation.
Perform all construction, deletion, and per collection state application outside the registry lock.
Acquire the write guard only to insert or replace the newly built entries and to remove the entries that are gone.
Route per collection state application through the collection handle instead of through the registry guard.

The snapshot application is replayed from the Raft snapshot after a restart, so an interrupted reconciliation converges on the next attempt. Each step should therefore remain idempotent.

Source: qdrant/qdrant

View original on GitHub View discussion on GitHub