Upgrade Qdrant Edge to 0.8.0 and make unreadable vector stores repairable (#982)

runtime/src/qdrant_edge_database.rs

@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::fs;
 use std::path::{Path, PathBuf};
 use std::sync::Mutex;
@@ -11,8 +11,9 @@ use qdrant_edge::external::uuid::Uuid;
 use qdrant_edge::{
     Condition, Distance, EdgeConfig, EdgeOptimizersConfig, EdgeShard, EdgeVectorParams,
     FieldCondition, Filter, HnswIndexConfig, Match, MatchValue, NamedQuery, Payload, PointId,
-    PointInsertOperations, PointOperations, PointStruct, QueryEnum, ScoredPoint, SearchRequest,
-    UpdateOperation, ValueVariants, VectorInternal, Vectors, WithPayloadInterface, WithVector,
+    PointInsertOperations, PointOperations, PointStruct, QueryEnum, QueryRequest, ScoredPoint,
+    ScoringQuery, UpdateOperation, ValueVariants, VectorInternal, Vectors, WithPayloadInterface,
+    WithVector,
 };
 use serde::{Deserialize, Serialize};
 use tauri::Manager;
@@ -32,6 +33,10 @@ const STORE_INITIALIZATION_MARKER_TEMP: &str = "store_name.tmp";
 const STORE_DISPLAY_NAME_MARKER: &str = "data_source_name.txt";
 const STORE_DISPLAY_NAME_MARKER_TEMP: &str = "data_source_name.tmp";
 
+/// Marks a response whose store exists on disk but cannot be opened. The .NET side keys its repair
+/// offer off this value instead of parsing `issue`, so rewording the message stays harmless.
+const ISSUE_CODE_STORE_UNREADABLE: &str = "store-unreadable";
+
 type QdrantEdgeResult<T> = Result<T, Box<dyn std::error::Error + Send + Sync>>;
 
 static QDRANT_EDGE_DATABASE: Lazy<Mutex<Option<QdrantEdgeDatabase>>> =
@@ -127,9 +132,39 @@ pub struct DeleteQdrantEdgeStoreRequest {
 pub struct QdrantEdgeResponse<T> {
     pub success: bool,
     pub issue: String,
+    pub issue_code: &'static str,
     pub data: Option<T>,
 }
 
+/// A vector store which is initialized on disk but which Qdrant Edge refuses to open.
+///
+/// This is deliberately its own error type rather than one more formatted string: a broken store
+/// is the one failure the user can act on, and the request layer has to recognize it to label the
+/// response. Nothing here deletes the store -- rebuilding the embeddings costs the user time and,
+/// with a cloud embedding provider, money, so that stays their decision.
+#[derive(Debug)]
+struct StoreUnreadableError {
+    store_name: String,
+    message: String,
+}
+
+impl StoreUnreadableError {
+    fn new(store_name: &str, path: &Path, source: impl std::fmt::Display) -> Self {
+        Self {
+            store_name: store_name.to_string(),
+            message: format!("Failed to load vector store '{store_name}' from '{}': {source}", path.display()),
+        }
+    }
+}
+
+impl std::fmt::Display for StoreUnreadableError {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str(&self.message)
+    }
+}
+
+impl std::error::Error for StoreUnreadableError {}
+
 #[derive(Serialize)]
 pub struct QdrantEdgeEnsureStoreResult {
     pub created: bool,
@@ -168,6 +203,10 @@ pub struct QdrantEdgeInfo {
 pub struct QdrantEdgeDatabase {
     base_path: PathBuf,
     shards: HashMap<String, EdgeShard>,
+
+    /// Stores whose unreadability has already been logged. A broken store is hit by every single
+    /// request against it, and one log line per request would bury everything else.
+    reported_unreadable_stores: HashSet<String>,
 }
 
 impl QdrantEdgeDatabase {
@@ -175,9 +214,16 @@ impl QdrantEdgeDatabase {
         Self {
             base_path,
             shards: HashMap::new(),
+            reported_unreadable_stores: HashSet::new(),
         }
     }
 
+    /// Whether this store's defect still has to be written to the log. True exactly once per store,
+    /// until the store loads again.
+    fn report_unreadable_store(&mut self, store_name: &str) -> bool {
+        self.reported_unreadable_stores.insert(store_name.to_string())
+    }
+
     fn store_path(&self, store_name: &str) -> QdrantEdgeResult<PathBuf> {
         validate_store_name(store_name)?;
         Ok(self.base_path.join("stores").join(store_directory_name(store_name)))
@@ -191,9 +237,10 @@ impl QdrantEdgeDatabase {
         }
 
         let shard = if is_initialized {
-            EdgeShard::load(&path, None).map_err(|error| {
-                format!("Failed to load vector store '{store_name}' from '{}': {error}", path.display())
-            })?
+            match EdgeShard::load(&path, None) {
+                Ok(shard) => shard,
+                Err(error) => return Err(StoreUnreadableError::new(store_name, &path, error).into()),
+            }
         } else {
             fs::create_dir_all(&path).map_err(|error| {
                 format!("Failed to create directory for vector store '{store_name}' at '{}': {error}", path.display())
@@ -215,6 +262,7 @@ impl QdrantEdgeDatabase {
             shard
         };
 
+        self.reported_unreadable_stores.remove(store_name);
         self.shards.insert(store_name.to_string(), shard);
         Ok((self.shards.get(store_name).unwrap(), !is_initialized))
     }
@@ -230,9 +278,12 @@ impl QdrantEdgeDatabase {
             return Ok(None);
         }
 
-        let shard = EdgeShard::load(&path, None).map_err(|error| {
-            format!("Failed to load vector store '{store_name}' from '{}': {error}", path.display())
-        })?;
+        let shard = match EdgeShard::load(&path, None) {
+            Ok(shard) => shard,
+            Err(error) => return Err(StoreUnreadableError::new(store_name, &path, error).into()),
+        };
+
+        self.reported_unreadable_stores.remove(store_name);
         self.shards.insert(store_name.to_string(), shard);
         Ok(self.shards.get(store_name))
     }
@@ -306,7 +357,7 @@ impl QdrantEdgeDatabase {
         shard.update(UpdateOperation::PointOperation(
             PointOperations::UpsertPoints(PointInsertOperations::PointsList(points)),
         ))?;
-        shard.flush();
+        shard.flush()?;
         Ok(())
     }
 
@@ -320,18 +371,19 @@ impl QdrantEdgeDatabase {
             return Ok(vec![]);
         };
 
-        let search_results = shard.search(SearchRequest {
-            query: QueryEnum::Nearest(NamedQuery::new(
+        let search_results = shard.query(QueryRequest {
+            prefetches: Vec::new(),
+            query: Some(ScoringQuery::Vector(QueryEnum::Nearest(NamedQuery::new(
                 VectorInternal::Dense(vector),
                 VECTOR_NAME,
-            )),
+            )))),
             filter: None,
-            params: None,
+            score_threshold: None,
             limit: max_matches,
             offset: 0,
-            with_payload: Some(WithPayloadInterface::Bool(true)),
-            with_vector: Some(WithVector::Bool(false)),
-            score_threshold: None,
+            params: None,
+            with_vector: WithVector::Bool(false),
+            with_payload: WithPayloadInterface::Bool(true),
         })?;
 
         Ok(search_results
@@ -348,7 +400,7 @@ impl QdrantEdgeDatabase {
         shard.update(UpdateOperation::PointOperation(
             PointOperations::DeletePointsByFilter(match_keyword_filter("file_path", file_path)?),
         ))?;
-        shard.flush();
+        shard.flush()?;
         Ok(())
     }
 
@@ -361,7 +413,7 @@ impl QdrantEdgeDatabase {
         if optimized {
             info!(Source = "Qdrant Edge"; "Optimized vector store '{}'.", store_name);
         }
-        shard.flush();
+        shard.flush()?;
         Ok(())
     }
 
@@ -503,6 +555,7 @@ where
         return Json(QdrantEdgeResponse {
             success: false,
             issue: "Qdrant Edge is not available.".to_string(),
+            issue_code: "",
             data: None,
         });
     };
@@ -511,14 +564,36 @@ where
         Ok(data) => Json(QdrantEdgeResponse {
             success: true,
             issue: String::new(),
+            issue_code: "",
             data: Some(data),
         }),
         Err(e) => {
             let issue = e.to_string();
-            error!(Source = "Qdrant Edge"; "Qdrant Edge request failed: {issue}");
+
+            //
+            // An unreadable store keeps failing for as long as the user leaves it alone, so it is
+            // logged once and then only answered. Every other failure is logged as it happens,
+            // because those are one-offs worth seeing each time.
+            //
+            let issue_code = match e.downcast_ref::<StoreUnreadableError>() {
+                Some(unreadable) => {
+                    if database.report_unreadable_store(&unreadable.store_name) {
+                        error!(Source = "Qdrant Edge"; "Qdrant Edge request failed: {issue}");
+                    }
+
+                    ISSUE_CODE_STORE_UNREADABLE
+                },
+
+                None => {
+                    error!(Source = "Qdrant Edge"; "Qdrant Edge request failed: {issue}");
+                    ""
+                },
+            };
+
             Json(QdrantEdgeResponse {
                 success: false,
                 issue,
+                issue_code,
                 data: None,
             })
         },
@@ -595,7 +670,7 @@ fn remove_obsolete_qdrant_path(path: &Path) {
 
 fn edge_config(vector_size: usize) -> EdgeConfig {
     EdgeConfig {
-        on_disk_payload: true,
+        on_disk_payload: Some(true),
         vectors: HashMap::from([(
             VECTOR_NAME.to_string(),
             EdgeVectorParams {
@@ -609,13 +684,20 @@ fn edge_config(vector_size: usize) -> EdgeConfig {
             },
         )]),
         sparse_vectors: HashMap::new(),
-        hnsw_config: hnsw_config(),
+        hnsw_config: Some(hnsw_config()),
         quantization_config: None,
-        optimizers: edge_optimizers_config(),
+        optimizers: Some(edge_optimizers_config()),
         wal_options: None,
+        max_search_threads: None,
+        search_pool_core: None,
     }
 }
 
+// `on_disk` is deprecated in favor of `memory`, but Qdrant Edge does not re-export the `Memory`
+// type, so the new field cannot be named from here. Leaving both unset is not an option either:
+// the effective placement would fall back to cached instead of on-disk, which is a real change
+// and would have the optimizers rebuild the HNSW graph.
+#[allow(deprecated)]
 fn hnsw_config() -> HnswIndexConfig {
     HnswIndexConfig {
         m: HNSW_M,
@@ -623,6 +705,7 @@ fn hnsw_config() -> HnswIndexConfig {
         full_scan_threshold: HNSW_FULL_SCAN_THRESHOLD_KB,
         max_indexing_threads: HNSW_MAX_INDEXING_THREADS,
         on_disk: Some(true),
+        memory: None,
         payload_m: None,
         inline_storage: None,
     }
@@ -902,6 +985,44 @@ mod tests {
         fs::remove_dir_all(test_directory).unwrap();
     }
 
+    #[test]
+    fn an_unreadable_store_is_reported_but_never_deleted() {
+        let test_directory = std::env::temp_dir().join(format!(
+            "ai-studio-qdrant-unreadable-{}-{}",
+            std::process::id(),
+            std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_nanos()
+        ));
+        let store_name = "rag_6cc665a82b1e4d42bc748015b7b391ec";
+
+        let mut database = QdrantEdgeDatabase::new(test_directory.clone());
+        assert!(database.ensure_store_exists(store_name, "Some source", 3).unwrap().created);
+        let store_path = database.store_path(store_name).unwrap();
+
+        // Release the shard before breaking it, so the files are not held open any more.
+        drop(database);
+        fs::write(store_path.join("edge_config.json"), "this is not a config").unwrap();
+
+        let mut database = QdrantEdgeDatabase::new(test_directory.clone());
+        let error = database.get_existing_store(store_name).unwrap_err();
+        assert!(
+            error.downcast_ref::<StoreUnreadableError>().is_some(),
+            "a store which cannot be opened has to be recognizable as such, not just a message"
+        );
+
+        // The whole point: the user's embeddings survive a defect until they ask for a rebuild.
+        assert!(store_path.join("segments").is_dir());
+        assert!(store_path.join(STORE_INITIALIZATION_MARKER).is_file());
+
+        // And the defect is logged once, not once per request.
+        assert!(database.report_unreadable_store(store_name));
+        assert!(!database.report_unreadable_store(store_name));
+
+        fs::remove_dir_all(test_directory).unwrap();
+    }
+
     #[test]
     fn point_ids_must_be_valid_uuids() {
         assert!(to_point_id("6cc665a8-2b1e-4d42-bc74-8015b7b391ec").is_ok());