fix: premature socket teardown on CLOSE + duplicate CLOSE echo

Two bugs when upstream closes:

1. Data loss: gatuna received CLOSE from gatunad and immediately
   RST'd the socket to curl while the drain pump still had queued
   payloads in _deliverChannel. Fix: OnRemoteClose() completes the
   channel writer and cancels the read pump, but lets the drain pump
   finish naturally. Drain pump no longer takes _cts.Token — it stops
   on channel completion and closes the socket gracefully (FIN, OS
   flushes in background).

2. Duplicate CLOSE: gatuna echoed CLOSE back to gatunad (server
   already knows — it initiated). SendClose() now guarded by
   _closeSent flag. OnRemoteClose() never sends CLOSE at all.

Server-side: remove session from store before sending CLOSE so the
retransmit timer stops before CLOSE is queued. Only log CLOSE if
the session actually existed (dedup).

.gitea/workflows/build.yml

@@ -0,0 +1,55 @@
+name: build
+on: [push, pull_request]
+
+jobs:
+  build:
+    runs-on: hugmaster
+    steps:
+      - name: Checkout
+        run: |
+          git clone "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}.git" .
+          git checkout "$GITHUB_SHA"
+
+      - name: Build Rust
+        run: |
+          cd gatunad
+          cargo build --release
+
+      - name: Build .NET
+        run: |
+          cd gatuna-win
+          dotnet build
+
+      - name: Publish Windows (framework-dependent)
+        if: ${{ startsWith(github.ref, 'refs/tags/v') }}
+        run: |
+          cd gatuna-win
+          dotnet publish -c Release -r win-x64 --no-self-contained -o ../publish/win
+
+      - name: Package release assets
+        if: ${{ startsWith(github.ref, 'refs/tags/v') }}
+        run: |
+          mkdir -p release
+          cp gatunad/target/release/gatunad release/gatunad-linux-amd64
+          cd publish/win && zip -r ../../release/gatuna-win.zip . && cd ../..
+
+      - name: Create Gitea release
+        if: ${{ startsWith(github.ref, 'refs/tags/v') }}
+        run: |
+          TAG="${GITHUB_REF#refs/tags/}"
+          RESP=$(curl -sS -X POST \
+            -H "Authorization: token ${GITHUB_TOKEN}" \
+            -H "Content-Type: application/json" \
+            -d "{\"tag_name\":\"$TAG\",\"name\":\"$TAG\",\"body\":\"Automated build for $TAG\"}" \
+            "${GITHUB_SERVER_URL}/api/v1/repos/${GITHUB_REPOSITORY}/releases")
+          RID=$(echo "$RESP" | jq .id)
+          curl -sS -X POST \
+            -H "Authorization: token ${GITHUB_TOKEN}" \
+            -H "Content-Type: application/octet-stream" \
+            --data-binary @release/gatunad-linux-amd64 \
+            "${GITHUB_SERVER_URL}/api/v1/repos/${GITHUB_REPOSITORY}/releases/${RID}/assets?name=gatunad-linux-amd64"
+          curl -sS -X POST \
+            -H "Authorization: token ${GITHUB_TOKEN}" \
+            -H "Content-Type: application/octet-stream" \
+            --data-binary @release/gatuna-win.zip \
+            "${GITHUB_SERVER_URL}/api/v1/repos/${GITHUB_REPOSITORY}/releases/${RID}/assets?name=gatuna-win.zip"

gatuna-win/SessionManager.cs

@@ -100,7 +100,7 @@ sealed class SessionManager : IDisposable
 
             case Frame.Close close:
                 if (_sessions.TryRemove(close.SessionId, out var s))
-                    s.Dispose();
+                    s.OnRemoteClose();
                 break;
 
             case Frame.Pong pong:
