vector-store: Move Usearch operations to a dedicated tasks

Usearch's API uses global locks. Operations like `add` and
`remove` can hold these locks for extended periods.

When these operations are executed on the Tokio runtime, they can block
worker threads. This leads to thread starvation, effectively halting all
asynchronous tasks, including handling HTTP requests, processing CDC
data, and serving metrics.

This commit moves the blocking `add` and `remove` calls to a dedicated
Rayon thread pool. By isolating these  operations, the Tokio runtime
remains unblocked and responsive, ensuring
the application's I/O continues to function correctly.

Author: knowack1

Cargo.lock

@@ -57,6 +57,7 @@ utoipa.workspace = true
 utoipa-axum.workspace = true
 utoipa-swagger-ui.workspace = true
 uuid.workspace = true
+rayon = "1.11.0"
 
 [dev-dependencies]
 mockall.workspace = true

crates/vector-store/Cargo.toml

@@ -31,6 +31,7 @@ use std::sync::atomic::Ordering;
 use std::time::Duration;
 use std::time::Instant;
 use tokio::sync::Notify;
+use tokio::sync::OwnedSemaphorePermit;
 use tokio::sync::Semaphore;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
@@ -326,14 +327,13 @@ impl UsearchIndex for RwLock<Simulator> {
         sim.wait_search(start);
 
         let keys = {
-            let keys = sim.keys.read().unwrap();
-            let len = keys.len() as u64;
+            let len = sim.keys.read().unwrap().len() as u64;
             if len == 0 {
                 Vec::new()
             } else {
                 iter::repeat_with(|| rand::random_range(0..len))
                     .map(Key)
-                    .filter(|key| keys.contains(key))
+                    .filter(|key| sim.keys.read().unwrap().contains(key))
                     .take(limit.0.get())
                     .collect()
             }
@@ -425,8 +425,7 @@ fn new(
                         let usearch_key = Arc::clone(&usearch_key);
                         async move {
                             crate::move_to_the_end_of_async_runtime_queue().await;
-                            process(msg, dimensions, idx, keys, usearch_key);
-                            drop(permit);
+                            process(msg, dimensions, idx, keys, usearch_key, permit);
                         }
                     });
                 }
