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

@@ -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 {

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<Option<DaemonMsg>> {
         self.output.set_recv_timeout(Some(wait))?;
         match self.output.recv() {

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<Option<DaemonMsg>>;
 
+    /// 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<Box<dyn PaneLeases>> {
@@ -742,9 +749,11 @@ fn read_pane(mut feed: Box<dyn PaneFeed>, down: mpsc::Sender<Down>) {
     // 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<Instant> = None;
+    let mut partial = 0;
     loop {
         if down.is_closed() {
             return;
@@ -766,12 +775,14 @@ fn read_pane(mut feed: Box<dyn PaneFeed>, down: mpsc::Sender<Down>) {
         // 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());
             }
         }

crates/tty7-gateway/tests/bridge.rs

@@ -31,8 +31,9 @@ struct FakeMachine {
     /// Every pane closed, with the machine it was closed on.
     closed: Mutex<Vec<(Option<String>, 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<Option<(Vec<DaemonMsg>, Vec<(usize, Duration)>)>>,
+    /// the messages it goes quiet before, for how long, and whether the
+    /// message is on its way meanwhile.
+    replay: Mutex<Option<(Vec<DaemonMsg>, Vec<Stall>)>>,
 }
 
 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<DaemonMsg>,
     /// 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<Stall>,
     sent: usize,
-    /// Until when it is quiet, once it has gone so.
-    stalled: Option<Instant>,
+    /// 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<Vec<u8>>,
     leases: Option<FakeLeases>,
     answers: std_mpsc::Receiver<DaemonMsg>,
@@ -240,22 +248,31 @@ impl PaneFeed for FakeFeed {
             .map(|l| Box::new(l) as Box<dyn PaneLeases>)
     }
 
+    fn partial(&self) -> usize {
+        self.partial
+    }
+
     fn recv(&mut self, wait: Duration) -> io::Result<Option<DaemonMsg>> {
-        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;