diff --git a/crates/tty7-core/src/client/pane.rs b/crates/tty7-core/src/client/pane.rs index 01208be6..52322eee 100644 --- a/crates/tty7-core/src/client/pane.rs +++ b/crates/tty7-core/src/client/pane.rs @@ -412,6 +412,11 @@ impl PaneOutput { self.reader.set_read_timeout(wait) } + /// How much of a message has been read without the rest of it yet. + pub fn partial(&self) -> usize { + self.buffered.len() + } + fn refusal_check(&mut self, request: &str, pane_id: u64) -> io::Result<()> { let mut scratch = [0u8; 4096]; loop { diff --git a/crates/tty7-gateway/src/daemon.rs b/crates/tty7-gateway/src/daemon.rs index 5beb5d2f..36572d96 100644 --- a/crates/tty7-gateway/src/daemon.rs +++ b/crates/tty7-gateway/src/daemon.rs @@ -400,6 +400,10 @@ impl PaneFeed for Observed { self.leases.take() } + fn partial(&self) -> usize { + self.output.partial() + } + fn recv(&mut self, wait: Duration) -> io::Result> { self.output.set_recv_timeout(Some(wait))?; match self.output.recv() { diff --git a/crates/tty7-gateway/src/serve.rs b/crates/tty7-gateway/src/serve.rs index 54b0a995..a869c356 100644 --- a/crates/tty7-gateway/src/serve.rs +++ b/crates/tty7-gateway/src/serve.rs @@ -93,6 +93,13 @@ pub trait PaneFeed: Send { /// The next message, or `Ok(None)` if none arrived within `wait`. fn recv(&mut self, wait: Duration) -> io::Result>; + /// How much of the next message has come in so far, without the rest of + /// it: a long one on a slow link is still arriving while `recv` comes + /// back empty. + fn partial(&self) -> usize { + 0 + } + /// The way back to the daemon on the same connection, for a take-over. /// Taken once, before the feed goes to its reader thread. fn leases(&mut self) -> Option> { @@ -718,6 +725,11 @@ async fn pane_stream( /// stale to stop, and the longest it waits while output keeps coming. const DRAW_SETTLE: Duration = Duration::from_millis(30); const DRAW_LIMIT: Duration = Duration::from_millis(200); +/// How long the replay that opens a pane's feed may go quiet before it is +/// taken to be over. The daemon marks no end to it: usually the pane's status +/// follows and says so, but a pane with none to report ends on its last +/// segment, and only this tells it from one held up on a slow link. +const REPLAY_SETTLE: Duration = Duration::from_millis(500); fn read_pane(mut feed: Box, down: mpsc::Sender) { // The daemon reports the agent and its status in separate messages; the @@ -731,6 +743,17 @@ fn read_pane(mut feed: Box, down: mpsc::Sender) { let mut mirror: Option = None; let mut size = (80, 24); let mut stale: Option = Some(Instant::now()); + // Whether the feed is still in the replay it opens with (sizes and + // segments of the ring, `ReplayRing::replay`), and when it last said + // anything. The ring runs oldest first, so a drawing made before its end + // shows the phone what the pane had on it long ago — a restored session's + // old screen, say — however long the replay takes over a slow link. The + // quiet only counts once the feed has said something: a link slow to + // deliver the first frame is not a replay that has ended. Nor is one + // still delivering a segment: part of a message is something said. + let mut replaying = true; + let mut heard: Option = None; + let mut partial = 0; loop { if down.is_closed() { return; @@ -749,13 +772,28 @@ fn read_pane(mut feed: Box, down: mpsc::Sender) { return; } }; + // The replay is over at the first thing it does not consist of, or + // once it has been quiet for long enough. + if replaying { + let said = msg.is_some() || feed.partial() != partial; + partial = feed.partial(); + replaying = match &msg { + Some(DaemonMsg::Size(_) | DaemonMsg::Snapshot(_)) => true, + Some(_) => false, + None => said || heard.is_none_or(|at| at.elapsed() < REPLAY_SETTLE), + }; + if said { + heard = Some(Instant::now()); + } + } // Drawn once the burst is over, or anything other than output is - // about to be said, or it has waited long enough. + // about to be said, or it has waited long enough — but never before + // the replay is all in. let burst = matches!( msg, Some(DaemonMsg::Size(_) | DaemonMsg::Snapshot(_) | DaemonMsg::Output(_)) ); - if stale.is_some_and(|since| !burst || since.elapsed() >= DRAW_LIMIT) { + if !replaying && stale.is_some_and(|since| !burst || since.elapsed() >= DRAW_LIMIT) { stale = None; if let Some(mirror) = &mirror { let (cols, rows) = size; @@ -781,7 +819,17 @@ fn read_pane(mut feed: Box, down: mpsc::Sender) { stale.get_or_insert_with(Instant::now); continue; } - DaemonMsg::Snapshot(bytes) | DaemonMsg::Output(bytes) => { + // A segment of the replay reaches the phone only in a drawing: it + // is a picture to rebuild the screen from, which the phone could + // not tell from what the pane says next. + DaemonMsg::Snapshot(bytes) => { + mirror + .get_or_insert_with(|| Mirror::new(size.0, size.1)) + .feed(&bytes); + stale.get_or_insert_with(Instant::now); + continue; + } + DaemonMsg::Output(bytes) => { let mirror = mirror.get_or_insert_with(|| Mirror::new(size.0, size.1)); if mirror.feed(&bytes) { stale.get_or_insert_with(Instant::now); diff --git a/crates/tty7-gateway/tests/bridge.rs b/crates/tty7-gateway/tests/bridge.rs index 22b10f76..d984b080 100644 --- a/crates/tty7-gateway/tests/bridge.rs +++ b/crates/tty7-gateway/tests/bridge.rs @@ -5,7 +5,7 @@ use std::io; use std::sync::mpsc as std_mpsc; use std::sync::{Arc, Mutex}; -use std::time::Duration; +use std::time::{Duration, Instant}; use iroh::Endpoint; use iroh::endpoint::presets; @@ -30,6 +30,10 @@ struct FakeMachine { leased: Arc>>, /// Every pane closed, with the machine it was closed on. closed: Mutex, u64)>>, + /// A replay to give the next pane opened instead of the usual one, and + /// the messages it goes quiet before, for how long, and whether the + /// message is on its way meanwhile. + replay: Mutex, Vec)>>, } impl FakeMachine { @@ -51,6 +55,7 @@ impl FakeMachine { typed: Mutex::new(None), leased: Arc::default(), closed: Mutex::default(), + replay: Mutex::default(), } } } @@ -110,18 +115,20 @@ impl Backend for FakeMachine { answer, seen: self.leased.clone(), }); + let (replay, stall) = self.replay.lock().unwrap().take().unwrap_or_else(|| { + ( + vec![DaemonMsg::Size(size(100, 30)), DaemonMsg::Snapshot(prompt)], + Vec::new(), + ) + }); Ok(Box::new(FakeFeed { leases, answers, - replay: vec![ - DaemonMsg::Size(WinSize { - cols: 100, - rows: 30, - cell_w: 8, - cell_h: 16, - }), - DaemonMsg::Snapshot(prompt), - ], + replay: replay.into(), + stalls: stall, + sent: 0, + stalled: None, + partial: 0, typed: rx, })) } @@ -184,8 +191,29 @@ impl Backend for FakeMachine { } } +fn size(cols: u16, rows: u16) -> WinSize { + WinSize { + cols, + rows, + cell_w: 8, + cell_h: 16, + } +} + +/// Before which message of a replay it is held up, for how long, and whether +/// that message trickles in meanwhile rather than nothing arriving at all. +type Stall = (usize, Duration, bool); + struct FakeFeed { - replay: Vec, + replay: std::collections::VecDeque, + /// The replay goes quiet before its messages at these indexes, the way + /// one held up on a slow link does; and how many it has given so far. + stalls: Vec, + sent: usize, + /// Until when it is quiet, once it has gone so, and whether the next + /// message is coming in meanwhile; how much of it has. + stalled: Option<(Instant, bool)>, + partial: usize, typed: std_mpsc::Receiver>, leases: Option, answers: std_mpsc::Receiver, @@ -220,9 +248,32 @@ impl PaneFeed for FakeFeed { .map(|l| Box::new(l) as Box) } + fn partial(&self) -> usize { + self.partial + } + fn recv(&mut self, wait: Duration) -> io::Result> { - if !self.replay.is_empty() { - return Ok(Some(self.replay.remove(0))); + if let Some(at) = self.stalls.iter().position(|(at, ..)| *at == self.sent) { + let (_, quiet, trickle) = self.stalls.remove(at); + self.stalled = Some((Instant::now() + quiet, trickle)); + } + // Quiet the way a socket is: each wait runs out with nothing read, + // or with some of a message read but not all of it. + if let Some((until, trickle)) = self.stalled { + let left = until.saturating_duration_since(Instant::now()); + if !left.is_zero() { + std::thread::sleep(left.min(wait)); + if trickle { + self.partial += 1; + } + return Ok(None); + } + self.stalled = None; + } + if let Some(msg) = self.replay.pop_front() { + self.sent += 1; + self.partial = 0; + return Ok(Some(msg)); } if let Ok(answer) = self.answers.try_recv() { return Ok(Some(answer)); @@ -389,6 +440,155 @@ async fn a_paired_phone_reads_the_tree_and_drives_a_pane() { assert!(err.to_string().contains("no such pane 99"), "{err}"); } +/// What the daemon replays of a pane whose ring holds an old session ahead of +/// the current one, the way a pane restored from scrollback does: each +/// segment at its own size, the later one starting from a reset. +fn two_sessions() -> Vec { + vec![ + DaemonMsg::Size(size(100, 30)), + DaemonMsg::Snapshot(b"OLD SESSION\r\n".to_vec()), + DaemonMsg::Size(size(100, 30)), + DaemonMsg::Snapshot(b"\x1bc NEW SESSION $ ".to_vec()), + ] +} + +/// Everything the phone is sent for a pane until `until` says to stop. +async fn screen_until( + screen: &mut tty7_mobile_client::PaneReader, + until: impl Fn(&PaneItem) -> bool, +) -> Vec { + let mut seen = Vec::new(); + loop { + let item = within(screen.next()) + .await + .unwrap() + .expect("stream ends early"); + let done = until(&item); + seen.push(item); + if done { + return seen; + } + } +} + +/// The first drawing among what the phone was sent, and that nothing else it +/// was sent carries the replay past the drawing. +fn first_drawing(seen: &[PaneItem]) -> String { + let mut outputs = seen.iter().filter_map(|item| match item { + PaneItem::Output(bytes) => Some(bytes), + _ => None, + }); + let first = drawing(outputs.next().cloned().map(PaneItem::Output)); + for bytes in outputs { + let text = String::from_utf8_lossy(bytes); + assert!( + !text.contains("SESSION"), + "the replay reached the phone raw, past its drawing: {text:?} (all: {seen:?})" + ); + } + first +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_replay_held_up_mid_ring_is_drawn_whole() { + let rig = Rig::new().await; + // The link goes quiet between the two segments for far longer than a + // drawing waits for live output to settle, but well short of what a + // replay is given (`REPLAY_SETTLE`). + let mut replay = two_sessions(); + replay.push(DaemonMsg::Cwd("/src/demo".into())); + *rig.machine.replay.lock().unwrap() = + Some((replay, vec![(3, Duration::from_millis(150), false)])); + let session = within(rig.paired()).await; + let (_keys, mut screen) = within(session.pane(None, 1)).await.unwrap(); + let seen = screen_until(&mut screen, |item| { + matches!(item, PaneItem::Event(PaneEvent::Cwd { .. })) + }) + .await; + let first = first_drawing(&seen); + assert!( + first.contains("NEW SESSION") && !first.contains("OLD SESSION"), + "the phone's first picture of the pane has to be its current screen, \ + not the front of the ring: {first:?} (all: {seen:?})" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_one_segment_replay_held_up_is_drawn_not_streamed() { + let rig = Rig::new().await; + // Held up after the size, before the only segment; and nothing follows + // it, so the replay's end is only told by the quiet after it. + let replay = vec![ + DaemonMsg::Size(size(100, 30)), + DaemonMsg::Snapshot(b"\x1bc NEW SESSION $ ".to_vec()), + ]; + *rig.machine.replay.lock().unwrap() = + Some((replay, vec![(1, Duration::from_millis(150), false)])); + let session = within(rig.paired()).await; + let (_keys, mut screen) = within(session.pane(None, 1)).await.unwrap(); + let seen = screen_until(&mut screen, |item| { + matches!(item, PaneItem::Output(bytes) if String::from_utf8_lossy(bytes).contains("SESSION")) + }) + .await; + let first = first_drawing(&seen); + assert!( + first.contains("NEW SESSION"), + "drawn before the replay was in, then the rest streamed raw past the \ + drawing: {first:?} (all: {seen:?})" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_replay_slow_to_start_is_still_drawn_whole() { + let rig = Rig::new().await; + // Nothing at all for longer than a replay is given to go quiet, then a + // pause mid-ring: the wait for the first frame is not a replay that ended. + let mut replay = two_sessions(); + replay.push(DaemonMsg::Cwd("/src/demo".into())); + *rig.machine.replay.lock().unwrap() = Some(( + replay, + vec![ + (0, Duration::from_millis(700), false), + (3, Duration::from_millis(150), false), + ], + )); + let session = within(rig.paired()).await; + let (_keys, mut screen) = within(session.pane(None, 1)).await.unwrap(); + let seen = screen_until(&mut screen, |item| { + matches!(item, PaneItem::Event(PaneEvent::Cwd { .. })) + }) + .await; + let first = first_drawing(&seen); + assert!( + first.contains("NEW SESSION") && !first.contains("OLD SESSION"), + "the phone's first picture of the pane has to be its current screen: \ + {first:?} (all: {seen:?})" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_segment_still_coming_in_is_not_the_end_of_the_replay() { + let rig = Rig::new().await; + // The last segment takes far longer than a replay is given to go quiet, + // the link pausing now and then as it comes: what has arrived of it so + // far is the replay still going, not over. + let mut replay = two_sessions(); + replay.push(DaemonMsg::Cwd("/src/demo".into())); + *rig.machine.replay.lock().unwrap() = + Some((replay, vec![(3, Duration::from_millis(900), true)])); + let session = within(rig.paired()).await; + let (_keys, mut screen) = within(session.pane(None, 1)).await.unwrap(); + let seen = screen_until(&mut screen, |item| { + matches!(item, PaneItem::Event(PaneEvent::Cwd { .. })) + }) + .await; + let first = first_drawing(&seen); + assert!( + first.contains("NEW SESSION") && !first.contains("OLD SESSION"), + "drawn while the last segment was still coming in: {first:?} (all: {seen:?})" + ); +} + #[tokio::test(flavor = "multi_thread")] async fn typing_that_fails_is_reported_not_dropped() { let rig = Rig::new().await;