fix: what a pane prints is bounded in the server's memory and time

Cargo.toml

@@ -17,6 +17,9 @@ serde = { version = "1.0.229", features = ["derive"] }
 serde_json = { version = "1.0.151", features = ["preserve_order"] }
 terminput = "0.5.15"
 terminput-crossterm = "0.4.11"
+# The width table alacritty_terminal's input uses, so a pane's zero-width chars are counted as it counts them.
+# The crate is already in the tree via alacritty_terminal, so this adds no new dependency.
+unicode-width = "0.2.2"
 
 [dev-dependencies]
 # tests/egress.rs reads the source to find every door out of nerd; all three are already in Cargo.lock.

docs/privacy-and-security.md

@@ -46,3 +46,5 @@ An agent in a pane runs shell commands as you, so it could type `nerd --stop` or
 The way in is the message socket. Panes of admitted accounts reach it through `/tmp/nerd-<uid>/`, a folder only nerd's owner can write, checked before use, and reach nothing else of nerd. No path in nerd lets an agent change names, routes, isolation or that switch while nerd runs; from an isolated pane, no file can either.
 
 This guards against accidents, not against code that means it: a program that detaches from the pane (a double fork) is no longer its descendant, and any program running as you can stop the server with a signal anyway.
+
+What a pane prints is bounded: it cannot hold more of the server's memory than its `scrollback` and a 2 MiB update buffer allow, and it gets one slice of parsing at a time while other panes wait. An OSC string keeps at most 4096 characters, as GNOME VTE does, its number and `;` included, so a window title keeps 4094; a program saves at most 8 titles, the oldest going first; a repeat request (REP) writes at most one row's width; a character carries at most 30 zero-width characters (combining marks and the like), the limit Unicode's Stream-Safe format sets for marks; OSC 8 hyperlinks are not kept. Output is parsed 512 bytes at a time, and once a turn's 4 ms are spent the other panes go before the next piece; an update that just ended may take up to 8 ms more, so it is drawn whole. A synchronized update (mode 2026) is drawn when it ends, 150 ms after it began, or once it passes 2 MiB. A pane whose program exits closes once its last output is drawn; under a flood of costly sequences, after at most 2 s, without the rest.

src/app/lifecycle.rs

@@ -7,6 +7,12 @@ const SCREEN_TEXT_LIMIT: usize = 1 << 20;
 /// How long an exit notice waits for the pane's last output.
 const EXIT_WAIT: Duration = Duration::from_millis(100);
 
+// ceiling: a pane whose program left closes 2 s after its exit even with output still unparsed (a flood of costly sequences)
+const DRAIN_CAP: Duration = Duration::from_secs(2);
+
+/// How long an Update or exit parses what every pane still has, all panes together; the rest is not drawn.
+const DRAIN_BUDGET: Duration = Duration::from_secs(1);
+
 /// Chunks a pane's output takes per round, so a flooding pane cannot hold back a quiet one.
 pub(super) const QUANTUM: usize = 16;
 
