fix(terminal): resume pasted input as soon as the PTY accepts more (#2220)

crates/ide-core/src/input.rs

@@ -4,10 +4,12 @@ use std::collections::VecDeque;
 use std::io::{self, Write};
 use std::sync::{Arc, Condvar, Mutex};
 use std::thread::JoinHandle;
+#[cfg(any(not(unix), test))]
 use std::time::Duration;
 
 const MAX_BYTES: usize = 4 * 1024 * 1024;
 const MAX_WRITES: usize = 1024;
+#[cfg(any(not(unix), test))]
 const RETRY: Duration = Duration::from_millis(4);
 
 #[derive(Default)]
@@ -28,10 +30,11 @@ pub struct InputWriter {
 }
 
 impl InputWriter {
-    /// Unix writers must first be made nonblocking with `nonblocking`.
+    /// On Unix, pass the readiness hook returned by `nonblocking` for this writer.
     pub fn new(
         name: String,
         mut writer: Box<dyn Write + Send>,
+        #[cfg(unix)] mut wait_writable: impl FnMut() -> io::Result<()> + Send + 'static,
         on_error: impl FnOnce(io::Error) + Send + 'static,
     ) -> io::Result<Self> {
         let shared = Arc::new((Mutex::new(State::default()), Condvar::new()));
@@ -74,11 +77,19 @@ impl InputWriter {
                             Ok(()) => {}
                             Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
                             Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
-                                let state = lock.lock().unwrap();
-                                if state.aborted {
-                                    return Ok(());
+                                #[cfg(unix)]
+                                match wait_writable() {
+                                    Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
+                                    result => result?,
+                                }
+                                #[cfg(not(unix))]
+                                {
+                                    let state = lock.lock().unwrap();
+                                    if state.aborted {
+                                        return Ok(());
+                                    }
+                                    drop(wake.wait_timeout(state, RETRY).unwrap());
                                 }
-                                drop(wake.wait_timeout(state, RETRY).unwrap());
                             }
                             Err(e) => return Err(e),
                         }
@@ -168,14 +179,46 @@ impl Drop for InputWriter {
     }
 }
 
-/// PTY clones share file status flags, so readers must also handle WouldBlock.
+/// Make a writer nonblocking and return its readiness hook. PTY clones share file
+/// status flags, so readers must also handle WouldBlock.
 #[cfg(unix)]
-pub fn nonblocking(fd: std::os::fd::RawFd) -> io::Result<()> {
+pub fn nonblocking(fd: std::os::fd::RawFd) -> io::Result<impl FnMut() -> io::Result<()> + Send> {
+    let wait = readiness(fd, libc::POLLOUT)?;
     let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) };
     if flags == -1 || unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) } == -1 {
         return Err(io::Error::last_os_error());
     }
-    Ok(())
+    Ok(wait)
+}
+
+/// Wait for output from a nonblocking PTY without delaying the child's next read.
+#[cfg(unix)]
+pub fn readable(fd: std::os::fd::RawFd) -> io::Result<impl FnMut() -> io::Result<()> + Send> {
+    readiness(fd, libc::POLLIN)
+}
+
+#[cfg(unix)]
+fn readiness(fd: std::os::fd::RawFd, events: libc::c_short) -> io::Result<impl FnMut() -> io::Result<()> + Send> {
+    use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
+
+    // Own a duplicate so the hook remains valid if the caller closes its fd.
+    let duplicate = unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, 0) };
+    if duplicate == -1 {
+        return Err(io::Error::last_os_error());
+    }
+    // SAFETY: fcntl returned a new, owned descriptor.
+    let fd = unsafe { OwnedFd::from_raw_fd(duplicate) };
+    Ok(move || {
+        let mut pollfd = libc::pollfd { fd: fd.as_raw_fd(), events, revents: 0 };
+        // Readiness wakes immediately; the timeout only bounds abort latency.
+        if unsafe { libc::poll(&mut pollfd, 1, 50) } == -1 {
+            return Err(io::Error::last_os_error());
+        }
+        if pollfd.revents & libc::POLLNVAL != 0 {
+            return Err(io::Error::from_raw_os_error(libc::EBADF));
+        }
+        Ok(())
+    })
 }
 
 #[cfg(test)]
