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"

PROTOCOL.md

@@ -3,7 +3,7 @@
 All frames ride Ethernet with ethertype `0x6969`. The ethertype is the sole
 discriminator; there is no magic number inside the payload.
 
-## Frame layout
+## Frame layout (non-DATA frames)
 
 ```
  0                   1                   2                   3
@@ -17,7 +17,7 @@ discriminator; there is no magic number inside the payload.
 +---------------------------------------------------------------+
 ```
 
-- **version** (u8): protocol version. Currently `1`.
+- **version** (u8): protocol version. Currently `2`.
 - **type** (u8): frame type, see table below.
 - **session_id** (u32, big-endian): `0` for non-session frames; the
   server-assigned ID for session-scoped frames.
@@ -27,13 +27,37 @@ discriminator; there is no magic number inside the payload.
   meet the minimum Ethernet frame size).
 - **payload** (`payload_len` bytes): type-dependent.
 
-No CRC, no retransmit, no ordering at this layer. TCP reliability is handled by
+## DATA frame layout
-the endpoints' TCP stacks; UDP (reserved) will rely on application-level
-mechanisms.
 
-Maximum payload: 1500 (Ethernet MTU) − 8 (our header) = **1492 bytes**. In
+DATA frames carry two additional fields after the common header for L2-level
-practice we cap at **1480** to stay conservative. Larger payloads are not
+reliability:
-emitted in v1.
+
+```
+ 0                   1                   2                   3
+ 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
++---------------+---------------+-------------------------------+
+| version=2     | type=0x06     | session_id                    |
++---------------+---------------+-------------------------------+
+| payload_len (big-endian)      | seq (big-endian)              |
++-------------------------------+-------------------------------+
+| ack_seq (big-endian)          | payload ...                   |
++-------------------------------+                               +
+|                                                               |
++---------------------------------------------------------------+
+```
+
+- **seq** (u32, big-endian): monotonically increasing per-session sequence
+  number. Wraps at 2^32 (same as TCP). Identifies this DATA frame's position
+  in the byte stream.
+- **ack_seq** (u32, big-endian): cumulative acknowledgment — the highest
+  contiguous `seq` that the sender of this frame has delivered to its local
+  TCP socket. The receiver uses this to advance its retransmit window.
+- **payload** (`payload_len` bytes): raw application bytes (0–1480). A
+  `payload_len = 0` DATA frame is a **pure ACK** — it carries no data, just
+  an acknowledgment. This mirrors TCP's empty-segment ACK.
+
+Maximum payload: 1500 (Ethernet MTU) − 8 (common header) − 8 (seq + ack_seq)
+= **1484 bytes**. In practice we cap at **1480** to stay conservative.
 
 ## Frame types
 
@@ -41,15 +65,15 @@ emitted in v1.
 |------|-------------|---------------|------------|----------------------------------|
 | 0x01 | DISCOVER    | C → broadcast | 0          | empty                            |
 | 0x02 | MANIFEST    | S → C         | 0          | `hostname_len:1, hostname:N, entries...` |
-| 0x03 | OPEN        | C → S         | 0          | `upstream_id:1`                  |
+| 0x03 | OPEN        | C → S         | 0          | `upstream_id:1, proto:1`        |
-| 0x04 | OPEN_ACK    | S → C         | assigned   | `upstream_id:1`                  |
+| 0x04 | OPEN_ACK    | S → C         | assigned   | `upstream_id:1, proto:1`        |
 | 0x05 | OPEN_NAK    | S → C         | 0          | `upstream_id:1, reason:1`        |
-| 0x06 | DATA        | both          | session    | raw bytes (≤1480)                |
+| 0x06 | DATA        | both          | session    | `seq:4, ack_seq:4, raw bytes`    |
 | 0x07 | CLOSE       | both          | session    | optional `reason:1`              |
 | 0x0B | PING        | C → S         | 0          | `nonce:8`                        |
 | 0x0C | PONG        | S → C         | 0          | `nonce:8` (echoed)               |
 
-Reserved (unimplemented in v1; parse returns Err, encode unimplemented):
+Reserved (unimplemented; parse returns Err, encode unimplemented):
 
 | Type | Name        |
 |------|-------------|
@@ -57,6 +81,91 @@ Reserved (unimplemented in v1; parse returns Err, encode unimplemented):
 | 0x09 | UDP_DATA    |
 | 0x0A | UDP_CLOSE   |
 
