diff --git a/internal/proxy/udprelay/loop.go b/internal/proxy/udprelay/loop.go index d5850bb..f5b1355 100644 --- a/internal/proxy/udprelay/loop.go +++ b/internal/proxy/udprelay/loop.go @@ -271,40 +271,14 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, const maxPayload = 1600 - // Watchdog-счётчики (always-on, в отличие от st, который считает только в - // debug). Детектят blackhole: pion при провале CreatePermission-refresh - // («Fail to refresh permissions») молча роняет permission на relay — пакеты - // к peer дропаются без ошибки, oneTURN висит на живом, но мёртвом relay. - var wTx, wRx atomic.Uint64 - + // PermDead закрывается при блэкхоле permission (см. turndial/permwatch.go) — + // отменяем turnctx, TURNLoop делает свежий allocate. wg.Go(func() { - // Стрим жив, пока tx растёт И rx идёт. Если tx растёт, а rx заморожен - // stallLimit подряд — relay блэкхолит: отменяем turnctx, oneTURN выходит, - // TURNLoop делает свежий DialTURN (новый allocate + permission). - const checkInterval = 5 * time.Second - const stallLimit = 20 * time.Second - ticker := time.NewTicker(checkInterval) - defer ticker.Stop() - var lastTx, lastRx uint64 - var stalled time.Duration - for { - select { - case <-turnctx.Done(): - return - case <-ticker.C: - tx, rx := wTx.Load(), wRx.Load() - if tx > lastTx && rx == lastRx { - stalled += checkInterval - if stalled >= stallLimit { - deps.log().Warnf("[STREAM %d] TURN relay stall: tx идёт, rx заморожен %v — рецикл allocation", streamID, stallLimit) - turncancel() - return - } - } else { - stalled = 0 - } - lastTx, lastRx = tx, rx - } + select { + case <-turnctx.Done(): + case <-stream.PermDead: + deps.log().Warnf("[STREAM %d] TURN permission refresh умер — рецикл allocation", streamID) + turncancel() } }) @@ -352,9 +326,6 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, written, err1 := relayConn.WriteTo(out, peer) st.AddTx(written) - if written > 0 { - wTx.Add(uint64(written)) - } if err1 != nil { return } @@ -389,9 +360,6 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, payload = p } st.AddRx(len(payload)) - if len(payload) > 0 { - wRx.Add(uint64(len(payload))) - } if _, err := conn2.WriteTo(payload, addr); err != nil { return } diff --git a/internal/transport/turndial/permwatch.go b/internal/transport/turndial/permwatch.go new file mode 100644 index 0000000..9497d97 --- /dev/null +++ b/internal/transport/turndial/permwatch.go @@ -0,0 +1,85 @@ +package turndial + +import ( + "strings" + "sync" + + "github.com/pion/logging" +) + +// VK relay при дропе permission молча роняет пакеты к peer без ошибки записи. +// pion на каждый проваленный цикл refresh пишет permFailMarker (Warnf, scope +// "turnc"). Считаем провалы подряд — точный per-stream детект блэкхола вместо +// трафик-эвристики (обратный путь WG привязан к endpoint, не к стриму). +const ( + permFailMarker = "Failed to refresh permissions" + permOKMarker = "Refresh permissions successful" + turncScope = "turnc" + // ~60s при PermissionRefreshInterval=30s, внутри 5-мин lifetime permission. + permFailThreshold = 2 +) + +// permWatchFactory: для scope "turnc" возвращает logger, который зовёт onDead +// после permFailThreshold проваленных циклов refresh подряд. +type permWatchFactory struct { + inner logging.LoggerFactory + onDead func() + threshold int +} + +func (f *permWatchFactory) NewLogger(scope string) logging.LeveledLogger { + inner := f.inner.NewLogger(scope) + if scope != turncScope { + return inner + } + return &permWatchLogger{LeveledLogger: inner, f: f} +} + +type permWatchLogger struct { + logging.LeveledLogger + f *permWatchFactory + mu sync.Mutex + fails int + fired bool +} + +// Маркеры — литералы в format pion, матчим format до подстановки args. +func (l *permWatchLogger) note(msg string) { + switch { + case strings.Contains(msg, permFailMarker): + l.mu.Lock() + l.fails++ + fire := !l.fired && l.fails >= l.f.threshold + if fire { + l.fired = true + } + l.mu.Unlock() + if fire && l.f.onDead != nil { + l.f.onDead() + } + case strings.Contains(msg, permOKMarker): + l.mu.Lock() + l.fails = 0 + l.mu.Unlock() + } +} + +func (l *permWatchLogger) Warn(msg string) { + l.note(msg) + l.LeveledLogger.Warn(msg) +} + +func (l *permWatchLogger) Warnf(format string, args ...any) { + l.note(format) + l.LeveledLogger.Warnf(format, args...) +} + +func (l *permWatchLogger) Debug(msg string) { + l.note(msg) + l.LeveledLogger.Debug(msg) +} + +func (l *permWatchLogger) Debugf(format string, args ...any) { + l.note(format) + l.LeveledLogger.Debugf(format, args...) +} diff --git a/internal/transport/turndial/permwatch_test.go b/internal/transport/turndial/permwatch_test.go new file mode 100644 index 0000000..811787a --- /dev/null +++ b/internal/transport/turndial/permwatch_test.go @@ -0,0 +1,72 @@ +package turndial + +import ( + "sync/atomic" + "testing" + + "github.com/pion/logging" +) + +func newTestWatch(threshold int) (*permWatchFactory, *atomic.Int32) { + var fired atomic.Int32 + f := &permWatchFactory{ + inner: logging.NewDefaultLoggerFactory(), + threshold: threshold, + onDead: func() { fired.Add(1) }, + } + return f, &fired +} + +func TestPermWatchFiresAfterThreshold(t *testing.T) { + f, fired := newTestWatch(2) + log := f.NewLogger(turncScope) + + log.Warnf(permFailMarker+": %s", "boom") + if fired.Load() != 0 { + t.Fatalf("fired too early after 1 fail: %d", fired.Load()) + } + log.Warnf(permFailMarker+": %s", "boom") + if fired.Load() != 1 { + t.Fatalf("expected 1 fire after threshold, got %d", fired.Load()) + } + // Дальнейшие провалы не должны фаерить повторно. + log.Warnf(permFailMarker+": %s", "boom") + if fired.Load() != 1 { + t.Fatalf("fired more than once: %d", fired.Load()) + } +} + +func TestPermWatchResetOnSuccess(t *testing.T) { + f, fired := newTestWatch(2) + log := f.NewLogger(turncScope) + + log.Warnf(permFailMarker + ": x") + log.Debug(permOKMarker) // сброс счётчика + log.Warnf(permFailMarker + ": x") + if fired.Load() != 0 { + t.Fatalf("reset failed: fired=%d (fail/ok/fail не должно фаерить)", fired.Load()) + } +} + +func TestPermWatchIgnoresOtherScopes(t *testing.T) { + f, fired := newTestWatch(1) + log := f.NewLogger("other") + if _, ok := log.(*permWatchLogger); ok { + t.Fatal("non-turnc scope must not be wrapped") + } + log.Warnf(permFailMarker) + if fired.Load() != 0 { + t.Fatalf("fired on non-turnc scope: %d", fired.Load()) + } +} + +func TestPermWatchIgnoresUnrelatedMessages(t *testing.T) { + f, fired := newTestWatch(1) + log := f.NewLogger(turncScope) + log.Debug("Started refresh permission timer") + log.Debug("No permission to refresh") + log.Warnf("Failed to refresh allocation: %s", "x") + if fired.Load() != 0 { + t.Fatalf("fired on unrelated message: %d", fired.Load()) + } +} diff --git a/internal/transport/turndial/turndial.go b/internal/transport/turndial/turndial.go index e9796c5..0a886ea 100644 --- a/internal/transport/turndial/turndial.go +++ b/internal/transport/turndial/turndial.go @@ -11,6 +11,7 @@ import ( "context" "fmt" "net" + "sync" "time" "github.com/pion/logging" @@ -36,7 +37,10 @@ type Stream struct { Relay net.PacketConn // ServerUDPAddr — резолвнутый UDP-адрес TURN-сервера (host:port). ServerUDPAddr *net.UDPAddr - close func() error + // PermDead закрывается при стойком провале refresh permission (relay + // блэкхолит) — вызывающий рециклит allocation. См. permwatch.go. + PermDead <-chan struct{} + close func() error } // Close освобождает аллокацию, TURN-клиент и транспорт. @@ -103,15 +107,25 @@ func Open(ctx context.Context, cfg Config, peer *net.UDPAddr, user, pass, rawAdd } else { addrFamily = turn.RequestedAddressFamilyIPv6 } + + // RefreshInterval 120s→30s: быстрее ловим блэкхол, внутри 5-мин lifetime. + permDead := make(chan struct{}) + var permOnce sync.Once + loggerFactory := &permWatchFactory{ + inner: logging.NewDefaultLoggerFactory(), + threshold: permFailThreshold, + onDead: func() { permOnce.Do(func() { close(permDead) }) }, + } client, err := turn.NewClient(&turn.ClientConfig{ - STUNServerAddr: turnServerAddr, - TURNServerAddr: turnServerAddr, - Conn: turnConn, - Net: netconn.New(), - Username: user, - Password: pass, - RequestedAddressFamily: addrFamily, - LoggerFactory: logging.NewDefaultLoggerFactory(), + STUNServerAddr: turnServerAddr, + TURNServerAddr: turnServerAddr, + Conn: turnConn, + Net: netconn.New(), + Username: user, + Password: pass, + RequestedAddressFamily: addrFamily, + PermissionRefreshInterval: 30 * time.Second, + LoggerFactory: loggerFactory, }) if err != nil { if cerr := closeConn(); cerr != nil { @@ -138,6 +152,7 @@ func Open(ctx context.Context, cfg Config, peer *net.UDPAddr, user, pass, rawAdd return &Stream{ Relay: relay, ServerUDPAddr: turnServerUDPAddr, + PermDead: permDead, close: func() error { var firstErr error if cerr := relay.Close(); cerr != nil {