@@ -221,6 +264,11 @@ mod tests {
         let writer = InputWriter::new(
             "test-input".into(),
             Box::new(SlowWriter { ready: ready.clone(), output: output.clone(), dropped: tx }),
+            #[cfg(unix)]
+            || {
+                std::thread::sleep(RETRY);
+                Ok(())
+            },
             |e| panic!("{e}"),
         )
         .unwrap();
@@ -243,6 +291,11 @@ mod tests {
         let writer = InputWriter::new(
             "test-input".into(),
             Box::new(SlowWriter { ready: ready.clone(), output: output.clone(), dropped: tx }),
+            #[cfg(unix)]
+            || {
+                std::thread::sleep(RETRY);
+                Ok(())
+            },
             |e| panic!("{e}"),
         )
         .unwrap();
@@ -256,6 +309,26 @@ mod tests {
         assert_eq!(*output.lock().unwrap(), format!("{paste}tail").as_bytes());
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn readiness_errors_are_reported_and_reject_following_writes() {
+        let (dropped, _rx) = mpsc::channel();
+        let (errors, error) = mpsc::channel();
+        let writer = InputWriter::new(
+            "test-error-input".into(),
+            Box::new(SlowWriter { ready: Arc::new(AtomicBool::new(false)), output: Arc::default(), dropped }),
+            || Err(io::Error::from_raw_os_error(libc::EBADF)),
+            move |e| errors.send(e.raw_os_error()).unwrap(),
+        )
+        .unwrap();
+        writer.write(b"input").unwrap();
+        assert_eq!(error.recv_timeout(Duration::from_secs(1)).unwrap(), Some(libc::EBADF));
+        assert_eq!(writer.write(b"later").unwrap_err().kind(), io::ErrorKind::BrokenPipe);
+        let state = writer.shared.0.lock().unwrap();
+        assert!(state.queue.is_empty());
+        assert_eq!(state.bytes, 0);
+    }
+
     #[test]
     fn tiny_writes_are_also_bounded() {
         let shared = Arc::new((Mutex::new(State::default()), Condvar::new()));
@@ -266,14 +339,59 @@ mod tests {
         assert_eq!(writer.write(b"x").unwrap_err().kind(), io::ErrorKind::WouldBlock);
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn writable_input_resumes_without_timer_delays() {
+        use std::io::Read;
+        use std::os::fd::AsRawFd;
+        use std::os::unix::net::UnixStream;
+
+        let (pipe, mut peer) = UnixStream::pair().unwrap();
+        let capacity: libc::c_int = 1024;
+        // Keep the OS buffer small enough that a timer per buffer takes seconds.
+        assert_eq!(
+            unsafe {
+                libc::setsockopt(
+                    pipe.as_raw_fd(),
+                    libc::SOL_SOCKET,
+                    libc::SO_SNDBUF,
+                    (&capacity as *const libc::c_int).cast(),
+                    std::mem::size_of_val(&capacity) as libc::socklen_t,
+                )
+            },
+            0
+        );
+        peer.set_read_timeout(Some(Duration::from_secs(3))).unwrap();
+        let wait_writable = nonblocking(pipe.as_raw_fd()).unwrap();
+        let writer =
+            InputWriter::new("test-ready-input".into(), Box::new(pipe), wait_writable, |e| panic!("{e}")).unwrap();
+        let bytes = vec![b'x'; MAX_BYTES];
+        writer.write(&bytes).unwrap();
+        writer.finish();
+        // Start with a full buffer, then make the fd writable from another thread.
+        std::thread::sleep(Duration::from_millis(50));
+        let (tx, rx) = mpsc::channel();
+        let started = Instant::now();
+        let reader = std::thread::spawn(move || {
+            let mut output = Vec::new();
+            peer.read_to_end(&mut output).unwrap();
+            tx.send(output).unwrap();
+        });
+        let output = rx.recv_timeout(Duration::from_secs(1)).expect("input resumes as soon as the fd is writable");
+        assert_eq!(output, bytes);
+        reader.join().unwrap();
+        eprintln!("4 MiB input drained in {:?}", started.elapsed());
+    }
+
     #[cfg(unix)]
     #[test]
     fn close_interrupts_a_full_os_pipe_while_the_peer_stays_open() {
         use std::os::fd::AsRawFd;
         use std::os::unix::net::UnixStream;
         let (pipe, _peer) = UnixStream::pair().unwrap();
-        nonblocking(pipe.as_raw_fd()).unwrap();
-        let mut writer = InputWriter::new("test-pipe-input".into(), Box::new(pipe), |e| panic!("{e}")).unwrap();
+        let wait_writable = nonblocking(pipe.as_raw_fd()).unwrap();
+        let mut writer =
+            InputWriter::new("test-pipe-input".into(), Box::new(pipe), wait_writable, |e| panic!("{e}")).unwrap();
         writer.write(&vec![b'x'; MAX_BYTES]).unwrap();
         std::thread::sleep(Duration::from_millis(50));
         assert!(!writer.worker.as_ref().unwrap().is_finished());

crates/ide-dap/src/session.rs

@@ -630,9 +630,15 @@ pub(crate) fn run_in_terminal(s: &Arc<Session>, args: &Value) -> Result<Value, S
         .openpty(PtySize { rows: 24, cols: 200, pixel_width: 0, pixel_height: 0 })
         .map_err(|e| format!("cannot open a terminal for {program}: {e}"))?;
     #[cfg(unix)]
-    if let Some(fd) = pair.master.as_raw_fd() {
-        ide_core::input::nonblocking(fd).map_err(|e| format!("terminal of {program}: {e}"))?;
-    }
+    let wait_writable = ide_core::input::nonblocking(
+        pair.master.as_raw_fd().ok_or_else(|| format!("terminal of {program}: PTY has no file descriptor"))?,
+    )
+    .map_err(|e| format!("terminal of {program}: {e}"))?;
+    #[cfg(unix)]
+    let wait_readable = ide_core::input::readable(
+        pair.master.as_raw_fd().ok_or_else(|| format!("terminal of {program}: PTY has no file descriptor"))?,
+    )
+    .map_err(|e| format!("terminal of {program}: {e}"))?;
     let mut cmd = CommandBuilder::new(program);
     // CodeLLDB's launcher clears the terminal first; the console keeps the session's earlier output.
     cmd.args(rest.iter().filter(|a| *a != "--clear-screen"));
@@ -662,16 +668,27 @@ pub(crate) fn run_in_terminal(s: &Arc<Session>, args: &Value) -> Result<Value, S
     drop(pair.slave);
     let reader = pair.master.try_clone_reader().map_err(|e| format!("terminal of {program}: {e}"))?;
     let writer = pair.master.take_writer().map_err(|e| format!("terminal of {program}: {e}"))?;
-    let writer = ide_core::input::InputWriter::new(format!("openidea-debug-input-{}", s.id.0), writer, |e| {
-        tracing::warn!("debug input: {e}");
-    })
+    let writer = ide_core::input::InputWriter::new(
+        format!("openidea-debug-input-{}", s.id.0),
+        writer,
+        #[cfg(unix)]
+        wait_writable,
+        |e| {
+            tracing::warn!("debug input: {e}");
+        },
+    )
     .inspect_err(|_| {
         let _ = child.kill();
         let _ = child.wait();
     })
     .map_err(|e| format!("terminal input worker: {e}"))?;
     let pid = child.process_id();
-    forward_terminal_output(s, reader);
+    forward_terminal_output(
+        s,
+        reader,
+        #[cfg(unix)]
+        wait_readable,
+    );
     if let Some(mut previous) = s.program.lock().replace(ProgramTerminal { writer, child, _master: pair.master }) {
         let _ = previous.child.kill();
     }
@@ -680,7 +697,11 @@ pub(crate) fn run_in_terminal(s: &Arc<Session>, args: &Value) -> Result<Value, S
 }
 
 /// The program's terminal output, as stdout lines (`\r\n` as `\n`; colours are kept).
-fn forward_terminal_output(s: &Session, mut reader: Box<dyn std::io::Read + Send>) {
+fn forward_terminal_output(
+    s: &Session,
+    mut reader: Box<dyn std::io::Read + Send>,
+    #[cfg(unix)] mut wait_readable: impl FnMut() -> std::io::Result<()> + Send + 'static,
+) {
     let session = s.id;
     let sink = s.owner.sink.clone();
     let _ = std::thread::Builder::new().name(format!("openidea-debuggee-{}", session.0)).spawn(move || {
@@ -692,6 +713,14 @@ fn forward_terminal_output(s: &Session, mut reader: Box<dyn std::io::Read + Send
                 Ok(n) => pending.extend_from_slice(&buf[..n]),
                 Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
                 Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                    #[cfg(unix)]
+                    if let Err(e) = wait_readable()
+                        && e.kind() != std::io::ErrorKind::Interrupted
+                    {
+                        tracing::warn!(%session, "debug terminal read: {e}");
+                        break;
+                    }
+                    #[cfg(not(unix))]
                     std::thread::sleep(std::time::Duration::from_millis(4));
                     continue;
                 }

crates/ide-run/src/process.rs

@@ -426,11 +426,16 @@ impl ProcessManager {
             Ok(match child.stdin.take() {
                 Some(pipe) if !redirected => {
                     #[cfg(unix)]
-                    {
+                    let wait_writable = {
                         use std::os::fd::AsRawFd;
-                        ide_core::input::nonblocking(pipe.as_raw_fd())?;
-                    }
-                    Some(input_writer(id, Box::new(pipe))?)
+                        ide_core::input::nonblocking(pipe.as_raw_fd())?
+                    };
+                    Some(input_writer(
+                        id,
+                        Box::new(pipe),
+                        #[cfg(unix)]
+                        wait_writable,
+                    )?)
                 }
                 _ => None,
             })
@@ -462,10 +467,24 @@ impl ProcessManager {
         let stderr = child.stderr.take();
         let result = (|| -> std::io::Result<()> {
             if let Some(out) = stdout {
-                spawn_reader(id, Stream::Stdout, out, tx.clone())?;
+                spawn_reader(
+                    id,
+                    Stream::Stdout,
+                    out,
+                    tx.clone(),
+                    #[cfg(unix)]
+                    None,
+                )?;
             }
             if let Some(err) = stderr {
-                spawn_reader(id, Stream::Stderr, err, tx.clone())?;
+                spawn_reader(
+                    id,
+                    Stream::Stderr,
+                    err,
+                    tx.clone(),
+                    #[cfg(unix)]
+                    None,
+                )?;
             }
             spawn_pump(rx, on_output.clone(), entry.clone(), sink, options.encoding.clone(), notice)?;
             Ok(())
@@ -538,9 +557,13 @@ impl ProcessManager {
             cmd.env(k, v);
         }
         #[cfg(unix)]
-        if let Some(fd) = pair.master.as_raw_fd() {
-            ide_core::input::nonblocking(fd)?;
-        }
+        let wait_writable = ide_core::input::nonblocking(
+            pair.master.as_raw_fd().ok_or_else(|| std::io::Error::other("PTY has no file descriptor"))?,
+        )?;
+        #[cfg(unix)]
+        let wait_readable = ide_core::input::readable(
+            pair.master.as_raw_fd().ok_or_else(|| std::io::Error::other("PTY has no file descriptor"))?,
+        )?;
         let mut child = pair.slave.spawn_command(cmd).map_err(|e| {
             let text = e.to_string();
             if text.contains("No such file") || text.contains("not found") {
@@ -555,7 +578,13 @@ impl ProcessManager {
         let writer = pair.master.take_writer().map_err(|e| IdeError::internal(format!("pty writer: {e}")))?;
 
         let id = RunId::next();
-        let writer = input_writer(id, writer).inspect_err(|_| {
+        let writer = input_writer(
+            id,
+            writer,
+            #[cfg(unix)]
+            wait_writable,
+        )
+        .inspect_err(|_| {
             let _ = child.kill();
             let _ = child.wait();
         })?;
@@ -577,8 +606,15 @@ impl ProcessManager {
         on_output(OutputChunk::System { text: format!("{}\n", inv.display) });
 
         let (tx, rx) = mpsc::channel();
-        let result = spawn_reader(id, Stream::Stdout, reader, tx.clone())
-            .and_then(|_| spawn_pump(rx, on_output, entry.clone(), sink, options.encoding.clone(), None));
+        let result = spawn_reader(
+            id,
+            Stream::Stdout,
+            reader,
+            tx.clone(),
+            #[cfg(unix)]
+            Some(Box::new(wait_readable)),
+        )
+        .and_then(|_| spawn_pump(rx, on_output, entry.clone(), sink, options.encoding.clone(), None));
         if let Err(e) = result {
             let _ = child.kill();
             let _ = child.wait();
@@ -789,11 +825,16 @@ fn run_followup(
         Ok(match child.stdin.take() {
             Some(pipe) => {
                 #[cfg(unix)]
-                {
+                let wait_writable = {
                     use std::os::fd::AsRawFd;
-                    ide_core::input::nonblocking(pipe.as_raw_fd())?;
-                }
-                Some(input_writer(entry.id, Box::new(pipe))?)
+                    ide_core::input::nonblocking(pipe.as_raw_fd())?
+                };
+                Some(input_writer(
+                    entry.id,
+                    Box::new(pipe),
+                    #[cfg(unix)]
+                    wait_writable,
+                )?)
             }
             None => None,
         })
@@ -805,10 +846,24 @@ fn run_followup(
     drop(pid);
     let readers = (|| {
         if let Some(out) = child.stdout.take() {
-            spawn_reader(entry.id, Stream::Stdout, out, tx.clone())?;
+            spawn_reader(
+                entry.id,
+                Stream::Stdout,
+                out,
+                tx.clone(),
+                #[cfg(unix)]
+                None,
+            )?;
         }
         if let Some(err) = child.stderr.take() {
-            spawn_reader(entry.id, Stream::Stderr, err, tx.clone())?;
+            spawn_reader(
+                entry.id,
+                Stream::Stderr,
+                err,
+                tx.clone(),
+                #[cfg(unix)]
+                None,
+            )?;
         }
         Ok::<_, std::io::Error>(())
     })();
@@ -982,6 +1037,7 @@ fn spawn_reader(
     stream: Stream,
     mut pipe: impl Read + Send + 'static,
     tx: Sender<Msg>,
+    #[cfg(unix)] mut wait_readable: Option<Box<dyn FnMut() -> std::io::Result<()> + Send>>,
 ) -> std::io::Result<()> {
     let label = if stream == Stream::Stdout { "out" } else { "err" };
     std::thread::Builder::new().name(format!("openidea-run-{label}-{}", id.0)).spawn(move || {
@@ -995,7 +1051,19 @@ fn spawn_reader(
                     }
                 }
                 Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
-                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => std::thread::sleep(BATCH_INTERVAL),
+                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                    #[cfg(unix)]
+                    if let Some(wait) = &mut wait_readable {
+                        if let Err(e) = wait()
+                            && e.kind() != std::io::ErrorKind::Interrupted
+                        {
+                            tracing::warn!(%id, "run read: {e}");
+                            break;
+                        }
+                        continue;
+                    }
+                    std::thread::sleep(BATCH_INTERVAL);
+                }
                 #[cfg(unix)]
                 Err(e) if e.raw_os_error() == Some(libc::EIO) => break,
                 Err(e) => {
@@ -1218,10 +1286,20 @@ fn signal_number(_description: &str) -> Option<i32> {
     None
 }
 
-fn input_writer(id: RunId, writer: Box<dyn std::io::Write + Send>) -> std::io::Result<ide_core::input::InputWriter> {
-    ide_core::input::InputWriter::new(format!("openidea-run-input-{}", id.0), writer, move |e| {
-        tracing::warn!(%id, "run input: {e}");
-    })
+fn input_writer(
+    id: RunId,
+    writer: Box<dyn std::io::Write + Send>,
+    #[cfg(unix)] wait_writable: impl FnMut() -> std::io::Result<()> + Send + 'static,
+) -> std::io::Result<ide_core::input::InputWriter> {
+    ide_core::input::InputWriter::new(
+        format!("openidea-run-input-{}", id.0),
+        writer,
+        #[cfg(unix)]
+        wait_writable,
+        move |e| {
+            tracing::warn!(%id, "run input: {e}");
+        },
+    )
 }
 
 #[cfg(test)]

crates/ide-terminal/src/lib.rs

@@ -163,9 +163,13 @@ impl TerminalManager {
         apply_powershell_env(&shell, &mut cmd);
 
         #[cfg(unix)]
-        if let Some(fd) = pair.master.as_raw_fd() {
-            ide_core::input::nonblocking(fd)?;
-        }
+        let wait_writable = ide_core::input::nonblocking(
+            pair.master.as_raw_fd().ok_or_else(|| std::io::Error::other("PTY has no file descriptor"))?,
+        )?;
+        #[cfg(unix)]
+        let wait_readable = ide_core::input::readable(
+            pair.master.as_raw_fd().ok_or_else(|| std::io::Error::other("PTY has no file descriptor"))?,
+        )?;
 
         let mut child = pair
             .slave
@@ -180,9 +184,15 @@ impl TerminalManager {
         let id = TerminalId::next();
         let info = TerminalInfo { id, title: shell_title(&shell), pid: child.process_id() };
         let exited = Arc::new(AtomicBool::new(false));
-        let writer = ide_core::input::InputWriter::new(format!("openidea-term-write-{}", id.0), writer, move |e| {
-            tracing::warn!(%id, "terminal input: {e}");
-        })
+        let writer = ide_core::input::InputWriter::new(
+            format!("openidea-term-write-{}", id.0),
+            writer,
+            #[cfg(unix)]
+            wait_writable,
+            move |e| {
+                tracing::warn!(%id, "terminal input: {e}");
+            },
+        )
         .inspect_err(|_| {
             let _ = child.kill();
             let _ = child.wait();
@@ -200,9 +210,15 @@ impl TerminalManager {
         self.sessions.lock().insert(id, session);
 
         let (tx, rx) = mpsc::channel::<Msg>();
-        let spawned = spawn_reader(id, reader, tx.clone())
-            .and_then(|_| spawn_waiter(id, child, tx))
-            .and_then(|_| spawn_pump(id, rx, on_output, exited, self.sessions.clone(), sink, Instant::now()));
+        let spawned = spawn_reader(
+            id,
+            reader,
+            tx.clone(),
+            #[cfg(unix)]
+            wait_readable,
+        )
+        .and_then(|_| spawn_waiter(id, child, tx))
+        .and_then(|_| spawn_pump(id, rx, on_output, exited, self.sessions.clone(), sink, Instant::now()));
         if let Err(e) = spawned {
             if let Some(s) = self.sessions.lock().remove(&id) {
                 let _ = s.killer.lock().kill();
@@ -407,7 +423,12 @@ fn terminate(session: &Session) {
     }
 }
 
-fn spawn_reader(id: TerminalId, mut reader: Box<dyn Read + Send>, tx: mpsc::Sender<Msg>) -> std::io::Result<()> {
+fn spawn_reader(
+    id: TerminalId,
+    mut reader: Box<dyn Read + Send>,
+    tx: mpsc::Sender<Msg>,
+    #[cfg(unix)] mut wait_readable: impl FnMut() -> std::io::Result<()> + Send + 'static,
+) -> std::io::Result<()> {
     std::thread::Builder::new().name(format!("openidea-term-read-{}", id.0)).spawn(move || {
         let mut buf = vec![0u8; 16 * 1024];
         loop {
@@ -419,7 +440,17 @@ fn spawn_reader(id: TerminalId, mut reader: Box<dyn Read + Send>, tx: mpsc::Send
                     }
                 }
                 Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
-                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => std::thread::sleep(BATCH_INTERVAL),
+                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                    #[cfg(unix)]
+                    if let Err(e) = wait_readable()
+                        && e.kind() != std::io::ErrorKind::Interrupted
+                    {
+                        tracing::warn!(%id, "terminal read: {e}");
+                        break;
+                    }
+                    #[cfg(not(unix))]
+                    std::thread::sleep(BATCH_INTERVAL);
+                }
                 // EIO is how Linux/macOS report "slave side closed".
                 #[cfg(unix)]
                 Err(e) if e.raw_os_error() == Some(libc::EIO) => break,