@@ -394,6 +394,7 @@ sealed class Session(
     readonly RecvState _recv = new();
     readonly object _sendLock = new();
     readonly object _recvLock = new();
+    volatile bool _closeSent;
 
     public void Start()
     {
@@ -403,6 +404,17 @@ sealed class Session(
             _ = RetransmitTimer();
     }
 
+    /// <summary>
+    /// Called when a CLOSE frame arrives from the server. The server has
+    /// already torn down its side — we just need to flush queued data to
+    /// the local socket and close gracefully. Do NOT echo CLOSE back.
+    /// </summary>
+    public void OnRemoteClose()
+    {
+        _cts.Cancel();
+        _deliverChannel.Writer.TryComplete();
+    }
+
     /// Handle a DATA frame from the tunnel.
     public void HandleData(Frame.Data data)
     {
@@ -440,10 +452,10 @@ sealed class Session(
 
     async Task PumpSocketToTunnel()
     {
+        var buf = new byte[Proto.MaxPayload];
         try
         {
             var stream = client.GetStream();
-            var buf = new byte[Proto.MaxPayload];
             using var reg = _cts.Token.Register(() => client.Dispose());
             while (!_cts.IsCancellationRequested)
             {
@@ -476,6 +488,8 @@ sealed class Session(
             }
         }
         catch { }
+        _cts.Cancel();
+        _deliverChannel.Writer.TryComplete();
         SendClose();
         onClosed();
     }
@@ -485,10 +499,11 @@ sealed class Session(
         try
         {
             var stream = client.GetStream();
-            await foreach (var payload in _deliverChannel.Reader.ReadAllAsync(_cts.Token))
-                await stream.WriteAsync(payload, _cts.Token);
+            await foreach (var payload in _deliverChannel.Reader.ReadAllAsync())
+                await stream.WriteAsync(payload);
         }
         catch { }
+        finally { client.Close(); }
     }
 
     async Task RetransmitTimer()
@@ -539,7 +554,12 @@ sealed class Session(
         link.SendTo(serverMac, new Frame.Data(sessionId, seq, ackSeq, []));
     }
 
-    void SendClose() => link.SendTo(serverMac, new Frame.Close(sessionId, null));
+    void SendClose()
+    {
+        if (_closeSent) return;
+        _closeSent = true;
+        link.SendTo(serverMac, new Frame.Close(sessionId, null));
+    }
 
     public void Dispose()
     {

gatunad/src/link.rs

@@ -12,6 +12,16 @@ use crate::frame::{ETHERTYPE, ETH_HEADER_LEN};
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct MacAddr(pub [u8; 6]);
 
+impl std::fmt::Display for MacAddr {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{:02X}:{:02X}:{:02X}:{:02X}:{:02X}:{:02X}",
+            self.0[0], self.0[1], self.0[2], self.0[3], self.0[4], self.0[5]
+        )
+    }
+}
+
 impl MacAddr {
     #[allow(dead_code)]
     pub fn broadcast() -> Self {

gatunad/src/main.rs

@@ -114,7 +114,7 @@ async fn main() -> ExitCode {
                 let frame = match Frame::parse(payload) {
                     Ok(f) => f,
                     Err(e) => {
-                        error!("decode from {src:?}: {e}");
+                        error!("decode from {src}: {e}");
                         continue;
                     }
                 };
@@ -145,16 +145,16 @@ async fn handle_frame(
 ) {
     match frame {
         Frame::Discover => {
-            info!("DISCOVER from {src:?}");
+            info!("DISCOVER from {src}");
             let manifest = Frame::Manifest {
                 hostname: (**hostname).clone(),
                 entries: table.entries(),
             };
             let _ = tx.send((src, manifest.encode())).await;
-            info!("MANIFEST sent to {src:?} ({} upstreams)", table.0.len());
+            info!("MANIFEST sent to {src} ({} upstreams)", table.0.len());
         }
         Frame::Open { upstream_id, proto: _ } => {
-            info!("OPEN upstream {upstream_id} from {src:?}");
+            info!("OPEN upstream {upstream_id} from {src}");
             let upstream = table.get(upstream_id);
             match upstream {
                 Some(upstream) => {
@@ -188,7 +188,7 @@ async fn handle_frame(
                             }
                             Err(e) => {
                                 error!("connect 127.0.0.1:{port} failed: {e}");
-                                info!("OPEN_NAK upstream {upstream_id} (connect_failed) to {src:?}");
+                                info!("OPEN_NAK upstream {upstream_id} (connect_failed) to {src}");
                                 let nak =
                                     Frame::OpenNak { upstream_id, reason: REASON_CONNECT_FAILED };
                                 let _ = tx.send((src, nak.encode())).await;
@@ -224,8 +224,9 @@ async fn handle_frame(
             }
         }
         Frame::Close { session_id, reason: _ } => {
-            info!("CLOSE session {session_id} from {src:?}");
-            store.lock().expect("store poisoned").remove(&session_id);
+            if store.lock().expect("store poisoned").remove(&session_id).is_some() {
+                info!("CLOSE session {session_id} from {src}");
+            }
         }
         Frame::Ping { nonce } => {
             let pong = Frame::Pong { nonce };

gatunad/src/session.rs

@@ -275,12 +275,12 @@ pub fn spawn_pump(
                 }
 
                 if should_close {
+                    store.lock().expect("store poisoned").remove(&session_id);
                     let close = Frame::Close {
                         session_id,
                         reason: Some(REASON_MAX_RETRIES),
                     };
                     let _ = tx.try_send((peer_mac, close.encode()));
-                    store.lock().expect("store poisoned").remove(&session_id);
                     break;
                 }
             }
@@ -330,8 +330,8 @@ pub fn spawn_pump(
                 break;
             }
         }
+        store.lock().expect("store poisoned").remove(&session_id);
         let close = Frame::Close { session_id, reason: None };
         let _ = tx.send((peer_mac, close.encode())).await;
-        store.lock().expect("store poisoned").remove(&session_id);
     });
 }