From d189dc79672be056df1c0b1ab5980124d512c501 Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Thu, 21 May 2026 10:28:55 +0800 Subject: [PATCH] fix(pipe): prevent error swallowing and goroutine leak in Pipe/Transport Pipe: remove context.WithCancel that caused premature ctx.Done() before the second pipeHalf could return its error. Replace with a completed counter that drains both errors even on cancellation. Transport: increase error channel buffer from 1 to 2 to prevent goroutine leak when both CopyBuffer goroutines complete. --- CLAUDE.md | 139 ++++++++++++++++++++++++++++++++++++++ README.md | 32 ++++++++- internal/net/pipe.go | 31 ++++----- internal/net/transport.go | 2 +- 4 files changed, 185 insertions(+), 19 deletions(-) create mode 100644 CLAUDE.md diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 00000000..bb312a78 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,139 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this directory. + +## Build & Verify + +```bash +cd x && go build ./... # verify compilation +cd x && go vet ./... # static analysis +cd x && go build ./... && go vet ./... # both +``` + +There are no tests in this module. Build + vet is the verification path. + +## Architecture + +`x/` is the **implementation layer** for the GOST proxy framework. Every interface defined in `core/` is implemented here. The module ties the whole system together: it provides ~25 handler implementations, ~25 listener implementations, ~20 dialer implementations, ~15 connector implementations, plus config parsing, registry management, and all cross-cutting concerns. + +### Component pattern (applies to every handler, listener, dialer, connector) + +Every component follows this exact pattern: + +```go +// 1. init() self-registration via blank import +func init() { + registry.HandlerRegistry().Register("http", NewHandler) +} + +// 2. Private struct holding options + parsed metadata +type httpHandler struct { + md metadata + options handler.Options +} + +// 3. Functional options constructor +func NewHandler(opts ...handler.Option) handler.Handler { + options := handler.Options{} + for _, opt := range opts { + opt(&options) + } + return &httpHandler{options: options} +} + +// 4. Init extracts typed values from generic metadata +func (h *httpHandler) Init(md md.Metadata) error { + return h.parseMetadata(md) +} +``` + +Each component package also defines a private `metadata` struct and a `parseMetadata(md)` method that uses `mdutil.GetBool/GetInt/GetString/GetDuration(md, keys...)` to extract configuration. Metadata keys are lowercased and multiple fallback keys are tried (e.g., `"observePeriod"`, `"observer.period"`, `"observer.observePeriod"`). + +### Config parsing and service composition + +`config/parsing/service/parse.go` is the critical wiring function. `ParseService(cfg)` does: + +1. Defaults listener type to `"tcp"`, handler type to `"auto"` +2. Sets up TLS config, network namespace, authentication, admission, bypass +3. Looks up the listener and handler factories from their registries, calls them with composed options +4. Creates a `service.Service` (which runs the accept loop calling `handler.Handle` per connection) +5. Wraps the listener in order: proxyproto → metrics → stats → admission → traffic limiter → connection limiter + +Each config sub-package (`config/parsing/{auth,bypass,admission,chain,hop,limiter,...}/`) has a `parse.go` that converts config structs into registry lookups and constructed objects. + +### Metadata system (`metadata/`) + +`metadata.NewMetadata(map[string]any)` wraps a raw config map. All keys are lowercased on lookup. The `mdutil` package (`metadata/util/`) provides typed accessors used everywhere: + +- `mdutil.GetBool(md, keys...)`, `GetInt`, `GetFloat`, `GetString`, `GetStrings` +- `mdutil.GetDuration(md, keys...)` — int values treated as seconds; strings parsed with `time.ParseDuration` +- `mdutil.GetStringMap(md, keys...)`, `GetStringMapString` + +Metadata key constants are defined in `config/parsing/parse.go` (e.g., `MDKeyProxyProtocol`, `MDKeySoMark`, `MDKeyInterface`, `MDKeyEnableStats`). + +### Internal packages (`internal/`) + +**Do not import from external modules.** Go's `internal/` restriction enforces this. Key internal packages: + +| Package | Purpose | +|---------|---------| +| `internal/ctx/` | Context values for session state during requests | +| `internal/io/` | Custom IO readers/writers | +| `internal/loader/` | `Loader`, `Lister`, `Mapper` interfaces for hot-reloadable components | +| `internal/matcher/` | Pattern matching utilities | +| `internal/net/` | Network utilities (resolver, dialer, proxyproto, HTTP helpers, UDP listener) | +| `internal/plugin/` | External plugin process management | +| `internal/util/sniffing/` | Protocol sniffing (TLS, HTTP, WebSocket) | +| `internal/util/stats/` | Connection stats tracking | +| `internal/util/tls/` | TLS config loading, cert management | +| `internal/util/ws/` | WebSocket utilities | + +### Context propagation (`ctx/`) + +`x/ctx/value.go` provides typed context key/value helpers used to pass per-request data through the handler chain: + +- `ContextWithSid` / `SidFromContext` — Session ID +- `ContextWithSrcAddr` / `SrcAddrFromContext` — Source address +- `ContextWithDstAddr` / `DstAddrFromContext` — Destination address +- `ContextWithClientID` / `ClientIDFromContext` — Client ID (for hash-based load balancing) +- `ContextWithHash` / `HashFromContext` — Hash source for selector + +### Hot-reloadable components + +Auth, bypass, admission, hosts, ingress, limiter, recorder, router, and SD components all support periodic reload from file/redis/http sources. The pattern: + +1. Options include `fileLoader`, `redisLoader`, `httpLoader` plus a `period` duration +2. A background goroutine runs `periodReload(ctx)` calling `loader.Load()` on each tick +3. The component swaps its internal data structure (e.g., `ipMatcher`, `hostMatcher`) atomically + +### Listener wrapper ordering + +Listeners wrap `net.Listener` in a fixed order. Getting this wrong breaks metrics/admission/limiting: +``` +proxyproto → metrics → stats → admission → traffic_limiter → conn_limiter → (protocol-specific) +``` + +### Registry + +All registries are in `x/registry/registry.go` — a single file with ~25 typed global singletons backed by `sync.Map`. Each is accessed via an exported function (e.g., `registry.HandlerRegistry()`, `registry.ListenerRegistry()`). Registration returns `ErrDup` on name collision; empty names are silently ignored. + +### Selectors (`selector/`) + +Load-balancing strategies for hops and chain groups: + +| Strategy | Behavior | +|----------|----------| +| `RoundRobinStrategy` | Atomic counter, modulo selection | +| `RandomStrategy` | Weighted random using `RandomWeighted` | +| `FIFOStrategy` | Always first element (stickiness) | +| `HashStrategy` | CRC32 of client ID or hash context value | + +Filters (`FailFilter`, `BackupFilter`) are applied before the strategy, reducing the candidate pool. + +### API (`api/`) + +Uses Gin framework. Registers CRUD endpoints under `/config` for all component types. Supports basic auth when an auther is configured. Swagger docs at `/docs`. + +### Metrics (`metrics/`) + +Prometheus-style metrics following `_total`, `_seconds` naming conventions. Constants for all metric names (e.g., `MetricServiceRequestsCounter`, `MetricNodeConnectDurationObserver`). `Enable(bool)` toggles globally. When disabled, `GetCounter/Gauge/Observer` return noop implementations. diff --git a/README.md b/README.md index 6da81857..2c4b6b21 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,33 @@ -# The extended (may be experimental) features outside the main gost tree. +# x +The implementation layer for the GOST proxy framework. Every interface defined in [core/](../core/) is implemented here — handlers, listeners, dialers, connectors, plus config parsing, registries, and all cross-cutting concerns. +## Package layout + +| Directory | Purpose | +|-----------|---------| +| `handler/` | Protocol handlers (http, socks4/5, ss, ssh, tunnel, tun, dns, redirect, etc.) | +| `listener/` | Protocol listeners (tcp, tls, ws, http2/3, quic, kcp, icmp, tun, udp, etc.) | +| `dialer/` | Outbound dialers (tcp, tls, ws, http2/3, quic, grpc, ssh, wg, kcp, icmp, etc.) | +| `connector/` | Destination connectors (http, socks4/5, ss, ssh, relay, tunnel, direct, etc.) | +| `config/` | Config struct, YAML/JSON parsing, and service construction | +| `registry/` | Typed global registries for all component types | +| `service/` | Accept loop wiring listener + handler | +| `chain/` | Multi-hop forwarding chain and route implementation | +| `router/` | Route table router (destination-based routing) | +| `hop/` | Node group with load-balanced selection | +| `selector/` | Load-balancing strategies (round-robin, random, weighted, hash) | +| `auth/`, `bypass/`, `admission/` | Authentication, bypass rules, admission control | +| `resolver/`, `hosts/` | DNS resolution and host mapping | +| `limiter/` | Traffic, connection, and rate limiters | +| `recorder/`, `observer/` | Traffic recording and observability | +| `ingress/`, `sd/`, `routing/` | Ingress control, service discovery, routing rules | +| `logger/`, `metrics/`, `api/` | Logging, Prometheus metrics, Web API | +| `metadata/`, `ctx/` | Metadata key-value system and context propagation | +| `internal/` | Shared internals — not importable externally | + +## Component pattern + +Every handler, listener, dialer, and connector follows the same pattern: `init()` registers a constructor into a global registry, the constructor takes functional options, and `Init(metadata.Metadata)` extracts typed configuration from the metadata key-value map. + +See [CLAUDE.md](CLAUDE.md) for detailed architecture and conventions. diff --git a/internal/net/pipe.go b/internal/net/pipe.go index 94270648..205bb401 100644 --- a/internal/net/pipe.go +++ b/internal/net/pipe.go @@ -15,34 +15,32 @@ const ( readTimeout = 30 * time.Second ) -// Pipe 在两个连接之间建立双向数据通道 func Pipe(ctx context.Context, rw1, rw2 io.ReadWriteCloser) error { - ctx, cancel := context.WithCancel(ctx) - defer cancel() - errCh := make(chan error, 2) - // 启动两个方向的传输 - go func() { - errCh <- pipeHalf(ctx, rw1, rw2) - }() + go func() { errCh <- pipeHalf(ctx, rw1, rw2) }() + go func() { errCh <- pipeHalf(ctx, rw2, rw1) }() - go func() { - errCh <- pipeHalf(ctx, rw2, rw1) - }() - - // 等待第一个错误或完成 var firstErr error - for i := 0; i < 2; i++ { + completed := 0 + for completed < 2 { select { case err := <-errCh: + completed++ if firstErr == nil && err != nil { firstErr = err - cancel() // 一个方向出错，取消另一个 } case <-ctx.Done(): - // 超时或主动取消 forceClose(rw1, rw2) + for completed < 2 { + if err := <-errCh; firstErr == nil && err != nil { + firstErr = err + } + completed++ + } + if firstErr != nil { + return firstErr + } return ctx.Err() } } @@ -60,7 +58,6 @@ func pipeHalf(ctx context.Context, src, dst io.ReadWriteCloser) error { buf := bufpool.Get(bufferSize / 2) defer bufpool.Put(buf) - // 创建带超时的读取器 reader := &readDeadliner{ Reader: src, diff --git a/internal/net/transport.go b/internal/net/transport.go index 8e8806d4..e69d7cf9 100644 --- a/internal/net/transport.go +++ b/internal/net/transport.go @@ -11,7 +11,7 @@ const ( ) func Transport(rw1, rw2 io.ReadWriter) error { - errc := make(chan error, 1) + errc := make(chan error, 2) go func() { errc <- CopyBuffer(rw1, rw2, bufferSize) }()