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

* tunnel: end the send side on an abrupt local close, not only on EOF

Issue 32. A finished outbound session could leave the SERVER counting it as
attached for 40 seconds or more on Linux, while macOS released in about 53 ms.

It was never a slow path. A dropped local socket reads as a clean zero-length
read on macOS and as an error on Linux. Only the EOF branch recorded the send
close, and only a recorded send close makes the writer emit Frame::Close. With
no such frame the server waited for bytes that could never arrive. A 750x
platform gap is a different code path, not a slower one.

PR 31 released the recycle guard on this branch so the stall could not pin the
endpoint, and deliberately left the teardown semantics alone. This finishes the
job: the abrupt branch now calls mark_send_closed, which records the final
offset and releases the guard. Only the send side closes, because the session
may still be writing queued remote output and a half-close is not a close.

Proved without Linux and without CI timing. The platform difference is one
decision, EOF against error, so a test injects it directly through the boxed
local reader. Before this change the abrupt case left send_closed at None while
the clean case set it to Some(5); now they agree. That turns a Linux-only, racy,
40-second failure into a deterministic assertion that runs anywhere in
milliseconds.

A second test asserts the server-side outcome end to end: after the client drops
its local end, the server's active attach count must reach zero. It carries a
positive control that the server really counted the attach first, so a server
that never counted the session cannot pass it for the wrong reason. That test
passes on macOS either way, because macOS always took the branch that worked; it
is the Linux regression guard.

The recycle-guard release is now asserted for BOTH endings beside the new case,
so a later change cannot fix one ending and quietly regress the other.

194 lib tests green.

* test: isolate the issue-32 defect with a quiet tunnel, not a shell

The first end-to-end test used a shell and passed on Linux CI with the defect
present, so it guarded nothing. A shell streams output back, so a dropped local
socket fails on the next WRITE and tears the session down by a second route that
hides the missing close frame.

The original trace was a quiet tunnel. With nothing to write back, the missing
frame is the only thing that could end the session, which is what isolates the
defect.

* test: stop claiming the end-to-end test guards issue 32

It does not, and I presented it as though it did. Twice.

Issue 32 lives in the branch where the local read returns an ERROR. Dropping a
UnixStream in-process closes it cleanly, so both platforms take the EOF branch
in this test and neither reaches the defect. Linux CI confirmed it twice against
the pre-fix teardown: the shell version passed, and so did the quiet-tunnel
version I wrote to replace it.

My masking explanation for the first failure was wrong too. The shell was not
hiding the defect behind a write failure; the test never reached the error
branch at all.

The real guard is the unit test, which injects the ending directly and fails on
every platform when the defect is present. This test is renamed to what it
actually proves: a CLEAN local close detaches the session on the server. That is
worth keeping and is not the issue-32 guard.

Author: myobie

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})"
+            );
+        }
+    }
 }