fix(cmux-tui-core): the reader's end, give-up and the owner's detach end the input port's reconnect wait (R-DAEMON-STATE S1.6c)

cmux-tui/crates/cmux-tui-core/src/surface/hosted_reader.rs

@@ -326,7 +326,7 @@ struct WakeInputOnReaderEnd(Arc<Surface>);
 impl Drop for WakeInputOnReaderEnd {
     fn drop(&mut self) {
         if let Some(pty) = self.0.as_pty() {
-            pty.stream_progress.notify();
+            pty.reader_gave_up();
             // A reader that gave up on its host leaves the terminal `failed`:
             // push the tab here, with no lock held, so clients show it.
             if matches!(

cmux-tui/crates/cmux-tui-core/src/surface/hosted_reader/reconnect.rs

@@ -299,7 +299,7 @@ impl HostedReader {
                 if !reconnect_mux.terminal_host_connection_lost(surface.id, &identity) {
                     pty.host_connection_state
                         .store(TerminalHostConnectionState::Failed as u8, Ordering::Release);
-                    pty.stream_progress.notify();
+                    pty.reader_gave_up();
                     return None;
                 }
                 // A resync's reconnect that failed is a loss clients have not heard of.

cmux-tui/crates/cmux-tui-core/src/surface/input_port.rs

@@ -181,7 +181,8 @@ struct PortState {
     /// The writer waits for a reconnect: a barrier does not wait then; the connection it would write to is gone, and
     /// the install it waits for may need the runtime lock that barrier's caller holds.
     reconnecting: bool,
-    /// Moves each time the terminal publishes a route ([`InputPort::route_changed`]): ends the writer's reconnect wait.
+    /// Moves each time the terminal publishes a route ([`InputPort::route_changed`]) or will install none
+    /// ([`InputPort::reconnect_ended`]): ends the writer's reconnect wait.
     route_epoch: u64,
     /// Tests: the writer waits here before it writes each frame.
     #[cfg(test)]
@@ -314,6 +315,17 @@ impl InputPort {
         self.shared.changed.notify_all();
     }
 
+    /// The terminal will install no host connection for now (its reader ended or gave the host up, or its owner
+    /// detaches it): a writer waiting for a reconnect checks again at once and fails its frame, as the old path's
+    /// write failed then (#35082 review 2).
+    pub(crate) fn reconnect_ended(&self) {
+        let mut state = self.shared.state.lock().unwrap_or_else(PoisonError::into_inner);
+        state.route_epoch = state.route_epoch.wrapping_add(1);
+        state.reconnecting = false;
+        drop(state);
+        self.shared.changed.notify_all();
+    }
+
     fn enqueue_frame(
         &self,
         terminal: &Arc<PtyTerminalRuntime>,
@@ -517,8 +529,9 @@ fn run_writer(shared: &PortShared, terminal: &Weak<PtyTerminalRuntime>) {
     }
 }
 
-/// Waits until the terminal publishes a route after the one seen at epoch `observed` ([`InputPort::route_changed`]),
-/// until the port closes or until `deadline`; true when a new route came. Marks the writer waiting for a reconnect
+/// Waits until the terminal publishes a route after the one seen at epoch `observed` ([`InputPort::route_changed`])
+/// or ends its reconnect ([`InputPort::reconnect_ended`]), until the port closes or until `deadline`; true when the
+/// epoch moved (the caller checks the route and the terminal's state again). Marks the writer waiting for a reconnect
 /// meanwhile, so barriers do not wait for it (a new route clears the mark at once).
 fn wait_for_route(shared: &PortShared, observed: u64, deadline: Instant) -> bool {
     let mut state = shared.state.lock().unwrap_or_else(PoisonError::into_inner);

cmux-tui/crates/cmux-tui-core/src/surface/reconnect_backoff.rs

@@ -85,7 +85,7 @@ impl TerminalHostReconnectBackoff {
                 host.disconnect();
             }
             // Input waiting out the reconnect (surface/input.rs) fails now.
-            pty.stream_progress.notify();
+            pty.reader_gave_up();
             return false;
         };
         #[cfg(test)]

cmux-tui/crates/cmux-tui-core/src/surface/shutdown.rs

@@ -134,6 +134,9 @@ impl Surface {
                 // racing termination by reconnecting a Surface that no longer
                 // exists in the mux topology.
                 pty.owner_detaching.store(true, Ordering::Release);
+                // A port writer waiting for a reconnect fails its frame now (#35082 review 2); only the port
+                // lock, which no holder of the runtime lock waits for.
+                pty.input_port.reconnect_ended();
                 {
                     let mut runtime = pty.runtime.lock().unwrap();
                     match &mut *runtime {
@@ -175,6 +178,9 @@ impl Surface {
             Surface::Pty(pty) => {
                 if let PtyRuntime::Hosted(host) = &*pty.runtime.lock().unwrap() {
                     pty.owner_detaching.store(true, Ordering::Release);
+                    // A port writer waiting for a reconnect fails its frame now (#35082 review 2); only the port
+                    // lock, which no holder of the runtime lock waits for.
+                    pty.input_port.reconnect_ended();
                     host.disconnect();
                     return;
                 }
@@ -187,6 +193,9 @@ impl Surface {
                     &*pty.runtime.lock().unwrap()
                 {
                     pty.owner_detaching.store(true, Ordering::Release);
+                    // A port writer waiting for a reconnect fails its frame now (#35082 review 2); only the port
+                    // lock, which no holder of the runtime lock waits for.
+                    pty.input_port.reconnect_ended();
                     holder.detach();
                     return;
                 }
@@ -208,6 +217,9 @@ impl Surface {
             let runtime = pty.runtime.lock().unwrap();
             if let PtyRuntime::Hosted(host) = &*runtime {
                 pty.owner_detaching.store(true, Ordering::Release);
+                // A port writer waiting for a reconnect fails its frame now (#35082 review 2); only the port
+                // lock, which no holder of the runtime lock waits for.
+                pty.input_port.reconnect_ended();
                 let mut gap = None;
                 if host.supports_journal_detach_fence()
                     && let Err(error) = host.detach_for_daemon_shutdown_until(deadline)

cmux-tui/crates/cmux-tui-core/src/surface/terminal_runtime.rs

@@ -79,6 +79,14 @@ impl PtyTerminalRuntime {
         self.input_port.route_changed();
     }
 
+    /// The reader will install no host connection now (it ended, gave the host up, or the owner detaches the
+    /// terminal): input that waits out a lost connection fails now, as the old path's write failed then.
+    pub(super) fn reader_gave_up(&self) {
+        self.stream_progress.notify();
+        // The input port's writer waits on its own route epoch, not on the stream (#35082 review 2).
+        self.input_port.reconnect_ended();
+    }
+
     /// The reader shut the current host connection down and keeps retrying (`Failed`): input is refused until a
     /// replacement connection is installed ([`Self::replace_runtime`] publishes its route), as the old path's write
     /// to the shut connection failed. Takes the held runtime so it is ordered with every replacement.

cmux-tui/crates/cmux-tui-core/src/surface/tests/hosted.rs

@@ -1161,3 +1161,33 @@ fn the_creating_input_flush_keeps_input_that_may_have_been_written() {
     let written = input_payloads(&frames_until_quiet(&mut host, Duration::from_millis(500)));
     assert_eq!(written, vec![b"a".to_vec()], "the input the flush kept reached the host once");
 }
+
+/// R-DAEMON-STATE S1.6c (#35082 review 2, should-do 1): a write that waits out a lost host connection fails as soon
+/// as the reader gives that host up (the link is `Failed` and nothing will install a replacement), as the old path's
+/// write failed then, not at its own reconnect deadline. The reader's wake reached only `stream_progress`, which the
+/// port's writer no longer waits on.
+#[cfg(unix)]
+#[test]
+fn a_write_waiting_for_a_reconnect_fails_when_the_reader_gives_up() {
+    let (_mux, surface, _host) = hosted_typing_surface("hosted-input-port-give-up", true);
+    let pty = surface.as_pty().expect("a terminal");
+    pty.input_port.set_reconnect_wait_for_test(Duration::from_secs(5));
+    pty.input_route.store(Some(Arc::new(dead_host_route())));
+    std::thread::scope(|scope| {
+        let writing = scope.spawn(|| {
+            let started = Instant::now();
+            (surface.write_bytes(b"g"), started.elapsed())
+        });
+        wait_until("the write waits for a reconnect", || pty.input_port.reconnecting_for_test());
+        // The reader gives the host up, as `wait_or_fail` does at the end of its budget.
+        pty.host_connection_state
+            .store(TerminalHostConnectionState::Failed as u8, Ordering::Release);
+        pty.reader_gave_up();
+        let (written, took) = writing.join().unwrap();
+        written.expect_err("a write to a host the reader gave up on answered ok");
+        assert!(
+            took < Duration::from_secs(2),
+            "a write to a host the reader gave up on waited {took:?}, its reconnect deadline, not the give-up"
+        );
+    });
+}