From 7fa50dd4e4bf6e5c01c4c15f8e4069513caef90f Mon Sep 17 00:00:00 2001 From: Mute Date: Thu, 13 Aug 2026 09:15:15 +0000 Subject: [PATCH] carry transport proto end-to-end through OPEN/OPEN_ACK MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OPEN and OPEN_ACK now carry proto:1 alongside upstream_id:1. The session is tagged with its transport proto and reliability semantics switch on the specific proto: - TCP (proto=1): full L2 reliability (seq, ack, retransmit, in-order) - UDP (proto=2, reserved): best-effort (no retransmit, no ordering, no pure ACKs — seq/ack_seq fields present but ignored) This is architecturally correct: instead of a generic 'reliable:bool' flag, each proto gets the semantics it needs. Today only TCP exists so every session is reliable, but the extension point is clean for when stateless protos are added. Updated: PROTOCOL.md, Rust frame.rs/session.rs/main.rs, C# Frame.cs/SessionManager.cs. --- PROTOCOL.md | 45 ++++++++++++++++++---------- gatuna-win/Frame.cs | 12 ++++---- gatuna-win/SessionManager.cs | 58 ++++++++++++++++++++++++------------ gatunad/src/frame.rs | 22 +++++++------- gatunad/src/main.rs | 15 ++++++---- gatunad/src/session.rs | 45 +++++++++++++++++++++------- 6 files changed, 129 insertions(+), 68 deletions(-) diff --git a/PROTOCOL.md b/PROTOCOL.md index 9d13e92..e4ca3d3 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -65,8 +65,8 @@ Maximum payload: 1500 (Ethernet MTU) − 8 (common header) − 8 (seq + ack_seq) |------|-------------|---------------|------------|----------------------------------| | 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` | -| 0x04 | OPEN_ACK | S → C | assigned | `upstream_id:1` | +| 0x03 | OPEN | C → S | 0 | `upstream_id:1, proto: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 | `seq:4, ack_seq:4, raw bytes` | | 0x07 | CLOSE | both | session | optional `reason:1` | @@ -83,11 +83,21 @@ Reserved (unimplemented; parse returns Err, encode unimplemented): ## 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). +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) @@ -197,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 - identified by the `session_id` field in the header, not the payload. +- **upstream_id** (u8): echoes the requested upstream. +- **proto** (u8): echoes the requested proto. The session is identified by + the `session_id` field in the header, not the payload. ### OPEN_NAK payload diff --git a/gatuna-win/Frame.cs b/gatuna-win/Frame.cs index 022e813..cea54a1 100644 --- a/gatuna-win/Frame.cs +++ b/gatuna-win/Frame.cs @@ -45,8 +45,8 @@ abstract record Frame { internal record Discover : Frame; internal record Manifest(string Hostname, UpstreamEntry[] Entries) : Frame; - internal record Open(byte UpstreamId) : Frame; - internal record OpenAck(uint SessionId, byte UpstreamId) : Frame; + internal record Open(byte UpstreamId, byte Proto) : Frame; + internal record OpenAck(uint SessionId, byte UpstreamId, byte Proto) : Frame; internal record OpenNak(byte UpstreamId, byte Reason) : Frame; internal record Data(uint SessionId, uint Seq, uint AckSeq, byte[] Payload) : Frame; internal record Close(uint SessionId, byte? Reason) : Frame; @@ -106,9 +106,9 @@ 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 => @@ -177,8 +177,8 @@ static class FrameCodec return type switch { Proto.TypeManifest => ParseManifest(payload2), - Proto.TypeOpenAck when payload2.Length == 1 => - new Frame.OpenAck(sessionId, payload2[0]), + Proto.TypeOpenAck when payload2.Length == 2 => + new Frame.OpenAck(sessionId, payload2[0], payload2[1]), Proto.TypeOpenNak when payload2.Length == 2 => new Frame.OpenNak(payload2[0], payload2[1]), Proto.TypeClose when payload2.Length is 0 or 1 => diff --git a/gatuna-win/SessionManager.cs b/gatuna-win/SessionManager.cs index 97813ea..bebc73b 100644 --- a/gatuna-win/SessionManager.cs +++ b/gatuna-win/SessionManager.cs @@ -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,10 +273,11 @@ 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. @@ -380,6 +381,7 @@ class RecvState sealed class Session( uint sessionId, + byte proto, TcpClient client, byte[] serverMac, TunnelLink link, @@ -387,6 +389,7 @@ sealed class Session( Action? log) : IDisposable { readonly CancellationTokenSource _cts = new(); + readonly bool _isTcp = proto == Proto.ProtoTcp; readonly SendState _send = new(); readonly RecvState _recv = new(); readonly object _sendLock = new(); @@ -396,13 +399,22 @@ sealed class Session( { _ = PumpSocketToTunnel(); _ = PumpTunnelToSocket(); - _ = RetransmitTimer(); + if (_isTcp) + _ = RetransmitTimer(); } - /// Handle a DATA frame from the tunnel: process ack, deliver in-order. + /// Handle a DATA frame from the tunnel. public void HandleData(Frame.Data data) { - // Process ack_seq to advance send window. + 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); @@ -438,20 +450,27 @@ sealed class Session( var n = await stream.ReadAsync(buf, _cts.Token); if (n == 0) break; - uint seq; - uint ackSeq; - lock (_sendLock) + uint seq = 0, ackSeq = 0; + byte[]? frameBytes = null; + + if (_isTcp) { - seq = _send.NextSeq(); - lock (_recvLock) - ackSeq = _recv.CurrentAckSeq; + 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); + if (_isTcp) + { + frameBytes = FrameCodec.Encode(frame); + lock (_sendLock) + _send.RecordSent(seq, frameBytes); + } link.SendTo(serverMac, frame); } @@ -508,6 +527,7 @@ sealed class Session( void SendPureAck() { + if (!_isTcp) return; uint seq, ackSeq; lock (_sendLock) { diff --git a/gatunad/src/frame.rs b/gatunad/src/frame.rs index e4f43ef..e05fea0 100644 --- a/gatunad/src/frame.rs +++ b/gatunad/src/frame.rs @@ -44,8 +44,8 @@ pub struct UpstreamEntry { pub enum Frame { Discover, Manifest { hostname: String, entries: Vec }, - Open { upstream_id: u8 }, - OpenAck { session_id: u32, upstream_id: u8 }, + Open { upstream_id: u8, proto: u8 }, + OpenAck { session_id: u32, upstream_id: u8, proto: u8 }, OpenNak { upstream_id: u8, reason: u8 }, Data { session_id: u32, seq: u32, ack_seq: u32, payload: Vec }, Close { session_id: u32, reason: Option }, @@ -125,9 +125,9 @@ impl Frame { } build(TYPE_MANIFEST, 0, payload) } - Frame::Open { upstream_id } => build(TYPE_OPEN, 0, vec![*upstream_id]), - Frame::OpenAck { session_id, upstream_id } => { - build(TYPE_OPEN_ACK, *session_id, vec![*upstream_id]) + Frame::Open { upstream_id, proto } => build(TYPE_OPEN, 0, vec![*upstream_id, *proto]), + Frame::OpenAck { session_id, upstream_id, proto } => { + build(TYPE_OPEN_ACK, *session_id, vec![*upstream_id, *proto]) } Frame::OpenNak { upstream_id, reason } => { build(TYPE_OPEN_NAK, 0, vec![*upstream_id, *reason]) @@ -225,16 +225,16 @@ impl Frame { Ok(Frame::Manifest { hostname, entries }) } TYPE_OPEN => { - if payload.len() != 1 { - return Err(DecodeError::BadPayload("OPEN payload must be 1 byte")); + if payload.len() != 2 { + 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 { - return Err(DecodeError::BadPayload("OPEN_ACK payload must be 1 byte")); + if payload.len() != 2 { + 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 { diff --git a/gatunad/src/main.rs b/gatunad/src/main.rs index d4d354b..f510669 100644 --- a/gatunad/src/main.rs +++ b/gatunad/src/main.rs @@ -147,10 +147,12 @@ async fn handle_frame( }; let _ = tx.send((src, manifest.encode())).await; } - Frame::Open { upstream_id } => { - let port = table.get(upstream_id).map(|u| u.port); - match port { - Some(port) => { + Frame::Open { upstream_id, proto } => { + let upstream = table.get(upstream_id); + 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 +167,16 @@ 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); + spawn_pump(r, sid, proto, src, tx, store); } Err(e) => { error!("connect 127.0.0.1:{port} failed: {e}"); diff --git a/gatunad/src/session.rs b/gatunad/src/session.rs index f825a09..7a017d1 100644 --- a/gatunad/src/session.rs +++ b/gatunad/src/session.rs @@ -131,6 +131,7 @@ impl RecvState { #[derive(Clone)] pub struct SessionHandle { pub upstream_id: u8, + pub proto: u8, pub write: Arc>, pub send_state: Arc>, pub recv_state: Arc>, @@ -145,9 +146,9 @@ pub enum WriteError { Io, } -/// 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). +/// 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, session_id: u32, @@ -165,7 +166,18 @@ pub async fn handle_data( return Err(WriteError::UnknownSession); }; - // Process ack_seq to advance send window. + // 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); + } + } + 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); @@ -202,6 +214,11 @@ pub fn send_pure_ack( }; 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"); @@ -220,17 +237,21 @@ pub fn send_pure_ack( } /// Spawn the socket→tunnel pump: reads from the localhost TCP stream in -/// 1480-byte chunks, tags each with seq, stores in retransmit buffer, and -/// emits DATA frames. Also spawns the retransmit timer. +/// 1480-byte chunks, tags each with seq, and emits DATA frames. For TCP +/// 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, ) { - // Retransmit timer task. - { + 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 { @@ -281,11 +302,13 @@ pub fn spawn_pump( }; let Some(handle) = handle else { break }; - let (seq, ack_seq) = { + 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 { @@ -296,8 +319,8 @@ pub fn spawn_pump( }; let frame_bytes = frame.encode(); - // Store in retransmit buffer before sending. - { + // 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()); }