[next] bug: communicating resources (#716)

* refactor: send claim to provisioner

* feat: recording of provisioned resources

This returns provisioned resources from runtime so that they can be
recorded by deployer. Closes #ENG-252.

* refactor: restore resources for idle containers waking up

* tests: next tests using wrong folder

* tests: swap rocket for the more stable poem

* refactor: pass auth_uri to runtime

* misc: update all patches

.circleci/config.yml

@@ -88,14 +88,17 @@ commands:
             cat\<< EOF > ~/.cargo/config.toml
             [patch.crates-io]
             shuttle-service = { path = "$PWD/service" }
+            shuttle-runtime = { path = "$PWD/runtime" }
+
             shuttle-aws-rds = { path = "$PWD/resources/aws-rds" }
             shuttle-persist = { path = "$PWD/resources/persist" }
-            shuttle-runtime = { path = "$PWD/runtime" }
             shuttle-shared-db = { path = "$PWD/resources/shared-db" }
             shuttle-secrets = { path = "$PWD/resources/secrets" }
             shuttle-static-folder = { path = "$PWD/resources/static-folder" }
+
             shuttle-axum = { path = "$PWD/services/shuttle-axum" }
             shuttle-actix-web = { path = "$PWD/services/shuttle-actix-web" }
+            shuttle-next = { path = "$PWD/services/shuttle-next" }
             shuttle-poem = { path = "$PWD/services/shuttle-poem" }
             shuttle-poise = { path = "$PWD/services/shuttle-poise" }
             shuttle-rocket = { path = "$PWD/services/shuttle-rocket" }

CONTRIBUTING.md

@@ -56,13 +56,26 @@ In order to test local changes to the library crates, you may want to add the be
 ```toml
 [patch.crates-io]
 shuttle-service = { path = "[base]/shuttle/service" }
-shuttle-common = { path = "[base]/shuttle/common" }
-shuttle-proto = { path = "[base]/shuttle/proto" }
+shuttle-runtime = { path = "[base]/shuttle/runtime" }
+
 shuttle-aws-rds = { path = "[base]/shuttle/resources/aws-rds" }
 shuttle-persist = { path = "[base]/shuttle/resources/persist" }
 shuttle-shared-db = { path = "[base]/shuttle/resources/shared-db" }
 shuttle-secrets = { path = "[base]/shuttle/resources/secrets" }
 shuttle-static-folder = { path = "[base]/shuttle/resources/static-folder" }
+
+shuttle-axum = { path = "[base]/shuttle/services/shuttle-axum" }
+shuttle-actix-web = { path = "[base]/shuttle/services/shuttle-actix-web" }
+shuttle-next = { path = "[base]/shuttle/services/shuttle-next" }
+shuttle-poem = { path = "[base]/shuttle/services/shuttle-poem" }
+shuttle-poise = { path = "[base]/shuttle/services/shuttle-poise" }
+shuttle-rocket = { path = "[base]/shuttle/services/shuttle-rocket" }
+shuttle-salvo = { path = "[base]/shuttle/services/shuttle-salvo" }
+shuttle-serenity = { path = "[base]/shuttle/services/shuttle-serenity" }
+shuttle-thruster = { path = "[base]/shuttle/services/shuttle-thruster" }
+shuttle-tide = { path = "[base]/shuttle/services/shuttle-tide" }
+shuttle-tower = { path = "[base]/shuttle/services/shuttle-tower" }
+shuttle-warp = { path = "[base]/shuttle/services/shuttle-warp" }
 ```
 
 Before we can login to our local instance of shuttle, we need to create a user.

cargo-shuttle/src/lib.rs

@@ -527,6 +527,7 @@ impl Shuttle {
             is_wasm,
             runtime::StorageManagerType::WorkingDir(working_directory.to_path_buf()),
             &format!("http://localhost:{}", run_args.port + 1),
+            None,
             run_args.port + 2,
             runtime_path,
         )
@@ -543,6 +544,7 @@ impl Shuttle {
                 .into_string()
                 .expect("to convert path to string"),
             service_name: service_name.clone(),
+            resources: Default::default(),
             secrets,
         });
         trace!("loading service");

cargo-shuttle/tests/integration/run.rs

