Decouple terminal rendering from transport

tests/client_terminal_runtime.rs

@@ -48,6 +48,7 @@ struct CachedCredentialWire {
 
 #[derive(Debug)]
 enum ServerObservation {
+    BulkSent,
     Input(Vec<u8>),
     Resize(u16, u16),
 }
@@ -159,6 +160,70 @@ fn native_client_terminal_round_trip_is_platform_complete() {
     }
 }
 
+#[test]
+fn terminal_output_backpressure_does_not_block_input_transport() {
+    const PROBE: &[u8] = b"DOSH_BACKPRESSURE_INPUT\r";
+
+    let dir = tempfile::tempdir().unwrap();
+    let home = dir.path().join("home");
+    let cache = dir.path().join("credentials");
+    fs::create_dir_all(home.join(".config/dosh")).unwrap();
+    fs::create_dir_all(&cache).unwrap();
+
+    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+    socket
+        .set_read_timeout(Some(Duration::from_secs(8)))
+        .unwrap();
+    let port = socket.local_addr().unwrap().port();
+    write_client_fixture(&home, &cache, port);
+    let (observation_tx, observation_rx) = mpsc::channel();
+    let server = thread::spawn(move || run_backpressure_server(socket, observation_tx, PROBE));
+
+    let pty = NativePtySystem::default();
+    let pair = pty
+        .openpty(PtySize {
+            rows: 24,
+            cols: 80,
+            pixel_width: 0,
+            pixel_height: 0,
+        })
+        .unwrap();
+    let mut reader = pair.master.try_clone_reader().unwrap();
+    let mut writer = pair.master.take_writer().unwrap();
+    let mut command = client_command(dir.path(), port);
+    command.env("HOME", home.to_string_lossy().to_string());
+    command.env("USERPROFILE", home.to_string_lossy().to_string());
+    command.env("APPDATA", home.to_string_lossy().to_string());
+    command.env("LOCALAPPDATA", home.to_string_lossy().to_string());
+    command.env("TERM", "xterm-256color");
+    let mut child = pair.slave.spawn_command(command).unwrap();
+    drop(pair.slave);
+
+    wait_for_observation(&observation_rx, Duration::from_secs(5), |observation| {
+        matches!(observation, ServerObservation::BulkSent)
+    });
+    thread::sleep(Duration::from_millis(100));
+    writer.write_all(PROBE).unwrap();
+    writer.flush().unwrap();
+    wait_for_input(&observation_rx, PROBE, Duration::from_secs(2));
+
+    let reader_thread = thread::spawn(move || {
+        let mut buf = [0u8; 16 * 1024];
+        while let Ok(n) = reader.read(&mut buf) {
+            if n == 0 {
+                break;
+            }
+        }
+    });
+    let status = child.wait().unwrap();
+    drop(writer);
+    drop(pair.master);
+    reader_thread.join().unwrap();
+    server.join().unwrap();
+
+    assert!(status.success(), "Dosh client exited with {status:?}");
+}
+
 fn write_client_fixture(home: &Path, cache: &Path, port: u16) {
     let config = ClientConfig {
         server: "local".to_string(),
@@ -344,6 +409,74 @@ fn run_fake_terminal_server(socket: UdpSocket, observations: mpsc::Sender<Server
     }
 }
 
+fn run_backpressure_server(
+    socket: UdpSocket,
+    observations: mpsc::Sender<ServerObservation>,
+    expected_input: &[u8],
+) {
+    let mut peer = None;
+    let mut bulk_sent = false;
+    let mut received_input = Vec::new();
+    let mut buf = [0u8; 65535];
+    loop {
+        let (n, source) = socket.recv_from(&mut buf).unwrap();
+        let packet = protocol::decode(&buf[..n]).unwrap();
+        match packet.header.kind {
+            PacketKind::ResumeRequest => {
+                peer = Some(source);
+                send_frame(
+                    &socket,
+                    source,
+                    PacketKind::ResumeOk,
+                    1,
+                    10,
+                    b"DOSH_BACKPRESSURE_READY",
+                    true,
+                    false,
+                );
+            }
+            PacketKind::Ack if !bulk_sent => {
+                let bulk = vec![b'x'; 60 * 1024];
+                send_frame(
+                    &socket,
+                    peer.unwrap_or(source),
+                    PacketKind::Frame,
+                    2,
+                    11,
+                    &bulk,
+                    false,
+                    false,
+                );
+                bulk_sent = true;
+                observations.send(ServerObservation::BulkSent).unwrap();
+            }
+            PacketKind::Input => {
+                let plain =
+                    protocol::decrypt_body(&packet, &SESSION_KEY, CLIENT_TO_SERVER).unwrap();
+                let input: Input = protocol::from_body(&plain).unwrap();
+                received_input.extend_from_slice(&input.bytes);
+                observations
+                    .send(ServerObservation::Input(input.bytes))
+                    .unwrap();
+                if contains(&received_input, expected_input) {
+                    send_frame(
+                        &socket,
+                        peer.unwrap_or(source),
+                        PacketKind::Frame,
+                        3,
+                        12,
+                        b"DOSH_BACKPRESSURE_DONE",
+                        false,
+                        true,
+                    );
+                    break;
+                }
+            }
+            _ => {}
+        }
+    }
+}
+
 #[allow(clippy::too_many_arguments)]
 fn send_frame(
     socket: &UdpSocket,
@@ -429,6 +562,22 @@ fn wait_for_input(
     panic!("terminal input was not delivered; got {input:?}");
 }
 
+fn wait_for_observation(
+    observations: &mpsc::Receiver<ServerObservation>,
+    timeout: Duration,
+    matches: impl Fn(&ServerObservation) -> bool,
+) {
+    let deadline = Instant::now() + timeout;
+    while let Some(remaining) = deadline.checked_duration_since(Instant::now()) {
+        match observations.recv_timeout(remaining) {
+            Ok(observation) if matches(&observation) => return,
+            Ok(_) => {}
+            Err(err) => panic!("expected server observation was not received: {err}"),
+        }
+    }
+    panic!("expected server observation was not received");
+}
+
 fn contains(haystack: &[u8], needle: &[u8]) -> bool {
     !needle.is_empty() && haystack.windows(needle.len()).any(|bytes| bytes == needle)
 }