fix: rds deploy crash (shuttle-hq#1068)

* misc: debug prints

* fix: debug rds bug

* feat: get state from runtime::LogItem

* feat: debug logs in client

* refactor: improve run.rs debug logs

* refactor: comment, unwrap to expect

* refactor: debug string in cargo-shuttle

Co-authored-by: Iulian Barbu <14218860+iulianbarbu@users.noreply.github.com>

---------

Co-authored-by: Iulian Barbu <14218860+iulianbarbu@users.noreply.github.com>

Makefile

@@ -24,7 +24,7 @@ BUILDX_FLAGS=$(BUILDX_OP) $(PLATFORM_FLAGS) $(CACHE_FLAGS)
 # ensuring all user crates are compiled with the same rustc toolchain
 RUSTUP_TOOLCHAIN=1.70.0
 
-TAG?=$(shell git describe --tags)
+TAG?=$(shell git describe --tags --abbrev=0)
 BACKEND_TAG?=$(TAG)
 DEPLOYER_TAG?=$(TAG)
 PROVISIONER_TAG?=$(TAG)

cargo-shuttle/src/lib.rs

@@ -1041,10 +1041,9 @@ impl Shuttle {
 
                             return Ok(CommandOutcome::DeploymentFailure);
                         }
-                        shuttle_common::deployment::State::Running
-                        | shuttle_common::deployment::State::Completed
-                        | shuttle_common::deployment::State::Stopped
-                        | shuttle_common::deployment::State::Unknown => {
+                        // Break on remaining end states: Running, Stopped, Completed or Unknown.
+                        end_state => {
+                            debug!(state = %end_state, "received end state, breaking deployment stream");
                             break;
                         }
                     };
@@ -1100,9 +1099,12 @@ impl Shuttle {
                         "State: Crashed - Deployment crashed after startup.".red()
                     );
                 }
-                _ => println!(
-                    "Deployment encountered an unexpected error - Please create a ticket to report this."
-                ),
+                state => {
+                    debug!("deployment logs stream received state: {state} when it expected to receive running state");
+                    println!(
+                    "Deployment entered an unexpected state - Please create a ticket to report this."
+                );
+                }
             }
 
             println!();

common/src/lib.rs

@@ -108,6 +108,8 @@ pub enum ParseError {
     Timestamp(#[from] prost_types::TimestampError),
     #[error("failed to parse serde: {0}")]
     Serde(#[from] serde_json::Error),
+    #[error("failed to parse state: {0}")]
+    State(String),
 }
 
 /// Holds the input for a DB resource

deployer/src/deployment/deploy_layer.rs

@@ -118,7 +118,7 @@ impl TryFrom<runtime::LogItem> for Log {
     fn try_from(log: runtime::LogItem) -> Result<Self, Self::Error> {
         Ok(Self {
             id: Default::default(),
-            state: State::Running,
+            state: State::from_str(&log.state).map_err(|err| ParseError::State(err.to_string()))?,
             level: runtime::LogLevel::from_i32(log.level)
                 .unwrap_or_default()
                 .into(),

deployer/src/deployment/run.rs

@@ -333,9 +333,10 @@ async fn load(
         load_request.extensions_mut().insert(claim);
     }
 
-    debug!("loading service");
+    debug!(service_name = %service_name, "loading service");
     let response = runtime_client.load(load_request).await;
 
+    debug!(service_name = %service_name, "service loaded");
     match response {
         Ok(response) => {
             let response = response.into_inner();

proto/runtime.proto

@@ -90,6 +90,7 @@ message LogItem {
   optional uint32 line = 6;
   string target = 7;
   bytes fields = 8;
+  string state = 9;
 }
 
 enum LogLevel {

proto/src/generated/runtime.rs

@@ -85,6 +85,8 @@ pub struct LogItem {
     pub target: ::prost::alloc::string::String,
     #[prost(bytes = "vec", tag = "8")]
     pub fields: ::prost::alloc::vec::Vec<u8>,
+    #[prost(string, tag = "9")]
+    pub state: ::prost::alloc::string::String,
 }
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
 #[repr(i32)]

proto/src/lib.rs

@@ -106,6 +106,7 @@ pub mod runtime {
     use prost_types::Timestamp;
     use shuttle_common::{
         claims::{ClaimLayer, ClaimService, InjectPropagation, InjectPropagationLayer},
+        deployment::State,
         ParseError,
     };
     use tokio::process;
@@ -178,6 +179,9 @@ pub mod runtime {
                 line,
                 target: log.target,
                 fields: log.fields,
+                // We can safely assume the state received from shuttle-next to be running,
+                // it will not currently load any resources.
+                state: State::Running.to_string(),
             }
         }
     }

runtime/src/logger.rs

@@ -1,8 +1,11 @@
-use std::time::SystemTime;
+use std::{
+    sync::{Arc, Mutex},
+    time::SystemTime,
+};
 
 use chrono::Utc;
 use prost_types::Timestamp;
-use shuttle_common::tracing::JsonVisitor;
+use shuttle_common::{deployment::State, tracing::JsonVisitor};
 use shuttle_proto::runtime::{LogItem, LogLevel};
 use tokio::sync::mpsc::UnboundedSender;
 use tracing::{
@@ -13,11 +16,15 @@ use tracing_subscriber::Layer;
 
 pub struct Logger {
     tx: UnboundedSender<LogItem>,
+    state: Arc<Mutex<State>>,
 }
 
 impl Logger {
     pub fn new(tx: UnboundedSender<LogItem>) -> Self {
-        Self { tx }
+        Self {
+            tx,
+            state: Arc::new(Mutex::new(State::Loading)),
+        }
     }
 }
 
@@ -61,6 +68,11 @@ where
                     .target
                     .unwrap_or_else(|| metadata.target().to_string()),
                 fields: serde_json::to_vec(&visitor.fields).unwrap(),
+                state: self
+                    .state
+                    .lock()
+                    .expect("state lock should not be poisoned")
+                    .to_string(),
             }
         };
 
@@ -72,6 +84,12 @@ where
         event: &tracing::Event<'_>,
         _ctx: tracing_subscriber::layer::Context<'_, S>,
     ) {
+        if event.metadata().name() == "start" {
+            *self
+                .state
+                .lock()
+                .expect("state lock should not be poisoned") = State::Running;
+        }
         let datetime = Utc::now();
 
         let item = {
@@ -89,6 +107,11 @@ where
                     .target
                     .unwrap_or_else(|| metadata.target().to_string()),
                 fields: serde_json::to_vec(&visitor.fields).unwrap(),
+                state: self
+                    .state
+                    .lock()
+                    .expect("state lock should not be poisoned")
+                    .to_string(),
             }
         };