fix(deployer): added runtime error handling

deployer/src/deployment/run.rs

@@ -74,6 +74,7 @@ pub async fn task(
                     active_deployment_getter.clone(),
                     runtime_manager.clone(),
                 );
+                let runtime_manager_clone = runtime_manager.clone();
                 let cleanup = move |response: Option<SubscribeStopResponse>| {
                     debug!(response = ?response,  "stop client response: ");
 
@@ -83,20 +84,23 @@ pub async fn task(
                             StopReason::End => completed_cleanup(&id),
                             StopReason::Crash => crashed_cleanup(
                                 &id,
+                                runtime_manager_clone,
                                 Error::Run(anyhow::Error::msg(response.message).into()),
-                            ),
+                            )
                         }
                     } else {
                         crashed_cleanup(
                             &id,
+                            runtime_manager_clone,
                             Error::Runtime(anyhow::anyhow!(
                                 "stop subscribe channel stopped unexpectedly"
                             )),
-                        )
+                        );
                     }
+
                 };
-                let runtime_manager = runtime_manager.clone();
 
+                let runtime_manager = runtime_manager.clone();
                 set.spawn(async move {
                     let parent_cx = global::get_text_map_propagator(|propagator| {
                         propagator.extract(&built.tracing_context)
@@ -180,12 +184,22 @@ fn stopped_cleanup(_id: &Uuid) {
     info!("{}", DEPLOYER_END_MSG_STOPPED);
 }
 
-#[instrument(name = "Cleaning up crashed deployment", skip(_id), fields(deployment_id = %_id, state = %State::Crashed))]
-fn crashed_cleanup(_id: &Uuid, error: impl std::error::Error + 'static) {
+#[instrument(name = "Cleaning up crashed deployment", skip(id, runtime_manager), fields(deployment_id = %id, state = %State::Crashed))]
+fn crashed_cleanup(
+    id: &Uuid,
+    runtime_manager: Arc<Mutex<RuntimeManager>>,
+    error: impl std::error::Error + 'static,
+) {
     error!(
         error = &error as &dyn std::error::Error,
         "{}", DEPLOYER_END_MSG_CRASHED
     );
+
+    // Fire a task which we'll not wait for. This initializes the runtime process killing.
+    let id = *id;
+    tokio::spawn(async move {
+        runtime_manager.lock().await.kill_process(id);
+    });
 }
 
 #[instrument(name = "Cleaning up startup crashed deployment", skip(_id), fields(deployment_id = %_id, state = %State::Crashed))]
@@ -263,7 +277,7 @@ impl Built {
         let runtime_client = runtime_manager
             .lock()
             .await
-            .get_runtime_client(
+            .create_runtime_client(
                 self.id,
                 project_path.as_path(),
                 self.service_name.clone(),
@@ -401,21 +415,34 @@ async fn run(
     deployment_updater: impl DeploymentUpdater,
     cleanup: impl FnOnce(Option<SubscribeStopResponse>) + Send + 'static,
 ) {
-    deployment_updater
-        .set_address(&id, &address)
-        .await
-        .expect("to set deployment address");
+    if let Err(err) = deployment_updater.set_address(&id, &address).await {
+        // Clean up based on a stop response built outside the runtime
+        cleanup(Some(SubscribeStopResponse {
+            reason: 2,
+            message: format!("errored while setting the new deployer address: {}", err),
+        }));
+        return;
+    }
 
     let start_request = tonic::Request::new(StartRequest {
         ip: address.to_string(),
     });
 
     // Subscribe to stop before starting to catch immediate errors
-    let mut stream = runtime_client
+    let mut stream = match runtime_client
         .subscribe_stop(tonic::Request::new(SubscribeStopRequest {}))
         .await
-        .unwrap()
-        .into_inner();
+    {
+        Ok(stream) => stream.into_inner(),
+        Err(err) => {
+            // Clean up based on a stop response built outside the runtime
+            cleanup(Some(SubscribeStopResponse {
+                reason: 2,
+                message: format!("errored while opening the StopSubscribe channel: {}", err),
+            }));
+            return;
+        }
+    };
 
     let response = runtime_client.start(start_request).await;
 
@@ -426,9 +453,14 @@ async fn run(
             }
 
             // Wait for stop reason
-            let reason = stream.message().await.expect("message from tonic stream");
-
-            cleanup(reason);
+            match stream.message().await {
+                Ok(reason) => cleanup(reason),
+                // Stream closed abruptly, most probably runtime crashed.
+                Err(err) => cleanup(Some(SubscribeStopResponse {
+                    reason: 2,
+                    message: format!("runtime StopSubscribe channel errored: {}", err),
+                })),
+            }
         }
         Err(ref status) if status.code() == Code::InvalidArgument => {
             cleanup(Some(SubscribeStopResponse {

deployer/src/runtime_manager.rs

@@ -18,7 +18,7 @@ use shuttle_proto::{
 use shuttle_service::Environment;
 use tokio::{io::AsyncBufReadExt, io::BufReader, process, sync::Mutex};
 use tonic::transport::Channel;
-use tracing::{debug, info, trace};
+use tracing::{debug, error, info, trace};
 use uuid::Uuid;
 
 const MANIFEST_DIR: &str = env!("CARGO_MANIFEST_DIR");
@@ -71,7 +71,7 @@ impl RuntimeManager {
         }))
     }
 
-    pub async fn get_runtime_client(
+    pub async fn create_runtime_client(
         &mut self,
         id: Uuid,
         project_path: &Path,
@@ -170,6 +170,18 @@ impl RuntimeManager {
         Ok(runtime_client)
     }
 
+    pub fn kill_process(&mut self, id: Uuid) {
+        if let Some((mut process, _)) = self.runtimes.lock().unwrap().remove(&id) {
+            match process.start_kill() {
+                Ok(_) => info!(deployment_id = %id, "initiated runtime process killing"),
+                Err(err) => error!(
+                    deployment_id = %id, "failed to start the killing of the runtime: {}",
+                    err
+                ),
+            }
+        }
+    }
+
     /// Send a kill / stop signal for a deployment to its running runtime
     pub async fn kill(&mut self, id: &Uuid) -> bool {
         let value = self.runtimes.lock().unwrap().remove(id);

proto/src/lib.rs

@@ -387,7 +387,7 @@ pub mod logger {
             }
         }
 
-        /// Background task to forward the items ones the batch capacity has been reached
+        /// Background task to forward the items once the batch capacity has been reached
         async fn batch(
             mut inner: I,
             mut rx: mpsc::UnboundedReceiver<I::Item>,