server: graceful shutdown

typhon/src/lib.rs

@@ -284,3 +284,15 @@ pub fn log_event(event: Event) {
         LISTENERS.lock().await.log(event);
     });
 }
+
+pub async fn shutdown() {
+    let res = tokio::signal::ctrl_c().await;
+    eprintln!("Typhon is shutting down...");
+    let _ = res.map_err(|e| log::error!("{}", e));
+    tokio::join!(
+        EVALUATIONS.shutdown(),
+        JOBS_BUILD.shutdown(),
+        JOBS_POST.shutdown(),
+        JOBS_PRE.shutdown(),
+    );
+}

typhon/src/main.rs

@@ -30,8 +30,15 @@ async fn main() -> std::io::Result<()> {
         .expect("failed to run migrations");
 
     // Run actix server
-    HttpServer::new(|| App::new().configure(typhon::api::config))
+    let actix = HttpServer::new(|| App::new().configure(typhon::api::config))
         .bind(("127.0.0.1", 8000))?
-        .run()
-        .await
+        .run();
+
+    // Graceful shutdown
+    tokio::select! {
+        _ = actix => panic!(),
+        _ = typhon::shutdown() => eprintln!("Good bye!"),
+    }
+
+    Ok(())
 }

typhon/src/tasks.rs

@@ -91,4 +91,18 @@ impl<Id: std::cmp::Eq + std::hash::Hash + std::clone::Clone + Send> Tasks<Id> {
             })
             .unwrap_or(false)
     }
+
+    pub async fn keys(&self) -> Vec<Id> {
+        let tasks = self.tasks.lock().await;
+        tasks.keys().cloned().collect()
+    }
+
+    pub async fn shutdown(&'static self) {
+        let mut set = tokio::task::JoinSet::new();
+        let keys = self.keys().await;
+        for id in keys {
+            set.spawn(self.cancel(id));
+        }
+        while let Some(_) = set.join_next().await {}
+    }
 }