+## L2 reliability
+
+The reliability semantics of a session depend on its transport proto,
+which is carried end-to-end through OPEN/OPEN_ACK:
+
+- **TCP (`proto = 1`):** DATA frames are delivered reliably and in-order via
+  a TCP-like sequence + cumulative ACK + retransmit mechanism. This ensures
+  that a dropped Ethernet frame does not permanently corrupt a TCP session
+  (which would otherwise happen because the local kernel TCP stack ACKs data
+  before it is chunked into DATA frames).
+- **UDP (`proto = 2`, reserved):** DATA frames are delivered best-effort. The
+  `seq` and `ack_seq` fields are present in the frame layout for uniformity
+  but are ignored — no retransmit, no in-order buffering, no pure ACKs. Drops
+  are tolerated because stateless protocols either don't care or handle
+  recovery at the application layer.
+
+The following sections describe the TCP reliability mechanism.
+
+### Sender state (per session)
+
+- `send_seq`: next seq to assign (starts at 0, increments per DATA frame).
+- `retransmit_buffer`: map of `seq → (payload, timestamp)`, holding all sent
+  but unacked frames.
+- `acked_seq`: highest seq acknowledged by the peer (initially `None`).
+
+On sending DATA:
+1. Assign `seq = send_seq; send_seq += 1`.
+2. Set `ack_seq` to the highest contiguous seq we have received from the peer
+   (our receive side's `deliver_seq`).
+3. Store `(payload, now)` in `retransmit_buffer[seq]`.
+4. Transmit the frame.
+
+On receiving an `ack_seq` in any DATA frame (including pure ACKs):
+1. Advance `acked_seq` to `max(acked_seq, ack_seq)`.
+2. Remove all entries from `retransmit_buffer` with `seq <= ack_seq`.
+
+Retransmit timer (per session, checked periodically):
+1. For each entry in `retransmit_buffer` older than `RETRANSMIT_TIMEOUT`
+   (default 5 ms), retransmit the frame and reset its timestamp.
+2. If any entry has been retransmitted more than `MAX_RETRIES` times (default
+   10), send `CLOSE` and tear down the session.
+
+### Receiver state (per session)
+
+- `expected_seq`: next seq expected (starts at 0).
+- `receive_buffer`: map of `seq → payload`, holding out-of-order frames.
+- `deliver_seq`: highest seq delivered to the local TCP socket (starts at
+  `None`; reported as `ack_seq` in outgoing DATA frames).
+
+On receiving DATA with `seq`:
+1. If `seq < expected_seq`: duplicate (already delivered). Discard the payload,
+   but still process the `ack_seq` field to advance the send window. Send a
+   pure ACK so the sender can converge.
+2. If `seq == expected_seq`: deliver payload to the TCP socket. Increment
+   `expected_seq`. Then check `receive_buffer` for the next contiguous seq and
+   deliver those too (drain the buffer in order). Update `deliver_seq`.
+3. If `seq > expected_seq`: store in `receive_buffer[seq]`. Do not deliver yet.
+   Send a pure ACK (re-ACKing `deliver_seq`) to trigger retransmit of the gap.
+
+### Pure ACK frames
+
+When a side needs to ACK but has no data to send, it sends a DATA frame with
+`payload_len = 0`. The `seq` field is set to `send_seq` (consuming a seq
+number, same as TCP's empty segment) and `ack_seq` carries the cumulative
+acknowledgment. The receiver processes the `ack_seq` and discards the empty
+payload without delivering to the TCP socket.
+
+### Acknowledgment timing
+
+- When data is flowing in both directions, each DATA frame carries the latest
+  `ack_seq` — no separate ACK frames needed.
+- When data is one-directional, the receiver sends a pure ACK after each DATA
+  frame (or after a small batch, implementation-defined).
+- On receiving a duplicate or out-of-order frame, the receiver immediately
+  sends a pure ACK to help the sender converge.
+
+### Why this is not the Two Generals Problem
+
+We do not need mutual consensus. We need one-sided reliable delivery: the
+sender retransmits until it gets an ACK. If the ACK is lost, the sender
+retransmits the data; the receiver sees a duplicate, discards it, and re-ACKs.
+This converges in O(1) round trips. The only unsolvable case — the last frame
+before a permanent link death — is handled by `MAX_RETRIES` → `CLOSE`, which is
+correct: a dead link should kill the session.
+
 ## Payload field encodings
 
 ### MANIFEST payload
@@ -98,23 +207,28 @@ the payload is exhausted. The number of entries is not carried explicitly.
 ### OPEN payload
 
 ```
-+---------------+
++---------------+---------------+
-| upstream_id   |
+| upstream_id   | proto         |
-+---------------+
++---------------+---------------+
 ```
 
 - **upstream_id** (u8): which MANIFEST entry to open.
+- **proto** (u8): the transport protocol of the upstream (`1 = TCP`,
+  `2 = UDP`). Carried end-to-end so both sides know which reliability
+  semantics apply to the session. Must match the proto advertised in the
+  MANIFEST for that upstream.
 
 ### OPEN_ACK payload
 
 ```
-+---------------+
++---------------+---------------+
-| upstream_id   |
+| upstream_id   | proto         |
-+---------------+
++---------------+---------------+
 ```
 
-- **upstream_id** (u8): echoes the requested upstream. The session is
+- **upstream_id** (u8): echoes the requested upstream.
-  identified by the `session_id` field in the header, not the payload.
+- **proto** (u8): echoes the requested proto. The session is identified by
+  the `session_id` field in the header, not the payload.
 
 ### OPEN_NAK payload
 
@@ -129,8 +243,10 @@ the payload is exhausted. The number of entries is not carried explicitly.
 
 ### DATA payload
 
-Raw application bytes. Up to 1480 bytes per frame. The `session_id` header
+`[ seq:4 ][ ack_seq:4 ][ raw bytes ]` — the `seq` and `ack_seq` fields are
-field identifies which session the bytes belong to.
+part of the DATA frame's extended header (between `payload_len` and the
+payload). The `payload_len` field counts only the raw bytes, not the seq/ack
+fields. Up to 1480 bytes of application data per frame.
 
 ### CLOSE payload
 
@@ -163,6 +279,7 @@ field identifies which session the bytes belong to.
 | 2     | connect_failed    |
 | 3     | oversize          |
 | 4     | unknown_session   |
+| 5     | max_retries       |
 
 ## Discovery flow
 
@@ -195,23 +312,27 @@ client                            server
   |   OPEN_NAK { reason }           |
   |<--------------------------------|   (on failure)
   |                                 |
-  | DATA { session_id, bytes }      |
+  | DATA { seq, ack_seq, bytes }    |
-  |<------------------------------->| DATA { session_id, bytes }
+  |<------------------------------->| DATA { seq, ack_seq, bytes }
   |                                 |
   | CLOSE { session_id, reason? }   |
-  |<------------------------------->|   (on EOF, RST, or error)
+  |<------------------------------->|   (on EOF, RST, or max_retries)
   |                                 |
 ```
 
 - `session_id` is allocated by the server as a monotonically increasing u32
   (starting at 1) from an atomic counter. Collision by wraparound is ignored.
+- `seq` starts at 0 on both sides of each session and increments per DATA
+  frame (including pure ACKs).
 - Either side may send `CLOSE`. The side receiving `CLOSE` tears down its half
   and stops emitting frames for that session.
 - The server's socket→tunnel pump reads `TcpStream` in 1480-byte chunks and
-  emits one `DATA` frame per chunk. On `read` returning 0 (FIN) or an error,
+  emits one DATA frame per chunk. On `read` returning 0 (FIN) or an error,
   it emits `CLOSE` and exits.
-- The server's tunnel→socket path writes `DATA` payloads to the `TcpStream`
+- The server's tunnel→socket path delivers DATA payloads in seq order to the
-  with `write_all`. On error it emits `CLOSE` and drops the session.
+  `TcpStream` with `write_all`. On error it emits `CLOSE` and drops the session.
+- A DATA frame is retransmitted if no ACK is received within 5 ms. After 10
+  failed retransmits, the session is closed with `reason = max_retries`.
 
 ## Network test (PING/PONG)
 
@@ -235,3 +356,14 @@ client                            server
   compute RTT, average latency, jitter (mean absolute delta of consecutive
   RTTs), and drop rate (unanswered PINGs).
 - PING/PONG frames use `session_id = 0`; they are independent of TCP sessions.
+
+## Version compatibility
+
+- Version 1: no `seq`/`ack_seq` in DATA frames, no L2 reliability. Dropped
+  DATA frames kill the TCP session.
+- Version 2: DATA frames carry `seq` + `ack_seq`, L2 retransmit. Dropped
+  frames are recovered.
+
+Version 2 is the current version. A receiver that sees `version != 2` rejects
+the frame. Both sides of a session must speak the same version; there is no
+negotiation.

README.md

@@ -1,9 +1,16 @@
 # gatuna
 
-A raw-Ethernet tunnel for reaching services on a peer machine when a WFP
+**gata** (Spanish: *cat*) + **atún** (Spanish: *tuna*) — a cat fishing a
-killswitch on the local machine blocks normal IP traffic. Two programs carry
+tunnel. The "tun" root is also the commonplace shorthand for a network tunnel
-bytes between their respective loopbacks over a private L2 protocol, bypassing
+(as in `/dev/net/tun`, `wintun`, `tuntap`), so the name works in both languages:
-the IP stack (and therefore the killswitch) entirely.
+a Spanish portmanteau that sounds like "gatuna" (feline), and an English pun on
+"cat-tun" — a cat that tunnels.
+
+## Attribution
+
+Entirely vibe-coded with [GLM 5.2](https://github.com/zai-org/GLM-5.2). No
+human-typed source code; all implementation was directed through natural
+language and reviewed by the human in the loop.
 
 ## Why this works
 
@@ -39,13 +46,16 @@ See [`PROTOCOL.md`](PROTOCOL.md) for the full wire format. Summary:
 
 - 8-byte header, big-endian:
   `[ version:1 ][ type:1 ][ session_id:4 ][ payload_len:2 ][ payload:N ]`.
-- `version` = `1`. `payload_len` lets the receiver ignore Ethernet padding
+- `version` = `2`. `payload_len` lets the receiver ignore Ethernet padding
   (frames under 60 bytes are zero-padded by the NIC).
+- DATA frames carry an extended 16-byte header with `seq` and `ack_seq` fields
+  for L2-level reliability (retransmit + in-order delivery), preventing lost
+  Ethernet frames from corrupting TCP sessions.
 - Discovery: client broadcasts `DISCOVER`; server unicasts `MANIFEST` (with
   hostname and upstream list) back.
 - Sessions: `OPEN` → `OPEN_ACK` (or `OPEN_NAK`) → `DATA`* ↔ `DATA`* → `CLOSE`.
 - Network test: `PING` (with 8-byte nonce) → `PONG` (nonce echoed).
-- v1 ships TCP only. UDP frame types are reserved but unimplemented.
+- TCP only. UDP frame types are reserved but unimplemented.
 
 ## `gatunad` usage
 
@@ -142,15 +152,14 @@ silent.
 
 **Client:** status line in the UI. Errors are not logged to disk.
 
-## v1 limitations
+## Limitations
 
 - TCP only. UDP wire types reserved, code paths stubbed.
-- No retransmit at the L2 layer. A dropped `DATA` frame breaks the TCP session
-  irrecoverably because the localhost socket already ACKed the bytes. Acceptable
-  on a healthy switched link.
 - No auth/crypto. Anyone on the same L2 segment can `DISCOVER` and `OPEN`.
 - Single server instance per interface.
 - One outstanding `OPEN` at a time on the client (serialized via queue).
+- L2 retransmit caps at 10 retries × 5 ms = 50 ms. A permanently dead link
+  closes the session with `reason = max_retries`.
 
 ## Repository layout
 
@@ -179,6 +188,24 @@ gatuna/
     └── PingTest.cs          # PING/PONG network test with stats
 ```
 
+## Versioning
+
+Both programs use `MAJOR.MINOR.PATCH` where:
+
+- **MAJOR** matches the wire protocol version. A v2.x program rejects any frame
+  with `version != 2`. No cross-major compatibility.
+- **MINOR** is the main change indicator for features and fixes within a major
+  version.
+- **PATCH** is reserved for bug fixes; generally unused.
+
+Current version: **2.0.0** (wire protocol v2).
+
+Check versions:
+```sh
+gatunad --version          # Rust server
+```
+The Windows client shows its version in the title bar and tray tooltip.
+
 ## License
 
 CC0 1.0 Universal. See [`LICENSE`](LICENSE).

gatuna-win/Frame.cs

@@ -6,8 +6,9 @@ static class Proto
 {
     public const ushort EtherType = 0x6969;
     public const int EthHeaderLen = 14;
-    public const byte Version = 1;
+    public const byte Version = 2;
     public const int HeaderLen = 8;
+    public const int DataHeaderLen = 16; // 8 common + 4 seq + 4 ack_seq
     public const int MaxPayload = 1480;
 
     public const byte TypeDiscover = 0x01;
@@ -28,6 +29,7 @@ static class Proto
     public const byte ReasonConnectFailed = 2;
     public const byte ReasonOversize = 3;
     public const byte ReasonUnknownSession = 4;
+    public const byte ReasonMaxRetries = 5;
 }
 
 readonly record struct UpstreamEntry(
@@ -43,10 +45,10 @@ abstract record Frame
 {
     internal record Discover : Frame;
     internal record Manifest(string Hostname, UpstreamEntry[] Entries) : Frame;
-    internal record Open(byte UpstreamId) : Frame;
+    internal record Open(byte UpstreamId, byte Proto) : Frame;
-    internal record OpenAck(uint SessionId, byte UpstreamId) : Frame;
+    internal record OpenAck(uint SessionId, byte UpstreamId, byte Proto) : Frame;
     internal record OpenNak(byte UpstreamId, byte Reason) : Frame;
-    internal record Data(uint SessionId, byte[] Payload) : Frame;
+    internal record Data(uint SessionId, uint Seq, uint AckSeq, byte[] Payload) : Frame;
     internal record Close(uint SessionId, byte? Reason) : Frame;
     internal record Ping(ulong Nonce) : Frame;
     internal record Pong(ulong Nonce) : Frame;
@@ -54,8 +56,7 @@ abstract record Frame
 
 static class FrameCodec
 {
-    /// Build the 8-byte header + payload. payload_len records the exact
+    /// Build a non-DATA frame: 8-byte common header + payload.
-    /// payload length so the receiver can ignore Ethernet padding.
     static byte[] Build(byte type, uint sessionId, byte[] payload)
     {
         var buf = new byte[Proto.HeaderLen + payload.Length];
@@ -71,6 +72,31 @@ static class FrameCodec
         return buf;
     }
 
+    /// Build a DATA frame: 8-byte common header + seq + ack_seq + payload.
+    /// payload_len counts only the raw bytes, not seq/ack_seq.
+    static byte[] BuildData(uint sessionId, uint seq, uint ackSeq, byte[] payload)
+    {
+        var buf = new byte[Proto.DataHeaderLen + payload.Length];
+        buf[0] = Proto.Version;
+        buf[1] = Proto.TypeData;
+        buf[2] = (byte)(sessionId >> 24);
+        buf[3] = (byte)(sessionId >> 16);
+        buf[4] = (byte)(sessionId >> 8);
+        buf[5] = (byte)(sessionId & 0xFF);
+        buf[6] = (byte)(payload.Length >> 8);
+        buf[7] = (byte)(payload.Length & 0xFF);
+        buf[8] = (byte)(seq >> 24);
+        buf[9] = (byte)(seq >> 16);
+        buf[10] = (byte)(seq >> 8);
+        buf[11] = (byte)(seq & 0xFF);
+        buf[12] = (byte)(ackSeq >> 24);
+        buf[13] = (byte)(ackSeq >> 16);
+        buf[14] = (byte)(ackSeq >> 8);
+        buf[15] = (byte)(ackSeq & 0xFF);
+        Buffer.BlockCopy(payload, 0, buf, Proto.DataHeaderLen, payload.Length);
+        return buf;
+    }
+
     public static byte[] Encode(Frame frame)
     {
         return frame switch
@@ -80,13 +106,13 @@ static class FrameCodec
             Frame.Manifest manifest =>
                 Build(Proto.TypeManifest, 0, BuildManifestPayload(manifest.Hostname, manifest.Entries)),
             Frame.Open open =>
-                Build(Proto.TypeOpen, 0, [open.UpstreamId]),
+                Build(Proto.TypeOpen, 0, [open.UpstreamId, open.Proto]),
             Frame.OpenAck ack =>
-                Build(Proto.TypeOpenAck, ack.SessionId, [ack.UpstreamId]),
+                Build(Proto.TypeOpenAck, ack.SessionId, [ack.UpstreamId, ack.Proto]),
             Frame.OpenNak nak =>
                 Build(Proto.TypeOpenNak, 0, [nak.UpstreamId, nak.Reason]),
             Frame.Data data =>
-                Build(Proto.TypeData, data.SessionId, data.Payload),
+                BuildData(data.SessionId, data.Seq, data.AckSeq, data.Payload),
             Frame.Close close =>
                 Build(Proto.TypeClose, close.SessionId,
                     close.Reason.HasValue ? [close.Reason.Value] : []),
@@ -130,27 +156,38 @@ static class FrameCodec
         var type = buf[1];
         var sessionId = (uint)(buf[2] << 24 | buf[3] << 16 | buf[4] << 8 | buf[5]);
         var payloadLen = (ushort)(buf[6] << 8 | buf[7]);
+
+        // DATA frames have seq + ack_seq after the common header.
+        if (type == Proto.TypeData)
+        {
+            if (buf.Length < Proto.DataHeaderLen + payloadLen)
+                return null;
+            var seq = (uint)(buf[8] << 24 | buf[9] << 16 | buf[10] << 8 | buf[11]);
+            var ackSeq = (uint)(buf[12] << 24 | buf[13] << 16 | buf[14] << 8 | buf[15]);
+            var payload = buf.Slice(Proto.DataHeaderLen, payloadLen);
+            if (payload.Length > Proto.MaxPayload)
+                return null;
+            return new Frame.Data(sessionId, seq, ackSeq, payload.ToArray());
+        }
+
         if (buf.Length < Proto.HeaderLen + payloadLen)
             return null;
-        // Slice exactly payloadLen bytes, ignoring any trailing Ethernet padding.
+        var payload2 = buf.Slice(Proto.HeaderLen, payloadLen);
-        var payload = buf.Slice(Proto.HeaderLen, payloadLen);
 
         return type switch
         {
-            Proto.TypeManifest => ParseManifest(payload),
+            Proto.TypeManifest => ParseManifest(payload2),
-            Proto.TypeOpenAck when payload.Length == 1 =>
+            Proto.TypeOpenAck when payload2.Length == 2 =>
-                new Frame.OpenAck(sessionId, payload[0]),
+                new Frame.OpenAck(sessionId, payload2[0], payload2[1]),
-            Proto.TypeOpenNak when payload.Length == 2 =>
+            Proto.TypeOpenNak when payload2.Length == 2 =>
-                new Frame.OpenNak(payload[0], payload[1]),
+                new Frame.OpenNak(payload2[0], payload2[1]),
-            Proto.TypeData when payload.Length <= Proto.MaxPayload =>
+            Proto.TypeClose when payload2.Length is 0 or 1 =>
-                new Frame.Data(sessionId, payload.ToArray()),
+                new Frame.Close(sessionId, payload2.Length == 1 ? payload2[0] : null),
-            Proto.TypeClose when payload.Length is 0 or 1 =>
+            Proto.TypePong when payload2.Length == 8 =>
-                new Frame.Close(sessionId, payload.Length == 1 ? payload[0] : null),
+                new Frame.Pong(ParseNonce(payload2)),
-            Proto.TypePong when payload.Length == 8 =>
+            Proto.TypePing when payload2.Length == 8 =>
-                new Frame.Pong(ParseNonce(payload)),
+                new Frame.Ping(ParseNonce(payload2)),
-            Proto.TypePing when payload.Length == 8 =>
+            Proto.TypeDiscover when payload2.Length == 0 =>
-                new Frame.Ping(ParseNonce(payload)),
-            Proto.TypeDiscover when payload.Length == 0 =>
                 new Frame.Discover(),
             _ => null,
         };

gatuna-win/MainForm.cs

@@ -17,7 +17,8 @@ public partial class MainForm : Form
 
     public MainForm()
     {
-        Text = "gatuna";
+        var ver = System.Reflection.Assembly.GetExecutingAssembly().GetName().Version!;
+        Text = $"gatuna {ver.Major}.{ver.Minor}";
         Width = 520;
         Height = 420;
         FormBorderStyle = FormBorderStyle.FixedSingle;

gatuna-win/Program.cs

@@ -9,10 +9,12 @@ static class Program
 
         var form = new MainForm();
 
+        var ver = System.Reflection.Assembly.GetExecutingAssembly().GetName().Version!;
+
         using var tray = new NotifyIcon
         {
             Icon = SystemIcons.GetStockIcon(StockIconId.NetworkConnect, 32),
-            Text = "gatuna",
+            Text = $"gatuna {ver.Major}.{ver.Minor}",
             Visible = true,
         };
 

gatuna-win/SessionManager.cs

@@ -85,14 +85,14 @@ sealed class SessionManager : IDisposable
                 lock (_openLock)
                 {
                     if (_pending != null)
-                        _pending.Client.Dispose();
+                        NetUtil.RstClose(_pending.Client);
                 }
                 ProcessQueue();
                 break;
 
             case Frame.Data data:
                 if (_sessions.TryGetValue(data.SessionId, out var session))
-                    session.Deliver(data.Payload);
+                    session.HandleData(data);
                 else if (_link != null && _serverMac != null)
                     _link.SendTo(_serverMac,
                         new Frame.Close(data.SessionId, Proto.ReasonUnknownSession));
@@ -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:
@@ -180,22 +180,22 @@ sealed class SessionManager : IDisposable
                 client = await state.Listener.AcceptTcpClientAsync();
             }
             catch { break; }
-            EnqueueOpen(client, state.Upstream.Id);
+            EnqueueOpen(client, state.Upstream.Id, state.Upstream.Protocol);
         }
     }
 
-    void EnqueueOpen(TcpClient client, byte upstreamId)
+    void EnqueueOpen(TcpClient client, byte upstreamId, byte proto)
     {
         lock (_openLock)
         {
             if (_pending == null)
             {
-                _pending = new PendingOpen(client, upstreamId);
+                _pending = new PendingOpen(client, upstreamId, proto);
                 SendOpen(_pending);
             }
             else
             {
-                _openQueue.Enqueue(new PendingOpen(client, upstreamId));
+                _openQueue.Enqueue(new PendingOpen(client, upstreamId, proto));
             }
         }
     }
@@ -208,7 +208,7 @@ sealed class SessionManager : IDisposable
             po.Client.Dispose();
             return;
         }
-        _link.SendTo(_serverMac, new Frame.Open(po.UpstreamId));
+        _link.SendTo(_serverMac, new Frame.Open(po.UpstreamId, po.Proto));
     }
 
     void ProcessQueue()
@@ -240,7 +240,7 @@ sealed class SessionManager : IDisposable
         }
 
         var session = new Session(
-            ack.SessionId, po.Client, srcMac, _link!,
+            ack.SessionId, ack.Proto, po.Client, srcMac, _link!,
             () => _sessions.TryRemove(ack.SessionId, out _),
             msg => Log?.Invoke(msg));
         _sessions[ack.SessionId] = session;
