Issue 32: end the send side on an abrupt local close, not only on EOF

src/daemon.rs

@@ -6143,4 +6143,106 @@ mod tests {
         server.shutdown().await?;
         Ok(())
     }
+
+    /// The SERVER must stop counting a session as attached once the client's
+    /// local end goes away.
+    ///
+    /// The test above proves the client releases its own gauge. It says nothing
+    /// about the far end, and the far end is what pins the server's endpoint.
+    /// This asserts the far end, on a quiet tunnel so no remote output can end
+    /// the session by a second route.
+    ///
+    /// **This is NOT a regression guard for issue 32, and it was wrong to
+    /// present it as one.** Issue 32 lives in the branch where the local read
+    /// returns an ERROR, and dropping a `UnixStream` in-process closes it
+    /// cleanly, so both platforms take the EOF branch here. I confirmed that on
+    /// Linux CI twice: with the pre-fix teardown restored, this test passed,
+    /// first with a shell and then with this quiet tunnel. Reaching the error
+    /// branch needs an abrupt close such as a killed process or an RST, which is
+    /// how the original trace produced it.
+    ///
+    /// The real guard for issue 32 is
+    /// `tunnel::tests::an_abrupt_local_close_reports_the_end_just_like_a_clean_eof`,
+    /// which injects the ending directly and therefore fails on every platform
+    /// when the defect is present. What this test still earns is the healthy
+    /// path: a clean local close must detach the session on the server promptly,
+    /// and that must not regress.
+    #[tokio::test]
+    async fn a_clean_local_close_detaches_a_quiet_session_on_the_server_too() -> Result<()> {
+        let server_dir = tempfile::tempdir()?;
+        let client_dir = tempfile::tempdir()?;
+        let server_home = FabricHome::new(server_dir.path());
+        let client_home = FabricHome::new(client_dir.path());
+        let server = FabricNode::start(server_home.clone()).await?;
+        let client = FabricNode::start(client_home.clone()).await?;
+        trust_test_peer(&server_home, &server, client.id(), "client", client.addr()).await?;
+        trust_test_peer(&client_home, &client, server.id(), "server", server.addr()).await?;
+
+        // A sink: it reads and never writes back. Anything it echoed would give
+        // the teardown a second route and mask exactly what this test isolates.
+        let sink_socket = server_dir.path().join("sink.sock");
+        let sink_listener = UnixListener::bind(&sink_socket)?;
+        tokio::spawn(async move {
+            while let Ok((mut stream, _)) = sink_listener.accept().await {
+                tokio::spawn(async move {
+                    let mut buf = [0u8; 64];
+                    while let Ok(read) = stream.read(&mut buf).await {
+                        if read == 0 {
+                            break;
+                        }
+                    }
+                });
+            }
+        });
+        server.expose("audit/sink", sink_socket.clone()).await?;
+
+        let client_state = client.state();
+        let server_state = server.state();
+        let socket = client_state
+            .dial_alpn("server", "audit/sink", b"audit/sink".to_vec(), false)
+            .await?;
+        let mut dial = UnixStream::connect(&socket).await?;
+        dial.write_all(b"open-the-session").await?;
+
+        // POSITIVE CONTROL. Prove the server really is holding an attach before
+        // asserting that it lets one go, or a server that never counted the
+        // session would pass the real check for the wrong reason.
+        let attached = tokio::time::timeout(Duration::from_secs(20), async {
+            loop {
+                if server_state.tunnel_sessions.stats().await.active_attaches > 0 {
+                    return;
+                }
+                tokio::time::sleep(Duration::from_millis(50)).await;
+            }
+        })
+        .await;
+        assert!(
+            attached.is_ok(),
+            "the server never counted the quiet session as attached, so this test \
+             could not observe it detaching either"
+        );
+
+        drop(dial);
+
+        // Bounded well under the 40s stall and well over the ~53ms healthy case,
+        // so this fails on the defect and does not flake on a slow machine.
+        let detached = tokio::time::timeout(Duration::from_secs(20), async {
+            loop {
+                if server_state.tunnel_sessions.stats().await.active_attaches == 0 {
+                    return;
+                }
+                tokio::time::sleep(Duration::from_millis(50)).await;
+            }
+        })
+        .await;
+        assert!(
+            detached.is_ok(),
+            "the server still counts the session as attached 20s after the client \
+             dropped its local end; a finished session pins the server's endpoint"
+        );
+
+        client.shutdown().await?;
+        server.shutdown().await?;
+        Ok(())
+    }
 }

src/tunnel.rs

