feat(deployer): load phase caching, automatic startup (#1640)

* feat(deployer): cache resources after provisioning for use on wakeups

* fix(deployer): fix old deployment handling, always start previous running deploy on wakeup

* nit: comment

* fix and speed up depleoyer integration

* fix deployer integration

Author: jonaro00

Makefile

@@ -134,14 +134,14 @@ DOCKER_COMPOSE_ENV=\
 	SHUTTLE_ENV=$(SHUTTLE_ENV)\
 	SHUTTLE_SERVICE_VERSION=$(SHUTTLE_SERVICE_VERSION)
 
-.PHONY: clean cargo-clean images the-shuttle-images shuttle-% postgres otel deploy test docker-compose.rendered.yml up down
+.PHONY: clean deep-clean images the-shuttle-images shuttle-% postgres otel deploy test docker-compose.rendered.yml up down
 
 clean:
 	rm .shuttle-*
 	rm docker-compose.rendered.yml
 
-cargo-clean:
-	find . -type d \( -name target -or -name .shuttle-executables \) | xargs rm -rf
+deep-clean:
+	find . -type d \( -name target -or -name .shuttle-executables -or -name node_modules \) | xargs rm -rf
 
 images: the-shuttle-images postgres otel
 

deployer/src/deployment/queue.rs

@@ -245,7 +245,7 @@ impl Queued {
             project_id: self.project_id,
             tracing_context: Default::default(),
             is_next,
-            claim: self.claim,
+            claim: Some(self.claim),
             secrets,
         };
 

deployer/src/deployment/run.rs

@@ -150,7 +150,7 @@ pub async fn task(
     }
 }
 
