feat(terminal): add opt-in flow control

.tegami/add-opt-in-flow-control.md

@@ -0,0 +1,11 @@
+---
+packages:
+  et:
+    type: patch
+---
+
+## Add opt-in terminal flow control
+
+Clients can now select lossless backpressure or oldest-output discard when
+terminal output outruns the network, keeping Ctrl-C and prompt responses
+bounded without changing the default session behavior.

crates/et-bin/Cargo.toml

@@ -39,6 +39,9 @@ signal-hook = "0.3.18"
 # Ctrl+C / console shutdown signalling, the Windows analogue of SIGINT/SIGTERM.
 ctrlc = "3.5"
 
+[dev-dependencies]
+socket2 = "0.6.1"
+
 [target.'cfg(target_os = "linux")'.dev-dependencies]
 nix = { version = "0.31.3", default-features = false, features = ["fs", "poll"] }
 rustix = { version = "1.1.4", features = ["process"] }

crates/et-bin/src/client_environment.rs

@@ -189,6 +189,7 @@ mod tests {
             jumphost: Some(false),
             reversetunnels: vec![request; 128],
             environmentvariables: Default::default(),
+            flowcontrol: None,
         };
         let mut locale = vec![
             ("LC_ALL".to_owned(), "C".to_owned()),

crates/et-bin/src/forward_config.rs

@@ -104,6 +104,7 @@ pub fn build(
             jumphost: Some(false),
             reversetunnels: reverse_tunnels,
             environmentvariables: std::collections::HashMap::new(),
+            flowcontrol: args.flow_control.protocol_value(),
         },
     })
 }

crates/et-bin/src/terminal_protocol.rs

@@ -2,7 +2,7 @@ use et_net::local::LocalStream;
 use std::io::{self, Read, Write};
 
 use et_core::packet::Packet;
-use et_core::proto::{TermInit, TerminalBuffer, TerminalInfo, TerminalPacketType};
+use et_core::proto::{FlowControlMode, TermInit, TerminalBuffer, TerminalInfo, TerminalPacketType};
 use et_net::local_packet::{read_local_packet, LocalPacketDecoder};
 use portable_pty::{MasterPty, PtySize};
 use prost::Message;
@@ -11,7 +11,12 @@ pub(crate) const MAX_ENVIRONMENT: usize = 128;
 pub(crate) const MAX_ENV_VALUE: usize = 4096;
 const READ_BUFFER: usize = 16 * 1024;
 
