[bug] Exhausting the replica set clock pool turns new updates into an unbounded busy wait

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

Labelsbugarea/collectionparked

Current Behavior

Every update on a replica set leases one of 64 clocks for the whole operation, and an update that cannot lease one polls for a free clock with yield_now instead of waiting for a notification. The lease is not a short tick allocation: it is held from before the operation is written to the WAL until the operation has been applied locally and acknowledged by every updatable remote replica, including any wait the update performs on the way.

Once all 64 clocks are leased, the next update cannot make progress. ShardReplicaSet::get_clock polls the clock set, receives None, yields to the scheduler and polls again. There is no notification when a clock is released, no queue that bounds an individual wait, and no timeout inside the loop; the only bound is the optional timeout the caller passed to the update. A request without a timeout therefore never returns while the pool is exhausted, and it keeps a runtime worker busy polling for the whole duration. A request with a timeout fails with Failed to acquire clock for update operation within ..., and its operation never reaches the WAL.

The pool is exhausted by ordinary conditions. The fan-out to a slow, partitioned or unreachable remote replica waits up to the per-request gRPC deadline, 60 seconds by default, and that wait consumes a clock. A wait=true write consumes a clock until its local apply completes, and under prevent_unoptimized it consumes one for the deferred-points wait as well. Because each wait consumes pool capacity, the situations that make updates slow are exactly the situations that prevent new updates from starting.

The runtime reproduction below leases all 64 clocks with updates parked on a fake remote replica, observes the acquisition failure on the next update, and then observes the same call completing once the clocks are released.

Environment

dev branch at commit e857c5d9 (1.19.2-dev), where the runtime reproduction was executed.
The same shape is present on the v1.18.0 and v1.17.1 lines (lib/collection/src/shards/replica_set/clock_set.rs:10, lib/collection/src/shards/replica_set/update.rs:233 and :238 on v1.18.0, and :225 and :230 on v1.17.1).
Tokio 1.53.1 as pinned by the workspace Cargo.lock.
The clock cap is 64 per replica set and is not configurable in the server path.

Steps to Reproduce

Place the PoC below at lib/storage/tests/clock_pool_exhaustion_stall.rs.

Run:

cargo +nightly-2026-03-21 test -p storage --test clock_pool_exhaustion_stall -- --nocapture

The test builds a TableOfContent whose channel service knows one fake peer, registered at a local TCP listener that accepts connections and never answers. It seeds a collection with one local shard, marks the local replica Active, and adds the fake peer as an Active remote replica, so every update fans out to the fake peer and waits for a reply that never comes.
It starts 64 concurrent updates with a long timeout. Each update leases one clock, applies locally, and then parks on the fake remote, so all 64 clocks are leased.
It probes the pool with one more update that has a one second timeout, repeating the probe until it observes the acquisition failure, and it asserts that the failure is the clock acquisition timeout, which proves that the probe never reached the WAL or the remote.
It aborts the 64 updates, which releases their clocks, and repeats the probe: the same call now completes, which proves that the stall came from the clock pool and not from another dependency.

PoC: lib/storage/tests/clock_pool_exhaustion_stall.rs

rust

//! PoC: a replica set leases one of 64 clocks for each in-flight update. Once all 64 are leased,
//! further updates acquire a clock by polling with `yield_now`, with no notification and no
//! timeout, so a slow remote replica stalls the shard write path and keeps runtime tasks busy.
//!
//! Integration location: lib/storage/tests/clock_pool_exhaustion_stall.rs
//! Run: cargo +nightly-2026-03-21 test -p storage --test clock_pool_exhaustion_stall -- --nocapture
//! Target: dev (1.19.2-dev).

use std::net::{SocketAddr, TcpListener};
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant};

