diff --git a/PROTOCOL.md b/PROTOCOL.md index eb9472b..9d13e92 100644 --- a/PROTOCOL.md +++ b/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 -the endpoints' TCP stacks; UDP (reserved) will rely on application-level -mechanisms. +## DATA frame layout -Maximum payload: 1500 (Ethernet MTU) − 8 (our header) = **1492 bytes**. In -practice we cap at **1480** to stay conservative. Larger payloads are not -emitted in v1. +DATA frames carry two additional fields after the common header for L2-level +reliability: + +``` + 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 @@ -44,12 +68,12 @@ emitted in v1. | 0x03 | OPEN | C → S | 0 | `upstream_id:1` | | 0x04 | OPEN_ACK | S → C | assigned | `upstream_id: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,81 @@ Reserved (unimplemented in v1; parse returns Err, encode unimplemented): | 0x09 | UDP_DATA | | 0x0A | UDP_CLOSE | +## L2 reliability + +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). + +### 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 @@ -129,8 +228,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 -field identifies which session the bytes belong to. +`[ seq:4 ][ ack_seq:4 ][ raw bytes ]` — the `seq` and `ack_seq` fields are +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 +264,7 @@ field identifies which session the bytes belong to. | 2 | connect_failed | | 3 | oversize | | 4 | unknown_session | +| 5 | max_retries | ## Discovery flow @@ -195,23 +297,27 @@ client server | OPEN_NAK { reason } | |<--------------------------------| (on failure) | | - | DATA { session_id, bytes } | - |<------------------------------->| DATA { session_id, bytes } + | DATA { seq, ack_seq, 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` - with `write_all`. On error it emits `CLOSE` and drops the session. +- The server's tunnel→socket path delivers DATA payloads in seq order to the + `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 +341,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. diff --git a/README.md b/README.md index 824b3ce..570b5ab 100644 --- a/README.md +++ b/README.md @@ -39,13 +39,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 +145,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 diff --git a/gatuna-win/Frame.cs b/gatuna-win/Frame.cs index b89a361..022e813 100644 --- a/gatuna-win/Frame.cs +++ b/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( @@ -46,7 +48,7 @@ abstract record Frame internal record Open(byte UpstreamId) : Frame; internal record OpenAck(uint SessionId, byte UpstreamId) : 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 - /// payload length so the receiver can ignore Ethernet padding. + /// Build a non-DATA frame: 8-byte common header + payload. 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 @@ -86,7 +112,7 @@ static class FrameCodec 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 payload = buf.Slice(Proto.HeaderLen, payloadLen); + var payload2 = buf.Slice(Proto.HeaderLen, payloadLen); return type switch { - Proto.TypeManifest => ParseManifest(payload), - Proto.TypeOpenAck when payload.Length == 1 => - new Frame.OpenAck(sessionId, payload[0]), - Proto.TypeOpenNak when payload.Length == 2 => - new Frame.OpenNak(payload[0], payload[1]), - Proto.TypeData when payload.Length <= Proto.MaxPayload => - new Frame.Data(sessionId, payload.ToArray()), - Proto.TypeClose when payload.Length is 0 or 1 => - new Frame.Close(sessionId, payload.Length == 1 ? payload[0] : null), - Proto.TypePong when payload.Length == 8 => - new Frame.Pong(ParseNonce(payload)), - Proto.TypePing when payload.Length == 8 => - new Frame.Ping(ParseNonce(payload)), - Proto.TypeDiscover when payload.Length == 0 => + Proto.TypeManifest => ParseManifest(payload2), + Proto.TypeOpenAck when payload2.Length == 1 => + new Frame.OpenAck(sessionId, payload2[0]), + Proto.TypeOpenNak when payload2.Length == 2 => + new Frame.OpenNak(payload2[0], payload2[1]), + Proto.TypeClose when payload2.Length is 0 or 1 => + new Frame.Close(sessionId, payload2.Length == 1 ? payload2[0] : null), + Proto.TypePong when payload2.Length == 8 => + new Frame.Pong(ParseNonce(payload2)), + Proto.TypePing when payload2.Length == 8 => + new Frame.Ping(ParseNonce(payload2)), + Proto.TypeDiscover when payload2.Length == 0 => new Frame.Discover(), _ => null, }; diff --git a/gatuna-win/SessionManager.cs b/gatuna-win/SessionManager.cs index f667e59..97813ea 100644 --- a/gatuna-win/SessionManager.cs +++ b/gatuna-win/SessionManager.cs @@ -92,7 +92,7 @@ sealed class SessionManager : IDisposable 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)); @@ -279,6 +279,105 @@ sealed class PendingOpen(TcpClient client, byte upstreamId) public byte UpstreamId { get; } = upstreamId; } +/// L2 reliability: sender-side state. +class SendState +{ + public uint SendSeq; + public uint AckedSeq; + // seq -> (frame_bytes, send_time_ticks, retry_count) + public readonly SortedList 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 resend, bool shouldClose) CheckRetransmit() + { + var resend = new List(); + 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 ReceiveBuffer = new(); + + /// Process an incoming DATA frame. Returns (payloads to deliver, needAck). + public (List 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 { 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, TcpClient client, @@ -288,20 +387,45 @@ sealed class Session( Action? log) : IDisposable { readonly CancellationTokenSource _cts = new(); - readonly Channel _incoming = Channel.CreateBounded(256); + readonly SendState _send = new(); + readonly RecvState _recv = new(); + readonly object _sendLock = new(); + readonly object _recvLock = new(); public void Start() { _ = PumpSocketToTunnel(); _ = PumpTunnelToSocket(); + _ = RetransmitTimer(); } - public void Deliver(byte[] payload) + /// Handle a DATA frame from the tunnel: process ack, deliver in-order. + public void HandleData(Frame.Data data) { - if (!_incoming.Writer.TryWrite(payload)) - log?.Invoke($"session {sessionId}: incoming channel full"); + // Process ack_seq to advance send window. + lock (_sendLock) + _send.ProcessAck(data.AckSeq); + + // Process seq for in-order delivery. + List 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 _deliverChannel = Channel.CreateBounded(256); + async Task PumpSocketToTunnel() { try @@ -313,7 +437,23 @@ sealed class Session( { var n = await stream.ReadAsync(buf, _cts.Token); if (n == 0) break; - link.SendTo(serverMac, new Frame.Data(sessionId, buf[..n])); + + uint seq; + uint ackSeq; + lock (_sendLock) + { + seq = _send.NextSeq(); + lock (_recvLock) + ackSeq = _recv.CurrentAckSeq; + } + + var frame = new Frame.Data(sessionId, seq, ackSeq, buf[..n]); + var frameBytes = FrameCodec.Encode(frame); + + lock (_sendLock) + _send.RecordSent(seq, frameBytes); + + link.SendTo(serverMac, frame); } } catch { } @@ -326,18 +466,65 @@ 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(_cts.Token)) await stream.WriteAsync(payload, _cts.Token); } catch { } } + async Task RetransmitTimer() + { + using var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(1)); + try + { + while (!_cts.IsCancellationRequested) + { + await timer.WaitForNextTickAsync(_cts.Token); + + List? 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 SendPureAck() + { + 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() => link.SendTo(serverMac, new Frame.Close(sessionId, null)); public void Dispose() { _cts.Cancel(); - _incoming.Writer.TryComplete(); + _deliverChannel.Writer.TryComplete(); SendClose(); try { client.Dispose(); } catch { } _cts.Dispose(); diff --git a/gatuna-win/TunnelLink.cs b/gatuna-win/TunnelLink.cs index d4ea6c0..186b81e 100644 --- a/gatuna-win/TunnelLink.cs +++ b/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); diff --git a/gatunad/src/frame.rs b/gatunad/src/frame.rs index 702ef29..e4f43ef 100644 --- a/gatunad/src/frame.rs +++ b/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 { @@ -45,7 +47,7 @@ pub enum Frame { Open { upstream_id: u8 }, OpenAck { session_id: u32, upstream_id: u8 }, OpenNak { upstream_id: u8, reason: u8 }, - Data { session_id: u32, payload: Vec }, + Data { session_id: u32, seq: u32, ack_seq: u32, payload: Vec }, Close { session_id: u32, reason: Option }, Ping { nonce: u64 }, Pong { nonce: u64 }, @@ -81,8 +83,7 @@ fn encode_entry(buf: &mut Vec, e: &UpstreamEntry) { buf.extend_from_slice(&label_bytes[..label_len as usize]); } -/// Build the 8-byte header + payload. The payload_len field records the -/// exact payload length so the receiver can ignore Ethernet padding. +/// Build a non-DATA frame: 8-byte common header + payload. fn build(type_byte: u8, session_id: u32, payload: Vec) -> Vec { 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) -> Vec { 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) -> Vec { + 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 { match self { @@ -116,7 +132,9 @@ impl Frame { 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() { @@ -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)), } diff --git a/gatunad/src/main.rs b/gatunad/src/main.rs index 4035a44..d4d354b 100644 --- a/gatunad/src/main.rs +++ b/gatunad/src/main.rs @@ -22,7 +22,7 @@ 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)] @@ -166,6 +166,9 @@ async fn handle_frame( SessionHandle { upstream_id, 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 }; @@ -187,9 +190,13 @@ async fn handle_frame( } } } - Frame::Data { session_id, payload } => { - match write_to_session(store, session_id, &payload).await { - Ok(()) => {} + Frame::Data { session_id, seq, ack_seq, payload } => { + match handle_data(store, session_id, seq, ack_seq, &payload, tx).await { + Ok(need_ack) => { + if need_ack { + send_pure_ack(store, session_id, tx); + } + } Err(session::WriteError::UnknownSession) => { let close = Frame::Close { session_id, diff --git a/gatunad/src/session.rs b/gatunad/src/session.rs index 4a3a5c5..f825a09 100644 --- a/gatunad/src/session.rs +++ b/gatunad/src/session.rs @@ -1,20 +1,140 @@ -//! 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)>; -#[allow(dead_code)] +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, 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>, +} + +impl SendState { + 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) { + 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>, 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 { + 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) -> (Vec>, 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)] pub struct SessionHandle { pub upstream_id: u8, - pub write: Arc>, + pub write: Arc>, + pub send_state: Arc>, + pub recv_state: Arc>, + pub peer_mac: MacAddr, } pub type SessionStore = Arc>>; @@ -25,28 +145,83 @@ pub enum WriteError { Io, } -pub async fn write_to_session( +/// Handle a DATA frame received from the tunnel. Delivers in-order payloads +/// to the TCP socket and processes the ack_seq. Returns whether a pure ACK +/// should be sent back (duplicate or out-of-order). +pub async fn handle_data( store: &SessionStore, - id: u32, + session_id: u32, + seq: u32, + ack_seq: u32, payload: &[u8], -) -> Result<(), WriteError> { - let write = { + tx: &TxChan, +) -> Result { + 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 mut w = w.lock().await; - w.write_all(payload).await.map_err(|_| WriteError::Io)?; - Ok(()) - } - None => Err(WriteError::UnknownSession), + + let Some(handle) = handle else { + return Err(WriteError::UnknownSession); + }; + + // 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 }; + + 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 -/// removes the session from the store. +/// 1480-byte chunks, tags each with seq, stores in retransmit buffer, and +/// emits DATA frames. Also spawns the retransmit timer. pub fn spawn_pump( mut read: OwnedReadHalf, session_id: u32, @@ -54,21 +229,81 @@ pub fn spawn_pump( tx: TxChan, store: SessionStore, ) { + // Retransmit timer task. + { + let store = Arc::clone(&store); + let tx = tx.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(RETRANSMIT_TICK); + loop { + interval.tick().await; + let handle = { + let store = store.lock().expect("store poisoned"); + store.get(&session_id).cloned() + }; + 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 { + 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; + } + } + }); + } + + // Socket read pump. tokio::spawn(async move { let mut buf = vec![0u8; MAX_PAYLOAD]; loop { - match read.read(&mut buf).await { + let n = match read.read(&mut buf).await { Ok(0) => break, - Ok(n) => { - let frame = Frame::Data { - session_id, - payload: buf[..n].to_vec(), - }; - if tx.send((peer_mac, frame.encode())).await.is_err() { - 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) = { + 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()) + }; + + 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. + { + 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; } } let close = Frame::Close { session_id, reason: None };