@@ -273,14 +273,115 @@ sealed class ListenerState(TcpListener listener, UpstreamEntry upstream)
     public UpstreamEntry Upstream { get; } = upstream;
 }
 
-sealed class PendingOpen(TcpClient client, byte upstreamId)
+sealed class PendingOpen(TcpClient client, byte upstreamId, byte proto)
 {
     public TcpClient Client { get; } = client;
     public byte UpstreamId { get; } = upstreamId;
+    public byte Proto { get; } = proto;
+}
+
+/// L2 reliability: sender-side state.
+class SendState
+{
+    public uint SendSeq;
+    public uint AckedSeq;
+    // seq -> (frame_bytes, send_time_ticks, retry_count)
+    public readonly SortedList<uint, (byte[], long, uint)> RetransmitBuffer = new();
+
+    public uint NextSeq()
+    {
+        var s = SendSeq;
+        SendSeq++;
+        return s;
+    }
+
+    public void RecordSent(uint seq, byte[] frameBytes)
+    {
+        RetransmitBuffer[seq] = (frameBytes, Environment.TickCount64, 0);
+    }
+
+    public void ProcessAck(uint ackSeq)
+    {
+        var toRemove = RetransmitBuffer.Keys
+            .Where(k => k <= ackSeq)
+            .ToList();
+        foreach (var k in toRemove)
+            RetransmitBuffer.Remove(k);
+        if (ackSeq > AckedSeq)
+            AckedSeq = ackSeq;
+    }
+
+    /// Returns frames to retransmit and whether the session should close.
+    public (List<byte[]> resend, bool shouldClose) CheckRetransmit()
+    {
+        var resend = new List<byte[]>();
+        var shouldClose = false;
+        var now = Environment.TickCount64;
+        const int timeoutMs = 5;
+        const uint maxRetries = 10;
+
+        foreach (var kv in RetransmitBuffer.ToList())
+        {
+            if (now - kv.Value.Item2 > timeoutMs)
+            {
+                if (kv.Value.Item3 >= maxRetries)
+                {
+                    shouldClose = true;
+                    break;
+                }
+                RetransmitBuffer[kv.Key] = (kv.Value.Item1, now, kv.Value.Item3 + 1);
+                resend.Add(kv.Value.Item1);
+            }
+        }
+        return (resend, shouldClose);
+    }
+}
+
+/// L2 reliability: receiver-side state.
+class RecvState
+{
+    public uint ExpectedSeq;
+    public uint DeliverSeq;
+    // seq -> payload (out-of-order buffer)
+    public readonly SortedList<uint, byte[]> ReceiveBuffer = new();
+
+    /// Process an incoming DATA frame. Returns (payloads to deliver, needAck).
+    public (List<byte[]> deliver, bool needAck) ProcessData(uint seq, byte[] payload)
+    {
+        if (seq < ExpectedSeq)
+        {
+            // Duplicate.
+            return ([], true);
+        }
+        if (seq == ExpectedSeq)
+        {
+            // In-order: deliver and drain buffer.
+            var deliver = new List<byte[]> { payload };
+            ExpectedSeq++;
+            DeliverSeq = ExpectedSeq - 1;
+            while (ReceiveBuffer.Remove(ExpectedSeq, out var buffered))
+            {
+                deliver.Add(buffered);
+                ExpectedSeq++;
+                DeliverSeq = ExpectedSeq - 1;
+            }
+            return (deliver, false);
+        }
+        else
+        {
+            // Out-of-order: buffer.
+            ReceiveBuffer[seq] = payload;
+            return ([], true);
+        }
+    }
+
+    /// The ack_seq to report in outgoing DATA frames.
+    public uint CurrentAckSeq => ExpectedSeq == 0 ? uint.MaxValue : ExpectedSeq - 1;
 }
 
 sealed class Session(
     uint sessionId,
+    byte proto,
     TcpClient client,
     byte[] serverMac,
     TunnelLink link,
@@ -288,35 +389,107 @@ sealed class Session(
     Action<string>? log) : IDisposable
 {
     readonly CancellationTokenSource _cts = new();
-    readonly Channel<byte[]> _incoming = Channel.CreateBounded<byte[]>(256);
+    readonly bool _isTcp = proto == Proto.ProtoTcp;
+    readonly SendState _send = new();
+    readonly RecvState _recv = new();
+    readonly object _sendLock = new();
+    readonly object _recvLock = new();
+    volatile bool _closeSent;
 
     public void Start()
     {
         _ = PumpSocketToTunnel();
         _ = PumpTunnelToSocket();
+        if (_isTcp)
+            _ = RetransmitTimer();
     }
 
-    public void Deliver(byte[] payload)
+    /// <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()
     {
-        if (!_incoming.Writer.TryWrite(payload))
+        _cts.Cancel();
-            log?.Invoke($"session {sessionId}: incoming channel full");
+        _deliverChannel.Writer.TryComplete();
     }
 
+    /// Handle a DATA frame from the tunnel.
+    public void HandleData(Frame.Data data)
+    {
+        if (!_isTcp)
+        {
+            // Best-effort: deliver directly, ignore seq/ack.
+            if (data.Payload.Length > 0)
+                _deliverChannel.Writer.TryWrite(data.Payload);
+            return;
+        }
+
+        // TCP: process ack_seq to advance send window.
+        lock (_sendLock)
+            _send.ProcessAck(data.AckSeq);
+
+        // Process seq for in-order delivery.
+        List<byte[]> deliver;
+        bool needAck;
+        lock (_recvLock)
+            (deliver, needAck) = _recv.ProcessData(data.Seq, data.Payload);
+
+        // Deliver to the TCP socket via the channel.
+        foreach (var chunk in deliver)
+        {
+            if (!_deliverChannel.Writer.TryWrite(chunk))
+                log?.Invoke($"session {sessionId}: deliver channel full");
+        }
+
+        // Send pure ACK if duplicate or out-of-order.
+        if (needAck)
+            SendPureAck();
+    }
+
+    readonly Channel<byte[]> _deliverChannel = Channel.CreateBounded<byte[]>(256);
+
     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)
             {
                 var n = await stream.ReadAsync(buf, _cts.Token);
                 if (n == 0) break;
-                link.SendTo(serverMac, new Frame.Data(sessionId, buf[..n]));
+
+                uint seq = 0, ackSeq = 0;
+                byte[]? frameBytes = null;
+
+                if (_isTcp)
+                {
+                    lock (_sendLock)
+                    {
+                        seq = _send.NextSeq();
+                        lock (_recvLock)
+                            ackSeq = _recv.CurrentAckSeq;
+                    }
+                }
+
+                var frame = new Frame.Data(sessionId, seq, ackSeq, buf[..n]);
+
+                if (_isTcp)
+                {
+                    frameBytes = FrameCodec.Encode(frame);
+                    lock (_sendLock)
+                        _send.RecordSent(seq, frameBytes);
+                }
+
+                link.SendTo(serverMac, frame);
             }
         }
         catch { }