use collection::operations::CollectionUpdateOperations;
use collection::operations::point_ops::{PointOperations, WriteOrdering};
use collection::operations::types::CollectionError;
use collection::shards::channel_service::ChannelService;
use collection::shards::replica_set::replica_set_state::ReplicaState;
use collection::shards::shard_trait::WaitUntil;
use common::budget::ResourceBudget;
use common::counter::hardware_accumulator::HwMeasurementAcc;
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::toc::TableOfContent;
use storage::rbac::{Access, AccessRequirements};
use storage::types::StorageConfig;
use tempfile::Builder;
use tokio::runtime::Runtime;

/// Peer ID of the replica set member whose address never answers.
const FAKE_PEER_ID: u64 = 2;
/// In-flight updates needed to lease every clock of the replica set.
const IN_FLIGHT_UPDATES: usize = 64;
/// How long each in-flight update waits for the fake remote before giving up.
const IN_FLIGHT_TIMEOUT: Duration = Duration::from_secs(30);

/// TCP listener that accepts connections and never replies, so gRPC calls to it hang until their
/// deadline.
fn hanging_listener() -> SocketAddr {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();

    std::thread::spawn(move || {
        // Keep every accepted connection open and silent.
        let mut held = Vec::new();
        for stream in listener.incoming() {
            match stream {
                Ok(stream) => held.push(stream),
                Err(_) => break,
            }
        }
    });

    addr
}

/// Built from JSON so field additions do not break the PoC across versions.
fn test_storage_config(storage_dir: &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")
}

/// Smallest valid update that reaches the WAL and the remote replication fan-out.
fn probe_operation() -> CollectionUpdateOperations {
    CollectionUpdateOperations::PointOperation(PointOperations::DeletePoints { ids: vec![1.into()] })
}

