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

@@ -85,7 +85,7 @@ sealed class SessionManager : IDisposable
                 lock (_openLock)
                 {
                     if (_pending != null)
-                        _pending.Client.Dispose();
+                        NetUtil.RstClose(_pending.Client);
                 }
                 ProcessQueue();
                 break;
@@ -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 foreach (var payload in _deliverChannel.Reader.ReadAllAsync())
-                await stream.WriteAsync(payload, _cts.Token);
+                await stream.WriteAsync(payload);
         }
         catch { }
+        finally { client.Close(); }
     }
 
     async Task RetransmitTimer()
@@ -539,14 +554,33 @@ 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()
     {
         _cts.Cancel();
         _deliverChannel.Writer.TryComplete();
         SendClose();
-        try { client.Dispose(); } catch { }
+        NetUtil.RstClose(client);
         _cts.Dispose();
     }
 }
+
+/// Close a TcpClient with a TCP RST instead of a FIN.
+static partial class NetUtil
+{
+    public static void RstClose(TcpClient c)
+    {
+        try
+        {
+            c.LingerState = new LingerOption(true, 0);
+            c.Close();
+        }
+        catch { }
+    }
+}

gatuna-win/app.manifest

@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="utf-8"?>
 <assembly manifestVersion="1.0" xmlns="urn:schemas-microsoft-com:asm.v1">
-  <assemblyIdentity version="2.0.0.0" name="gatuna" />
+  <assemblyIdentity version="2.1.0.0" name="gatuna" />
   <trustInfo xmlns="urn:schemas-microsoft-com:asm.v2">
     <security>
       <requestedPrivileges xmlns="urn:schemas-microsoft-com:asm.v3">

gatuna-win/gatuna.csproj

@@ -5,7 +5,7 @@
     <TargetFramework>net8.0-windows</TargetFramework>
     <AssemblyName>gatuna</AssemblyName>
     <RootNamespace>gatuna</RootNamespace>
-    <Version>2.0.0</Version>
+    <Version>2.1.0</Version>
     <Nullable>enable</Nullable>
     <UseWindowsForms>true</UseWindowsForms>
     <ImplicitUsings>enable</ImplicitUsings>

gatunad/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "gatuna"
-version = "2.0.0"
+version = "2.1.0"
 edition = "2021"
 license = "CC0-1.0"
 
@@ -10,7 +10,7 @@ path = "src/main.rs"
 
 [dependencies]
 libc = "0.2"
-tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "sync", "io-util"] }
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "sync", "io-util", "time"] }
 clap = { version = "4", features = ["derive"] }
 pnet_datalink = "0.35"
 tracing = "0.1"

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

@@ -16,7 +16,7 @@ use std::sync::{Arc, Mutex};
 use tokio::net::TcpStream;
 use tokio::sync::mpsc;
 use tokio::sync::Mutex as AsyncMutex;
-use tracing::{error, Level};
+use tracing::{error, info, Level};
 
 use crate::frame::{
     Frame, REASON_CONNECT_FAILED, REASON_UNKNOWN_SESSION, REASON_UNKNOWN_UPSTREAM,
@@ -33,17 +33,21 @@ struct Args {
     /// One or more TCP upstreams as PORT[:label], relayed to 127.0.0.1:PORT.
     #[arg(num_args = 1..)]
     ports: Vec<String>,
+    /// Verbose logging (lifecycle events to stdout).
+    #[arg(short, long)]
+    verbose: bool,
 }
 
 #[tokio::main]
 async fn main() -> ExitCode {
+    let args = Args::parse();
+
+    let level = if args.verbose { Level::INFO } else { Level::ERROR };
     tracing_subscriber::fmt()
-        .with_max_level(Level::ERROR)
+        .with_max_level(level)
         .with_writer(|| std::io::stdout())
         .init();
 
-    let args = Args::parse();
-
     let table = match build_table(&args.ports) {
         Ok(t) => Arc::new(t),
         Err(e) => {
@@ -110,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;
                     }
                 };
@@ -141,13 +145,16 @@ async fn handle_frame(
 ) {
     match frame {
         Frame::Discover => {
+            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());
         }
-        Frame::Open { upstream_id, proto } => {
+        Frame::Open { upstream_id, proto: _ } => {
+            info!("OPEN upstream {upstream_id} from {src}");
             let upstream = table.get(upstream_id);
             match upstream {
                 Some(upstream) => {
@@ -176,10 +183,12 @@ async fn handle_frame(
                                 );
                                 let ack = Frame::OpenAck { session_id: sid, upstream_id, proto };
                                 let _ = tx.send((src, ack.encode())).await;
+                                info!("session {sid} established (upstream {upstream_id})");
                                 spawn_pump(r, sid, proto, src, tx, store);
                             }
                             Err(e) => {
                                 error!("connect 127.0.0.1:{port} failed: {e}");
+                                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;
@@ -215,7 +224,9 @@ async fn handle_frame(
             }
         }
         Frame::Close { session_id, reason: _ } => {
-            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

@@ -33,7 +33,7 @@ pub struct RecvState {
 }
 
 impl SendState {
-    fn new() -> Self {
+    pub fn new() -> Self {
         SendState {
             send_seq: 0,
             acked_seq: 0,
@@ -83,7 +83,7 @@ impl SendState {
 }
 
 impl RecvState {
-    fn new() -> Self {
+    pub fn new() -> Self {
         RecvState {
             expected_seq: 0,
             deliver_seq: 0,
@@ -129,6 +129,7 @@ impl RecvState {
 }
 
 #[derive(Clone)]
+#[allow(dead_code)]
 pub struct SessionHandle {
     pub upstream_id: u8,
     pub proto: u8,
@@ -155,7 +156,7 @@ pub async fn handle_data(
     seq: u32,
     ack_seq: u32,
     payload: &[u8],
-    tx: &TxChan,
+    _tx: &TxChan,
 ) -> Result<bool, WriteError> {
     let handle = {
         let store = store.lock().expect("store lock poisoned");
@@ -274,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;
                 }
             }
@@ -329,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);
     });
 }