Improve error reporting (#7)

* Improve error reporting

* Help clippy

Author: GavinMendelGleason

src/indexer.rs

@@ -1,6 +1,6 @@
 #![allow(unused, dead_code)]
 use crate::{
-    openai::embeddings_for,
+    openai::{embeddings_for, EmbeddingError},
     server::Operation,
     vecmath::{self, Embedding},
     vectors::{LoadedVec, VectorStore},
@@ -93,9 +93,9 @@ pub async fn operations_to_point_operations(
     vector_store: &VectorStore,
     structs: Vec<Result<Operation, std::io::Error>>,
     key: &str,
-) -> Vec<PointOperation> {
+) -> Result<Vec<PointOperation>, IndexError> {
     // Should not unwrap here -
-    let ops: Vec<Operation> = structs.into_iter().map(|ro| ro.unwrap()).collect();
+    let ops: Vec<Operation> = structs.into_iter().collect::<Result<Vec<_>, _>>()?;
     let tuples: Vec<(Op, String, String)> = ops
         .iter()
         .flat_map(|o| match o {
@@ -112,12 +112,10 @@ pub async fn operations_to_point_operations(
     let vecs: Vec<Embedding> = if strings.is_empty() {
         Vec::new()
     } else {
-        embeddings_for(key, &strings).await.unwrap()
+        embeddings_for(key, &strings).await?
     };
-    let domain = vector_store.get_domain(domain).unwrap();
-    let loaded_vecs = vector_store
-        .add_and_load_vecs(&domain, vecs.iter())
-        .unwrap();
+    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 {
             Op::Insert => PointOperation::Insert {
@@ -136,7 +134,7 @@ pub async fn operations_to_point_operations(
         })
         .collect();
     new_ops.append(&mut delete_ops);
-    new_ops
+    Ok(new_ops)
 }
 
 pub struct IndexIdentifier {
@@ -145,9 +143,14 @@ pub struct IndexIdentifier {
     pub domain: String,
 }
 
-#[derive(Debug)]
-enum IndexError {
+#[derive(Debug, Error)]
+pub enum IndexError {
+    #[error("Indexing failed")]
     Failed,
+    #[error("Indexing failed with io error: {0:?}")]
+    IoError(#[from] std::io::Error),
+    #[error("Embedding error: {0:?}")]
+    EmbeddingError(#[from] EmbeddingError),
 }
 
 /*

src/main.rs

@@ -216,7 +216,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(&domain.clone(), &store, structs, &key).await?;
                 hnsw = start_indexing_from_operations(hnsw, new_ops).unwrap();
             }
             let index_id = create_index_name(&domain, &commit);

src/server.rs

@@ -39,6 +39,7 @@ use crate::indexer::deserialize_index;
 use crate::indexer::operations_to_point_operations;
 use crate::indexer::search;
 use crate::indexer::serialize_index;
+use crate::indexer::IndexError;
 use crate::indexer::Point;
 use crate::indexer::PointOperation;
 use crate::indexer::SearchError;
@@ -244,7 +245,7 @@ fn uri_to_spec(uri: &Uri) -> Result<ResourceSpec, SpecParseError> {
 #[derive(Clone, Debug)]
 pub enum TaskStatus {
     Pending(f32),
-    Error,
+    Error(String),
     Completed,
 }
 
@@ -422,6 +423,7 @@ impl Service {
         api_key: String,
     ) -> Result<(), StartIndexError> {
         let content_endpoint = self.content_endpoint.clone();
+        let internal_task_id = task_id.clone();
         if let Some(content_endpoint) = content_endpoint {
             tokio::spawn(async move {
                 let index_id = create_index_name(&domain, &commit);
@@ -436,13 +438,24 @@ impl Service {
                     .await
                     .unwrap()
                     .chunks(100);
-                    let (id, hnsw) = self
+                    match self
                         .process_operation_chunks(
                             opstream, domain, commit, previous, &index_id, &task_id, &api_key,
                         )
-                        .await;
-                    self.set_index(id, hnsw.into()).await;
-                    self.clear_pending(&index_id).await;
+                        .await
+                    {
+                        Ok((id, hnsw)) => {
+                            self.set_index(id, hnsw.into()).await;
+                            self.clear_pending(&index_id).await;
+                        }
+                        Err(err) => {
+                            self.set_task_status(
+                                internal_task_id,
+                                TaskStatus::Error(err.to_string()),
+                            )
+                            .await;
+                        }
+                    }
                 }
                 self.set_task_status(task_id, TaskStatus::Completed).await;
             });
@@ -471,6 +484,7 @@ impl Service {
         Ok(())
     }
 
+    #[allow(clippy::too_many_arguments)]
     async fn process_operation_chunks(
         self: &Arc<Self>,
         mut opstream: futures::stream::Chunks<
@@ -482,7 +496,7 @@ impl Service {
         index_id: &str,
         task_id: &str,
         api_key: &str,
-    ) -> (String, HnswIndex) {
+    ) -> Result<(String, HnswIndex), IndexError> {
         let id = create_index_name(&domain, &commit);
         let mut hnsw = self
             .load_hnsw_for_indexing(IndexIdentifier {
@@ -500,14 +514,14 @@ impl Service {
                 structs,
                 api_key,
             )
-            .await;
-            hnsw = start_indexing_from_operations(hnsw, new_ops).unwrap();
+            .await?;
+            hnsw = start_indexing_from_operations(hnsw, new_ops)?;
         }
         self.set_task_status(task_id.to_string(), TaskStatus::Pending(0.8))
             .await;
         let path = self.path.clone();
-        serialize_index(path, index_id, hnsw.clone()).unwrap();
-        (id, hnsw)
+        serialize_index(path, index_id, hnsw.clone())?;
+        Ok((id, hnsw))
     }
 
     async fn get_start_index(
@@ -526,7 +540,7 @@ impl Service {
 
     async fn get(self: Arc<Self>, req: Request<Body>) -> Result<Response<Body>, Infallible> {
         let uri = req.uri();
-        match dbg!(uri_to_spec(uri)) {
+        match uri_to_spec(uri) {
             Ok(ResourceSpec::StartIndex {
                 domain,
                 commit,
@@ -557,8 +571,9 @@ impl Service {
                         TaskStatus::Pending(f) => {
                             Ok(Response::builder().body(format!("{}", f).into()).unwrap())
                         }
-                        TaskStatus::Error => Ok(Response::builder()
-                            .body(format!("{:?}", state).into())
+                        TaskStatus::Error(msg) => Ok(Response::builder()
+                            .status(StatusCode::INTERNAL_SERVER_ERROR)
+                            .body(format!("{:?}", msg).into())
                             .unwrap()),
                         TaskStatus::Completed => {
                             Ok(Response::builder().body(format!("{}", 1.0).into()).unwrap())