Add logging and fix events (#11)

rrooij · Matthijs van Otterdijk · web-flow · commit 060f5a479b75 · 2023-07-20T15:46:25.000+02:00
* log every request

* slightly more robust error handling in start_indexing

* return a 404 for unknown task status

* Fix formatting

---------

Co-authored-by: Matthijs van Otterdijk &lt;matthijs@terminusdb.com&gt;
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -31,6 +31,7 @@ packed_simd = {version = "0.3.8", optional=true}
 aligned_box = "0.2"
 tiktoken-rs = "0.4"
 itertools = "0.10"
+chrono = "0.4.26"
 
 [features]
 simd = ["packed_simd"]
diff --git a/src/server.rs b/src/server.rs
@@ -395,6 +395,12 @@ impl Service {
     }
 
     async fn serve(self: Arc<Self>, req: Request<Body>) -> Result<Response<Body>, Infallible> {
+        eprintln!(
+            "{:?}: {:?} {:?}",
+            chrono::offset::Local::now(),
+            req.method(),
+            req.uri()
+        );
         match *req.method() {
             Method::POST => self.post(req).await,
             Method::GET => self.get(req).await,
@@ -414,6 +420,32 @@ impl Service {
         }
     }
 
+    async fn start_indexing_inner(
+        self: Arc<Self>,
+        domain: String,
+        commit: String,
+        previous: Option<String>,
+        task_id: &str,
+        api_key: String,
+        index_id: &str,
+        content_endpoint: String,
+    ) -> Result<(String, HnswIndex), IndexError> {
+        let internal_task_id = task_id.clone();
+        let opstream = get_operations_from_content_endpoint(
+            content_endpoint.to_string(),
+            self.user_forward_header.clone(),
+            domain.clone(),
+            commit.clone(),
+            previous.clone(),
+        )
+        .await?
+        .chunks(100);
+        self.process_operation_chunks(
+            opstream, domain, commit, previous, index_id, task_id, &api_key,
+        )
+        .await
+    }
+
     fn start_indexing(
         self: Arc<Self>,
         domain: String,
@@ -428,36 +460,39 @@ impl Service {
             tokio::spawn(async move {
                 let index_id = create_index_name(&domain, &commit);
                 if self.test_and_set_pending(index_id.clone()).await {
-                    let opstream = get_operations_from_content_endpoint(
-                        content_endpoint.to_string(),
-                        self.user_forward_header.clone(),
-                        domain.clone(),
-                        commit.clone(),
-                        previous.clone(),
-                    )
-                    .await
-                    .unwrap()
-                    .chunks(100);
                     match self
-                        .process_operation_chunks(
-                            opstream, domain, commit, previous, &index_id, &task_id, &api_key,
+                        .clone()
+                        .start_indexing_inner(
+                            domain,
+                            commit,
+                            previous,
+                            &task_id,
+                            api_key,
+                            &index_id,
+                            content_endpoint,
                         )
                         .await
                     {
                         Ok((id, hnsw)) => {
                             self.set_index(id, hnsw.into()).await;
+                            self.set_task_status(task_id, TaskStatus::Completed).await;
                             self.clear_pending(&index_id).await;
                         }
                         Err(err) => {
+                            eprintln!(
+                                "{:?}: error while indexing: {:?}",
+                                chrono::offset::Local::now(),
+                                err
+                            );
                             self.set_task_status(
                                 internal_task_id,
                                 TaskStatus::Error(err.to_string()),
                             )
                             .await;
+                            self.clear_pending(&index_id).await;
                         }
                     }
                 }
-                self.set_task_status(task_id, TaskStatus::Completed).await;
             });
             Ok(())
         } else {
@@ -577,7 +612,7 @@ impl Service {
                         }
                     }
                 } else {
-                    Ok(Response::builder().body(format!("{}", 1.0).into()).unwrap())
+                    Ok(Response::builder().status(404).body(Body::empty()).unwrap())
                 }
             }
             Ok(ResourceSpec::DuplicateCandidates {