@@ -92,6 +92,7 @@ async fn rocket_secrets() {
 
 // This example uses a shared Postgres. Thus local runs should create a docker container for it.
 #[tokio::test(flavor = "multi_thread")]
+#[ignore]
 async fn rocket_postgres() {
     let url = cargo_shuttle_run("../examples/rocket/postgres", false).await;
     let client = reqwest::Client::new();
@@ -281,7 +282,6 @@ async fn poem_hello_world() {
 
 // This example uses a shared Postgres. Thus local runs should create a docker container for it.
 #[tokio::test(flavor = "multi_thread")]
-#[ignore]
 async fn poem_postgres() {
     let url = cargo_shuttle_run("../examples/poem/postgres", false).await;
     let client = reqwest::Client::new();

common/Cargo.toml

@@ -31,8 +31,8 @@ reqwest = { version = "0.11.13", optional = true }
 rmp-serde = { version = "1.1.1", optional = true }
 rustrict = { version = "0.5.5", optional = true }
 serde = { workspace = true, features = ["derive", "std"] }
-serde_json = { workspace = true, optional = true }
-strum = { workspace = true, features = ["derive"], optional = true }
+serde_json = { workspace = true }
+strum = { workspace = true, features = ["derive"] }
 thiserror = { workspace = true, optional = true }
 tonic = { version = "0.8.3", optional = true }
 tower = { workspace = true, optional = true }
@@ -47,10 +47,10 @@ uuid = { workspace = true, features = ["v4", "serde"], optional = true }
 backend = ["async-trait", "axum/matched-path", "claims", "hyper/client", "opentelemetry-otlp", "thiserror", "tower-http", "tracing-subscriber/env-filter", "tracing-subscriber/fmt", "ttl_cache"]
 claims = ["bytes", "chrono/clock", "headers", "http", "http-body", "jsonwebtoken", "opentelemetry", "opentelemetry-http", "pin-project", "tower", "tracing", "tracing-opentelemetry"]
 display = ["chrono/clock", "comfy-table", "crossterm"]
-error = ["prost-types", "serde_json", "thiserror", "uuid"]
-models = ["anyhow", "async-trait", "display", "http", "reqwest", "serde_json", "service"]
-service = ["chrono/serde", "once_cell", "rustrict", "serde/derive", "strum", "uuid"]
-tracing = ["serde_json"]
+error = ["prost-types", "thiserror", "uuid"]
+models = ["anyhow", "async-trait", "display", "http", "reqwest", "service"]
+service = ["chrono/serde", "once_cell", "rustrict", "serde/derive", "uuid"]
+tracing = []
 wasm = ["chrono/clock", "http-serde", "http", "rmp-serde", "tracing", "tracing-subscriber"]
 
 [dev-dependencies]
@@ -59,7 +59,6 @@ base64 = "0.13.1"
 cap-std = "1.0.2"
 hyper = { workspace = true }
 ring = { workspace = true }
-serde_json = { workspace = true }
 tokio = { version = "1.22.0", features = ["macros", "rt-multi-thread"] }
 tower = { workspace = true, features = ["util"] }
 tracing-fluent-assertions = "0.3.0"

common/src/database.rs

@@ -3,14 +3,14 @@ use std::fmt::Display;
 use serde::{Deserialize, Serialize};
 use strum::Display;
 
-#[derive(Clone, Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize, Eq, PartialEq)]
 #[serde(rename_all = "lowercase")]
 pub enum Type {
     AwsRds(AwsRdsEngine),
     Shared(SharedEngine),
 }
 
-#[derive(Clone, Debug, Deserialize, Display, Serialize)]
+#[derive(Clone, Debug, Deserialize, Display, Serialize, Eq, PartialEq)]
 #[serde(rename_all = "lowercase")]
 #[strum(serialize_all = "lowercase")]
 pub enum AwsRdsEngine {
@@ -19,7 +19,7 @@ pub enum AwsRdsEngine {
     MariaDB,
 }
 
-#[derive(Clone, Debug, Deserialize, Display, Serialize)]
+#[derive(Clone, Debug, Deserialize, Display, Serialize, Eq, PartialEq)]
 #[serde(rename_all = "lowercase")]
 #[strum(serialize_all = "lowercase")]
 pub enum SharedEngine {

common/src/lib.rs

@@ -2,7 +2,6 @@
 pub mod backends;
 #[cfg(feature = "claims")]
 pub mod claims;
-#[cfg(feature = "service")]
 pub mod database;
 #[cfg(feature = "service")]
 pub mod deployment;
@@ -12,6 +11,7 @@ pub mod log;
 pub mod models;
 #[cfg(feature = "service")]
 pub mod project;
+pub mod resource;
 #[cfg(feature = "service")]
 pub mod storage_manager;
 #[cfg(feature = "tracing")]

common/src/models/mod.rs

@@ -1,7 +1,6 @@
 pub mod deployment;
 pub mod error;
 pub mod project;
-pub mod resource;
 pub mod secret;
 pub mod service;
 pub mod stats;

common/src/models/service.rs

@@ -1,6 +1,6 @@
 use crate::{
-    models::{deployment, resource, resource::ResourceInfo, secret},
-    DatabaseReadyInfo,
+    models::{deployment, secret},
+    resource::{self, ResourceInfo},
 };
 
 use comfy_table::{
@@ -34,12 +34,6 @@ pub struct Summary {
     pub uri: String,
 }
 
-impl ResourceInfo for DatabaseReadyInfo {
-    fn connection_string_public(&self) -> String {
-        self.connection_string_public()
-    }
-}
-
 impl Display for Detailed {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let deploys = get_deployments_table(&self.deployments, &self.name);

common/src/models/resource.rs → common/src/resource.rs

@@ -2,13 +2,11 @@ use std::fmt::Display;
 
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use uuid::Uuid;
 
 use crate::{database, DatabaseReadyInfo};
 
-#[derive(Deserialize, Serialize)]
+#[derive(Clone, Deserialize, Serialize)]
 pub struct Response {
-    pub service_id: Uuid,
     pub r#type: Type,
     pub data: Value,
 }
@@ -17,9 +15,22 @@ pub struct Response {
 pub trait ResourceInfo {
     /// String to connect to this resource from a public location
     fn connection_string_public(&self) -> String;
+
+    /// String to connect to this resource from within shuttle
+    fn connection_string_private(&self) -> String;
+}
+
+impl ResourceInfo for DatabaseReadyInfo {
+    fn connection_string_public(&self) -> String {
+        self.connection_string_public()
+    }
+
+    fn connection_string_private(&self) -> String {
+        self.connection_string_private()
+    }
 }
 
-#[derive(Clone, Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize, Eq, PartialEq)]
 #[serde(rename_all = "lowercase")]
 pub enum Type {
     Database(database::Type),
@@ -33,6 +44,14 @@ impl Response {
             }
         }
     }
+
+    pub fn into_bytes(self) -> Vec<u8> {
+        serde_json::to_vec(&self).expect("to turn resource into a vec")
+    }
+
+    pub fn from_bytes(bytes: Vec<u8>) -> Self {
+        serde_json::from_slice(&bytes).expect("to turn bytes into a resource")
+    }
 }
 
 impl Display for Type {

deployer/src/deployment/deploy_layer.rs

@@ -310,7 +310,10 @@ mod tests {
         time::Duration,
     };
 
-    use crate::{persistence::DeploymentUpdater, RuntimeManager};
+    use crate::{
+        persistence::{DeploymentUpdater, Resource, ResourceManager},
+        RuntimeManager,
+    };
     use async_trait::async_trait;
     use axum::body::Bytes;
     use ctor::ctor;
@@ -467,7 +470,7 @@ mod tests {
         let path = tmp_dir.into_path();
         let (tx, _rx) = crossbeam_channel::unbounded();
 
-        RuntimeManager::new(path, format!("http://{}", provisioner_addr), tx)
+        RuntimeManager::new(path, format!("http://{}", provisioner_addr), None, tx)
     }
 
     #[async_trait::async_trait]
@@ -553,6 +556,21 @@ mod tests {
         }
     }
 
+    #[derive(Clone)]
+    struct StubResourceManager;
+
+    #[async_trait::async_trait]
+    impl ResourceManager for StubResourceManager {
+        type Err = std::io::Error;
+
+        async fn insert_resource(&self, _resource: &Resource) -> Result<(), Self::Err> {
+            Ok(())
+        }
+        async fn get_resources(&self, _service_id: &Uuid) -> Result<Vec<Resource>, Self::Err> {
+            Ok(Vec::new())
+        }
+    }
+
     async fn test_states(id: &Uuid, expected_states: Vec<StateLog>) {
         loop {
             let states = RECORDER.lock().unwrap().get_deployment_states(id);
@@ -861,6 +879,7 @@ mod tests {
             .active_deployment_getter(StubActiveDeploymentGetter)
             .artifacts_path(PathBuf::from("/tmp"))
             .secret_getter(StubSecretGetter)
+            .resource_manager(StubResourceManager)
             .runtime(get_runtime_manager())
             .deployment_updater(StubDeploymentUpdater)
             .queue_client(StubBuildQueueClient)

deployer/src/deployment/mod.rs

@@ -12,7 +12,7 @@ use tracing::{instrument, Span};
 use tracing_opentelemetry::OpenTelemetrySpanExt;
 
 use crate::{
-    persistence::{DeploymentUpdater, SecretGetter, SecretRecorder, State},
+    persistence::{DeploymentUpdater, ResourceManager, SecretGetter, SecretRecorder, State},
     RuntimeManager,
 };
 use tokio::sync::{mpsc, Mutex};
@@ -23,24 +23,26 @@ use self::{deploy_layer::LogRecorder, gateway_client::BuildQueueClient};
 const QUEUE_BUFFER_SIZE: usize = 100;
 const RUN_BUFFER_SIZE: usize = 100;
 
-pub struct DeploymentManagerBuilder<LR, SR, ADG, DU, SG, QC> {
+pub struct DeploymentManagerBuilder<LR, SR, ADG, DU, SG, RM, QC> {
     build_log_recorder: Option<LR>,
     secret_recorder: Option<SR>,
     active_deployment_getter: Option<ADG>,
     artifacts_path: Option<PathBuf>,
     runtime_manager: Option<Arc<Mutex<RuntimeManager>>>,
     deployment_updater: Option<DU>,
     secret_getter: Option<SG>,
+    resource_manager: Option<RM>,
     queue_client: Option<QC>,
 }
 
-impl<LR, SR, ADG, DU, SG, QC> DeploymentManagerBuilder<LR, SR, ADG, DU, SG, QC>
+impl<LR, SR, ADG, DU, SG, RM, QC> DeploymentManagerBuilder<LR, SR, ADG, DU, SG, RM, QC>
 where
     LR: LogRecorder,
     SR: SecretRecorder,
     ADG: ActiveDeploymentsGetter,
     DU: DeploymentUpdater,
     SG: SecretGetter,
+    RM: ResourceManager,
     QC: BuildQueueClient,
 {
     pub fn build_log_recorder(mut self, build_log_recorder: LR) -> Self {
@@ -79,6 +81,12 @@ where
         self
     }
 
+    pub fn resource_manager(mut self, resource_manager: RM) -> Self {
+        self.resource_manager = Some(resource_manager);
+
+        self
+    }
+
     pub fn runtime(mut self, runtime_manager: Arc<Mutex<RuntimeManager>>) -> Self {
         self.runtime_manager = Some(runtime_manager);
 
@@ -110,6 +118,7 @@ where
             .deployment_updater
             .expect("a deployment updater to be set");
         let secret_getter = self.secret_getter.expect("a secret getter to be set");
+        let resource_manager = self.resource_manager.expect("a resource manager to be set");
 
         let (queue_send, queue_recv) = mpsc::channel(QUEUE_BUFFER_SIZE);
         let (run_send, run_recv) = mpsc::channel(RUN_BUFFER_SIZE);
@@ -132,6 +141,7 @@ where
             deployment_updater,
             active_deployment_getter,
             secret_getter,
+            resource_manager,
             storage_manager.clone(),
         ));
 
@@ -169,7 +179,8 @@ pub struct DeploymentManager {
 impl DeploymentManager {
     /// Create a new deployment manager. Manages one or more 'pipelines' for
     /// processing service building, loading, and deployment.
-    pub fn builder<LR, SR, ADG, DU, SG, QC>() -> DeploymentManagerBuilder<LR, SR, ADG, DU, SG, QC> {
+    pub fn builder<LR, SR, ADG, DU, SG, RM, QC>(
+    ) -> DeploymentManagerBuilder<LR, SR, ADG, DU, SG, RM, QC> {
         DeploymentManagerBuilder {
             build_log_recorder: None,
             secret_recorder: None,
@@ -178,6 +189,7 @@ impl DeploymentManager {
             runtime_manager: None,
             deployment_updater: None,
             secret_getter: None,
+            resource_manager: None,
             queue_client: None,
         }
     }