/// Whether an error is the clock acquisition timeout from `ShardReplicaSet::get_clock`.
fn is_clock_timeout(err: &CollectionError) -> bool {
    matches!(
        err,
        CollectionError::Timeout { description } if description.contains("Failed to acquire clock"),
    )
}

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

    // Register the fake peer before building the table of contents, so the collection sees it.
    let fake_addr = hanging_listener();
    let channel_service = ChannelService::default();
    channel_service.id_to_address.write().insert(
        FAKE_PEER_ID,
        format!("http://{fake_addr}").parse().unwrap(),
    );

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

    // Seed one collection with a single local shard.
    let create_collection: CreateCollection = serde_json::from_value(serde_json::json!({
        "vectors": { "size": 4, "distance": "Dot" },
        "shard_number": 1,
        "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 rt = Runtime::new().unwrap();
    let access = Access::full("clock pool exhaustion PoC");
    let pass = access
        .check_collection_access("seed", AccessRequirements::new())
        .unwrap();
    let collection = rt.block_on(toc.get_collection(&pass)).unwrap();

    let replica_set = rt
        .block_on(async { collection.shards_holder().read().await.get_shard(0).cloned() })
        .unwrap();

    // Make the local replica active and add the fake peer as an active replica. Every update now
    // fans out to the fake peer and waits for a reply that never comes.
    rt.block_on(replica_set.set_replica_state(0, ReplicaState::Active))
        .unwrap();
    rt.block_on(replica_set.add_remote(FAKE_PEER_ID, ReplicaState::Active))
        .unwrap();

    // Lease every clock: each update holds its clock until the fake remote answers or its own
    // timeout fires.
    let mut in_flight = Vec::with_capacity(IN_FLIGHT_UPDATES);
    for _ in 0..IN_FLIGHT_UPDATES {
        let replica_set = replica_set.clone();
        in_flight.push(rt.spawn(async move {
            replica_set
                .update_with_consistency(
                    probe_operation(),
                    WaitUntil::Wal,
                    Some(IN_FLIGHT_TIMEOUT),
                    WriteOrdering::Weak,
                    false,
                    HwMeasurementAcc::new(),
                )
                .await
        }));
    }

    // Give the in-flight updates time to acquire a clock and park on the fake remote.
    rt.block_on(async { tokio::time::sleep(Duration::from_secs(1)).await });

    // Probe the pool: with all clocks leased, this update cannot even acquire one. The probe
    // never reaches the remote, so it cannot mark the fake peer as failed either.
    let mut saturated_error = None;
    let deadline = Instant::now() + Duration::from_secs(20);
    while Instant::now() < deadline {
        let result = rt.block_on(replica_set.update_with_consistency(
            probe_operation(),
            WaitUntil::Wal,
            Some(Duration::from_secs(1)),
            WriteOrdering::Weak,
            false,
            HwMeasurementAcc::new(),
        ));
        match result {
            Err(err) if is_clock_timeout(&err) => {
                saturated_error = Some(err);
                break;
            }
            _ => rt.block_on(async { tokio::time::sleep(Duration::from_millis(50)).await }),
        }
    }
    assert!(
        saturated_error.is_some(),
        "with {IN_FLIGHT_UPDATES} updates in flight, a new update must fail to acquire a clock",
    );

    // Release the clocks and probe again: the same call now gets past clock acquisition.
    for task in in_flight.drain(..) {
        task.abort();
    }

    let mut released_result = None;
    let deadline = Instant::now() + Duration::from_secs(20);
    while Instant::now() < deadline {
        let result = rt.block_on(replica_set.update_with_consistency(
            probe_operation(),
            WaitUntil::Wal,
            Some(Duration::from_secs(1)),
            WriteOrdering::Weak,
            false,
            HwMeasurementAcc::new(),
        ));
        let still_saturated = matches!(&result, Err(err) if is_clock_timeout(err));
        if still_saturated {
            rt.block_on(async { tokio::time::sleep(Duration::from_millis(50)).await });
        } else {
            released_result = Some(result);
            break;
        }
    }
    assert!(
        released_result.is_some(),
        "after the in-flight updates are aborted, the clock pool must accept new updates again",
    );

    println!("in-flight updates: {IN_FLIGHT_UPDATES}");
    println!(
        "clock pool probe while saturated: {:?}",
        saturated_error.unwrap()
    );
    println!(
        "clock pool probe after release: {:?}",
        released_result.unwrap()
    );
    println!("clock pool exhaustion stall test done");
}

Observed output

$ cargo +nightly-2026-03-21 test -p storage --test clock_pool_exhaustion_stall -- --nocapture
   Compiling storage v0.2.0
    Finished `test` profile [unoptimized + debuginfo] target(s) in 4.31s
     Running tests/clock_pool_exhaustion_stall.rs (target/debug/deps/clock_pool_exhaustion_stall-9e6eaca7e9831d16)

running 1 test
in-flight updates: 64
clock pool probe while saturated: Timeout { description: "Operation 'Failed to acquire clock for update operation within Some(1s)' timed out after 1s" }
clock pool probe after release: Ok(UpdateResult { operation_id: Some(1), status: Acknowledged, clock_tag: Some(ClockTag { peer_id: 0, clock_id: 0, clock_tick: 1, token: 9472931496320471396, force: false }) })
clock pool exhaustion stall test done
test clock_pool_exhaustion_stalls_updates ... ok

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

The output shows that the probe failed at clock acquisition while 64 updates were in flight, with the exact timeout the source produces, and that the same call completed with operation_id: Some(1) once the clocks were released.

Root Cause

The clock pool bounds the number of parallel updates by design, but its acquisition path does not wait for capacity: it polls. The lease is held for the whole update, so the pool is drained by remote latency and by local waits, and the polling loops of every waiting update keep the runtime busy without making progress. Line ranges refer to the dev branch at commit e857c5d9.

1. Each replica set caps in-flight updates at 64 clocks

rust

// lib/collection/src/shards/replica_set/clock_set.rs:9-14
const DEFAULT_MAX_CLOCKS: usize = 64;

#[derive(Clone, Debug)]
pub struct ClockSet {
    clocks: Vec<Arc<Clock>>,
    max_clocks: usize,
}

rust

// lib/collection/src/shards/replica_set/clock_set.rs:42-57
    /// Get the first available clock from the set, or create a new one.
    pub fn get_clock(&mut self) -> Option<ClockGuard> {
        for (id, clock) in self.clocks.iter().enumerate() {
            if clock.try_lock() {
                return Some(ClockGuard::new(id as u32, clock.clone()));
            }
        }

        let id = self.clocks.len();
        if id >= self.max_clocks {
            return None;
        }

        let clock = Arc::new(Clock::new_locked());

        self.clocks.push(clock.clone());

        Some(ClockGuard::new(id as u32, clock))
    }

The module states the intent of the limit: the number of clocks bounds the number of parallel update operations, and the limit is expected to be high enough for reasonable use cases. ShardReplicaSet constructs the set with the default, so 64 is the production cap (lib/collection/src/shards/replica_set/mod.rs:128, :231, :364).

2. A new update that cannot lease a clock polls for one

rust

// lib/collection/src/shards/replica_set/update.rs:242-249
    pub async fn get_clock(&self) -> ClockGuard {
        loop {
            match self.clock_set.lock().await.get_clock() {
                Some(clock) => return clock,
                // Prevent blocking async runtime with spinlock
                None => yield_now().await,
            }
        }
    }

yield_now hands the task back to the scheduler and makes it immediately runnable, so this is a busy wait rather than a wait on a notification. The clock set has no primitive that wakes waiters when a ClockGuard is released.

3. The wait is unbounded unless the caller supplied a timeout

rust

// lib/collection/src/shards/replica_set/update.rs:270-278
        let clock_timeout = timeout.unwrap_or(Duration::MAX);
        let mut clock = tokio::time::timeout(clock_timeout, self.get_clock())
            .await
            .map_err(|_| {
                CollectionError::timeout(
                    clock_timeout,
                    format!("Failed to acquire clock for update operation within {timeout:?}"),
                )
            })?;

An update issued without a timeout waits forever, spinning. An update with a timeout fails after the timeout without the operation reaching the WAL, and the error is the only signal the caller receives.

4. The lease spans the whole update, including remote replication

rust

// lib/collection/src/shards/replica_set/update.rs:382-390 (abridged)
        let current_clock_tick = clock.tick_once();
        let clock_tag = ClockTag::new(this_peer_id, clock.id() as _, current_clock_tick);
        ...
        let local_wait = if is_listener { WaitUntil::Wal } else { wait };

rust

// lib/collection/src/shards/replica_set/update.rs:404-415 (abridged)
        let remote_futures: Vec<_> = updatable_remote_shards
            .into_iter()
            .map(|remote| {
                let operation = operation.clone();
                let hw_acc = hw_measurement_acc.clone();
                async move {
                    let peer_id = remote.peer_id;
                    remote
                        .update(operation, wait, timeout, hw_acc)
                        .await
                        .map(|ok| (peer_id, ok))
                        .map_err(|err| (peer_id, err))
                }
            })
            .collect();

rust

// lib/collection/src/shards/replica_set/update.rs:873-889
async fn run_update_futures<F>(
    update_futures: Vec<F>,
    update_concurrency: Option<NonZeroUsize>,
) -> Vec<Result<(PeerId, UpdateResult), (PeerId, CollectionError)>>
where
    F: Future<Output = Result<(PeerId, UpdateResult), (PeerId, CollectionError)>>,
{
    match update_concurrency {
        Some(concurrency) => {
            futures::stream::iter(update_futures)
                .buffer_unordered(concurrency.get())
                .collect()
                .await
        }
        None => FuturesUnordered::from_iter(update_futures).collect().await,
    }
}

The fan-out collects every remote result, so the update future, and with it the clock guard owned by update, stays alive until the slowest remote answers. Write ordering does not change this: even WriteOrdering::Weak updates are replicated to every updatable remote and therefore lease a clock for the round trip.

5. Remote latency sets the lease length

rust

// lib/collection/src/shards/remote_shard.rs:668-687 (abridged)
        let default_timeout = self.channel_service.request_timeout();
        // We want to know if operation wrote into WAL before we break connection on client side
        // So, we always want to

Source: qdrant/qdrant

View original on GitHub View discussion on GitHub