ensure domain exists before writing out hsnw index

Author: Matthijs van Otterdijk

src/indexer.rs

@@ -3,7 +3,7 @@ use crate::{
     openai::{embeddings_for, EmbeddingError},
     server::Operation,
     vecmath::{self, Embedding},
-    vectors::{LoadedVec, VectorStore},
+    vectors::{Domain, LoadedVec, VectorStore},
 };
 use hnsw::{Hnsw, Searcher};
 use rand_pcg::Lcg128Xsl64;
@@ -89,7 +89,7 @@ enum Op {
 }
 
 pub async fn operations_to_point_operations(
-    domain: &str,
+    domain: &Domain,
     vector_store: &VectorStore,
     structs: Vec<Result<Operation, std::io::Error>>,
     key: &str,
@@ -114,7 +114,6 @@ pub async fn operations_to_point_operations(
     } else {
         embeddings_for(key, &strings).await?
     };
-    let domain = vector_store.get_domain(domain)?;
     let loaded_vecs = vector_store.add_and_load_vecs(&domain, vecs.iter())?;
     let mut new_ops: Vec<PointOperation> = zip(tuples, loaded_vecs)
         .map(|((op, _, id), vec)| match op {

src/main.rs

@@ -200,6 +200,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
             let dirpath = Path::new(&directory);
             let mut hnsw: HnswIndex = Hnsw::new(OpenAI);
             let store = VectorStore::new(dirpath, size);
+            let resolved_domain = store.get_domain(&domain)?;
 
             let f = File::options().read(true).open(path)?;
 
@@ -216,7 +217,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
             for structs in opstream {
                 let structs: Vec<_> = structs.collect();
                 let new_ops =
-                    operations_to_point_operations(&domain.clone(), &store, structs, &key).await?;
+                    operations_to_point_operations(&resolved_domain, &store, structs, &key).await?;
                 hnsw = start_indexing_from_operations(hnsw, new_ops).unwrap();
             }
             let index_id = create_index_name(&domain, &commit);

src/server.rs

@@ -505,16 +505,13 @@ impl Service {
                 previous,
             })
             .await;
+        let domain = self.vector_store.get_domain(&domain)?;
         self.set_task_status(task_id.to_string(), TaskStatus::Pending(0.3))
             .await;
         while let Some(structs) = opstream.next().await {
-            let new_ops = operations_to_point_operations(
-                &domain.clone(),
-                &self.vector_store,
-                structs,
-                api_key,
-            )
-            .await?;
+            let new_ops =
+                operations_to_point_operations(&domain, &self.vector_store, structs, api_key)
+                    .await?;
             hnsw = start_indexing_from_operations(hnsw, new_ops)?;
         }
         self.set_task_status(task_id.to_string(), TaskStatus::Pending(0.8))