carry transport proto end-to-end through OPEN/OPEN_ACK
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.
This commit is contained in:
+30
-15
@@ -65,8 +65,8 @@ Maximum payload: 1500 (Ethernet MTU) − 8 (common header) − 8 (seq + ack_seq)
|
|||||||
|------|-------------|---------------|------------|----------------------------------|
|
|------|-------------|---------------|------------|----------------------------------|
|
||||||
| 0x01 | DISCOVER | C → broadcast | 0 | empty |
|
| 0x01 | DISCOVER | C → broadcast | 0 | empty |
|
||||||
| 0x02 | MANIFEST | S → C | 0 | `hostname_len:1, hostname:N, entries...` |
|
| 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` |
|
| 0x05 | OPEN_NAK | S → C | 0 | `upstream_id:1, reason:1` |
|
||||||
| 0x06 | DATA | both | session | `seq:4, ack_seq:4, raw bytes` |
|
| 0x06 | DATA | both | session | `seq:4, ack_seq:4, raw bytes` |
|
||||||
| 0x07 | CLOSE | both | session | optional `reason:1` |
|
| 0x07 | CLOSE | both | session | optional `reason:1` |
|
||||||
@@ -83,11 +83,21 @@ Reserved (unimplemented; parse returns Err, encode unimplemented):
|
|||||||
|
|
||||||
## L2 reliability
|
## L2 reliability
|
||||||
|
|
||||||
DATA frames are delivered reliably and in-order via a TCP-like sequence +
|
The reliability semantics of a session depend on its transport proto,
|
||||||
cumulative ACK + retransmit mechanism. This ensures that a dropped Ethernet
|
which is carried end-to-end through OPEN/OPEN_ACK:
|
||||||
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
|
- **TCP (`proto = 1`):** DATA frames are delivered reliably and in-order via
|
||||||
frames).
|
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)
|
### Sender state (per session)
|
||||||
|
|
||||||
@@ -197,23 +207,28 @@ the payload is exhausted. The number of entries is not carried explicitly.
|
|||||||
### OPEN payload
|
### OPEN payload
|
||||||
|
|
||||||
```
|
```
|
||||||
+---------------+
|
+---------------+---------------+
|
||||||
| upstream_id |
|
| upstream_id | proto |
|
||||||
+---------------+
|
+---------------+---------------+
|
||||||
```
|
```
|
||||||
|
|
||||||
- **upstream_id** (u8): which MANIFEST entry to open.
|
- **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
|
### 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
|
### OPEN_NAK payload
|
||||||
|
|
||||||
|
|||||||
+6
-6
@@ -45,8 +45,8 @@ abstract record Frame
|
|||||||
{
|
{
|
||||||
internal record Discover : Frame;
|
internal record Discover : Frame;
|
||||||
internal record Manifest(string Hostname, UpstreamEntry[] Entries) : 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 OpenNak(byte UpstreamId, byte Reason) : Frame;
|
||||||
internal record Data(uint SessionId, uint Seq, uint AckSeq, 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 Close(uint SessionId, byte? Reason) : Frame;
|
||||||
@@ -106,9 +106,9 @@ static class FrameCodec
|
|||||||
Frame.Manifest manifest =>
|
Frame.Manifest manifest =>
|
||||||
Build(Proto.TypeManifest, 0, BuildManifestPayload(manifest.Hostname, manifest.Entries)),
|
Build(Proto.TypeManifest, 0, BuildManifestPayload(manifest.Hostname, manifest.Entries)),
|
||||||
Frame.Open open =>
|
Frame.Open open =>
|
||||||
Build(Proto.TypeOpen, 0, [open.UpstreamId]),
|
Build(Proto.TypeOpen, 0, [open.UpstreamId, open.Proto]),
|
||||||
Frame.OpenAck ack =>
|
Frame.OpenAck ack =>
|
||||||
Build(Proto.TypeOpenAck, ack.SessionId, [ack.UpstreamId]),
|
Build(Proto.TypeOpenAck, ack.SessionId, [ack.UpstreamId, ack.Proto]),
|
||||||
Frame.OpenNak nak =>
|
Frame.OpenNak nak =>
|
||||||
Build(Proto.TypeOpenNak, 0, [nak.UpstreamId, nak.Reason]),
|
Build(Proto.TypeOpenNak, 0, [nak.UpstreamId, nak.Reason]),
|
||||||
Frame.Data data =>
|
Frame.Data data =>
|
||||||
@@ -177,8 +177,8 @@ static class FrameCodec
|
|||||||
return type switch
|
return type switch
|
||||||
{
|
{
|
||||||
Proto.TypeManifest => ParseManifest(payload2),
|
Proto.TypeManifest => ParseManifest(payload2),
|
||||||
Proto.TypeOpenAck when payload2.Length == 1 =>
|
Proto.TypeOpenAck when payload2.Length == 2 =>
|
||||||
new Frame.OpenAck(sessionId, payload2[0]),
|
new Frame.OpenAck(sessionId, payload2[0], payload2[1]),
|
||||||
Proto.TypeOpenNak when payload2.Length == 2 =>
|
Proto.TypeOpenNak when payload2.Length == 2 =>
|
||||||
new Frame.OpenNak(payload2[0], payload2[1]),
|
new Frame.OpenNak(payload2[0], payload2[1]),
|
||||||
Proto.TypeClose when payload2.Length is 0 or 1 =>
|
Proto.TypeClose when payload2.Length is 0 or 1 =>
|
||||||
|
|||||||
@@ -180,22 +180,22 @@ sealed class SessionManager : IDisposable
|
|||||||
client = await state.Listener.AcceptTcpClientAsync();
|
client = await state.Listener.AcceptTcpClientAsync();
|
||||||
}
|
}
|
||||||
catch { break; }
|
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)
|
lock (_openLock)
|
||||||
{
|
{
|
||||||
if (_pending == null)
|
if (_pending == null)
|
||||||
{
|
{
|
||||||
_pending = new PendingOpen(client, upstreamId);
|
_pending = new PendingOpen(client, upstreamId, proto);
|
||||||
SendOpen(_pending);
|
SendOpen(_pending);
|
||||||
}
|
}
|
||||||
else
|
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();
|
po.Client.Dispose();
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
_link.SendTo(_serverMac, new Frame.Open(po.UpstreamId));
|
_link.SendTo(_serverMac, new Frame.Open(po.UpstreamId, po.Proto));
|
||||||
}
|
}
|
||||||
|
|
||||||
void ProcessQueue()
|
void ProcessQueue()
|
||||||
@@ -240,7 +240,7 @@ sealed class SessionManager : IDisposable
|
|||||||
}
|
}
|
||||||
|
|
||||||
var session = new Session(
|
var session = new Session(
|
||||||
ack.SessionId, po.Client, srcMac, _link!,
|
ack.SessionId, ack.Proto, po.Client, srcMac, _link!,
|
||||||
() => _sessions.TryRemove(ack.SessionId, out _),
|
() => _sessions.TryRemove(ack.SessionId, out _),
|
||||||
msg => Log?.Invoke(msg));
|
msg => Log?.Invoke(msg));
|
||||||
_sessions[ack.SessionId] = session;
|
_sessions[ack.SessionId] = session;
|
||||||
@@ -273,10 +273,11 @@ sealed class ListenerState(TcpListener listener, UpstreamEntry upstream)
|
|||||||
public UpstreamEntry Upstream { get; } = 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 TcpClient Client { get; } = client;
|
||||||
public byte UpstreamId { get; } = upstreamId;
|
public byte UpstreamId { get; } = upstreamId;
|
||||||
|
public byte Proto { get; } = proto;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// L2 reliability: sender-side state.
|
/// L2 reliability: sender-side state.
|
||||||
@@ -380,6 +381,7 @@ class RecvState
|
|||||||
|
|
||||||
sealed class Session(
|
sealed class Session(
|
||||||
uint sessionId,
|
uint sessionId,
|
||||||
|
byte proto,
|
||||||
TcpClient client,
|
TcpClient client,
|
||||||
byte[] serverMac,
|
byte[] serverMac,
|
||||||
TunnelLink link,
|
TunnelLink link,
|
||||||
@@ -387,6 +389,7 @@ sealed class Session(
|
|||||||
Action<string>? log) : IDisposable
|
Action<string>? log) : IDisposable
|
||||||
{
|
{
|
||||||
readonly CancellationTokenSource _cts = new();
|
readonly CancellationTokenSource _cts = new();
|
||||||
|
readonly bool _isTcp = proto == Proto.ProtoTcp;
|
||||||
readonly SendState _send = new();
|
readonly SendState _send = new();
|
||||||
readonly RecvState _recv = new();
|
readonly RecvState _recv = new();
|
||||||
readonly object _sendLock = new();
|
readonly object _sendLock = new();
|
||||||
@@ -396,13 +399,22 @@ sealed class Session(
|
|||||||
{
|
{
|
||||||
_ = PumpSocketToTunnel();
|
_ = PumpSocketToTunnel();
|
||||||
_ = PumpTunnelToSocket();
|
_ = 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)
|
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)
|
lock (_sendLock)
|
||||||
_send.ProcessAck(data.AckSeq);
|
_send.ProcessAck(data.AckSeq);
|
||||||
|
|
||||||
@@ -438,20 +450,27 @@ sealed class Session(
|
|||||||
var n = await stream.ReadAsync(buf, _cts.Token);
|
var n = await stream.ReadAsync(buf, _cts.Token);
|
||||||
if (n == 0) break;
|
if (n == 0) break;
|
||||||
|
|
||||||
uint seq;
|
uint seq = 0, ackSeq = 0;
|
||||||
uint ackSeq;
|
byte[]? frameBytes = null;
|
||||||
lock (_sendLock)
|
|
||||||
|
if (_isTcp)
|
||||||
{
|
{
|
||||||
seq = _send.NextSeq();
|
lock (_sendLock)
|
||||||
lock (_recvLock)
|
{
|
||||||
ackSeq = _recv.CurrentAckSeq;
|
seq = _send.NextSeq();
|
||||||
|
lock (_recvLock)
|
||||||
|
ackSeq = _recv.CurrentAckSeq;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
var frame = new Frame.Data(sessionId, seq, ackSeq, buf[..n]);
|
var frame = new Frame.Data(sessionId, seq, ackSeq, buf[..n]);
|
||||||
var frameBytes = FrameCodec.Encode(frame);
|
|
||||||
|
|
||||||
lock (_sendLock)
|
if (_isTcp)
|
||||||
_send.RecordSent(seq, frameBytes);
|
{
|
||||||
|
frameBytes = FrameCodec.Encode(frame);
|
||||||
|
lock (_sendLock)
|
||||||
|
_send.RecordSent(seq, frameBytes);
|
||||||
|
}
|
||||||
|
|
||||||
link.SendTo(serverMac, frame);
|
link.SendTo(serverMac, frame);
|
||||||
}
|
}
|
||||||
@@ -508,6 +527,7 @@ sealed class Session(
|
|||||||
|
|
||||||
void SendPureAck()
|
void SendPureAck()
|
||||||
{
|
{
|
||||||
|
if (!_isTcp) return;
|
||||||
uint seq, ackSeq;
|
uint seq, ackSeq;
|
||||||
lock (_sendLock)
|
lock (_sendLock)
|
||||||
{
|
{
|
||||||
|
|||||||
+11
-11
@@ -44,8 +44,8 @@ pub struct UpstreamEntry {
|
|||||||
pub enum Frame {
|
pub enum Frame {
|
||||||
Discover,
|
Discover,
|
||||||
Manifest { hostname: String, entries: Vec<UpstreamEntry> },
|
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 },
|
OpenNak { upstream_id: u8, reason: u8 },
|
||||||
Data { session_id: u32, seq: u32, ack_seq: u32, payload: Vec<u8> },
|
Data { session_id: u32, seq: u32, ack_seq: u32, payload: Vec<u8> },
|
||||||
Close { session_id: u32, reason: Option<u8> },
|
Close { session_id: u32, reason: Option<u8> },
|
||||||
@@ -125,9 +125,9 @@ impl Frame {
|
|||||||
}
|
}
|
||||||
build(TYPE_MANIFEST, 0, payload)
|
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 } => {
|
Frame::OpenNak { upstream_id, reason } => {
|
||||||
build(TYPE_OPEN_NAK, 0, vec![*upstream_id, *reason])
|
build(TYPE_OPEN_NAK, 0, vec![*upstream_id, *reason])
|
||||||
@@ -225,16 +225,16 @@ impl Frame {
|
|||||||
Ok(Frame::Manifest { hostname, entries })
|
Ok(Frame::Manifest { hostname, entries })
|
||||||
}
|
}
|
||||||
TYPE_OPEN => {
|
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 => {
|
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 => {
|
TYPE_OPEN_NAK => {
|
||||||
if payload.len() != 2 {
|
if payload.len() != 2 {
|
||||||
|
|||||||
+9
-6
@@ -147,10 +147,12 @@ async fn handle_frame(
|
|||||||
};
|
};
|
||||||
let _ = tx.send((src, manifest.encode())).await;
|
let _ = tx.send((src, manifest.encode())).await;
|
||||||
}
|
}
|
||||||
Frame::Open { upstream_id } => {
|
Frame::Open { upstream_id, proto } => {
|
||||||
let port = table.get(upstream_id).map(|u| u.port);
|
let upstream = table.get(upstream_id);
|
||||||
match port {
|
match upstream {
|
||||||
Some(port) => {
|
Some(upstream) => {
|
||||||
|
let port = upstream.port;
|
||||||
|
let proto = upstream.proto.as_u8();
|
||||||
// Spawn so connect() doesn't block the rx loop.
|
// Spawn so connect() doesn't block the rx loop.
|
||||||
let tx = tx.clone();
|
let tx = tx.clone();
|
||||||
let store = Arc::clone(store);
|
let store = Arc::clone(store);
|
||||||
@@ -165,15 +167,16 @@ async fn handle_frame(
|
|||||||
sid,
|
sid,
|
||||||
SessionHandle {
|
SessionHandle {
|
||||||
upstream_id,
|
upstream_id,
|
||||||
|
proto,
|
||||||
write: w,
|
write: w,
|
||||||
send_state: Arc::new(Mutex::new(session::SendState::new())),
|
send_state: Arc::new(Mutex::new(session::SendState::new())),
|
||||||
recv_state: Arc::new(Mutex::new(session::RecvState::new())),
|
recv_state: Arc::new(Mutex::new(session::RecvState::new())),
|
||||||
peer_mac: src,
|
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;
|
let _ = tx.send((src, ack.encode())).await;
|
||||||
spawn_pump(r, sid, src, tx, store);
|
spawn_pump(r, sid, proto, src, tx, store);
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
error!("connect 127.0.0.1:{port} failed: {e}");
|
error!("connect 127.0.0.1:{port} failed: {e}");
|
||||||
|
|||||||
+34
-11
@@ -131,6 +131,7 @@ impl RecvState {
|
|||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
pub struct SessionHandle {
|
pub struct SessionHandle {
|
||||||
pub upstream_id: u8,
|
pub upstream_id: u8,
|
||||||
|
pub proto: u8,
|
||||||
pub write: Arc<AsyncMutex<OwnedWriteHalf>>,
|
pub write: Arc<AsyncMutex<OwnedWriteHalf>>,
|
||||||
pub send_state: Arc<Mutex<SendState>>,
|
pub send_state: Arc<Mutex<SendState>>,
|
||||||
pub recv_state: Arc<Mutex<RecvState>>,
|
pub recv_state: Arc<Mutex<RecvState>>,
|
||||||
@@ -145,9 +146,9 @@ pub enum WriteError {
|
|||||||
Io,
|
Io,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Handle a DATA frame received from the tunnel. Delivers in-order payloads
|
/// Handle a DATA frame received from the tunnel. For TCP sessions, delivers
|
||||||
/// to the TCP socket and processes the ack_seq. Returns whether a pure ACK
|
/// in-order via the reliability layer. For stateless protos, delivers
|
||||||
/// should be sent back (duplicate or out-of-order).
|
/// best-effort. Returns whether a pure ACK should be sent back.
|
||||||
pub async fn handle_data(
|
pub async fn handle_data(
|
||||||
store: &SessionStore,
|
store: &SessionStore,
|
||||||
session_id: u32,
|
session_id: u32,
|
||||||
@@ -165,7 +166,18 @@ pub async fn handle_data(
|
|||||||
return Err(WriteError::UnknownSession);
|
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");
|
let mut ss = handle.send_state.lock().expect("send_state poisoned");
|
||||||
ss.process_ack(ack_seq);
|
ss.process_ack(ack_seq);
|
||||||
@@ -202,6 +214,11 @@ pub fn send_pure_ack(
|
|||||||
};
|
};
|
||||||
let Some(handle) = handle else { return };
|
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 (seq, ack_seq) = {
|
||||||
let mut ss = handle.send_state.lock().expect("send_state poisoned");
|
let mut ss = handle.send_state.lock().expect("send_state poisoned");
|
||||||
let rs = handle.recv_state.lock().expect("recv_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
|
/// Spawn the socket→tunnel pump: reads from the localhost TCP stream in
|
||||||
/// 1480-byte chunks, tags each with seq, stores in retransmit buffer, and
|
/// 1480-byte chunks, tags each with seq, and emits DATA frames. For TCP
|
||||||
/// emits DATA frames. Also spawns the retransmit timer.
|
/// sessions, also stores in retransmit buffer and spawns the retransmit
|
||||||
|
/// timer. For stateless protos, best-effort (no retransmit).
|
||||||
pub fn spawn_pump(
|
pub fn spawn_pump(
|
||||||
mut read: OwnedReadHalf,
|
mut read: OwnedReadHalf,
|
||||||
session_id: u32,
|
session_id: u32,
|
||||||
|
proto: u8,
|
||||||
peer_mac: MacAddr,
|
peer_mac: MacAddr,
|
||||||
tx: TxChan,
|
tx: TxChan,
|
||||||
store: SessionStore,
|
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 store = Arc::clone(&store);
|
||||||
let tx = tx.clone();
|
let tx = tx.clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
@@ -281,11 +302,13 @@ pub fn spawn_pump(
|
|||||||
};
|
};
|
||||||
let Some(handle) = handle else { break };
|
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 mut ss = handle.send_state.lock().expect("send_state poisoned");
|
||||||
let rs = handle.recv_state.lock().expect("recv_state poisoned");
|
let rs = handle.recv_state.lock().expect("recv_state poisoned");
|
||||||
let seq = ss.next_seq();
|
let seq = ss.next_seq();
|
||||||
(seq, rs.current_ack_seq())
|
(seq, rs.current_ack_seq())
|
||||||
|
} else {
|
||||||
|
(0, 0)
|
||||||
};
|
};
|
||||||
|
|
||||||
let frame = Frame::Data {
|
let frame = Frame::Data {
|
||||||
@@ -296,8 +319,8 @@ pub fn spawn_pump(
|
|||||||
};
|
};
|
||||||
let frame_bytes = frame.encode();
|
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");
|
let mut ss = handle.send_state.lock().expect("send_state poisoned");
|
||||||
ss.record_sent(seq, frame_bytes.clone());
|
ss.record_sent(seq, frame_bytes.clone());
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user