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

@@ -15,7 +15,7 @@ pub struct Tasks<Id> {
     tasks: Mutex<HashMap<Id, TaskHandle>>,
 }
 
-impl<Id: std::cmp::Eq + std::hash::Hash + std::clone::Clone + Send> Tasks<Id> {
+impl<Id: std::cmp::Eq + std::hash::Hash + std::clone::Clone + Send + Sync> Tasks<Id> {
     pub fn new() -> Self {
         Tasks {
             tasks: Mutex::new(HashMap::new()),
@@ -83,4 +83,17 @@ impl<Id: std::cmp::Eq + std::hash::Hash + std::clone::Clone + Send> Tasks<Id> {
             .flatten()
             .is_some()
     }
+
+    pub async fn shutdown(&'static self) {
+        let tasks = self.tasks.lock().await;
+        let mut set = tokio::task::JoinSet::new();
+        for id in tasks.keys() {
+            set.spawn({
+                let id = id.clone();
+                async move { self.wait(&id).await }
+            });
+            self.cancel(id).await;
+        }
+        while let Some(_) = set.join_next().await {}
+    }
 }