refactor(logx): заменить Deps{Debug,Debugf} на интерфейс logx.Logger

Добавляет internal/logx с Debugf/Infof/Warnf/Errorf + DebugEnabled().
Протягивает logx.Logger через bondclient, bondserver, proxy/udp, proxy/vless,
vkauth. Убирает глобалы isDebug/debugf из обоих бинарей. Переносит
globalBondRegistry из package init в main (инициализация после
config.ParseServer), закрывая давний footgun "Debug=false на init".
This commit is contained in:
samosvalishe
2026-05-19 17:54:19 +03:00
parent acbd90f52e
commit 7fd2d9594d
10 changed files with 157 additions and 111 deletions
+6 -14
View File
@@ -22,6 +22,7 @@ import (
"github.com/cacggghp/vk-turn-proxy/internal/client/vkauth"
"github.com/cacggghp/vk-turn-proxy/internal/config"
"github.com/cacggghp/vk-turn-proxy/internal/dtlsdial"
"github.com/cacggghp/vk-turn-proxy/internal/logx"
udpproxy "github.com/cacggghp/vk-turn-proxy/internal/proxy/udp"
"github.com/cacggghp/vk-turn-proxy/internal/proxy/vless"
"github.com/cacggghp/vk-turn-proxy/internal/wrap"
@@ -38,7 +39,6 @@ var (
HandshakeSem: make(chan struct{}, 3),
}
vlessDtlsDialer = &dtlsdial.Dialer{HandshakeTimeout: 30 * time.Second}
isDebug bool
)
var appDialer net.Dialer
@@ -46,12 +46,6 @@ var appDialer net.Dialer
// vkAuth is the lazily-initialized VK auth client. Set once in main().
var vkAuth *vkauth.Client
func debugf(format string, v ...any) {
if isDebug {
log.Printf(format, v...)
}
}
// manualCaptchaSolver bridges the vkauth.ManualSolveFunc contract to the
// local captcha bouncer (internal/client/captcha/manual).
func manualCaptchaSolver(_ context.Context, e *captcha.Error, d net.Dialer) (string, string, error) {
@@ -118,7 +112,7 @@ func main() {
log.Printf("WRAP mode enabled: peer server must use matching -wrap-key")
}
isDebug = cfg.Debug
logger := logx.New(cfg.Debug)
manualcaptcha.Debug = cfg.Debug
vkAuth = vkauth.New(vkauth.Config{
@@ -127,7 +121,7 @@ func main() {
StreamsPerCache: cfg.StreamsPerCred,
StreamsAlive: func() int32 { return connectedStreams.Load() },
ManualSolver: manualCaptchaSolver,
Debugf: debugf,
Log: logger,
})
getCreds := getCredsFunc(vkAuth.GetCredentials)
@@ -142,11 +136,10 @@ func main() {
}
if cfg.VLESSMode {
bondH := &bondclient.Handler{Deps: bondclient.Deps{Debug: isDebug, Debugf: debugf}}
bondH := &bondclient.Handler{Deps: bondclient.Deps{Log: logger}}
vlessDeps := &vless.Deps{
DTLSDialer: vlessDtlsDialer,
Debug: isDebug,
Debugf: debugf,
Log: logger,
BondHandler: bondH.Handle,
}
vlessParams := &vless.Params{
@@ -226,8 +219,7 @@ func main() {
udpDeps := &udpproxy.Deps{
DTLSDialer: udpDtlsDialer,
Auth: vkAuth,
Debug: isDebug,
Debugf: debugf,
Log: logger,
ActiveLocalPeer: &activeLocalPeer,
ConnectedStreams: &connectedStreams,
AppCancel: cancel,
+24 -27
View File
@@ -15,6 +15,7 @@ import (
"github.com/cacggghp/vk-turn-proxy/internal/bond"
bondserver "github.com/cacggghp/vk-turn-proxy/internal/bond/server"
"github.com/cacggghp/vk-turn-proxy/internal/config"
"github.com/cacggghp/vk-turn-proxy/internal/logx"
"github.com/cacggghp/vk-turn-proxy/internal/stats"
"github.com/cacggghp/vk-turn-proxy/internal/wrap"
"github.com/cacggghp/vk-turn-proxy/tcputil"
@@ -23,20 +24,18 @@ import (
"github.com/xtaci/smux"
)
var isDebug bool
func debugf(format string, v ...any) {
if isDebug {
log.Printf(format, v...)
}
}
var (
logger logx.Logger = logx.Nop()
globalBondRegistry *bondserver.Registry
)
func main() {
cfg, err := config.ParseServer(os.Args[1:], os.Stderr)
if err != nil {
log.Panicf("%v", err)
}
isDebug = cfg.Debug
logger = logx.New(cfg.Debug)
globalBondRegistry = bondserver.NewRegistry(bondserver.Deps{Log: logger})
if cfg.GenWrapKey {
key, gerr := wrap.GenKeyHex()
@@ -123,7 +122,7 @@ func main() {
log.Printf("failed to close incoming connection: %s", closeErr)
}
}()
debugf("Connection from %s\n", conn.RemoteAddr())
logger.Debugf("Connection from %s\n", conn.RemoteAddr())
// Perform the handshake with a 30-second timeout
ctx1, cancel1 := context.WithTimeout(ctx, 30*time.Second)
@@ -134,12 +133,12 @@ func main() {
log.Println("Type error: expected *dtls.Conn")
return
}
debugf("Start handshake")
logger.Debugf("Start handshake")
if err := dtlsConn.HandshakeContext(ctx1); err != nil {
log.Printf("Handshake failed: %v", err)
return
}
debugf("Handshake done")
logger.Debugf("Handshake done")
if cfg.VLESSMode {
handleVLESSConnection(ctx, dtlsConn, cfg.Connect)
@@ -147,13 +146,11 @@ func main() {
handleUDPConnection(ctx, conn, cfg.Connect)
}
debugf("Connection closed: %s\n", conn.RemoteAddr())
logger.Debugf("Connection closed: %s\n", conn.RemoteAddr())
})
}
}
var globalBondRegistry = bondserver.NewRegistry(bondserver.Deps{Debug: isDebug, Debugf: debugf})
type prefixedConn struct {
net.Conn
prefix []byte
@@ -183,10 +180,10 @@ func handleUDPConnection(ctx context.Context, conn net.Conn, connectAddr string)
var wg sync.WaitGroup
ctx2, cancel2 := context.WithCancel(ctx)
st := stats.New(isDebug)
st := stats.New(logger.DebugEnabled())
go st.LogEvery(
ctx2,
debugf,
logger.Debugf,
fmt.Sprintf("[DTLS %s]", conn.RemoteAddr()),
"dtls-to-backend",
"backend-to-dtls",
@@ -271,10 +268,10 @@ func handleVLESSConnection(ctx context.Context, dtlsConn net.Conn, connectAddr s
// 1. Create KCP session over DTLS
statsCtx, statsCancel := context.WithCancel(ctx)
defer statsCancel()
st := stats.New(isDebug)
st := stats.New(logger.DebugEnabled())
go st.LogEvery(
statsCtx,
debugf,
logger.Debugf,
fmt.Sprintf("[VLESS %s]", dtlsConn.RemoteAddr()),
"to-client",
"from-client",
@@ -290,7 +287,7 @@ func handleVLESSConnection(ctx context.Context, dtlsConn net.Conn, connectAddr s
log.Printf("failed to close KCP session: %v", closeErr)
}
}()
debugf("KCP session established (server)")
logger.Debugf("KCP session established (server)")
// 2. Create smux server session over KCP
smuxSess, err := smux.Server(kcpSess, tcputil.DefaultSmuxConfig())
@@ -303,7 +300,7 @@ func handleVLESSConnection(ctx context.Context, dtlsConn net.Conn, connectAddr s
log.Printf("failed to close smux session: %v", err)
}
}()
debugf("smux session established (server)")
logger.Debugf("smux session established (server)")
// 3. Accept smux streams and forward to backend via TCP
var wg sync.WaitGroup
@@ -329,7 +326,7 @@ func handleVLESSConnection(ctx context.Context, dtlsConn net.Conn, connectAddr s
return
}
if string(prefix[:]) == bond.Magic {
debugf("auto-detected bond smux stream")
logger.Debugf("auto-detected bond smux stream")
globalBondRegistry.HandleStreamAfterMagic(ctx, s, connectAddr, prefix)
return
}
@@ -366,10 +363,10 @@ func pipeConn(ctx context.Context, c1, c2 net.Conn) {
context.AfterFunc(ctx2, func() {
if err := c1.SetDeadline(time.Now()); err != nil {
debugf("pipeConn: failed to set deadline c1: %v", err)
logger.Debugf("pipeConn: failed to set deadline c1: %v", err)
}
if err := c2.SetDeadline(time.Now()); err != nil {
debugf("pipeConn: failed to set deadline c2: %v", err)
logger.Debugf("pipeConn: failed to set deadline c2: %v", err)
}
})
@@ -377,13 +374,13 @@ func pipeConn(ctx context.Context, c1, c2 net.Conn) {
wg.Go(func() {
if _, err := io.Copy(c1, c2); err != nil {
debugf("pipeConn: c1<-c2 copy error: %v", err)
logger.Debugf("pipeConn: c1<-c2 copy error: %v", err)
}
})
wg.Go(func() {
if _, err := io.Copy(c2, c1); err != nil {
debugf("pipeConn: c2<-c1 copy error: %v", err)
logger.Debugf("pipeConn: c2<-c1 copy error: %v", err)
}
})
@@ -391,9 +388,9 @@ func pipeConn(ctx context.Context, c1, c2 net.Conn) {
// Reset deadlines (best-effort; connection may already be closed)
if err := c1.SetDeadline(time.Time{}); err != nil {
debugf("pipeConn: failed to reset deadline c1: %v", err)
logger.Debugf("pipeConn: failed to reset deadline c1: %v", err)
}
if err := c2.SetDeadline(time.Time{}); err != nil {
debugf("pipeConn: failed to reset deadline c2: %v", err)
logger.Debugf("pipeConn: failed to reset deadline c2: %v", err)
}
}
+18 -17
View File
@@ -17,6 +17,7 @@ import (
"time"
"github.com/cacggghp/vk-turn-proxy/internal/bond"
"github.com/cacggghp/vk-turn-proxy/internal/logx"
"github.com/cacggghp/vk-turn-proxy/internal/proxy/vless"
"github.com/cacggghp/vk-turn-proxy/internal/stats"
"github.com/xtaci/smux"
@@ -24,14 +25,14 @@ import (
// Deps groups host-process dependencies needed by the bond client.
type Deps struct {
Debug bool
Debugf func(format string, v ...any)
Log logx.Logger
}
func (d *Deps) debugf(format string, v ...any) {
if d.Debugf != nil {
d.Debugf(format, v...)
func (d *Deps) log() logx.Logger {
if d.Log == nil {
return logx.Nop()
}
return d.Log
}
// Handler binds Deps and exposes Handle, matching the vless.BondHandler signature.
@@ -83,23 +84,23 @@ func (h *Handler) Handle(ctx context.Context, tcpConn net.Conn, connID uint64, c
}
context.AfterFunc(ctx, func() {
now := time.Now()
if err := tcpConn.SetDeadline(now); err != nil && h.Deps.Debug {
log.Printf("[bond %d] local TCP deadline error: %v", connID, err)
if err := tcpConn.SetDeadline(now); err != nil {
h.Deps.log().Debugf("[bond %d] local TCP deadline error: %v", connID, err)
}
for _, l := range lanes {
if err := l.stream.SetDeadline(now); err != nil && h.Deps.Debug {
log.Printf("[bond %d] session %d stream deadline error: %v", connID, l.ps.ID, err)
if err := l.stream.SetDeadline(now); err != nil {
h.Deps.log().Debugf("[bond %d] session %d stream deadline error: %v", connID, l.ps.ID, err)
}
}
})
h.Deps.debugf("[bond %d] TCP accept from=%s lanes=%d [%s]", connID, tcpConn.RemoteAddr(), len(lanes), strings.Join(laneIDs, ","))
h.Deps.log().Debugf("[bond %d] TCP accept from=%s lanes=%d [%s]", connID, tcpConn.RemoteAddr(), len(lanes), strings.Join(laneIDs, ","))
defer func() {
for _, l := range lanes {
_ = l.stream.Close()
active := l.ps.Active.Add(-1)
closed := l.ps.Closed.Add(1)
h.Deps.debugf("[bond %d] lane session %d close active=%d closed=%d totals: to-session=%s from-session=%s",
h.Deps.log().Debugf("[bond %d] lane session %d close active=%d closed=%d totals: to-session=%s from-session=%s",
connID, l.ps.ID, active, closed,
stats.FormatByteCount(l.ps.ToSession.Load()), stats.FormatByteCount(l.ps.FromSession.Load()))
}
@@ -117,7 +118,7 @@ func (h *Handler) Handle(ctx context.Context, tcpConn net.Conn, connID uint64, c
case <-ctx.Done():
default:
if err != io.EOF {
h.Deps.debugf("[bond %d] session %d read frame error: %v", connID, l.ps.ID, err)
h.Deps.log().Debugf("[bond %d] session %d read frame error: %v", connID, l.ps.ID, err)
}
}
return
@@ -165,8 +166,8 @@ func (h *Handler) copyTCPToBond(ctx context.Context, connID uint64, tcpConn net.
seq++
}
if err != nil {
if h.Deps.Debug && err != io.EOF {
log.Printf("[bond %d] local TCP read finished with error: %v", connID, err)
if err != io.EOF {
h.Deps.log().Debugf("[bond %d] local TCP read finished with error: %v", connID, err)
}
for _, l := range lanes {
if l.dead.Load() {
@@ -179,7 +180,7 @@ func (h *Handler) copyTCPToBond(ctx context.Context, connID uint64, tcpConn net.
log.Printf("[bond %d] session %d write FIN error: %v", connID, l.ps.ID, writeErr)
}
}
h.Deps.debugf("[bond %d] upload finished chunks=%d", connID, seq)
h.Deps.log().Debugf("[bond %d] upload finished chunks=%d", connID, seq)
return
}
select {
@@ -222,8 +223,8 @@ func (h *Handler) copyBondToTCP(ctx context.Context, connID uint64, tcpConn net.
for {
if finSeq != nil && expect == *finSeq {
bond.CloseWrite(tcpConn, h.Deps.debugf)
h.Deps.debugf("[bond %d] download finished chunks=%d", connID, expect)
bond.CloseWrite(tcpConn, h.Deps.log().Debugf)
h.Deps.log().Debugf("[bond %d] download finished chunks=%d", connID, expect)
return
}
+13 -12
View File
@@ -13,6 +13,7 @@ import (
"time"
"github.com/cacggghp/vk-turn-proxy/internal/bond"
"github.com/cacggghp/vk-turn-proxy/internal/logx"
"github.com/xtaci/smux"
)
@@ -27,14 +28,14 @@ type laneStream interface {
// Deps groups host-process dependencies needed by the bond server.
type Deps struct {
Debug bool
Debugf func(format string, v ...any)
Log logx.Logger
}
func (d *Deps) debugf(format string, v ...any) {
if d.Debugf != nil {
d.Debugf(format, v...)
func (d *Deps) log() logx.Logger {
if d.Log == nil {
return logx.Nop()
}
return d.Log
}
// Registry deduplicates concurrent lanes that share a ConnID into a single
@@ -145,7 +146,7 @@ func (c *serverConn) addLane(l *serverLane, laneCount uint16) {
c.lanes = append(c.lanes, l)
count := len(c.lanes)
c.lanesMu.Unlock()
c.deps.debugf("[bond %d] lane %d attached (lanes=%d)", c.id, l.index, count)
c.deps.log().Debugf("[bond %d] lane %d attached (lanes=%d)", c.id, l.index, count)
select {
case c.ready <- struct{}{}:
default:
@@ -193,7 +194,7 @@ func (c *serverConn) waitForInitialLanes() {
return
case <-c.ready:
case <-timer.C:
c.deps.debugf("[bond %d] starting with %d/%d lanes after attach timeout", c.id, count, want)
c.deps.log().Debugf("[bond %d] starting with %d/%d lanes after attach timeout", c.id, count, want)
return
}
}
@@ -208,7 +209,7 @@ func (c *serverConn) readLane(l *serverLane) {
case <-c.ctx.Done():
default:
if err != io.EOF {
c.deps.debugf("[bond %d] lane %d read error: %v (lanes=%d)", c.id, l.index, err, left)
c.deps.log().Debugf("[bond %d] lane %d read error: %v (lanes=%d)", c.id, l.index, err, left)
}
if left == 0 {
c.cancel()
@@ -251,7 +252,7 @@ func (c *serverConn) run() {
}
}
})
c.deps.debugf("[bond %d] backend connected", c.id)
c.deps.log().Debugf("[bond %d] backend connected", c.id)
var wg sync.WaitGroup
wg.Go(func() {
@@ -271,8 +272,8 @@ func (c *serverConn) copyBondToBackend(backendConn net.Conn) {
for {
if finSeq != nil && expect == *finSeq {
bond.CloseWrite(backendConn, c.deps.debugf)
c.deps.debugf("[bond %d] upload to backend finished chunks=%d", c.id, expect)
bond.CloseWrite(backendConn, c.deps.log().Debugf)
c.deps.log().Debugf("[bond %d] upload to backend finished chunks=%d", c.id, expect)
return
}
@@ -336,7 +337,7 @@ func (c *serverConn) copyBackendToBond(backendConn net.Conn) {
log.Printf("[bond %d] lane %d write FIN error: %v", c.id, lane.index, writeErr)
}
}
c.deps.debugf("[bond %d] download from backend finished chunks=%d", c.id, seq)
c.deps.log().Debugf("[bond %d] download from backend finished chunks=%d", c.id, seq)
return
}
select {
+2 -2
View File
@@ -56,7 +56,7 @@ func (f *fakeStream) writtenFrames(t *testing.T) []bond.Frame {
func newTestConn(t *testing.T) *serverConn {
t.Helper()
deps := &Deps{Debug: false, Debugf: func(string, ...any) {}}
deps := &Deps{}
ctx, cancel := context.WithCancel(t.Context())
t.Cleanup(cancel)
return &serverConn{
@@ -248,7 +248,7 @@ func TestWriteToNextLaneAllFail(t *testing.T) {
// TestRegistryGetDedup ensures concurrent get() calls for the same ConnID
// share one serverConn, and that completion removes it from the registry.
func TestRegistryGetDedup(t *testing.T) {
r := NewRegistry(Deps{Debugf: func(string, ...any) {}})
r := NewRegistry(Deps{})
ctx := t.Context()
c1 := r.get(ctx, 7, "127.0.0.1:1")
+9 -7
View File
@@ -11,6 +11,8 @@ import (
"sync/atomic"
"time"
"github.com/cacggghp/vk-turn-proxy/internal/logx"
tlsclient "github.com/bogdanfinn/tls-client"
)
@@ -38,8 +40,8 @@ type Config struct {
AutoSolver AutoSolveFunc
ManualSolver ManualSolveFunc
// Debugf is the gated debug log sink. nil -> no-op.
Debugf func(format string, v ...any)
// Log is the leveled logger. nil -> no-op.
Log logx.Logger
}
// Client is the VK auth + creds-cache facade used by callers. It owns the
@@ -52,7 +54,7 @@ type Client struct {
streamsFn func() int32
autoSolver AutoSolveFunc
manualSolve ManualSolveFunc
debugf func(format string, v ...any)
log logx.Logger
store *Store
@@ -80,14 +82,14 @@ func New(cfg Config) *Client {
streamsFn: cfg.StreamsAlive,
autoSolver: cfg.AutoSolver,
manualSolve: cfg.ManualSolver,
debugf: cfg.Debugf,
log: cfg.Log,
store: NewStore(cfg.StreamsPerCache),
}
if len(c.credentials) == 0 {
c.credentials = DefaultCredentials
}
if c.debugf == nil {
c.debugf = func(string, ...any) {}
if c.log == nil {
c.log = logx.Nop()
}
if c.streamsFn == nil {
c.streamsFn = func() int32 { return 1 }
@@ -111,7 +113,7 @@ func (c *Client) GetCredentials(ctx context.Context, link string, streamID int)
u, p := cache.creds.Username, cache.creds.Password
addr := cache.creds.ServerAddrs[streamID%len(cache.creds.ServerAddrs)]
cache.mutex.RUnlock()
c.debugf("[STREAM %d] [VK Auth] Using cached credentials (cache=%d, expires in %v, server=%s)", streamID, cacheID, expires, addr)
c.log.Debugf("[STREAM %d] [VK Auth] Using cached credentials (cache=%d, expires in %v, server=%s)", streamID, cacheID, expires, addr)
return u, p, addr, nil
}
cache.mutex.RUnlock()
+51
View File
@@ -0,0 +1,51 @@
// Package logx is a minimal leveled logger over stdlib log. It replaces the
// per-package Deps{Debug bool, Debugf func} pair: callers receive a Logger and
// call Debugf/Infof/Warnf/Errorf directly. Output format is bit-compatible with
// the previous `log.Printf`-gated-by-isDebug pattern.
package logx
import "log"
// Logger is the leveled-log interface used across the proxy. Debugf is gated by
// the constructor's debug flag; the other levels always print via stdlib log.
type Logger interface {
Debugf(format string, v ...any)
Infof(format string, v ...any)
Warnf(format string, v ...any)
Errorf(format string, v ...any)
// DebugEnabled reports whether Debugf will produce output. Hot paths
// (stats counters, condition-gated branches) use this to skip work the
// logger would discard anyway.
DebugEnabled() bool
}
type stdLogger struct {
debug bool
}
// New returns a Logger that prints via stdlib log. If debug is false, Debugf is
// a no-op (matches the legacy `if isDebug { log.Printf(...) }` shape).
func New(debug bool) Logger {
return &stdLogger{debug: debug}
}
// Nop returns a Logger whose every method discards its input. Useful in tests.
func Nop() Logger { return nopLogger{} }
func (l *stdLogger) Debugf(format string, v ...any) {
if l.debug {
log.Printf(format, v...)
}
}
func (l *stdLogger) Infof(format string, v ...any) { log.Printf(format, v...) }
func (l *stdLogger) Warnf(format string, v ...any) { log.Printf(format, v...) }
func (l *stdLogger) Errorf(format string, v ...any) { log.Printf(format, v...) }
func (l *stdLogger) DebugEnabled() bool { return l.debug }
type nopLogger struct{}
func (nopLogger) Debugf(string, ...any) {}
func (nopLogger) Infof(string, ...any) {}
func (nopLogger) Warnf(string, ...any) {}
func (nopLogger) Errorf(string, ...any) {}
func (nopLogger) DebugEnabled() bool { return false }
+10 -9
View File
@@ -17,6 +17,7 @@ import (
"time"
"github.com/cacggghp/vk-turn-proxy/internal/dtlsdial"
"github.com/cacggghp/vk-turn-proxy/internal/logx"
"github.com/cacggghp/vk-turn-proxy/internal/stats"
"github.com/cacggghp/vk-turn-proxy/internal/turnpipe"
"github.com/cacggghp/vk-turn-proxy/internal/wrap"
@@ -66,17 +67,17 @@ type Params struct {
type Deps struct {
DTLSDialer *dtlsdial.Dialer
Auth AuthHandler
Debug bool
Debugf func(format string, v ...any)
Log logx.Logger
ActiveLocalPeer *atomic.Value
ConnectedStreams *atomic.Int32
AppCancel func()
}
func (d *Deps) debugf(format string, v ...any) {
if d.Debugf != nil {
d.Debugf(format, v...)
func (d *Deps) log() logx.Logger {
if d.Log == nil {
return logx.Nop()
}
return d.Log
}
// DTLSLoop keeps a single DTLS termination alive for streamID, restarting it
@@ -265,7 +266,7 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
return
}
relayConn := stream.Relay
deps.debugf("[STREAM %d] TURN server IP: %s", streamID, stream.ServerUDPAddr.IP)
deps.log().Debugf("[STREAM %d] TURN server IP: %s", streamID, stream.ServerUDPAddr.IP)
deps.Auth.ResetErrors(streamID)
@@ -277,14 +278,14 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
}
}()
if deps.Debug {
if deps.log().DebugEnabled() {
log.Printf("[STREAM %d] relayed-address=%s", streamID, relayConn.LocalAddr().String())
}
wg := sync.WaitGroup{}
turnctx, turncancel := context.WithCancel(ctx)
st := stats.New(deps.Debug)
go st.LogEvery(turnctx, deps.debugf, fmt.Sprintf("[STREAM %d] TURN", streamID), "to-turn", "from-turn")
st := stats.New(deps.log().DebugEnabled())
go st.LogEvery(turnctx, deps.log().Debugf, fmt.Sprintf("[STREAM %d] TURN", streamID), "to-turn", "from-turn")
context.AfterFunc(turnctx, func() {
if err := relayConn.SetDeadline(time.Now()); err != nil {
+20 -19
View File
@@ -11,6 +11,7 @@ import (
"github.com/cacggghp/vk-turn-proxy/internal/dtlsdial"
"github.com/cacggghp/vk-turn-proxy/internal/ish"
"github.com/cacggghp/vk-turn-proxy/internal/logx"
"github.com/cacggghp/vk-turn-proxy/internal/stats"
"github.com/cacggghp/vk-turn-proxy/internal/turnpipe"
"github.com/cacggghp/vk-turn-proxy/internal/wrap"
@@ -40,15 +41,15 @@ type BondHandler func(ctx context.Context, tcpConn net.Conn, connID uint64, lane
// Deps groups host-process dependencies needed by the VLESS loop.
type Deps struct {
DTLSDialer *dtlsdial.Dialer
Debug bool
Debugf func(format string, v ...any)
Log logx.Logger
BondHandler BondHandler
}
func (d *Deps) debugf(format string, v ...any) {
if d.Debugf != nil {
d.Debugf(format, v...)
func (d *Deps) log() logx.Logger {
if d.Log == nil {
return logx.Nop()
}
return d.Log
}
// Run is the VLESS-mode entrypoint. It spawns numSessions maintainers, waits
@@ -146,7 +147,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis
connID := pool.NextConnID()
opened := ps.Opened.Add(1)
active := ps.Active.Add(1)
deps.debugf("[session %d] TCP accept #%d from=%s active=%d opened=%d pool=%d",
deps.log().Debugf("[session %d] TCP accept #%d from=%s active=%d opened=%d pool=%d",
ps.ID, connID, tcpConn.RemoteAddr(), active, opened, pool.Count())
tc, sessRef, cid := tcpConn, ps, connID
@@ -155,7 +156,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis
defer func() {
active := sessRef.Active.Add(-1)
closed := sessRef.Closed.Add(1)
deps.debugf("[session %d] TCP close #%d active=%d closed=%d totals: to-session=%s from-session=%s",
deps.log().Debugf("[session %d] TCP close #%d active=%d closed=%d totals: to-session=%s from-session=%s",
sessRef.ID, cid, active, closed,
stats.FormatByteCount(sessRef.ToSession.Load()), stats.FormatByteCount(sessRef.FromSession.Load()))
}()
@@ -169,7 +170,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis
fromSession, toSession := pipe(deps, ctx, tc, stream)
sessRef.FromSession.Add(uint64(fromSession))
sessRef.ToSession.Add(uint64(toSession))
deps.debugf("[session %d] TCP done #%d local<-session=%s local->session=%s",
deps.log().Debugf("[session %d] TCP done #%d local<-session=%s local->session=%s",
sessRef.ID, cid, stats.FormatByteCount(uint64(fromSession)), stats.FormatByteCount(uint64(toSession)))
})
}
@@ -246,8 +247,8 @@ func createSmuxSession(ctx context.Context, deps *Deps, params *Params, peer *ne
}
cleanupFns = append(cleanupFns, func() { _ = stream.Close() })
relayConn := stream.Relay
deps.debugf("[session %d] TURN server IP: %s", id, stream.ServerUDPAddr.IP)
deps.debugf("relayed-address=%s", relayConn.LocalAddr().String())
deps.log().Debugf("[session %d] TURN server IP: %s", id, stream.ServerUDPAddr.IP)
deps.log().Debugf("relayed-address=%s", relayConn.LocalAddr().String())
var relayWC *wrap.Conn
if len(params.WrapKey) == wrap.KeyLen {
@@ -264,12 +265,12 @@ func createSmuxSession(ctx context.Context, deps *Deps, params *Params, peer *ne
return nil, nil, fmt.Errorf("DTLS handshake: %w", err)
}
cleanupFns = append(cleanupFns, func() { _ = dtlsConn.Close() })
deps.debugf("DTLS connection established")
deps.log().Debugf("DTLS connection established")
statsCtx, statsCancel := context.WithCancel(ctx)
cleanupFns = append(cleanupFns, statsCancel)
st := stats.New(deps.Debug)
go st.LogEvery(statsCtx, deps.debugf, fmt.Sprintf("[session %d] VLESS", id), "to-turn", "from-turn")
st := stats.New(deps.log().DebugEnabled())
go st.LogEvery(statsCtx, deps.log().Debugf, fmt.Sprintf("[session %d] VLESS", id), "to-turn", "from-turn")
kcpSess, err := tcputil.NewKCPOverDTLS(&stats.CountingConn{Conn: dtlsConn, Stats: st}, false)
if err != nil {
@@ -277,7 +278,7 @@ func createSmuxSession(ctx context.Context, deps *Deps, params *Params, peer *ne
return nil, nil, fmt.Errorf("KCP session: %w", err)
}
cleanupFns = append(cleanupFns, func() { _ = kcpSess.Close() })
deps.debugf("KCP session established")
deps.log().Debugf("KCP session established")
smuxSess, err := smux.Client(kcpSess, tcputil.DefaultSmuxConfig())
if err != nil {
@@ -285,7 +286,7 @@ func createSmuxSession(ctx context.Context, deps *Deps, params *Params, peer *ne
return nil, nil, fmt.Errorf("smux client: %w", err)
}
cleanupFns = append(cleanupFns, func() { _ = smuxSess.Close() })
deps.debugf("smux session established")
deps.log().Debugf("smux session established")
return smuxSess, cleanup, nil
}
@@ -310,7 +311,7 @@ func pipe(deps *Deps, ctx context.Context, c1, c2 net.Conn) (int64, int64) {
defer cancel()
n, err := io.Copy(c1, c2)
c1FromC2 = n
if err != nil && deps.Debug {
if err != nil && deps.log().DebugEnabled() {
log.Printf("pipe: c1<-c2 copy error: %v", err)
}
})
@@ -318,15 +319,15 @@ func pipe(deps *Deps, ctx context.Context, c1, c2 net.Conn) (int64, int64) {
defer cancel()
n, err := io.Copy(c2, c1)
c2FromC1 = n
if err != nil && deps.Debug {
if err != nil && deps.log().DebugEnabled() {
log.Printf("pipe: c2<-c1 copy error: %v", err)
}
})
wg.Wait()
if err := c1.SetDeadline(time.Time{}); err != nil && deps.Debug {
if err := c1.SetDeadline(time.Time{}); err != nil && deps.log().DebugEnabled() {
log.Printf("pipe: failed to reset deadline c1: %v", err)
}
if err := c2.SetDeadline(time.Time{}); err != nil && deps.Debug {
if err := c2.SetDeadline(time.Time{}); err != nil && deps.log().DebugEnabled() {
log.Printf("pipe: failed to reset deadline c2: %v", err)
}
return c1FromC2, c2FromC1
+4 -4
View File
@@ -263,10 +263,10 @@ internal/
### Этап V2-3 — `internal/logx` + удаление isDebug-глобалов
- [ ] **V2-3.1** Создать `internal/logx/logx.go`: интерфейс `Logger{Debug(format,...), Info, Warn, Error}` + конструктор из `slog.Level`.
- [ ] **V2-3.2** Заменить `Deps{Debug bool, Debugf func}` на `Deps{Log logx.Logger}` во всех пакетах: `bondclient`, `bondserver`, `proxy/udp`, `proxy/vless`, `stats`.
- [ ] **V2-3.3** Удалить `var isDebug bool` + `func debugf` из обоих main. В main: `log := logx.New(cfg.Debug)`.
- [ ] **V2-3.4** `globalBondRegistry` → локальная в `cmd/vk-turn-server/main.go`, инициализация **после** `cfg := config.ParseServer`.
- [x] **V2-3.1** Создан `internal/logx/logx.go`: `Logger{Debugf, Infof, Warnf, Errorf, DebugEnabled}`. stdLogger пишет через `log.Printf` (bit-compat с прежним поведением). Nop для тестов/zero-value.
- [x] **V2-3.2** `Deps{Debug, Debugf}` → `Deps{Log logx.Logger}` в `bondclient`, `bondserver`, `proxy/udp`, `proxy/vless`, `vkauth` (`Config.Debugf` → `Config.Log`). `stats.New(bool)` остался — теперь принимает `logger.DebugEnabled()`. `stats.LogEvery(ctx, logf, ...)` сигнатура не меняется — передаётся `logger.Debugf` как method-value.
- [x] **V2-3.3** `var isDebug` + `func debugf` удалены из обоих main. `logger := logx.New(cfg.Debug)` в `cmd/vk-turn-client/main.go`; package-var `logger` (init Nop, в main переустанавливается) в `cmd/vk-turn-server/main.go` (нужно потому что обработчики handleUDPConnection/handleVLESSConnection берут его без проброса).
- [x] **V2-3.4** `globalBondRegistry` из package init переехал в `main` — инициализируется **после** `config.ParseServer`. Bug-by-design (init с `isDebug=false`) закрыт.
### Этап V2-4 — переименования (механика, без логики)