From 31878a72ec666c8dd5e14c7d989ca0d024a12efc Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Tue, 2 Jun 2026 18:36:52 +0800 Subject: [PATCH] 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) --- handler/tunnel/bind.go | 17 ++++++++++ handler/tunnel/connect.go | 26 ++++++++++++++++ handler/tunnel/connector.go | 30 ++++++++++++++++++ handler/tunnel/dialer.go | 14 +++++++++ handler/tunnel/entrypoint.go | 19 ++++++++++-- handler/tunnel/ephttp.go | 15 +++++++++ handler/tunnel/eprelay.go | 14 +++++++++ handler/tunnel/eptls.go | 10 ++++++ handler/tunnel/handler.go | 57 +++++++++++++++++++++++++++++++++- handler/tunnel/handler_test.go | 41 ++++++++++++++++++++++++ handler/tunnel/id.go | 4 +++ handler/tunnel/tunnel.go | 20 ++++++++++++ 12 files changed, 264 insertions(+), 3 deletions(-) diff --git a/handler/tunnel/bind.go b/handler/tunnel/bind.go index ca0dfc93..9449c0d2 100644 --- a/handler/tunnel/bind.go +++ b/handler/tunnel/bind.go @@ -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, diff --git a/handler/tunnel/connect.go b/handler/tunnel/connect.go index 6826aac7..4c9a6dd8 100644 --- a/handler/tunnel/connect.go +++ b/handler/tunnel/connect.go @@ -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), diff --git a/handler/tunnel/connector.go b/handler/tunnel/connector.go index 31a3153d..0cd361ab 100644 --- a/handler/tunnel/connector.go +++ b/handler/tunnel/connector.go @@ -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 diff --git a/handler/tunnel/dialer.go b/handler/tunnel/dialer.go index b03afbb2..b0a4f8e4 100644 --- a/handler/tunnel/dialer.go +++ b/handler/tunnel/dialer.go @@ -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 diff --git a/handler/tunnel/entrypoint.go b/handler/tunnel/entrypoint.go index c5119e4a..a63d489c 100644 --- a/handler/tunnel/entrypoint.go +++ b/handler/tunnel/entrypoint.go @@ -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 diff --git a/handler/tunnel/ephttp.go b/handler/tunnel/ephttp.go index 0a4c07e8..18096b13 100644 --- a/handler/tunnel/ephttp.go +++ b/handler/tunnel/ephttp.go @@ -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) diff --git a/handler/tunnel/eprelay.go b/handler/tunnel/eprelay.go index 247ba58a..36f2783b 100644 --- a/handler/tunnel/eprelay.go +++ b/handler/tunnel/eprelay.go @@ -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 { diff --git a/handler/tunnel/eptls.go b/handler/tunnel/eptls.go index ba009aca..674d1076 100644 --- a/handler/tunnel/eptls.go +++ b/handler/tunnel/eptls.go @@ -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)) diff --git a/handler/tunnel/handler.go b/handler/tunnel/handler.go index d7d40797..e0ad08a2 100644 --- a/handler/tunnel/handler.go +++ b/handler/tunnel/handler.go @@ -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() diff --git a/handler/tunnel/handler_test.go b/handler/tunnel/handler_test.go index fec60dd7..0e17ac08 100644 --- a/handler/tunnel/handler_test.go +++ b/handler/tunnel/handler_test.go @@ -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 } diff --git a/handler/tunnel/id.go b/handler/tunnel/id.go index 99c19f89..15efb160 100644 --- a/handler/tunnel/id.go +++ b/handler/tunnel/id.go @@ -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 ( diff --git a/handler/tunnel/tunnel.go b/handler/tunnel/tunnel.go index 2cddd65c..99055a7f 100644 --- a/handler/tunnel/tunnel.go +++ b/handler/tunnel/tunnel.go @@ -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