-#[instrument(skip(active_deployment_getter, deployment_id, runtime_manager))]
+#[instrument(skip_all)]
 async fn kill_old_deployments(
     service_id: Ulid,
     deployment_id: Uuid,
@@ -230,16 +230,14 @@ pub struct Built {
     pub project_id: Ulid,
     pub tracing_context: HashMap<String, String>,
     pub is_next: bool,
-    pub claim: Claim,
+    /// must be set if this run will perform requests to backends
+    pub claim: Option<Claim>,
     pub secrets: HashMap<String, String>,
 }
 
 impl Built {
-    #[instrument(
-        name = "Loading resources",
-        skip(self, resource_manager, runtime_manager, kill_old_deployments, cleanup, provisioner_client),
-        fields(deployment_id = %self.id, state = %State::Loading)
-    )]
+    #[instrument(name = "Loading resources", skip_all, fields(deployment_id = %self.id, state = %State::Loading))]
+    #[allow(clippy::too_many_arguments)]
     pub async fn handle(
         self,
         mut resource_manager: impl ResourceManager,
@@ -256,6 +254,9 @@ impl Built {
         let executable_path = project_path
             .join(EXECUTABLE_DIRNAME)
             .join(self.id.to_string());
+        let cached_resources_path = project_path
+            .join(EXECUTABLE_DIRNAME)
+            .join(format!("{}.resources", self.id));
 
         // Let the runtime expose its user HTTP port on port 8000
         let address = SocketAddr::new(Ipv4Addr::UNSPECIFIED.into(), 8000);
@@ -279,62 +280,87 @@ impl Built {
             .await
             .map_err(Error::Runtime)?;
 
-        info!("Loading resources");
+        // Check for cached resources for this deployment id. This only succeeds on wakeup from idle or project restart.
+        let resources = if let Some(bytes) = std::fs::read(&cached_resources_path)
+            .ok()
+            .and_then(|bytes| serde_json::from_slice(bytes.as_slice()).ok())
+        {
+            info!("Using cached resources");
 
-        let mut new_secrets = self.secrets;
-        let prev_resources = resource_manager
-            .get_resources(&self.service_id, self.claim.clone())
-            .await
-            .map_err(|err| Error::Load(err.to_string()))?
-            .resources
-            .into_iter()
-            .map(resource::Response::try_from)
-            // Ignore and trace the errors for resources with corrupted data, returning just the valid resources.
-            // TODO: investigate how the resource data can get corrupted.
-            .filter_map(|resource| {
-                resource
-                    .map_err(|err| {
-                        error!(error = ?err, "failed to parse resource data");
-                    })
-                    .ok()
-            })
-            // inject old secrets into the secrets added in this deployment
-            .inspect(|r| {
-                if r.r#type == shuttle_common::resource::Type::Secrets {
-                    match serde_json::from_value::<SecretStore>(r.data.clone()) {
-                        Ok(ss) => {
-                            // Combine old and new, but insert old first so that new ones override.
-                            let mut combined = HashMap::from_iter(ss.into_iter());
-                            combined.extend(new_secrets.clone().into_iter());
-                            new_secrets = combined;
-                        }
-                        Err(err) => {
-                            error!(error = ?err, "failed to parse old secrets data");
+            bytes
+        }
+        // Default case for handling resources and provisioning
+        else {
+            info!("Loading resources");
+            let claim = self
+                .claim
+                .expect("claim must be present when loading resources");
+
+            let mut new_secrets = self.secrets;
+            let prev_resources = resource_manager
+                .get_resources(&self.service_id, claim.clone())
+                .await
+                .map_err(|err| Error::Load(err.to_string()))?
+                .resources
+                .into_iter()
+                .map(resource::Response::try_from)
+                // Ignore and trace the errors for resources with corrupted data, returning just the valid resources.
+                // TODO: investigate how the resource data can get corrupted.
+                .filter_map(|resource| {
+                    resource
+                        .map_err(|err| {
+                            error!(error = ?err, "failed to parse resource data");
+                        })
+                        .ok()
+                })
+                // inject old secrets into the secrets added in this deployment
+                .inspect(|r| {
+                    if r.r#type == shuttle_common::resource::Type::Secrets {
+                        match serde_json::from_value::<SecretStore>(r.data.clone()) {
+                            Ok(ss) => {
+                                // Combine old and new, but insert old first so that new ones override.
+                                let mut combined = HashMap::from_iter(ss.into_iter());
+                                combined.extend(new_secrets.clone().into_iter());
+                                new_secrets = combined;
+                            }
+                            Err(err) => {
+                                error!(error = ?err, "failed to parse old secrets data");
+                            }
                         }
                     }
-                }
-            })
-            .collect::<Vec<_>>();
-
-        let resources = load(
-            self.service_name.clone(),
-            runtime_client.clone(),
-            &new_secrets,
-        )
-        .await?;
-
-        let resources = provision(
-            self.service_name.as_str(),
-            self.service_id,
-            provisioner_client,
-            resource_manager,
-            self.claim,
-            prev_resources,
-            resources,
-            new_secrets,
-        )
-        .await
-        .map_err(Error::Provision)?;
+                })
+                .collect::<Vec<_>>();
+
+            let resources = load(
+                self.service_name.clone(),
+                runtime_client.clone(),
+                &new_secrets,
+            )
+            .await?;
+
+            let resources = provision(
+                self.service_name.as_str(),
+                self.service_id,
+                provisioner_client,
+                resource_manager,
+                claim,
+                prev_resources,
+                resources,
+                new_secrets,
+            )
+            .await
+            .map_err(Error::Provision)?;
+
+            // cache the final resources output for use in wakeups
+            // this should only happen on deployment, and not on wakeups
+            std::fs::write(
+                &cached_resources_path,
+                serde_json::to_vec(&resources).expect("resources to serialize"),
+            )
+            .map_err(|_| Error::Load("Failed to save resource cache".into()))?;
+
+            resources
+        };
 
         kill_old_deployments.await?;
 

deployer/src/handlers/mod.rs

@@ -39,7 +39,7 @@ use shuttle_proto::logger::LogsRequest;
 
 use crate::persistence::{Deployment, Persistence, State};
 use crate::{
-    deployment::{Built, DeploymentManager, Queued},
+    deployment::{DeploymentManager, Queued},
     persistence::resource::ResourceManager,
 };
 pub use {self::error::Error, self::error::Result, self::local::set_jwt_bearer};
@@ -430,32 +430,10 @@ pub async fn delete_deployment(
     }
 }
 
-#[instrument(skip_all, fields(shuttle.project.name = %project_name, %deployment_id))]
-pub async fn start_deployment(
-    Extension(persistence): Extension<Persistence>,
-    Extension(deployment_manager): Extension<DeploymentManager>,
-    Extension(claim): Extension<Claim>,
-    Extension(project_id): Extension<Ulid>,
-    CustomErrorPath((project_name, deployment_id)): CustomErrorPath<(String, Uuid)>,
-) -> Result<()> {
-    if let Some(deployment) = persistence.get_runnable_deployment(&deployment_id).await? {
-        let built = Built {
-            id: deployment.id,
-            service_name: deployment.service_name,
-            service_id: deployment.service_id,
-            project_id,
-            tracing_context: Default::default(),
-            is_next: deployment.is_next,
-            claim,
-            secrets: Default::default(),
-        };
-        deployment_manager.run_push(built).await;
-
-        Ok(())
-    } else {
-        Err(Error::NotFound("deployment not found".to_string()))
-    }
-}
+/// Deprecated.
+/// Now always starts the last running deployment on start up / wake up.
+/// Kept around for compatibility.
+pub async fn start_deployment() {}
 
 #[instrument(skip_all, fields(shuttle.project.name = %project_name, %deployment_id))]
 pub async fn get_logs(

deployer/src/lib.rs

@@ -17,7 +17,7 @@ mod runtime_manager;
 
 pub use crate::args::Args;
 pub use crate::deployment::state_change_layer::StateChangeLayer;
-use crate::deployment::DeploymentManager;
+use crate::deployment::{Built, DeploymentManager};
 use shuttle_common::backends::client::gateway;
 
 const VERSION: &str = env!("CARGO_PKG_VERSION");
@@ -29,6 +29,9 @@ pub async fn start(
     log_fetcher: logger::Client,
     args: Args,
 ) {
+    let project_id = Ulid::from_string(args.project_id.as_str())
+        .expect("to have a valid ULID as project_id arg");
+
     // when _set is dropped once axum exits, the deployment tasks will be aborted.
     let deployment_manager = DeploymentManager::builder()
         .build_log_recorder(log_recorder)
@@ -47,18 +50,32 @@ pub async fn start(
 
     persistence.cleanup_invalid_states().await.unwrap();
 
-    let runnable_deployments = persistence.get_all_runnable_deployments().await.unwrap();
-    info!(count = %runnable_deployments.len(), "stopping all but last running deploy");
-
-    // Make sure we don't stop the last running deploy. This works because they are returned in descending order.
-    let project_id = Ulid::from_string(args.project_id.as_str())
-        .expect("to have a valid ULID as project_id arg");
-    for existing_deployment in runnable_deployments.into_iter().skip(1) {
+    let deployments = persistence.get_all_runnable_deployments().await.unwrap();
+    info!(count = %deployments.len(), "Deployments considered in the running state");
+    // This works because they are returned in descending order.
+    let mut deployments = deployments.into_iter();
+    let last_running_deployment = deployments.next();
+    info!("Marking all but last running deployment as stopped");
+    for older_deployment in deployments {
         persistence
-            .stop_running_deployment(existing_deployment)
+            .stop_running_deployment(older_deployment)
             .await
             .unwrap();
     }
+    if let Some(deployment) = last_running_deployment {
+        info!("Starting up last running deployment");
+        let built = Built {
+            id: deployment.id,
+            service_name: deployment.service_name,
+            service_id: deployment.service_id,
+            project_id,
+            tracing_context: Default::default(),
+            is_next: deployment.is_next,
+            claim: None,
+            secrets: Default::default(),
+        };
+        deployment_manager.run_push(built).await;
+    }
 
     let mut builder = handlers::RouterBuilder::new(
         persistence,

deployer/src/persistence/mod.rs

@@ -222,7 +222,7 @@ impl Persistence {
     }
 
     pub async fn get_active_deployment(&self, service_id: &Ulid) -> Result<Option<Deployment>> {
-        sqlx::query_as("SELECT * FROM deployments WHERE service_id = ? AND state = ?")
+        sqlx::query_as("SELECT * FROM deployments WHERE service_id = ? AND state = ? ORDER BY last_update DESC")
             .bind(service_id.to_string())
             .bind(State::Running)
             .fetch_optional(&self.pool)
@@ -301,24 +301,6 @@ impl Persistence {
         .map_err(Error::from)
     }
 
-    /// Gets a deployment if it is runnable
-    pub async fn get_runnable_deployment(&self, id: &Uuid) -> Result<Option<DeploymentRunnable>> {
-        sqlx::query_as(
-            r#"SELECT d.id, service_id, s.name AS service_name, d.is_next
-                FROM deployments AS d
-                JOIN services AS s ON s.id = d.service_id
-                WHERE state IN (?, ?, ?)
-                AND d.id = ?"#,
-        )
-        .bind(State::Running)
-        .bind(State::Stopped)
-        .bind(State::Completed)
-        .bind(id)
-        .fetch_optional(&self.pool)
-        .await
-        .map_err(Error::from)
-    }
-
     pub async fn stop_running_deployment(&self, deployable: DeploymentRunnable) -> Result<()> {
         update_deployment(
             &self.pool,
@@ -931,20 +913,6 @@ mod tests {
             p.insert_deployment(deployment).await.unwrap();
         }
 
-        let runnable = p.get_runnable_deployment(&id_1).await.unwrap();
-        assert_eq!(
-            runnable,
-            Some(DeploymentRunnable {
-                id: id_1,
-                service_name: "foo".to_string(),
-                service_id: foo_id,
-                is_next: false,
-            })
-        );
-
-        let runnable = p.get_runnable_deployment(&id_crashed).await.unwrap();
-        assert_eq!(runnable, None);
-
         let runnable = p.get_all_runnable_deployments().await.unwrap();
         assert_eq!(
             runnable,