Merge pull request #176 from tower/develop

v0.3.43 release

Author: bradhe

Cargo.lock

@@ -4,7 +4,7 @@ resolver = "2"
 
 [workspace.package]
 edition = "2021"
-version = "0.3.42"
+version = "0.3.43"
 description = "Tower is the best way to host Python data apps in production"
 rust-version = "1.81"
 authors = ["Brad Heller <brad@tower.dev>"]
@@ -15,6 +15,7 @@ repository = "https://github.com/tower/tower-cli"
 aes-gcm = "0.10"
 anyhow = "1.0.95"
 async-compression = { version = "0.4", features = ["tokio", "gzip"] }
+async-trait = "0.1.89"
 async_zip = { version = "0.0.16", features = ["tokio", "tokio-fs", "deflate"] }
 axum = "0.8.4"
 base64 = "0.22"

Cargo.toml

@@ -21,7 +21,9 @@ fn extract_aid_from_jwt(jwt: &str) -> Option<String> {
     let payload = parts[1];
     let decoded = URL_SAFE_NO_PAD.decode(payload).ok()?;
     let json: serde_json::Value = serde_json::from_slice(&decoded).ok()?;
-    json.get("https://tower.dev/aid")?.as_str().map(String::from)
+    json.get("https://tower.dev/aid")?
+        .as_str()
+        .map(String::from)
 }
 
 const DEFAULT_TOWER_URL: &str = "https://api.tower.dev";

crates/config/src/session.rs

@@ -367,6 +367,17 @@ pub enum LogStreamError {
     Unknown,
 }
 
+impl std::fmt::Display for LogStreamError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            LogStreamError::Reqwest(err) => write!(f, "{err}"),
+            LogStreamError::Unknown => write!(f, "unknown log stream error"),
+        }
+    }
+}
+
+impl std::error::Error for LogStreamError {}
+
 impl From<reqwest_eventsource::CannotCloneRequestError> for LogStreamError {
     fn from(err: reqwest_eventsource::CannotCloneRequestError) -> Self {
         debug!("Failed to clone request {:?}", err);
@@ -399,10 +410,30 @@ async fn drain_run_logs_stream(mut source: EventSource, tx: mpsc::Sender<LogStre
                     }
                     "warning" => {
                         let event_warning = serde_json::from_str(&message.data);
-                        if let Ok(event) = event_warning {
-                            tx.send(LogStreamEvent::EventWarning(event)).await.ok();
-                        } else {
-                            debug!("Failed to parse warning message: {:?}", message.data);
+                        match event_warning {
+                            Ok(event) => {
+                                tx.send(LogStreamEvent::EventWarning(event)).await.ok();
+                            }
+                            Err(err) => {
+                                let warning_data = serde_json::from_str(&message.data);
+                                match warning_data {
+                                    Ok(data) => {
+                                        let event = tower_api::models::EventWarning {
+                                            data,
+                                            event: tower_api::models::event_warning::Event::Warning,
+                                            id: None,
+                                            retry: None,
+                                        };
+                                        tx.send(LogStreamEvent::EventWarning(event)).await.ok();
+                                    }
+                                    Err(_) => {
+                                        debug!(
+                                            "Failed to parse warning message: {:?}. Error: {}",
+                                            message.data, err
+                                        );
+                                    }
+                                }
+                            }
                         }
                     }
                     _ => {