docs(handler/tunnel): add architecture documentation and fix observeStats event loss
- Add package-level architecture doc in handler.go covering: - NAT traversal reverse proxy architecture - CmdBind/CmdConnect roles and data flow - Connector lifecycle and waitClose semantics - Entrypoint protocol dispatch (first-byte sniffing) - SD fallback behavior - Add data-flow-oriented doc comments to all business files: - entrypoint.go: protocol dispatch, dial flow - connector.go: Connector/ConnectorPool semantics - dialer.go: two-phase dial strategy - tunnel.go: MaxWeight semantics, selection algorithm - bind.go: 6-step CmdBind flow - connect.go: CmdConnect flow with relay framing - ephttp.go, eptls.go, eprelay.go: per-protocol entrypoint flow - Fix observeStats: after successful error retry, also flush new events instead of skipping the current tick (handler.go:261-271) - Add test for observeStats retry-then-flush (handler_test.go)
This commit is contained in:
ginuerzh
@@ -15,6 +15,23 @@ import (
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// handleBind handles a CmdBind request from an internal client.
|
||||
//
|
||||
// Flow:
|
||||
// 1. Generate a random connector ID (copies weight from tunnelID).
|
||||
// 2. Compute an 8-hex-char endpoint from md5(tunnelID) for ingress routing.
|
||||
// 3. Send relay response with address + tunnel features back to client.
|
||||
// 4. Upgrade the TCP connection to a mux.ClientSession (smux).
|
||||
// 5. Create a Connector wrapping the mux session.
|
||||
// 6. Register:
|
||||
// a. Add Connector to ConnectorPool (under tunnelID).
|
||||
// b. If ingress is configured, set rules: endpoint → tunnelID, and
|
||||
// the bind address host → tunnelID.
|
||||
// c. If SD is configured, register the service (tunnelID, node, network).
|
||||
//
|
||||
// The mux session ownership is transferred to the Connector — conn is NOT
|
||||
// closed after this function returns (no defer conn.Close()). The Connector's
|
||||
// waitClose goroutine handles session lifecycle.
|
||||
func (h *tunnelHandler) handleBind(ctx context.Context, conn net.Conn, network, address string, tunnelID relay.TunnelID, log logger.Logger) (err error) {
|
||||
resp := relay.Response{
|
||||
Version: relay.Version1,
|
||||
|
||||
@@ -13,6 +13,32 @@ import (
|
||||
xnet "github.com/go-gost/x/internal/net"
|
||||
)
|
||||
|
||||
// handleConnect handles a CmdConnect request.
|
||||
//
|
||||
// This is the "pull" path: a public user's connection arrives via the tunnel
|
||||
// handler's listener, and the handler creates a stream into the tunnel to
|
||||
// reach the internal service.
|
||||
//
|
||||
// Flow:
|
||||
// 1. Bypass check against dstAddr.
|
||||
// 2. Ingress routing: if the tunnel is the public entryPointID, look up the
|
||||
// destination host in the ingress table to find the target tunnel ID.
|
||||
// For direct tunnels, the tunnelID from the request is used directly.
|
||||
// 3. Dialer.Dial() → pool.Get() → GetConn() → mux.OpenStream()
|
||||
// (or SD fallback if no local connector).
|
||||
// 4. Relay protocol framing:
|
||||
// a. Local node (node == h.id): write StatusOK response to the public
|
||||
// connection, then write a StatusOK response with src/dst address
|
||||
// features to the mux stream. The internal client uses these address
|
||||
// features to know where to connect.
|
||||
// b. Remote node (SD fallback): write the original relay request
|
||||
// directly to the mux stream for the remote node to process.
|
||||
// 5. Pipe(publicConn, muxStream) — bidirectional data relay until either
|
||||
// side closes.
|
||||
//
|
||||
// The mux stream (cc) is closed via defer cc.Close() when this function
|
||||
// returns. The public connection (conn) is closed by the caller's
|
||||
// defer conn.Close() in Handle().
|
||||
func (h *tunnelHandler) handleConnect(ctx context.Context, req *relay.Request, conn net.Conn, network, srcAddr string, dstAddr string, tunnelID relay.TunnelID, log logger.Logger) error {
|
||||
log = log.WithFields(map[string]any{
|
||||
"dst": fmt.Sprintf("%s/%s", dstAddr, network),
|
||||
|
||||
@@ -25,6 +25,24 @@ type ConnectorOptions struct {
|
||||
limiter traffic.TrafficLimiter
|
||||
}
|
||||
|
||||
// Connector represents one mux-session tunnel endpoint registered by an
|
||||
// internal client via CmdBind.
|
||||
//
|
||||
// Each Connector wraps a smux.Session created by mux.ClientSession. From the
|
||||
// server side, Connector.GetConn() calls OpenStream() to create a stream to
|
||||
// the internal client.
|
||||
//
|
||||
// Connector.waitClose is a background goroutine that accepts and discards
|
||||
// unexpected inbound streams. This is a safety guard: the Connector's mux
|
||||
// session is supposed to be used unidirectionally (server→client), and any
|
||||
// stream arriving from the client side is anomalous and is safely discarded.
|
||||
// Normal request streams are created by the server side via OpenStream and
|
||||
// consumed by the internal client's Accept loop — they never pass through
|
||||
// waitClose.
|
||||
//
|
||||
// When the mux session is closed (client disconnect or network error),
|
||||
// waitClose detects the Accept error, closes the Connector, and deregisters
|
||||
// from SD if configured.
|
||||
type Connector struct {
|
||||
id relay.ConnectorID
|
||||
tid relay.TunnelID
|
||||
@@ -127,6 +145,18 @@ func (c *Connector) IsClosed() bool {
|
||||
return c.s.IsClosed()
|
||||
}
|
||||
|
||||
// ConnectorPool manages all Tunnel objects for this tunnel handler node.
|
||||
//
|
||||
// Hierarchy:
|
||||
//
|
||||
// ConnectorPool (per node) → map[tunnelID]*Tunnel → []*Connector
|
||||
// └── *mux.Session
|
||||
//
|
||||
// Each Tunnel holds Connectors sharing the same tunnel ID. Tunnels that have
|
||||
// no active connectors for 15 minutes are removed by closeIdles.
|
||||
//
|
||||
// Methods: Add (insert connector, creating Tunnel on demand), Get (select best
|
||||
// connector by network type), Close (stop idle ticker, remove all tunnels).
|
||||
type ConnectorPool struct {
|
||||
node string
|
||||
tunnels map[string]*Tunnel
|
||||
|
||||
@@ -9,6 +9,20 @@ import (
|
||||
"github.com/go-gost/core/sd"
|
||||
)
|
||||
|
||||
// Dialer resolves a tunnel connector and returns a stream to it.
|
||||
//
|
||||
// Dial strategy (two-phase):
|
||||
// 1. Local pool: try ConnectorPool.Get() up to retry times. Each attempt
|
||||
// calls GetConn() (= mux.OpenStream) on the same connector — if the
|
||||
// connector is dead, all retries fail until Tunnel.clean() removes it
|
||||
// (up to TTL, default 15s).
|
||||
// 2. SD fallback: if pool returns nil AND sd is configured, query service
|
||||
// discovery for remote nodes. Filter out self (d.node) and mismatched
|
||||
// networks. Establish a raw TCP connection to the remote address,
|
||||
// bypassing the mux layer entirely.
|
||||
//
|
||||
// The returned node and connector ID identify which node/hop the stream
|
||||
// is connected to — used by callers to decide the relay protocol framing.
|
||||
type Dialer struct {
|
||||
node string
|
||||
pool *ConnectorPool
|
||||
|
||||
@@ -24,8 +24,23 @@ import (
|
||||
)
|
||||
|
||||
// entrypoint is a public tunnel entry point that accepts external connections
|
||||
// and routes them through the tunnel network. It supports three protocols
|
||||
// determined by the first byte of the connection: relay (Version1), TLS, or HTTP.
|
||||
// and routes them through the tunnel network.
|
||||
//
|
||||
// Protocol dispatch is done by peeking at the first byte of the connection:
|
||||
// - relay.Version1 (0x52 'R') → relay protocol (handleConnect)
|
||||
// - dissector.Handshake (0x16) → TLS passthrough (handleTLS)
|
||||
// - otherwise → HTTP proxy (handleHTTP)
|
||||
//
|
||||
// For HTTP/TLS paths, the dial flow is:
|
||||
// ep.dial() → ingress lookup (host → tunnelID) → Dialer.Dial()
|
||||
// → ConnectorPool.Get() → Connector.GetConn() → mux.OpenStream()
|
||||
// → relay.Response{src, dst} written to mux stream (when local node)
|
||||
//
|
||||
// The relay protocol path handles its own tunnel ID extraction from the
|
||||
// relay request frame.
|
||||
//
|
||||
// When Dialer falls back to SD (no local connector), the mux layer is
|
||||
// bypassed and a raw TCP connection is established to the remote node.
|
||||
type entrypoint struct {
|
||||
node string
|
||||
service string
|
||||
|
||||
@@ -26,6 +26,21 @@ import (
|
||||
"golang.org/x/net/http/httpguts"
|
||||
)
|
||||
|
||||
// handleHTTP processes an HTTP request arriving at the entrypoint.
|
||||
//
|
||||
// Flow:
|
||||
// 1. Read HTTP request from the public connection.
|
||||
// 2. Record request metadata in ro.HTTP.
|
||||
// 3. httpRoundTrip: shallow-copy ro, check forwarding loop,
|
||||
// build http.Request, call ep.transport.RoundTrip() which
|
||||
// uses ep.dial() as DialContext — that resolves ingress rules
|
||||
// to a tunnelID, calls Dialer.Dial(), and writes relay address
|
||||
// features into the mux stream.
|
||||
// 4. Handle WebSocket upgrade via handleUpgradeResponse (sniffing).
|
||||
// 5. Continue in a loop for keep-alive: read next request, repeat.
|
||||
//
|
||||
// HTTP request body recording: if recorder options specify HTTPBody,
|
||||
// the request body is wrapped in a xhttp.Body for capture.
|
||||
func (ep *entrypoint) handleHTTP(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) {
|
||||
pStats := xstats.Stats{}
|
||||
conn = stats_wrapper.WrapConn(conn, &pStats)
|
||||
|
||||
@@ -12,6 +12,20 @@ import (
|
||||
xrecorder "github.com/go-gost/x/recorder"
|
||||
)
|
||||
|
||||
// handleConnect (entrypoint relay) processes a relay-protocol connection
|
||||
// arriving at the entrypoint.
|
||||
//
|
||||
// Flow:
|
||||
// 1. Read relay.Request (extracts src/dst address, tunnelID, network).
|
||||
// 2. Dialer.Dial() → pool.Get() → GetConn() → mux.OpenStream().
|
||||
// 3. Write StatusOK response to the public connection.
|
||||
// 4. Write relay.Response with src/dst address features to the mux stream.
|
||||
// 5. Pipe(publicConn, muxStream).
|
||||
//
|
||||
// Unlike tunnelHandler.handleConnect, this path does not use ingress routing
|
||||
// or bypass checks — the relay request already contains the tunnel ID.
|
||||
// Also unlike handleConnect, there is no local-vs-remote framing difference:
|
||||
// the entrypoint always writes StatusOK + address features regardless of node.
|
||||
func (ep *entrypoint) handleConnect(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) {
|
||||
req := relay.Request{}
|
||||
if _, err := req.ReadFrom(conn); err != nil {
|
||||
|
||||
@@ -18,6 +18,16 @@ import (
|
||||
xrecorder "github.com/go-gost/x/recorder"
|
||||
)
|
||||
|
||||
// handleTLS processes a TLS connection arriving at the entrypoint.
|
||||
//
|
||||
// Flow:
|
||||
// 1. Parse ClientHello (tee-reads first bytes for recording).
|
||||
// 2. Extract SNI hostname → ingress lookup → tunnelID.
|
||||
// 3. ep.dial() → Dialer.Dial() → mux stream (or SD TCP connection).
|
||||
// 4. Write buffered ClientHello bytes to the mux stream.
|
||||
// 5. Parse ServerHello from the mux stream (for TLS recording).
|
||||
// 6. Write ServerHello bytes back to the public connection.
|
||||
// 7. Pipe(publicConn, muxStream) — bidirectional TLS passthrough.
|
||||
func (ep *entrypoint) handleTLS(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
|
||||
buf := new(bytes.Buffer)
|
||||
clientHello, err := dissector.ParseClientHello(io.TeeReader(conn, buf))
|
||||
|
||||
@@ -1,3 +1,54 @@
|
||||
// Package tunnel implements a reverse proxy tunnel handler for NAT traversal.
|
||||
//
|
||||
// Architecture overview
|
||||
//
|
||||
// The tunnel handler is deployed on the public-facing (server) side. It acts as a
|
||||
// bridge between external clients and internal services behind NAT/firewall.
|
||||
//
|
||||
// There are two main roles:
|
||||
//
|
||||
// 1. Internal client (CmdBind) — connects to the tunnel handler and registers a
|
||||
// multiplexed session (mux.Session via smux) as a Connector. Once bound, this
|
||||
// client passively waits to receive streams from the public side.
|
||||
//
|
||||
// 2. Public entrypoints (CmdConnect + entrypoint) — accept incoming requests from
|
||||
// the Internet and forward them through the tunnel to the internal client via
|
||||
// mux streams (OpenStream).
|
||||
//
|
||||
// Data flow (normal direction: public → internal):
|
||||
//
|
||||
// Public request → tunnelHandler.Handle() / entrypoint.Handle()
|
||||
// → Dialer.Dial()
|
||||
// → ConnectorPool.Get() → Tunnel.GetConnector() → Connector.GetConn()
|
||||
// → mux.Session.OpenStream() ← creates stream to internal side
|
||||
// → Pipe(publicConn, muxStream)
|
||||
//
|
||||
// Internal client side:
|
||||
//
|
||||
// mux.Session.AcceptStream() ← receives the stream
|
||||
// → processes request, sends response back through the same stream
|
||||
//
|
||||
// Connector lifecycle (CmdBind):
|
||||
//
|
||||
// Internal client sends CmdBind → handleBind() creates mux.ClientSession
|
||||
// → NewConnector (stores session, starts waitClose goroutine)
|
||||
// → ConnectorPool.Add() → Tunnel.AddConnector()
|
||||
// → ingress rules + SD service registered
|
||||
//
|
||||
// The waitClose goroutine (Connector.waitClose) discards unexpected inbound
|
||||
// streams on the Connector's mux session. This is a safety guard — normal
|
||||
// request streams arrive via OpenStream from the public side and are handled
|
||||
// by the internal client's Accept loop, NOT by waitClose.
|
||||
//
|
||||
// Entrypoint protocol dispatch (first-byte sniffing):
|
||||
//
|
||||
// relay.Version1 (0x52 'R') → handleConnect (relay protocol)
|
||||
// dissector.Handshake (0x16) → handleTLS (TLS passthrough)
|
||||
// otherwise → handleHTTP (HTTP forward proxy)
|
||||
//
|
||||
// SD fallback: when ConnectorPool.Get() returns nil (no local tunnel registered),
|
||||
// Dialer queries service discovery for a remote node address and establishes a
|
||||
// direct TCP connection, bypassing the mux session entirely.
|
||||
package tunnel
|
||||
|
||||
import (
|
||||
@@ -261,8 +312,12 @@ func (h *tunnelHandler) observeStats(ctx context.Context) {
|
||||
if len(events) > 0 {
|
||||
if err := h.options.Observer.Observe(ctx, events); err == nil {
|
||||
events = nil
|
||||
// Retry succeeded — also flush new events this tick.
|
||||
// Fall through without break to reach the normal path.
|
||||
} else {
|
||||
// Retry still failing — skip new events, try again next tick.
|
||||
break
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
evs := h.stats.Events()
|
||||
|
||||
@@ -3,6 +3,7 @@ package tunnel
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"testing"
|
||||
@@ -269,18 +270,58 @@ func TestHandler_observeStats(t *testing.T) {
|
||||
h.observeStats(ctx)
|
||||
// Should return without blocking
|
||||
})
|
||||
|
||||
t.Run("retry after observer error flushes new events", func(t *testing.T) {
|
||||
observeC := make(chan []observer.Event, 4)
|
||||
callCount := 0
|
||||
h := &tunnelHandler{
|
||||
md: metadata{
|
||||
observerPeriod: 50 * time.Millisecond,
|
||||
observerResetTraffic: false,
|
||||
},
|
||||
stats: stats_util.NewHandlerStats("test", false),
|
||||
}
|
||||
h.options.Observer = &fakeObserver{
|
||||
observeC: observeC,
|
||||
errFunc: func() bool {
|
||||
callCount++
|
||||
// Fail on first call, succeed on subsequent calls
|
||||
return callCount == 1
|
||||
},
|
||||
}
|
||||
|
||||
// Trigger stats events
|
||||
s := h.stats.Stats("client1")
|
||||
s.Add(stats.KindTotalConns, 1)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
go h.observeStats(ctx)
|
||||
|
||||
// First tick: Observe fails, events are stored as pending.
|
||||
// Second tick: Observe succeeds on pending events, then also flushes new events.
|
||||
<-observeC // first call (fails)
|
||||
<-observeC // second call (pending retry succeeds)
|
||||
<-observeC // third call (new events flushed in same tick)
|
||||
|
||||
cancel()
|
||||
})
|
||||
}
|
||||
|
||||
// fakeObserver implements the observer.Observer interface for testing.
|
||||
type fakeObserver struct {
|
||||
observeC chan []observer.Event
|
||||
err error
|
||||
errFunc func() bool
|
||||
}
|
||||
|
||||
func (o *fakeObserver) Observe(ctx context.Context, events []observer.Event, opts ...observer.Option) error {
|
||||
if o.observeC != nil {
|
||||
o.observeC <- events
|
||||
}
|
||||
if o.errFunc != nil && o.errFunc() {
|
||||
return errors.New("simulated observer error")
|
||||
}
|
||||
return o.err
|
||||
}
|
||||
|
||||
|
||||
@@ -1,3 +1,7 @@
|
||||
// Package tunnel implements the GOST relay tunnel handler for NAT traversal,
|
||||
// connecting public entrypoints to internal services behind NAT/firewall.
|
||||
//
|
||||
// See handler.go for a full architecture overview.
|
||||
package tunnel
|
||||
|
||||
import (
|
||||
|
||||
@@ -10,10 +10,30 @@ import (
|
||||
"github.com/go-gost/x/selector"
|
||||
)
|
||||
|
||||
// MaxWeight is the exclusive takeover weight (255). A connector at MaxWeight
|
||||
// triggers rw.Reset() in GetConnector, clearing all previously added connectors
|
||||
// so that only this one is selected. If that MaxWeight connector later dies
|
||||
// (IsClosed), GetConnector returns nil until Tunnel.clean() removes it.
|
||||
const (
|
||||
MaxWeight uint8 = 0xff
|
||||
)
|
||||
|
||||
// Tunnel groups Connectors that share the same tunnel ID.
|
||||
//
|
||||
// Connector selection (GetConnector):
|
||||
// - Single connector: fast path, no weighted random.
|
||||
// - Multiple connectors: weighted random selection by network type (tcp/udp).
|
||||
// - MaxWeight (255): exclusive — triggers Reset() and only the MaxWeight
|
||||
// connector is added. If the MaxWeight connector closes, selection returns
|
||||
// nil until clean() removes it.
|
||||
//
|
||||
// Tunnel lifecycle:
|
||||
// - NewTunnel starts a clean() goroutine that runs every TTL (default 15s).
|
||||
// Each tick removes closed connectors and renews SD registrations.
|
||||
// A tunnel with 0 connectors gets cleaned up.
|
||||
// - CloseOnIdle closes the tunnel if it has 0 connectors (called by
|
||||
// ConnectorPool.closeIdles on a 15-minute ticker).
|
||||
// - Close() closes all connectors and signals shutdown.
|
||||
type Tunnel struct {
|
||||
node string
|
||||
id relay.TunnelID
|
||||
|
||||
Reference in New Issue
Block a user