fix(udprelay): детект блэкхола permission по логу pion вместо трафик-эвристики

Трафик-watchdog (tx растёт, rx заморожен) ложно рециклил здоровые
sender-only стримы: обратный путь WireGuard привязан к endpoint, а не к
стриму, поэтому rx==0 на отправляющем стриме — норма, не блэкхол.

permWatchFactory считает проваленные циклы refresh permission в логе pion
(scope turnc) и закрывает Stream.PermDead после порога. RefreshInterval
урезан 120s→30s. oneTURN рециклит allocation по PermDead.

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
 				}

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...)
+}

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())
+	}
+}

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 {