diff --git a/src/components/backend-core/src/cli.rs b/src/components/backend-core/src/cli.rs index 16d769a3..900e3a54 100644 --- a/src/components/backend-core/src/cli.rs +++ b/src/components/backend-core/src/cli.rs @@ -6,6 +6,7 @@ use base64::Engine; use futures_util::{SinkExt, StreamExt}; use serde_json::{json, Value}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::sync::watch; use tokio_tungstenite::connect_async; use tokio_tungstenite::tungstenite::Message; @@ -124,11 +125,21 @@ async fn ssh_broker_cli(args: Vec) -> anyhow::Result<()> { }); ws_write.send(Message::Text(open.to_string())).await?; + let (opened_tx, mut opened_rx) = watch::channel(false); let stdin_eot_on_end = payload .get("stdinEotOnEnd") .and_then(Value::as_bool) .unwrap_or(false); let write_task = tokio::spawn(async move { + // tcp-pool input frames sent before provider-gateway starts the host + // SSH process are dropped because the provider has no session yet. + if opened_rx + .wait_for(|opened| *opened) + .await + .is_err() + { + return; + } let mut stdin = tokio::io::stdin(); let mut buffer = vec![0_u8; 8192]; loop { @@ -207,6 +218,9 @@ async fn ssh_broker_cli(args: Vec) -> anyhow::Result<()> { match parsed.get("type").and_then(Value::as_str).unwrap_or("") { "ssh.opened" | "ssh.dispatched" => { opened = true; + if parsed.get("type").and_then(Value::as_str) == Some("ssh.opened") { + let _ = opened_tx.send(true); + } } "ssh.data" => { opened = true;