-pub fn read_initial_environment(router: &mut LocalStream) -> Result<Vec<(String, String)>, String> {
+pub struct TerminalInitialization {
+    pub environment: Vec<(String, String)>,
+    pub flow_control: FlowControlMode,
+}
+
+pub fn read_initialization(router: &mut LocalStream) -> Result<TerminalInitialization, String> {
     let packet = read_local_packet(router)
         .map_err(|error| format!("could not read terminal initialization: {error}"))?;
     if packet.is_encrypted() || packet.header() != TerminalPacketType::TerminalInit as u8 {
@@ -24,7 +29,12 @@ pub fn read_initial_environment(router: &mut LocalStream) -> Result<Vec<(String,
     {
         return Err("TERMINAL_INIT environment lists are invalid".to_owned());
     }
-    init.environmentnames
+    let flow_control = init
+        .flowcontrol
+        .and_then(|value| FlowControlMode::try_from(value).ok())
+        .unwrap_or(FlowControlMode::None);
+    let environment = init
+        .environmentnames
         .into_iter()
         .zip(init.environmentvalues)
         .map(|(name, value)| {
@@ -34,7 +44,11 @@ pub fn read_initial_environment(router: &mut LocalStream) -> Result<Vec<(String,
             }
             Ok((name, value))
         })
-        .collect()
+        .collect::<Result<_, _>>()?;
+    Ok(TerminalInitialization {
+        environment,
+        flow_control,
+    })
 }
 
 pub fn read_ready_packet(
@@ -182,18 +196,22 @@ mod tests {
             TermInit {
                 environmentnames: vec!["A".to_owned()],
                 environmentvalues: Vec::new(),
+                flowcontrol: None,
             },
             TermInit {
                 environmentnames: vec!["BAD-NAME".to_owned()],
                 environmentvalues: vec!["value".to_owned()],
+                flowcontrol: None,
             },
             TermInit {
                 environmentnames: vec!["VALID".to_owned()],
                 environmentvalues: vec!["bad\0value".to_owned()],
+                flowcontrol: None,
             },
             TermInit {
                 environmentnames: vec!["VALID".to_owned()],
                 environmentvalues: vec!["x".repeat(MAX_ENV_VALUE + 1)],
+                flowcontrol: None,
             },
         ] {
             let packet = Packet::new(TerminalPacketType::TerminalInit as u8, init.encode_to_vec());
@@ -204,6 +222,24 @@ mod tests {
     fn read_environment_packet(packet: Packet) -> Result<Vec<(String, String)>, String> {
         let (mut reader, mut writer) = et_net::local::wake_pair().unwrap();
         write_local_packet(&mut writer, &packet).unwrap();
-        read_initial_environment(&mut reader)
+        read_initialization(&mut reader).map(|initialization| initialization.environment)
+    }
+
+    #[test]
+    fn initialization_retains_the_typed_flow_control_mode() {
+        let (mut terminal, mut server) = et_net::local::wake_pair().unwrap();
+        let init = TermInit {
+            environmentnames: Vec::new(),
+            environmentvalues: Vec::new(),
+            flowcontrol: Some(FlowControlMode::Discard as i32),
+        };
+        write_local_packet(
+            &mut server,
+            &Packet::new(TerminalPacketType::TerminalInit as u8, init.encode_to_vec()),
+        )
+        .unwrap();
+
+        let initialization = read_initialization(&mut terminal).unwrap();
+        assert_eq!(initialization.flow_control, FlowControlMode::Discard);
     }
 }

crates/et-bin/src/terminal_pty.rs

@@ -3,7 +3,7 @@ use std::sync::mpsc;
 use std::thread;
 
 use et_core::packet::Packet;
-use et_core::proto::{TerminalBuffer, TerminalPacketType};
+use et_core::proto::{FlowControlMode, TerminalBuffer, TerminalPacketType};
 use et_net::local::LocalStream;
 use et_net::local_packet::{write_local_packet, LocalPacketDecoder};
 #[cfg(unix)]
@@ -19,15 +19,21 @@ use sysinfo::{Pid as SystemPid, ProcessesToUpdate, Signal as SystemSignal, Syste
 
 const MAX_OUTPUT_CHUNK: usize = 16 * 1024;
 
-use crate::terminal_protocol::{handle_packet, read_initial_environment, read_ready_packet};
+use crate::terminal_protocol::{handle_packet, read_initialization, read_ready_packet};
 
 enum WorkerEvent {
     Output(Result<(), String>),
     Child(Result<u32, String>),
 }
 
 pub fn run(mut router: LocalStream, term: &str) -> Result<i32, String> {
-    let environment = read_initial_environment(&mut router)?;
+    let initialization = read_initialization(&mut router)?;
+    if initialization.flow_control != FlowControlMode::None {
+        // Keep terminal output in the server's bounded application queue,
+        // rather than a large opaque local-socket queue (upstream PR #730).
+        et_net::local::minimize_terminal_output_buffering(&router)
+            .map_err(|error| format!("could not bound terminal output buffering: {error}"))?;
+    }
     // Upstream issue #257: show the login banner an interactive ssh would have
     // printed, before the shell writes anything.
     #[cfg(unix)]
@@ -44,7 +50,7 @@ pub fn run(mut router: LocalStream, term: &str) -> Result<i32, String> {
     #[cfg(unix)]
     command.arg("-l");
     command.env("TERM", term);
-    for (name, value) in environment {
+    for (name, value) in initialization.environment {
         command.env(name, value);
     }
     let mut child = pair

crates/et-bin/tests/client_bootstrap.rs

@@ -444,6 +444,7 @@ fn posix_client_bounds_locale_to_local_terminal_packet() {
     let term_init = TermInit {
         environmentnames: environment.keys().cloned().collect(),
         environmentvalues: environment.values().cloned().collect(),
+        flowcontrol: None,
     };
     let packet = Packet::new(
         TerminalPacketType::TerminalInit as u8,

crates/et-bin/tests/flow_control_tty_qa.rs

@@ -0,0 +1,157 @@
+#![cfg(unix)]
+#![forbid(unsafe_code)]
+
+mod flow_control_tty_support;
+
+use std::fs;
+use std::io::{Read, Write};
+use std::sync::mpsc;
+use std::thread;
+use std::time::{Duration, Instant};
+
+use flow_control_tty_support::{
+    receive_bytes, receive_until, Stack, ThrottleProxy, MAX_PROMPT_LATENCY, SATURATION_BYTES,
+    THROTTLE_BYTES_PER_SECOND,
+};
+use portable_pty::{native_pty_system, CommandBuilder, PtySize};
+
+#[test]
+fn flow_control_keeps_ctrl_c_and_prompt_responsive_on_a_slow_link() {
+    let evidence = std::env::var_os("ET_FLOW_QA_EVIDENCE_DIR").map(std::path::PathBuf::from);
+    if let Some(directory) = &evidence {
+        fs::create_dir_all(directory).unwrap();
+    }
+
+    for mode in ["none", "backpressure", "discard"] {
+        let stack = Stack::start();
+        let bytes_per_second = THROTTLE_BYTES_PER_SECOND;
+        let proxy = ThrottleProxy::start(stack.port, bytes_per_second);
+        let pair = native_pty_system()
+            .openpty(PtySize {
+                rows: 24,
+                cols: 80,
+                pixel_width: 800,
+                pixel_height: 480,
+            })
+            .unwrap();
+        let mut client = CommandBuilder::new(env!("CARGO_BIN_EXE_et"));
+        client.args([
+            "--flow-control",
+            mode,
+            "--terminal-path",
+            stack.terminal.to_str().unwrap(),
+            "--serverfifo",
+            stack.router.to_str().unwrap(),
+            "-p",
+            &proxy.port.to_string(),
+            "127.0.0.1",
+        ]);
+        client.env(
+            "PATH",
+            format!(
+                "{}:{}",
+                stack.directory.display(),
+                std::env::var("PATH").unwrap()
+            ),
+        );
+        client.env("TERM", "xterm-256color");
+        let mut child = pair.slave.spawn_command(client).unwrap();
+        drop(pair.slave);
+
+        let mut writer = pair.master.take_writer().unwrap();
+        let mut reader = pair.master.try_clone_reader().unwrap();
+        // Keep the test harness from adding its own half-megabyte output
+        // queue on top of the ET pipeline being measured.
+        let (sender, receiver) = mpsc::sync_channel(4);
+        let reader_thread = thread::spawn(move || {
+            let mut chunk = [0u8; 8192];
+            loop {
+                match reader.read(&mut chunk) {
+                    Ok(0) | Err(_) => return,
+                    Ok(count) if sender.send(chunk[..count].to_vec()).is_err() => return,
+                    Ok(_) => {}
+                }
+            }
+        });
+
+        writer
+            .write_all(
+                b"printf 'FLOW-%s\\n' START; while :; do printf \
+                  '0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef'; done\n",
+            )
+            .unwrap();
+        let startup_timeout = Duration::from_secs(10);
+        let output = match receive_until(&receiver, Vec::new(), b"FLOW-START\r\n", startup_timeout)
+        {
+            Ok(output) => output,
+            Err(error) => {
+                child.kill().unwrap();
+                drop(writer);
+                let _ = child.wait();
+                reader_thread.join().unwrap();
+                panic!("{mode}: waiting for FLOW-START: {error}");
+            }
+        };
+        let mut output =
+            match receive_bytes(&receiver, output, SATURATION_BYTES, Duration::from_secs(40)) {
+                Ok(output) => output,
+                Err(error) => {
+                    child.kill().unwrap();
+                    drop(writer);
+                    let _ = child.wait();
+                    reader_thread.join().unwrap();
+                    panic!("{mode}: saturating throttled link: {error}");
+                }
+            };
+        let interrupted = Instant::now();
+        writer
+            .write_all(b"\x03printf 'FLOW-%s\\n' PROMPT\n")
+            .unwrap();
+        let prompt_timeout = MAX_PROMPT_LATENCY;
+        let prompt = receive_until(
+            &receiver,
+            output.clone(),
+            b"FLOW-PROMPT\r\n",
+            prompt_timeout,
+        );
+        let latency = interrupted.elapsed();
+        if mode == "none" {
+            assert!(
+                prompt.is_err(),
+                "none baseline unexpectedly met the {MAX_PROMPT_LATENCY:?} latency criterion"
+            );
+        } else {
+            output = prompt.unwrap_or_else(|error| {
+                panic!("{mode}: waiting for Ctrl-C prompt within {prompt_timeout:?}: {error}")
+            });
+            assert!(
+                latency <= MAX_PROMPT_LATENCY,
+                "{mode} Ctrl-C-to-prompt latency {latency:?} exceeded {MAX_PROMPT_LATENCY:?}"
+            );
+        }
+
+        child.kill().unwrap();
+        drop(writer);
+        let _ = child.wait();
+        while let Ok(chunk) = receiver.recv() {
+            output.extend(chunk);
+        }
+        reader_thread.join().unwrap();
+        proxy.finish().unwrap();
+
+        if let Some(directory) = &evidence {
+            fs::write(directory.join(format!("{mode}.ansi")), &output).unwrap();
+            fs::write(
+                directory.join(format!("{mode}.json")),
+                format!(
+                    "{{\"mode\":\"{mode}\",\"rate_bytes_per_second\":{bytes_per_second},\
+                     \"saturation_bytes\":{SATURATION_BYTES},\"ctrl_c_prompt_millis\":{},\
+                     \"expected_latency_failure\":{},\"scenario_pass\":true}}\n",
+                    latency.as_millis(),
+                    mode == "none"
+                ),
+            )
+            .unwrap();
+        }
+    }
+}