@@ -100,9 +106,10 @@ impl App {
         let mut out = Vec::new();
         let trace = self.trace;
         let theme = self.theme;
+        let deadline = Some(Instant::now() + DRAIN_BUDGET);
         for p in self.spaces.iter_mut().flat_map(|s| s.tabs.iter_mut()).flat_map(|t| t.panes.values_mut()) {
             // Only what is queued now: a pane that keeps printing must not hold the Update.
-            p.pump(crate::pty::QUEUE, trace, &theme);
+            p.drain(deadline, trace, &theme);
             let fd = p.pty.export_fd()?;
             let (since, unseen) = p.term.activity();
             crate::paths::write_private(&crate::session::handover_dump_path(p.id), "tmp", &p.term.dump())?;
@@ -195,8 +202,9 @@ impl App {
         }
         let trace = self.trace;
         let theme = self.theme;
+        let deadline = Some(Instant::now() + DRAIN_BUDGET);
         for p in self.spaces.iter_mut().flat_map(|s| s.tabs.iter_mut()).flat_map(|t| t.panes.values_mut()) {
-            p.pump(crate::pty::QUEUE, trace, &theme);
+            p.drain(deadline, trace, &theme);
             let text = p.term.text(SCREEN_TEXT_LIMIT);
             if let Err(e) = crate::paths::write_private(&crate::session::screen_path(p.id), "tmp", &text) {
                 crate::log::error(&format!("screens: write failed for pane {}: {e}", p.id));
@@ -219,12 +227,13 @@ impl App {
             }
         }
         let mut changed = false;
+        let deadline = Some(start + budget);
         loop {
             let mut more = false;
             for p in self.spaces.iter_mut().flat_map(|s| s.tabs.iter_mut()).flat_map(|t| t.panes.values_mut()) {
-                let took = p.pump(QUANTUM, trace, &theme);
-                changed |= took > 0;
-                more |= took == QUANTUM;
+                let (parsed, left) = p.pump(QUANTUM, deadline, trace, &theme);
+                changed |= parsed;
+                more |= left;
             }
             if !more || start.elapsed() >= budget {
                 return (changed, more);
@@ -325,18 +334,27 @@ impl App {
         }
     }
 
-    /// Closes a pane only once its reader ended or the wait ran out, so its last lines reach the screen.
+    /// Closes a pane once its reader ended or the wait ran out, and `pump_pty` has parsed its last output, so its
+    /// last lines reach the screen; under a flood of costly sequences, DRAIN_CAP after the exit, without the rest.
     pub fn settle_exits(&mut self) -> bool {
-        let trace = self.trace;
-        let theme = self.theme;
+        self.settle_exits_at(Instant::now())
+    }
+
+    pub(super) fn settle_exits_at(&mut self, now: Instant) -> bool {
         let mut closed = false;
         for (id, until) in std::mem::take(&mut self.exits) {
             let Some(p) = self.pane_mut(id) else {
                 self.exit_how.remove(&id);
                 continue;
             };
-            p.pump(crate::pty::QUEUE, trace, &theme);
-            if p.pty.ended || Instant::now() >= until {
+            // Its program is gone: an update it left open will not end.
+            p.term.release_hold();
+            let cut = now >= until.checked_sub(EXIT_WAIT).unwrap_or(until) + DRAIN_CAP;
+            if (p.pty.ended || now >= until) && ((!p.term.queued() && !p.behind) || cut) {
+                if p.term.queued() || p.behind {
+                    crate::log::warn_every("pane output dropped", Duration::from_secs(10), &format!("pane {id}: its program left and its last output took over 2 s to draw; the rest was dropped"));
+                    p.term.discard();
+                }
                 // The pane's process is its shell: when that leaves, there is no pane left.
                 let how = self.exit_how.remove(&id).unwrap_or_else(|| "it left during the update, how is not known".into());
                 self.drop_pane(id, Some(&how));
@@ -424,9 +442,12 @@ impl App {
         let mut timers = false;
         let mut stalled = false;
         let mut panes = HashMap::new();
+        let now = Instant::now();
         for (si, sp) in self.spaces.iter_mut().enumerate() {
             for (ti, tab) in sp.tabs.iter_mut().enumerate() {
                 for p in tab.panes.values_mut() {
+                    // An update that never ends is drawn as it is; the next pump_pty parses it.
+                    p.term.release_stale(now);
                     stalled |= std::mem::take(&mut p.pty.stalled);
                     let before = (p.term.shown(), p.agent.clone(), p.command.clone(), p.activity_key);
                     changed |= p.term.tick_state();

src/app/mod.rs

@@ -1495,6 +1495,8 @@ mod tests {
             activity_key: [None, None],
             mouse_left_on: false,
             isolation: Default::default(),
+            unanswered: None,
+            behind: false,
         }
     }
 
@@ -5172,7 +5174,195 @@ mod tests {
         assert_eq!(app.pump_pty(Duration::ZERO), (true, true), "one round leaves the flood queued");
         assert!(app.pane(2).unwrap().term.screen().contents().contains("quiet"), "the quiet pane had its turn");
         let p = app.spaces[0].tabs[0].panes.get_mut(&1).unwrap();
-        assert_eq!(std::iter::from_fn(|| p.pty.next()).count(), crate::pty::QUEUE - QUANTUM, "the flood gave up one quantum");
+        assert!(p.term.queued(), "the flood stopped inside its first chunk");
+        assert_eq!(std::iter::from_fn(|| p.pty.next()).count(), crate::pty::QUEUE - 1, "the flood gave up after one slice");
+    }
+
+    #[test]
+    fn three_flooding_panes_each_parse_one_slice_a_round() {
+        let mut app = with_one_space();
+        let mut keep = Vec::new();
+        for (tab, id) in [(0, 1), (0, 2), (1, 3)] {
+            let (pane, tx) = fed_pane(&app, id, None);
+            for _ in 0..crate::pty::QUEUE {
+                tx.send(vec![b'y'; 8192]).unwrap();
+            }
+            app.spaces[0].tabs[tab].panes.insert(id, pane);
+            keep.push(tx);
+        }
+        let (quiet, qtx) = fed_pane(&app, 4, None);
+        app.spaces[0].tabs[1].panes.insert(4, quiet);
+        qtx.send(b"quiet".to_vec()).unwrap();
+        assert_eq!(app.pump_pty(Duration::ZERO), (true, true));
+        for id in [1, 2, 3] {
+            assert_eq!(app.pane(id).unwrap().term.queued_len(), 8192 - crate::term::SLICE, "pane {id}");
+        }
+        assert!(app.pane(4).unwrap().term.screen().contents().contains("quiet"));
+    }
+
+    #[test]
+    fn a_flood_in_chunks_smaller_than_a_slice_gives_way_after_one() {
+        let mut app = with_one_space();
+        let (flood, tx) = fed_pane(&app, 1, None);
+        app.spaces[0].tabs[0].panes.insert(1, flood);
+        for _ in 0..crate::pty::QUEUE {
+            tx.send(vec![b'y'; crate::term::SLICE / 2]).unwrap();
+        }
+        assert_eq!(app.pump_pty(Duration::ZERO), (true, true));
+        let p = app.spaces[0].tabs[0].panes.get_mut(&1).unwrap();
+        assert_eq!(std::iter::from_fn(|| p.pty.next()).count(), crate::pty::QUEUE - 1);
+    }
+
+    /// A stream sent as 8 KiB chunks and pumped to the end, each turn with `deadline`.
+    fn pumped(app: &App, raw: &[u8], deadline: Option<Instant>) -> (String, Vec<u8>) {
+        let (mut pane, tx) = fed_pane(app, 1, None);
+        pane.term = Term::new(120, 40, 1000);
+        let mut chunks = raw.chunks(8192);
+        let theme = app.theme;
+        loop {
+            let sent = chunks.next().map(|c| tx.send(c.to_vec()).unwrap()).is_some();
+            let (_, left) = pane.pump(QUANTUM, deadline, false, &theme);
+            if !sent && !left {
+                break;
+            }
+        }
+        pane.term.finish();
+        (pane.term.full_state(), pane.term.dump())
+    }
+
+    #[test]
+    fn output_parsed_a_slice_a_turn_ends_as_it_does_parsed_at_once() {
+        let app = with_one_space();
+        for (name, raw) in crate::term::corpus() {
+            let sliced = pumped(&app, &raw, Some(Instant::now()));
+            let whole = pumped(&app, &raw, None);
+            assert!(sliced == whole, "{name}");
+        }
+    }
+
+    #[test]
+    fn a_chunk_is_answered_once_after_its_last_slice() {
+        use std::io::Read;
+        let mut app = with_one_space();
+        let (ours, theirs) = std::os::unix::net::UnixStream::pair().unwrap();
+        let (pane, tx) = fed_pane(&app, 1, Some(std::fs::File::from(std::os::fd::OwnedFd::from(theirs))));
+        app.spaces[0].tabs[0].panes.insert(1, pane);
+        let chunk = [&b"\r\n"[..], &vec![b'x'; 3 * crate::term::SLICE], b"\x1b[6n"].concat();
+        let mut want = Term::new(80, 24, 100);
+        want.feed(&chunk);
+        let (row, col) = want.screen().cursor_position();
+        tx.send(chunk).unwrap();
+        let mut turns = 0;
+        while app.pump_pty(Duration::ZERO).1 {
+            turns += 1;
+        }
+        assert!(turns >= 3, "{turns} turns");
+        ours.set_read_timeout(Some(Duration::from_millis(300))).unwrap();
+        let mut got = Vec::new();
+        let mut b = [0u8; 64];
+        while let Ok(n) = (&ours).read(&mut b) {
+            got.extend_from_slice(&b[..n]);
+        }
+        assert_eq!(String::from_utf8_lossy(&got), format!("\x1b[{};{}R", row + 1, col + 1), "one answer, from after the whole chunk");
+    }
+
+    #[test]
+    fn an_update_that_ends_is_parsed_whole_beside_a_flood() {
+        let mut app = with_one_space();
+        let (flood, ftx) = fed_pane(&app, 1, None);
+        let (frame, tx) = fed_pane(&app, 2, None);
+        app.spaces[0].tabs[0].panes.insert(1, flood);
+        app.spaces[0].tabs[1].panes.insert(2, frame);
+        for _ in 0..crate::pty::QUEUE {
+            ftx.send(b"\x1b#8".repeat(2730)).unwrap();
+        }
+        let update = [&b"\x1b[?2026h"[..], &b"0123456789abcdef\r\n".repeat(40 * 1024 / 18), b"\x1b[?2026l"].concat();
+        for c in update.chunks(8192) {
+            tx.send(c.to_vec()).unwrap();
+        }
+        app.pump_pty(Duration::ZERO);
+        assert!(app.pane(2).unwrap().term.holding(), "the update is held while it lasts");
+        // The round that takes its end: the update is released and parsed in that same turn.
+        for _ in 0..8 {
+            app.pump_pty(Duration::ZERO);
+            if !app.pane(2).unwrap().term.holding() {
+                break;
+            }
+        }
+        let t = &app.pane(2).unwrap().term;
+        assert!(!t.holding() && !t.queued(), "the ended update was parsed whole in the turn that took its end");
+        assert!(t.screen().contents().contains("0123456789abcdef"));
+        assert!(app.pane(1).unwrap().term.queued(), "the flood still gave way after one slice");
+    }
+
+    /// S6: `cargo test --release --bin nerd -- --ignored --nocapture measure_rounds`.
+    #[test]
+    #[ignore]
+    fn measure_rounds() {
+        let mut app = with_one_space();
+        let mut keep = Vec::new();
+        for id in 1..=4 {
+            let (mut pane, tx) = fed_pane(&app, id, None);
+            pane.term = Term::new(600, 170, 1000);
+            for _ in 0..8 {
+                tx.send(b"\x1b#8".repeat(2730)).unwrap();
+            }
+            app.spaces[0].tabs[id % 2].panes.insert(id, pane);
+            keep.push(tx);
+        }
+        let frame = crate::term::corpus().into_iter().find(|(n, _)| n == "agent-wide").unwrap().1;
+        let (start, end) = (frame.windows(8).position(|w| w == b"\x1b[?2026h").unwrap(), frame.windows(8).position(|w| w == b"\x1b[?2026l").unwrap() + 8);
+        let (mut pane, tx) = fed_pane(&app, 5, None);
+        pane.term = Term::new(600, 170, 1000);
+        app.spaces[0].tabs[1].panes.insert(5, pane);
+        tx.send(frame[start..end].to_vec()).unwrap();
+        let (mut longest, mut rounds) = (Duration::ZERO, 0);
+        loop {
+            let t = Instant::now();
+            let (_, more) = app.pump_pty(PUMP_BUDGET);
+            longest = longest.max(t.elapsed());
+            rounds += 1;
+            if rounds == 1 {
+                println!("frame of {} bytes beside four DECALN floods: parsed whole in round 1: {}", end - start, !app.pane(5).unwrap().term.queued());
+            }
+            if !more {
+                break;
+            }
+        }
+        println!("four panes flooding DECALN at 600x170: {rounds} rounds, longest pump_pty {:.1} ms", longest.as_secs_f64() * 1e3);
+    }
+
+    #[test]
+    fn a_pane_whose_program_left_parses_its_flood_a_turn_at_a_time_then_closes() {
+        let mut app = with_one_space();
+        let (mut pane, tx) = fed_pane(&app, 1, None);
+        pane.argv = vec!["claude".into()];
+        app.spaces[0].tabs[0].panes.insert(1, pane);
+        for _ in 0..crate::pty::QUEUE {
+            tx.send(b"\x1b#8".repeat(2730)).unwrap();
+        }
+        app.exited((1, None));
+        let t0 = Instant::now();
+        app.pump_pty(Duration::ZERO);
+        assert!(!app.settle_exits_at(t0 + Duration::from_millis(150)), "past its wait, output still queued: it stays");
+        assert!(app.pane(1).unwrap().term.queued());
+        drop(tx);
+        while app.pump_pty(Duration::from_millis(50)).1 {}
+        assert!(app.settle_exits_at(t0 + Duration::from_millis(200)), "the reader ended and the queue drained");
+        assert!(app.pane(1).is_none());
+
+        let (mut pane, tx) = fed_pane(&app, 2, None);
+        pane.argv = vec!["claude".into()];
+        app.spaces[0].tabs[0].panes.insert(2, pane);
+        for _ in 0..crate::pty::QUEUE {
+            tx.send(b"\x1b#8".repeat(2730)).unwrap();
+        }
+        app.exited((2, None));
+        let t0 = Instant::now();
+        app.pump_pty(Duration::ZERO);
+        assert!(!app.settle_exits_at(t0 + Duration::from_millis(1900)));
+        assert!(app.settle_exits_at(t0 + Duration::from_millis(2100)), "2 s after the exit the rest goes");
+        assert!(app.pane(2).is_none());
     }
 
     #[test]
@@ -5184,7 +5374,7 @@ mod tests {
         app.spaces[0].tabs[0].panes.insert(1, pane);
         tx.send(b"ab\x1b[6n".to_vec()).unwrap();
         tx.send(b"\r\n\x1b[6n".to_vec()).unwrap();
-        app.pump_pty(Duration::ZERO);
+        while app.pump_pty(Duration::ZERO).1 {}
         ours.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
         let mut got = Vec::new();
         while got.len() < 12 {
@@ -5202,12 +5392,15 @@ mod tests {
         app.spaces[0].tabs[0].panes.insert(1, pane);
         tx.send(b"queued before the notice\r\n".to_vec()).unwrap();
         app.exited((1, None));
+        app.pump_pty(PUMP_BUDGET);
         assert!(!app.settle_exits(), "its reader is still there");
         tx.send(b"resume with: claude --resume x".to_vec()).unwrap();
+        app.pump_pty(PUMP_BUDGET);
         assert!(!app.settle_exits());
         let text = app.pane(1).unwrap().term.screen().contents();
         assert!(text.contains("queued before the notice") && text.contains("claude --resume x"), "{text}");
         drop(tx);
+        app.pump_pty(PUMP_BUDGET);
         assert!(app.settle_exits(), "the reader ended: the pane goes at once");
         assert!(app.pane(1).is_none());
 
@@ -5398,7 +5591,7 @@ mod tests {
             let mut first_drop = None;
             for i in 0..crate::pty::WRITES + 64 {
                 tx.send(b"\x1b[6n".repeat(2000)).unwrap();
-                app.pump_pty(Duration::ZERO);
+                while app.pump_pty(Duration::ZERO).1 {}
                 if first_drop.is_none() && log().contains("replies dropped") {
                     first_drop = Some(i);
                 }