From 847f5334bb9bb7c30127a21cd5973076b83bbae6 Mon Sep 17 00:00:00 2001 From: l0ng-ai <24760907+l0ng-ai@users.noreply.github.com> Date: Fri, 16 Oct 2026 12:02:05 +0800 Subject: [PATCH] fix(gateway): count a segment still arriving as the replay going on A replay segment can be megabytes over a slow link. PaneOutput buffers a frame until it is whole and returns WouldBlock on any read gap longer than the 30ms draw settle, so the serve loop saw silence mid-segment and, once half a second had passed since the last whole message, drew the mirror before the replay was in. Bytes of the next message now count as the feed having said something. --- crates/tty7-core/src/client/pane.rs | 5 +++ crates/tty7-gateway/src/daemon.rs | 4 ++ crates/tty7-gateway/src/serve.rs | 17 +++++-- crates/tty7-gateway/tests/bridge.rs | 70 +++++++++++++++++++++++------ 4 files changed, 79 insertions(+), 17 deletions(-) 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 9077aafb..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> { @@ -742,9 +749,11 @@ fn read_pane(mut feed: Box, down: mpsc::Sender) { // 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. + // 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; @@ -766,12 +775,14 @@ fn read_pane(mut feed: Box, down: mpsc::Sender) { // 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 => heard.is_none_or(|at| at.elapsed() < REPLAY_SETTLE), + None => said || heard.is_none_or(|at| at.elapsed() < REPLAY_SETTLE), }; - if msg.is_some() { + if said { heard = Some(Instant::now()); } } diff --git a/crates/tty7-gateway/tests/bridge.rs b/crates/tty7-gateway/tests/bridge.rs index 3555a992..d984b080 100644 --- a/crates/tty7-gateway/tests/bridge.rs +++ b/crates/tty7-gateway/tests/bridge.rs @@ -31,8 +31,9 @@ struct FakeMachine { /// 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. - replay: Mutex, Vec<(usize, Duration)>)>>, + /// the messages it goes quiet before, for how long, and whether the + /// message is on its way meanwhile. + replay: Mutex, Vec)>>, } impl FakeMachine { @@ -127,6 +128,7 @@ impl Backend for FakeMachine { stalls: stall, sent: 0, stalled: None, + partial: 0, typed: rx, })) } @@ -198,14 +200,20 @@ fn size(cols: u16, rows: u16) -> WinSize { } } +/// 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: 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<(usize, Duration)>, + stalls: Vec, sent: usize, - /// Until when it is quiet, once it has gone so. - stalled: Option, + /// 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, @@ -240,22 +248,31 @@ 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 let Some(at) = self.stalls.iter().position(|(at, _)| *at == self.sent) { - let (_, quiet) = self.stalls.remove(at); - self.stalled = Some(Instant::now() + quiet); + 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. - if let Some(until) = self.stalled { + // 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() { @@ -480,7 +497,8 @@ async fn a_replay_held_up_mid_ring_is_drawn_whole() { // 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))])); + *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| { @@ -504,7 +522,8 @@ async fn a_one_segment_replay_held_up_is_drawn_not_streamed() { 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))])); + *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| { @@ -529,8 +548,8 @@ async fn a_replay_slow_to_start_is_still_drawn_whole() { *rig.machine.replay.lock().unwrap() = Some(( replay, vec![ - (0, Duration::from_millis(700)), - (3, Duration::from_millis(150)), + (0, Duration::from_millis(700), false), + (3, Duration::from_millis(150), false), ], )); let session = within(rig.paired()).await; @@ -547,6 +566,29 @@ async fn a_replay_slow_to_start_is_still_drawn_whole() { ); } +#[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;