@@ -445,24 +444,32 @@ fn new(
 fn process(
     msg: Index,
     dimensions: Dimensions,
-    idx: Arc<impl UsearchIndex>,
+    idx: Arc<impl UsearchIndex + Send + Sync + 'static>,
     keys: Arc<RwLock<BiMap<PrimaryKey, Key>>>,
     usearch_key: Arc<AtomicU64>,
+    permit: OwnedSemaphorePermit,
 ) {
     match msg {
         Index::AddOrReplace {
             primary_key,
             embedding,
-            in_progress: _in_progress,
+            in_progress,
         } => {
-            add_or_replace(idx, keys, usearch_key, primary_key, embedding);
+            rayon::spawn(move || {
+                add_or_replace(idx, keys, usearch_key, primary_key, embedding);
+                drop(in_progress);
+                drop(permit);
+            });
         }
 
         Index::Remove {
             primary_key,
             in_progress: _in_progress,
         } => {
-            remove(idx, keys, primary_key);
+            rayon::spawn(move || {
+                remove(idx, keys, primary_key);
+                drop(permit);
+            });
         }
 
         Index::Ann {

crates/vector-store/src/index/usearch.rs

@@ -5,6 +5,7 @@
 
 use crate::usearch::{setup_store, setup_store_and_wait_for_index};
 use crate::wait_for;
+use vector_store::Config;
 use vector_store::httproutes::NodeStatus;
 
 #[tokio::test]
@@ -19,7 +20,7 @@ async fn status_is_serving_after_creation() {
 #[tokio::test]
 async fn status_is_bootstrapping_while_discovering_indexes() {
     crate::enable_tracing();
-    let (run, _index, db, _node_state) = setup_store().await;
+    let (run, _index, db, _node_state) = setup_store(Config::default()).await;
     db.simulate_endless_get_indexes_processing();
     let (client, _server, _config_tx) = run.await;
 

crates/vector-store/tests/integration/status.rs

@@ -3,6 +3,7 @@
  * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
  */
 
+use crate::Duration;
 use crate::db_basic;
 use crate::db_basic::DbBasic;
 use crate::db_basic::Index;
@@ -28,7 +29,9 @@ use vector_store::SpaceType;
 use vector_store::Vector;
 use vector_store::node_state::NodeState;
 
-pub(crate) async fn setup_store() -> (
+pub(crate) async fn setup_store(
+    config: Config,
+) -> (
     impl std::future::Future<Output = (HttpClient, impl Sized, impl Sized)>,
     IndexMetadata,
     DbBasic,
@@ -76,13 +79,12 @@ pub(crate) async fn setup_store() -> (
     )
     .unwrap();
 
-    let (_, rx) = watch::channel(Arc::new(Config::default()));
-    let index_factory = vector_store::new_index_factory_usearch(rx).unwrap();
+    let (config_tx, config_rx) = watch::channel(Arc::new(config));
+    let index_factory = vector_store::new_index_factory_usearch(config_rx.clone()).unwrap();
 
     let run = {
         let node_state = node_state.clone();
         async move {
-            let (_config_tx, config_rx) = watch::channel(Arc::new(vector_store::Config::default()));
             let (server, addr) = vector_store::run(
                 SocketAddr::from(([127, 0, 0, 1], 0)).into(),
                 node_state,
@@ -93,7 +95,7 @@ pub(crate) async fn setup_store() -> (
             .await
             .unwrap();
 
-            (HttpClient::new(addr), server, _config_tx)
+            (HttpClient::new(addr), server, config_tx)
         }
     };
 
@@ -107,7 +109,7 @@ pub(crate) async fn setup_store_and_wait_for_index() -> (
     impl Sized,
     Sender<NodeState>,
 ) {
-    let (run, index, db, node_state) = setup_store().await;
+    let (run, index, db, node_state) = setup_store(Config::default()).await;
     let (client, server, _config_tx) = run.await;
 
     wait_for(
@@ -123,7 +125,7 @@ pub(crate) async fn setup_store_and_wait_for_index() -> (
 async fn simple_create_search_delete_index() {
     crate::enable_tracing();
 
-    let (run, index, db, _node_state) = setup_store().await;
+    let (run, index, db, _node_state) = setup_store(Config::default()).await;
     let (client, _server, _config_tx) = run.await;
 
     db.insert_values(
@@ -349,7 +351,7 @@ async fn ann_returns_bad_request_when_provided_vector_size_is_not_eq_index_dimen
 #[tokio::test]
 async fn ann_fail_while_building() {
     crate::enable_tracing();
-    let (run, index, db, _node_state) = setup_store().await;
+    let (run, index, db, _node_state) = setup_store(Config::default()).await;
     db.set_next_full_scan_progress(vector_store::Progress::InProgress(
         Percentage::try_from(33.333).unwrap(),
     ));
@@ -393,3 +395,37 @@ async fn ann_works_with_embedding_field_name() {
 
     assert_eq!(response.status(), StatusCode::OK);
 }
+
+#[tokio::test]
+#[ntest::timeout(10_000)]
+async fn http_server_is_responsive_when_index_add_hangs() {
+    crate::enable_tracing();
+    let config = Config {
+        usearch_simulator: Some(vec![
+            Duration::from_secs(0),
+            Duration::from_secs(20), // Simulate long add operation (longer than test timeout).
+            Duration::from_secs(0),
+        ]),
+        ..Default::default()
+    };
+    let (run, index, db, _node_state) = setup_store(config).await;
+    // Insert a value before starting the vector store. The mock DB does not support
+    // adding embeddings while it's running, so it must be populated beforehand.
+    db.insert_values(
+        &index.keyspace_name,
+        &index.table_name,
+        &index.target_column,
+        vec![(
+            vec![CqlValue::Int(1), CqlValue::Text("one".to_string())].into(),
+            Some(vec![1., 1., 1.].into()),
+            OffsetDateTime::from_unix_timestamp(10).unwrap().into(),
+        )],
+    )
+    .unwrap();
+    let (client, _server, _config_tx) = run.await;
+
+    // Ensure the HTTP server stays responsive while the (simulated) embedding add is long-running.
+    let status = client.status().await.unwrap();
+
+    assert_eq!(status, vector_store::httproutes::NodeStatus::Serving);
+}