fix: drop stale ssh data channels

src/components/backend-core/src/ssh_bridge.rs

@@ -13,7 +13,7 @@ use crate::db::provider_supports;
 use crate::http::json_response;
 use crate::ssh_data_channel::{
     active_ssh_data_channel_count, claim_ssh_data_channel, close_ssh_data_session,
-    idle_ssh_data_channel_count, send_ssh_data_frame,
+    idle_ssh_data_channel_count, remove_ssh_data_channel, send_ssh_data_frame,
 };
 use crate::state::{AppState, SshClientConnection, SshDataFrame};
 
@@ -285,20 +285,53 @@ pub async fn forward_ssh_provider_message(state: &Arc<AppState>, message: &Value
         .get("sessionId")
         .and_then(Value::as_str)
         .unwrap_or("");
-    let clients = state.active_ssh_clients.read().await;
-    let Some(client) = clients.get(session_id) else {
+    let message_type = message.get("type").and_then(Value::as_str).unwrap_or("");
+    let message_text = message
+        .get("message")
+        .and_then(Value::as_str)
+        .unwrap_or("ssh session error");
+    let client = {
+        let clients = state.active_ssh_clients.read().await;
+        clients.get(session_id).map(|client| {
+            (
+                client.provider_id.clone(),
+                client.data_channel_id.clone(),
+                client.sender.clone(),
+            )
+        })
+    };
+    let Some((provider_id, data_channel_id, client_sender)) = client else {
         state.log("warn", "ssh_client_missing", Some(json!({ "providerId": message.get("providerId").cloned().unwrap_or(Value::Null), "sessionId": session_id, "type": message.get("type").cloned().unwrap_or(Value::Null) })));
         return;
     };
-    let message_type = message.get("type").and_then(Value::as_str).unwrap_or("");
-    let outbound = json!({
+    let stale_channel_not_ready = message_type == "host_ssh_error"
+        && message_text.contains("requested ssh tcp data channel is not ready");
+    if message_type == "host_ssh_error" {
+        let removed = remove_ssh_data_channel(state, &provider_id, &data_channel_id).await;
+        state.log(
+            if removed { "warn" } else { "info" },
+            "ssh_data_channel_removed_after_control_error",
+            Some(json!({
+                "providerId": provider_id.clone(),
+                "sessionId": session_id,
+                "dataChannelId": data_channel_id.clone(),
+                "removed": removed,
+                "message": message_text,
+            })),
+        );
+    }
+    let mut outbound = json!({
         "type": "ssh.error",
-        "message": message.get("message").and_then(Value::as_str).unwrap_or("ssh session error"),
+        "message": message_text,
         "transport": "tcp-pool",
         "controlFallback": true,
     });
-    let _ = client.sender.send(Message::Text(outbound.to_string()));
-    drop(clients);
+    if stale_channel_not_ready {
+        outbound["failureKind"] = Value::String("provider-data-channel-missing".to_string());
+        outbound["providerId"] = Value::String(provider_id);
+        outbound["dataChannelId"] = Value::String(data_channel_id);
+    }
+    let _ = client_sender.send(Message::Text(outbound.to_string()));
     if message_type == "host_ssh_error" {
         let state = state.clone();
         let session_id = session_id.to_string();

src/components/backend-core/src/ssh_data_channel.rs

@@ -101,7 +101,9 @@ pub async fn claim_ssh_data_channel(
     let channel = channels
         .values_mut()
         .filter(|channel| channel.provider_id == provider_id && channel.active_session_id.is_none())
-        .min_by(|left, right| left.channel_id.cmp(&right.channel_id))?;
+        // Prefer the newest channel so a freshly reconnected pool is not
+        // starved behind older stale entries that have not failed out yet.
+        .max_by_key(|channel| channel.connected_at_millis)?;
     channel.active_session_id = Some(session_id.to_string());
     Some((channel.channel_id.clone(), channel.writer.clone()))
 }
@@ -112,10 +114,16 @@ pub async fn send_ssh_data_frame(
     channel_id: &str,
     frame: SshDataFrame,
 ) -> bool {
-    let channels = state.active_ssh_data_channels.lock().await;
-    channels
-        .get(&channel_key(provider_id, channel_id))
-        .is_some_and(|channel| channel.writer.send(frame).is_ok())
+    let key = channel_key(provider_id, channel_id);
+    let mut channels = state.active_ssh_data_channels.lock().await;
+    let Some(channel) = channels.get(&key) else {
+        return false;
+    };
+    if channel.writer.send(frame).is_ok() {
+        return true;
+    }
+    channels.remove(&key);
+    false
 }
 
 pub async fn release_ssh_data_channel(
@@ -129,6 +137,19 @@ pub async fn release_ssh_data_channel(
     }
 }
 
+pub async fn remove_ssh_data_channel(
+    state: &Arc<AppState>,
+    provider_id: &str,
+    channel_id: &str,
+) -> bool {
+    state
+        .active_ssh_data_channels
+        .lock()
+        .await
+        .remove(&channel_key(provider_id, channel_id))
+        .is_some()
+}
+
 pub async fn close_ssh_data_session(state: &Arc<AppState>, session_id: &str) {
     let client = {
         let clients = state.active_ssh_clients.read().await;
@@ -294,6 +315,7 @@ async fn handle_ssh_data_stream(state: Arc<AppState>, stream: TcpStream) {
             channel_id: channel_id.clone(),
             writer: tx,
             active_session_id: None,
+            connected_at_millis: chrono::Utc::now().timestamp_millis(),
         },
     );
     state.log(

src/components/backend-core/src/state.rs

@@ -36,6 +36,7 @@ pub struct SshDataChannel {
     pub channel_id: String,
     pub writer: mpsc::UnboundedSender<SshDataFrame>,
     pub active_session_id: Option<String>,
+    pub connected_at_millis: i64,
 }
 
 pub struct HttpTunnelResponse {