+        _cts.Cancel();
+        _deliverChannel.Writer.TryComplete();
         SendClose();
         onClosed();
     }
@@ -326,20 +499,88 @@ sealed class Session(
         try
         {
             var stream = client.GetStream();
-            await foreach (var payload in _incoming.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()
+    {
+        using var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(1));
+        try
+        {
+            while (!_cts.IsCancellationRequested)
+            {
+                await timer.WaitForNextTickAsync(_cts.Token);
+
+                List<byte[]>? resend = null;
+                bool shouldClose = false;
+                lock (_sendLock)
+                    (resend, shouldClose) = _send.CheckRetransmit();
+
+                if (resend != null)
+                {
+                    foreach (var frameBytes in resend)
+                    {
+                        // Send raw bytes directly (already encoded).
+                        link.SendRaw(serverMac, frameBytes);
+                    }
+                }
+
+                if (shouldClose)
+                {
+                    link.SendTo(serverMac, new Frame.Close(sessionId, Proto.ReasonMaxRetries));
+                    onClosed();
+                    break;
+                }
+            }
         }
         catch { }
     }
 
-    void SendClose() => link.SendTo(serverMac, new Frame.Close(sessionId, null));
+    void SendPureAck()
+    {
+        if (!_isTcp) return;
+        uint seq, ackSeq;
+        lock (_sendLock)
+        {
+            seq = _send.NextSeq();
+            lock (_recvLock)
+                ackSeq = _recv.CurrentAckSeq;
+        }
+        // Pure ACK: empty payload. Not stored in retransmit buffer.
+        link.SendTo(serverMac, new Frame.Data(sessionId, seq, ackSeq, []));
+    }
+
+    void SendClose()
+    {
+        if (_closeSent) return;
+        _closeSent = true;
+        link.SendTo(serverMac, new Frame.Close(sessionId, null));
+    }
 
     public void Dispose()
     {
         _cts.Cancel();
-        _incoming.Writer.TryComplete();
+        _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/TunnelLink.cs

@@ -72,7 +72,9 @@ sealed class TunnelLink : IDisposable
         SendRaw(dstMac, FrameCodec.Encode(frame));
     }
 
-    void SendRaw(byte[] dstMac, byte[] payload)
+    /// Send a pre-encoded protocol payload (for retransmit, where we already
+    /// have the exact bytes and want to avoid re-encoding).
+    public void SendRaw(byte[] dstMac, byte[] payload)
     {
         var frame = new byte[Proto.EthHeaderLen + payload.Length];
         Buffer.BlockCopy(dstMac, 0, frame, 0, 6);

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="0.1.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,6 +5,7 @@
     <TargetFramework>net8.0-windows</TargetFramework>
     <AssemblyName>gatuna</AssemblyName>
     <RootNamespace>gatuna</RootNamespace>
+    <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 = "0.1.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/frame.rs

@@ -1,9 +1,10 @@
-//! gatuna wire protocol frame encode/decode.
+//! gatuna wire protocol frame encode/decode (v2).
 
 pub const ETHERTYPE: u16 = 0x6969;
 pub const ETH_HEADER_LEN: usize = 14;
-pub const VERSION: u8 = 1;
+pub const VERSION: u8 = 2;
 pub const HEADER_LEN: usize = 8;
+pub const DATA_HEADER_LEN: usize = 16; // 8 common + 4 seq + 4 ack_seq
 pub const MAX_PAYLOAD: usize = 1480;
 
 pub const TYPE_DISCOVER: u8 = 0x01;
@@ -29,6 +30,7 @@ pub const REASON_CONNECT_FAILED: u8 = 2;
 #[allow(dead_code)]
 pub const REASON_OVERSIZE: u8 = 3;
 pub const REASON_UNKNOWN_SESSION: u8 = 4;
+pub const REASON_MAX_RETRIES: u8 = 5;
 
 #[derive(Clone, Debug)]
 pub struct UpstreamEntry {
@@ -42,10 +44,10 @@ pub struct UpstreamEntry {
 pub enum Frame {
     Discover,
     Manifest { hostname: String, entries: Vec<UpstreamEntry> },
-    Open { upstream_id: u8 },
+    Open { upstream_id: u8, proto: u8 },
-    OpenAck { session_id: u32, upstream_id: u8 },
+    OpenAck { session_id: u32, upstream_id: u8, proto: u8 },
     OpenNak { upstream_id: u8, reason: u8 },
-    Data { session_id: u32, payload: Vec<u8> },
+    Data { session_id: u32, seq: u32, ack_seq: u32, payload: Vec<u8> },
     Close { session_id: u32, reason: Option<u8> },
     Ping { nonce: u64 },
     Pong { nonce: u64 },
@@ -81,8 +83,7 @@ fn encode_entry(buf: &mut Vec<u8>, e: &UpstreamEntry) {
     buf.extend_from_slice(&label_bytes[..label_len as usize]);
 }
 
-/// Build the 8-byte header + payload. The payload_len field records the
+/// Build a non-DATA frame: 8-byte common header + payload.
-/// exact payload length so the receiver can ignore Ethernet padding.
 fn build(type_byte: u8, session_id: u32, payload: Vec<u8>) -> Vec<u8> {
     let len = payload.len() as u16;
     let mut buf = Vec::with_capacity(HEADER_LEN + payload.len());
@@ -94,6 +95,21 @@ fn build(type_byte: u8, session_id: u32, payload: Vec<u8>) -> Vec<u8> {
     buf
 }
 
+/// Build a DATA frame: 8-byte common header + seq + ack_seq + payload.
+/// payload_len counts only the raw bytes, not seq/ack_seq.
+fn build_data(session_id: u32, seq: u32, ack_seq: u32, payload: Vec<u8>) -> Vec<u8> {
+    let len = payload.len() as u16;
+    let mut buf = Vec::with_capacity(DATA_HEADER_LEN + payload.len());
+    buf.push(VERSION);
+    buf.push(TYPE_DATA);
+    buf.extend_from_slice(&session_id.to_be_bytes());
+    buf.extend_from_slice(&len.to_be_bytes());
+    buf.extend_from_slice(&seq.to_be_bytes());
+    buf.extend_from_slice(&ack_seq.to_be_bytes());
+    buf.extend_from_slice(&payload);
+    buf
+}
+
 impl Frame {
     pub fn encode(&self) -> Vec<u8> {
         match self {
@@ -109,14 +125,16 @@ impl Frame {
                 }
                 build(TYPE_MANIFEST, 0, payload)
             }
-            Frame::Open { upstream_id } => build(TYPE_OPEN, 0, vec![*upstream_id]),
+            Frame::Open { upstream_id, proto } => build(TYPE_OPEN, 0, vec![*upstream_id, *proto]),
-            Frame::OpenAck { session_id, upstream_id } => {
+            Frame::OpenAck { session_id, upstream_id, proto } => {
-                build(TYPE_OPEN_ACK, *session_id, vec![*upstream_id])
+                build(TYPE_OPEN_ACK, *session_id, vec![*upstream_id, *proto])
             }
             Frame::OpenNak { upstream_id, reason } => {
                 build(TYPE_OPEN_NAK, 0, vec![*upstream_id, *reason])
             }
-            Frame::Data { session_id, payload } => build(TYPE_DATA, *session_id, payload.clone()),
+            Frame::Data { session_id, seq, ack_seq, payload } => {
+                build_data(*session_id, *seq, *ack_seq, payload.clone())
+            }
             Frame::Close { session_id, reason } => {
                 let p = match reason {
                     Some(r) => vec![*r],
@@ -140,11 +158,31 @@ impl Frame {
         let typ = buf[1];
         let session_id = u32::from_be_bytes([buf[2], buf[3], buf[4], buf[5]]);
         let payload_len = u16::from_be_bytes([buf[6], buf[7]]) as usize;
+
+        // DATA frames have seq + ack_seq after the common header.
+        if typ == TYPE_DATA {
+            if buf.len() < DATA_HEADER_LEN + payload_len {
+                return Err(DecodeError::BadPayload("DATA: payload_len exceeds available data"));
+            }
+            let seq = u32::from_be_bytes([buf[8], buf[9], buf[10], buf[11]]);
+            let ack_seq = u32::from_be_bytes([buf[12], buf[13], buf[14], buf[15]]);
+            let payload = &buf[DATA_HEADER_LEN..DATA_HEADER_LEN + payload_len];
+            if payload.len() > MAX_PAYLOAD {
+                return Err(DecodeError::BadPayload("DATA payload exceeds max"));
+            }
+            return Ok(Frame::Data {
+                session_id,
+                seq,
+                ack_seq,
+                payload: payload.to_vec(),
+            });
+        }
+
         if buf.len() < HEADER_LEN + payload_len {
             return Err(DecodeError::BadPayload("payload_len exceeds available data"));
         }
-        // Slice exactly payload_len bytes, ignoring any trailing Ethernet padding.
         let payload = &buf[HEADER_LEN..HEADER_LEN + payload_len];
+
         match typ {
             TYPE_DISCOVER => {
                 if !payload.is_empty() {
@@ -187,16 +225,16 @@ impl Frame {
                 Ok(Frame::Manifest { hostname, entries })
             }
             TYPE_OPEN => {
-                if payload.len() != 1 {
+                if payload.len() != 2 {
-                    return Err(DecodeError::BadPayload("OPEN payload must be 1 byte"));
+                    return Err(DecodeError::BadPayload("OPEN payload must be 2 bytes"));
                 }
-                Ok(Frame::Open { upstream_id: payload[0] })
+                Ok(Frame::Open { upstream_id: payload[0], proto: payload[1] })
             }
             TYPE_OPEN_ACK => {
-                if payload.len() != 1 {
+                if payload.len() != 2 {
-                    return Err(DecodeError::BadPayload("OPEN_ACK payload must be 1 byte"));
+                    return Err(DecodeError::BadPayload("OPEN_ACK payload must be 2 bytes"));
                 }
-                Ok(Frame::OpenAck { session_id, upstream_id: payload[0] })
+                Ok(Frame::OpenAck { session_id, upstream_id: payload[0], proto: payload[1] })
             }
             TYPE_OPEN_NAK => {
                 if payload.len() != 2 {
@@ -204,12 +242,6 @@ impl Frame {
                 }
                 Ok(Frame::OpenNak { upstream_id: payload[0], reason: payload[1] })
             }
-            TYPE_DATA => {
-                if payload.len() > MAX_PAYLOAD {
-                    return Err(DecodeError::BadPayload("DATA payload exceeds max"));
-                }
-                Ok(Frame::Data { session_id, payload: payload.to_vec() })
-            }
             TYPE_CLOSE => {
                 let reason = match payload.len() {
                     0 => None,
@@ -233,7 +265,7 @@ impl Frame {
                 Ok(Frame::Pong { nonce })
             }
             TYPE_UDP_OPEN | TYPE_UDP_DATA | TYPE_UDP_CLOSE => {
-                Err(DecodeError::BadPayload("UDP frame types not implemented in v1"))
+                Err(DecodeError::BadPayload("UDP frame types not implemented"))
             }
             other => Err(DecodeError::UnknownType(other)),
         }

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,13 +16,13 @@ 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,
 };
 use crate::link::Link;
-use crate::session::{spawn_pump, write_to_session, SessionHandle, SessionStore};
+use crate::session::{spawn_pump, handle_data, send_pure_ack, SessionHandle, SessionStore};
 use crate::upstream::build_table;
 
 #[derive(Parser)]
@@ -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,16 +145,21 @@ 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 } => {
+        Frame::Open { upstream_id, proto: _ } => {
-            let port = table.get(upstream_id).map(|u| u.port);
+            info!("OPEN upstream {upstream_id} from {src}");
-            match port {
+            let upstream = table.get(upstream_id);
-                Some(port) => {
+            match upstream {
+                Some(upstream) => {
+                    let port = upstream.port;
+                    let proto = upstream.proto.as_u8();
                     // Spawn so connect() doesn't block the rx loop.
                     let tx = tx.clone();
                     let store = Arc::clone(store);
@@ -165,15 +174,21 @@ async fn handle_frame(
                                     sid,
                                     SessionHandle {
                                         upstream_id,
+                                        proto,
                                         write: w,
+                                        send_state: Arc::new(Mutex::new(session::SendState::new())),
+                                        recv_state: Arc::new(Mutex::new(session::RecvState::new())),
+                                        peer_mac: src,
                                     },
                                 );
-                                let ack = Frame::OpenAck { session_id: sid, upstream_id };
+                                let ack = Frame::OpenAck { session_id: sid, upstream_id, proto };
                                 let _ = tx.send((src, ack.encode())).await;
-                                spawn_pump(r, sid, src, tx, store);
+                                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;
@@ -187,9 +202,13 @@ async fn handle_frame(
                 }
             }
         }
-        Frame::Data { session_id, payload } => {
+        Frame::Data { session_id, seq, ack_seq, payload } => {
-            match write_to_session(store, session_id, &payload).await {
+            match handle_data(store, session_id, seq, ack_seq, &payload, tx).await {
-                Ok(()) => {}
+                Ok(need_ack) => {
+                    if need_ack {
+                        send_pure_ack(store, session_id, tx);
+                    }
+                }
                 Err(session::WriteError::UnknownSession) => {
                     let close = Frame::Close {
                         session_id,
@@ -205,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

@@ -1,20 +1,142 @@
-//! Per-session state and the localhost→tunnel pump.
+//! Per-session state: L2 reliability layer + localhost→tunnel pump.
 
-use crate::frame::{Frame, MAX_PAYLOAD};
+use crate::frame::{Frame, MAX_PAYLOAD, REASON_MAX_RETRIES};
 use crate::link::MacAddr;
-use std::collections::HashMap;
+use std::collections::{HashMap, BTreeMap};
 use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::net::tcp::OwnedReadHalf;
+use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 use tokio::sync::mpsc::Sender;
 use tokio::sync::Mutex as AsyncMutex;
 
 pub type TxChan = Sender<(MacAddr, Vec<u8>)>;
 
+const RETRANSMIT_TIMEOUT: Duration = Duration::from_millis(5);
+const MAX_RETRIES: u32 = 10;
+const RETRANSMIT_TICK: Duration = Duration::from_millis(1);
+
+/// Sender-side reliability state (per session).
+pub struct SendState {
+    send_seq: u32,
+    acked_seq: u32,
+    /// seq -> (frame_bytes, send_time, retry_count)
+    retransmit_buffer: BTreeMap<u32, (Vec<u8>, Instant, u32)>,
+}
+
+/// Receiver-side reliability state (per session).
+pub struct RecvState {
+    expected_seq: u32,
+    deliver_seq: u32,
+    /// seq -> payload (out-of-order buffer)
+    receive_buffer: BTreeMap<u32, Vec<u8>>,
+}
+
+impl SendState {
+    pub fn new() -> Self {
+        SendState {
+            send_seq: 0,
+            acked_seq: 0,
+            retransmit_buffer: BTreeMap::new(),
+        }
+    }
+
+    /// Record a sent DATA frame in the retransmit buffer.
+    fn record_sent(&mut self, seq: u32, frame_bytes: Vec<u8>) {
+        self.retransmit_buffer
+            .insert(seq, (frame_bytes, Instant::now(), 0));
+    }
+
+    /// Process an incoming ack_seq: advance window, remove acked frames.
+    fn process_ack(&mut self, ack_seq: u32) {
+        self.retransmit_buffer.retain(|&seq, _| seq > ack_seq);
+        if ack_seq > self.acked_seq || self.retransmit_buffer.is_empty() {
+            self.acked_seq = ack_seq.max(self.acked_seq);
+        }
+    }
+
+    /// Check for frames needing retransmit. Returns frames to resend and
+    /// whether the session should be closed (max retries exceeded).
+    fn check_retransmit(&mut self) -> (Vec<Vec<u8>>, bool) {
+        let mut resend = Vec::new();
+        let mut should_close = false;
+        let now = Instant::now();
+        for (_seq, (frame_bytes, send_time, retries)) in self.retransmit_buffer.iter_mut() {
+            if now.duration_since(*send_time) > RETRANSMIT_TIMEOUT {
+                if *retries >= MAX_RETRIES {
+                    should_close = true;
+                    break;
+                }
+                *retries += 1;
+                *send_time = now;
+                resend.push(frame_bytes.clone());
+            }
+        }
+        (resend, should_close)
+    }
+
+    fn next_seq(&mut self) -> u32 {
+        let s = self.send_seq;
+        self.send_seq = self.send_seq.wrapping_add(1);
+        s
+    }
+}
+
+impl RecvState {
+    pub fn new() -> Self {
+        RecvState {
+            expected_seq: 0,
+            deliver_seq: 0,
+            receive_buffer: BTreeMap::new(),
+        }
+    }
+
+    /// Process an incoming DATA frame's seq. Returns:
+    /// - `Ok((payloads, need_ack))` — payloads to deliver (may be empty if
+    ///   out-of-order), need_ack is true if the frame was duplicate or
+    ///   out-of-order and the caller should send a pure ACK.
+    /// - `Err(())` — should not happen with current callers.
+    fn process_data(&mut self, seq: u32, payload: Vec<u8>) -> (Vec<Vec<u8>>, bool) {
+        if seq < self.expected_seq {
+            // Duplicate — already delivered.
+            return (vec![], true);
+        }
+        if seq == self.expected_seq {
+            // In-order: deliver immediately, then drain the receive buffer.
+            let mut deliver = vec![payload];
+            self.expected_seq = self.expected_seq.wrapping_add(1);
+            self.deliver_seq = self.expected_seq.wrapping_sub(1);
+            // Drain contiguous buffered frames.
+            while let Some(payload) = self.receive_buffer.remove(&self.expected_seq) {
+                self.expected_seq = self.expected_seq.wrapping_add(1);
+                self.deliver_seq = self.expected_seq.wrapping_sub(1);
+                deliver.push(payload);
+            }
+            (deliver, false)
+        } else {
+            // Out-of-order: buffer it.
+            self.receive_buffer.insert(seq, payload);
+            (vec![], true)
+        }
+    }
+
+    /// The ack_seq to report in outgoing DATA frames (highest contiguous
+    /// delivered seq). If nothing delivered yet, report expected_seq - 1
+    /// (wrapping), which is the last seq we can cumulatively ACK.
+    fn current_ack_seq(&self) -> u32 {
+        self.expected_seq.wrapping_sub(1)
+    }
+}
+
+#[derive(Clone)]
 #[allow(dead_code)]
 pub struct SessionHandle {
     pub upstream_id: u8,
-    pub write: Arc<AsyncMutex<tokio::net::tcp::OwnedWriteHalf>>,
+    pub proto: u8,
+    pub write: Arc<AsyncMutex<OwnedWriteHalf>>,
+    pub send_state: Arc<Mutex<SendState>>,
+    pub recv_state: Arc<Mutex<RecvState>>,
+    pub peer_mac: MacAddr,
 }
 
 pub type SessionStore = Arc<Mutex<HashMap<u32, SessionHandle>>>;
@@ -25,54 +147,191 @@ pub enum WriteError {
     Io,
 }
 
-pub async fn write_to_session(
+/// Handle a DATA frame received from the tunnel. For TCP sessions, delivers
+/// in-order via the reliability layer. For stateless protos, delivers
+/// best-effort. Returns whether a pure ACK should be sent back.
+pub async fn handle_data(
     store: &SessionStore,
-    id: u32,
+    session_id: u32,
+    seq: u32,
+    ack_seq: u32,
     payload: &[u8],
-) -> Result<(), WriteError> {
+    _tx: &TxChan,
-    let write = {
+) -> Result<bool, WriteError> {
+    let handle = {
         let store = store.lock().expect("store lock poisoned");
-        store.get(&id).map(|h| h.write.clone())
+        store.get(&session_id).cloned()
     };
-    match write {
+
-        Some(w) => {
+    let Some(handle) = handle else {
-            let mut w = w.lock().await;
+        return Err(WriteError::UnknownSession);
-            w.write_all(payload).await.map_err(|_| WriteError::Io)?;
+    };
-            Ok(())
+
+    // Non-TCP protos: best-effort delivery, no reliability machinery.
+    if handle.proto != crate::frame::PROTO_TCP {
+        if !payload.is_empty() {
+            let mut w = handle.write.lock().await;
+            if w.write_all(payload).await.is_err() {
+                return Err(WriteError::Io);
             }
-        None => Err(WriteError::UnknownSession),
         }
+        return Ok(false);
+    }
+
+    // TCP: full reliability — process ack_seq to advance send window.
+    {
+        let mut ss = handle.send_state.lock().expect("send_state poisoned");
+        ss.process_ack(ack_seq);
+    }
+
+    // Process seq for in-order delivery.
+    let need_ack;
+    {
+        let mut rs = handle.recv_state.lock().expect("recv_state poisoned");
+        let (deliver, ack) = rs.process_data(seq, payload.to_vec());
+        if !deliver.is_empty() {
+            let mut w = handle.write.lock().await;
+            for chunk in deliver {
+                if w.write_all(&chunk).await.is_err() {
+                    return Err(WriteError::Io);
+                }
+            }
+        }
+        need_ack = ack;
+    }
+
+    Ok(need_ack)
+}
+
+/// Send a pure ACK (DATA frame with empty payload) for the given session.
+pub fn send_pure_ack(
+    store: &SessionStore,
+    session_id: u32,
+    tx: &TxChan,
+) {
+    let handle = {
+        let store = store.lock().expect("store lock poisoned");
+        store.get(&session_id).cloned()
+    };
+    let Some(handle) = handle else { return };
+
+    // Pure ACKs are only meaningful for reliable (TCP) sessions.
+    if handle.proto != crate::frame::PROTO_TCP {
+        return;
+    }
+
+    let (seq, ack_seq) = {
+        let mut ss = handle.send_state.lock().expect("send_state poisoned");
+        let rs = handle.recv_state.lock().expect("recv_state poisoned");
+        let seq = ss.next_seq();
+        (seq, rs.current_ack_seq())
+    };
+
+    // Pure ACKs are not stored in the retransmit buffer (no payload to lose).
+    let frame = Frame::Data {
+        session_id,
+        seq,
+        ack_seq,
+        payload: Vec::new(),
+    };
+    let _ = tx.try_send((handle.peer_mac, frame.encode()));
 }
 
 /// Spawn the socket→tunnel pump: reads from the localhost TCP stream in
-/// 1480-byte chunks and emits DATA frames. On EOF/error sends CLOSE and
+/// 1480-byte chunks, tags each with seq, and emits DATA frames. For TCP
-/// removes the session from the store.
+/// sessions, also stores in retransmit buffer and spawns the retransmit
+/// timer. For stateless protos, best-effort (no retransmit).
 pub fn spawn_pump(
     mut read: OwnedReadHalf,
     session_id: u32,
+    proto: u8,
     peer_mac: MacAddr,
     tx: TxChan,
     store: SessionStore,
 ) {
+    let is_tcp = proto == crate::frame::PROTO_TCP;
+
+    // Retransmit timer task (TCP only).
+    if is_tcp {
+        let store = Arc::clone(&store);
+        let tx = tx.clone();
         tokio::spawn(async move {
-        let mut buf = vec![0u8; MAX_PAYLOAD];
+            let mut interval = tokio::time::interval(RETRANSMIT_TICK);
             loop {
-            match read.read(&mut buf).await {
+                interval.tick().await;
-                Ok(0) => break,
+                let handle = {
-                Ok(n) => {
+                    let store = store.lock().expect("store poisoned");
-                    let frame = Frame::Data {
+                    store.get(&session_id).cloned()
-                        session_id,
-                        payload: buf[..n].to_vec(),
                 };
-                    if tx.send((peer_mac, frame.encode())).await.is_err() {
+                let Some(handle) = handle else { break };
+
+                let (resend, should_close) = {
+                    let mut ss = handle.send_state.lock().expect("send_state poisoned");
+                    ss.check_retransmit()
+                };
+
+                for frame_bytes in resend {
+                    let _ = tx.try_send((peer_mac, frame_bytes));
+                }
+
+                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()));
                     break;
                 }
             }
-                Err(_) => break,
+        });
-            }
+    }
-        }
+
-        let close = Frame::Close { session_id, reason: None };
+    // Socket read pump.
-        let _ = tx.send((peer_mac, close.encode())).await;
+    tokio::spawn(async move {
-        store.lock().expect("store poisoned").remove(&session_id);
+        let mut buf = vec![0u8; MAX_PAYLOAD];
+        loop {
+            let n = match read.read(&mut buf).await {
+                Ok(0) => break,
+                Ok(n) => n,
+                Err(_) => break,
+            };
+
+            let handle = {
+                let store = store.lock().expect("store poisoned");
+                store.get(&session_id).cloned()
+            };
+            let Some(handle) = handle else { break };
+
+            let (seq, ack_seq) = if is_tcp {
+                let mut ss = handle.send_state.lock().expect("send_state poisoned");
+                let rs = handle.recv_state.lock().expect("recv_state poisoned");
+                let seq = ss.next_seq();
+                (seq, rs.current_ack_seq())
+            } else {
+                (0, 0)
+            };
+
+            let frame = Frame::Data {
+                session_id,
+                seq,
+                ack_seq,
+                payload: buf[..n].to_vec(),
+            };
+            let frame_bytes = frame.encode();
+
+            // Store in retransmit buffer before sending (TCP only).
+            if is_tcp {
+                let mut ss = handle.send_state.lock().expect("send_state poisoned");
+                ss.record_sent(seq, frame_bytes.clone());
+            }
+
+            if tx.send((peer_mac, frame_bytes)).await.is_err() {
+                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;
     });
 }