@@ -535,16 +535,24 @@ impl TunnelSession {
                 Err(error) => {
                     // An abrupt local close ends local input just as surely as a
                     // clean EOF does: no further bytes can arrive on this session.
-                    // Release the recycle guard so a stalled remote teardown cannot
-                    // pin the endpoint. Deliberately ONLY the guard — send_closed
-                    // and every other teardown semantic stay exactly as they were,
-                    // because this error path is not the place to change them.
+                    // So end the send side the same way, which both releases the
+                    // recycle guard and records the final offset.
+                    //
+                    // Recording the offset is the part that matters remotely. Only
+                    // a recorded close makes the writer emit `Frame::Close`, and
+                    // only that frame lets the server stop counting this session
+                    // as attached. Without it the server waits for bytes that can
+                    // never arrive.
                     //
-                    // This path is why the first version of the fix was incomplete.
                     // Dropping a local socket yields a clean zero-length read on
-                    // macOS and an error on Linux, so releasing on clean EOF alone
-                    // left the Linux case pinned and the CI failure unchanged.
-                    self.release_attach_gauge().await;
+                    // macOS and an error on Linux, so this path used to run only
+                    // on Linux. That is the whole of the platform difference: the
+                    // 40-second stall in issue 32 was never a slow path, it was a
+                    // message the server never received.
+                    //
+                    // Only the send side closes. This session may still be writing
+                    // queued remote output, and a half-close is not a close.
+                    self.mark_send_closed().await;
                     return Err(error.into());
                 }
             };
@@ -2075,4 +2083,95 @@ mod tests {
 
         assert!(kill.is_cancelled());
     }
+
+    /// A local reader that hands over `bytes`, then ends the way the test asks.
+    ///
+    /// This is the whole platform difference in issue 32, made explicit. Dropping
+    /// a local socket gives a clean zero-length read on macOS and an error on
+    /// Linux. Injecting the ending directly turns a Linux-only, racy, 40-second
+    /// CI failure into a decision this test makes on any operating system.
+    struct EndingReader {
+        bytes: Vec<u8>,
+        fail_at_end: bool,
+    }
+
+    impl AsyncRead for EndingReader {
+        fn poll_read(
+            mut self: std::pin::Pin<&mut Self>,
+            _cx: &mut std::task::Context<'_>,
+            buf: &mut tokio::io::ReadBuf<'_>,
+        ) -> std::task::Poll<std::io::Result<()>> {
+            if !self.bytes.is_empty() {
+                let take = self.bytes.len().min(buf.remaining());
+                let chunk: Vec<u8> = self.bytes.drain(..take).collect();
+                buf.put_slice(&chunk);
+                return std::task::Poll::Ready(Ok(()));
+            }
+            if self.fail_at_end {
+                // What Linux reports when the local peer drops the socket.
+                return std::task::Poll::Ready(Err(std::io::Error::new(
+                    std::io::ErrorKind::ConnectionReset,
+                    "connection reset by peer",
+                )));
+            }
+            // What macOS reports for the same event: a clean EOF.
+            std::task::Poll::Ready(Ok(()))
+        }
+    }
+
+    async fn session_after_local_input_ends(fail_at_end: bool) -> Arc<TunnelSession> {
+        let (_write_peer, write) = duplex(64);
+        let reader = EndingReader {
+            bytes: b"hello".to_vec(),
+            fail_at_end,
+        };
+        let (session, local_read) =
+            TunnelSession::new_parts(session_id(9), peer_id(), Box::new(reader), Box::new(write));
+        // Returns Ok on EOF and Err on an abrupt close. Either way the local
+        // input is over, which is the only thing that matters here.
+        let _ = session.clone().run_local_reader(local_read).await;
+        session
+    }
+
+    /// Both endings must tell the remote that local input is over.
+    ///
+    /// Only a recorded `send_closed` makes the writer emit `Frame::Close`, and
+    /// only that frame lets the server stop counting the session as attached.
+    /// Without it the server waits for bytes that can never arrive, which is the
+    /// 40-second Linux stall in issue 32: not a slow path, a missing message.
+    #[tokio::test]
+    async fn an_abrupt_local_close_reports_the_end_just_like_a_clean_eof() {
+        let clean = session_after_local_input_ends(false).await;
+        let abrupt = session_after_local_input_ends(true).await;
+
+        let clean_closed = clean.state.lock().await.send_closed;
+        let abrupt_closed = abrupt.state.lock().await.send_closed;
+
+        assert_eq!(
+            clean_closed,
+            Some(5),
+            "a clean EOF must close the send side after the 5 bytes it read"
+        );
+        assert_eq!(
+            abrupt_closed, clean_closed,
+            "an abrupt local close must end the send side exactly like a clean EOF; \
+             leaving it open is what makes the server hold the session attached"
+        );
+    }
+
+    /// The recycle guard must come off on both endings too.
+    ///
+    /// This part already worked. It is asserted beside the case above so a later
+    /// change cannot fix one ending and quietly regress the other.
+    #[tokio::test]
+    async fn both_endings_release_the_recycle_guard() {
+        for fail_at_end in [false, true] {
+            let session = session_after_local_input_ends(fail_at_end).await;
+            assert!(
+                session.state.lock().await.attach_gauge.is_none(),
+                "the recycle guard must be released when local input ends \
+                 (fail_at_end={fail_at_end})"
+            );
+        }
+    }
 }