diff --git a/crates/tty7-gateway/src/serve.rs b/crates/tty7-gateway/src/serve.rs index 2ad8fc5f..70751ef0 100644 --- a/crates/tty7-gateway/src/serve.rs +++ b/crates/tty7-gateway/src/serve.rs @@ -349,12 +349,16 @@ async fn pane_stream( backend: Arc, ) -> io::Result<()> { let (down_tx, mut down) = mpsc::channel::(PANE_BACKLOG); + // Weak, so the stream still ends when the pane's feed does. + let refused = down_tx.downgrade(); std::thread::Builder::new() .name(format!("gateway-pane-{pane_id}")) .spawn(move || read_pane(feed, down_tx))?; // Keystrokes go through one thread so they reach the pane in the order - // they were typed; each `send_input` is its own daemon round trip. + // they were typed; each `send_input` is its own daemon round trip. One + // that fails is said to the phone, which would otherwise go on showing a + // live pane while what it types goes nowhere. let (input_tx, input_rx) = std_mpsc::channel::>(); std::thread::Builder::new() .name(format!("gateway-input-{pane_id}")) @@ -362,6 +366,10 @@ async fn pane_stream( for bytes in input_rx { if let Err(e) = backend.send_input(machine.as_deref(), pane_id, &bytes) { log::debug!("mobile gateway: input to pane {pane_id}: {e}"); + let message = format!("typing didn't reach the pane: {e}"); + if let Some(down) = refused.upgrade() { + let _ = down.blocking_send(Down::Event(PaneEvent::Error { message })); + } return; } } @@ -370,10 +378,11 @@ async fn pane_stream( loop { tokio::select! { frame = read_frame(&mut recv) => match frame? { + // Once typing has failed, the phone has been told; later + // keys are dropped rather than ending the stream, which would + // read as the pane closing. Some(Frame::Bytes(bytes)) => { - if input_tx.send(bytes).is_err() { - return Ok(()); - } + let _ = input_tx.send(bytes); } // No upstream messages yet; one from a newer app is ignored // rather than fatal, so the terminal keeps working. diff --git a/crates/tty7-gateway/tests/bridge.rs b/crates/tty7-gateway/tests/bridge.rs index d4c7dec9..f47b99b6 100644 --- a/crates/tty7-gateway/tests/bridge.rs +++ b/crates/tty7-gateway/tests/bridge.rs @@ -111,6 +111,9 @@ impl Backend for FakeMachine { } fn send_input(&self, _machine: Option<&str>, _pane_id: u64, bytes: &[u8]) -> io::Result<()> { + if bytes == b"refuse" { + return Err(io::Error::other("the daemon hung up")); + } let typed = self.typed.lock().unwrap(); typed .as_ref() @@ -296,6 +299,32 @@ async fn a_paired_phone_reads_the_tree_and_drives_a_pane() { assert!(err.to_string().contains("no such pane 99"), "{err}"); } +#[tokio::test(flavor = "multi_thread")] +async fn typing_that_fails_is_reported_not_dropped() { + let rig = Rig::new().await; + let session = within(rig.paired()).await; + let (mut keys, mut screen) = within(session.pane(None, 1)).await.unwrap(); + within(screen.next()).await.unwrap(); // size + within(screen.next()).await.unwrap(); // prompt + + keys.input(b"refuse").await.unwrap(); + let Some(PaneItem::Event(PaneEvent::Error { message })) = within(screen.next()).await.unwrap() + else { + panic!("expected the failure to be reported"); + }; + assert!(message.contains("the daemon hung up"), "{message}"); + + // Keys after it go nowhere, but the stream stays up: its end would say + // the pane closed, which it did not. + keys.input(b"ls\r").await.unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(300), screen.next()) + .await + .is_err(), + "nothing more arrives" + ); +} + #[tokio::test(flavor = "multi_thread")] async fn a_revoked_phone_is_cut_off_at_its_next_stream() { let rig = Rig::new().await; diff --git a/mobile/src-tauri/src/lib.rs b/mobile/src-tauri/src/lib.rs index 38bac344..a50a123f 100644 --- a/mobile/src-tauri/src/lib.rs +++ b/mobile/src-tauri/src/lib.rs @@ -254,9 +254,19 @@ async fn pane_open( let (input_tx, mut input_rx) = mpsc::unbounded_channel::>(); state.panes.lock().await.insert(handle, input_tx); + // A write that fails is said on the pane's channel, and the handle goes, + // so the next key is refused too rather than queued behind a dead stream. + let refused = on_output.clone(); + let app_state = state.inner().clone(); tokio::spawn(async move { while let Some(bytes) = input_rx.recv().await { - if keys.input(&bytes).await.is_err() { + if let Err(e) = keys.input(&bytes).await { + app_state.panes.lock().await.remove(&handle); + let event = PaneEvent::Error { + message: format!("typing didn't reach the computer: {}", err(e)), + }; + let json = serde_json::to_string(&event).expect("events serialize"); + let _ = refused.send(InvokeResponseBody::Json(json)); return; } }