fix: improve TURN relay failover

core/allocation_gate.go

@@ -0,0 +1,44 @@
+package core
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+const allocateGateInterval = 100 * time.Millisecond
+
+// allocationGate spaces TURN Allocate attempts. A Core owns one gate, so its
+// workers share a quota without unrelated Core instances blocking each other.
+type allocationGate struct {
+	mu       sync.Mutex
+	next     time.Time
+	interval time.Duration
+}
+
+func newAllocationGate(interval time.Duration) *allocationGate {
+	return &allocationGate{interval: interval}
+}
+
+func (g *allocationGate) Wait(ctx context.Context) error {
+	g.mu.Lock()
+	due := time.Now()
+	if g.next.After(due) {
+		due = g.next
+	}
+	g.next = due.Add(g.interval)
+	g.mu.Unlock()
+
+	wait := time.Until(due)
+	if wait <= 0 {
+		return nil
+	}
+	t := time.NewTimer(wait)
+	defer t.Stop()
+	select {
+	case <-t.C:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}

core/allocation_gate_test.go

@@ -0,0 +1,59 @@
+package core
+
+import (
+	"context"
+	"errors"
+	"sort"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestAllocationGatePacesConcurrentAttempts(t *testing.T) {
+	interval := 25 * time.Millisecond
+	g := newAllocationGate(interval)
+	start := make(chan struct{})
+	attempts := make(chan time.Time, 3)
+	var wg sync.WaitGroup
+	for range 3 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			if err := g.Wait(context.Background()); err != nil {
+				t.Error(err)
+				return
+			}
+			attempts <- time.Now()
+		}()
+	}
+	close(start)
+	wg.Wait()
+	close(attempts)
+
+	got := make([]time.Time, 0, 3)
+	for at := range attempts {
+		got = append(got, at)
+	}
+	if len(got) != 3 {
+		t.Fatalf("completed attempts = %d, want 3", len(got))
+	}
+	sort.Slice(got, func(i, j int) bool { return got[i].Before(got[j]) })
+	for i := 1; i < len(got); i++ {
+		if elapsed := got[i].Sub(got[i-1]); elapsed < interval/2 {
+			t.Fatalf("attempts %d and %d were not paced: %v", i-1, i, elapsed)
+		}
+	}
+}
+
+func TestAllocationGateWaitCancels(t *testing.T) {
+	g := newAllocationGate(time.Second)
+	if err := g.Wait(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if err := g.Wait(ctx); !errors.Is(err, context.Canceled) {
+		t.Fatalf("Wait() error = %v, want context.Canceled", err)
+	}
+}

core/main.go

@@ -80,9 +80,10 @@ type Core struct {
 	captchaMode       atomic.Value
 	vkAuthMode        atomic.Value
 	events            chan Event
-	startOnce         sync.Once // guard от двойного Start()
-	startErr          atomic.Value // *startFailure — ошибка первого Start(), если он провалился
-	once              sync.Once // guard от двойного Stop()
+	allocateGate *allocationGate
+	startOnce    sync.Once // guard от двойного Start()
+	startErr     atomic.Value // *startFailure — ошибка первого Start(), если он провалился
+	once         sync.Once // guard от двойного Stop()
 }
 
 // startFailure — обёртка для atomic.Value (чтобы разные error-типы не паниковали).
@@ -129,6 +130,7 @@ func New(cfg Config) *Core {
 		cancel:            cancel,
 		CaptchaResultChan: make(chan string, 1),
 		events:            make(chan Event, 1024), // увеличен буфер для логов
+		allocateGate:      newAllocationGate(allocateGateInterval),
 	}
 	c.captchaMode.Store(normalizeCaptchaMode(cfg.CaptchaMode))
 	c.vkAuthMode.Store(normalizeVKAuthMode("vkcalls"))
@@ -228,14 +230,7 @@ func (c *Core) start() error {
 	}
 
 	// Нормализация воркеров
-	n := c.cfg.Workers
-	if n > 108 {
-		n = 108
-	}
-	if n < workersPerGroup {
-		n = workersPerGroup
-	}
-	n = (n / workersPerGroup) * workersPerGroup
+	n := normalizeWorkers(c.cfg.Workers)
 
 	tp := &TurnParams{
 		Host:         c.cfg.TurnHost,
@@ -244,6 +239,7 @@ func (c *Core) start() error {
 		WrapKey:      wrapKey,
 		ObfsMode:     c.cfg.ObfsMode,
 		TCPTransport: c.cfg.TurnTCP,
+		allocateGate: c.allocateGate,
 	}
 
 	// Локальный UDP сокет
@@ -573,7 +569,8 @@ func classifyLevel(msg string) string {
 	switch {
 	case strings.Contains(low, "fatal") || strings.Contains(low, "ошибка") || strings.Contains(low, "error"):
 		return "ERROR"
-	case strings.Contains(low, "warn") || strings.Contains(low, "не удалось") || strings.Contains(low, "retry"):
+	case strings.Contains(low, "warn") || strings.Contains(low, "не удалось") || strings.Contains(low, "retry") ||
+		strings.Contains(low, "[failover]") || strings.Contains(low, "[reconnect]"):
 		return "WARN"
 	default:
 		return "INFO"

core/session.go

@@ -8,6 +8,7 @@ import (
 	"net"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/cbeuw/connutil"
@@ -104,6 +105,7 @@ func RunSession(
 	stats *Stats,
 ) (bool, *SessionError) {
 	configDelivered := false
+	var firstWireRead atomic.Bool
 
 	if turnAddr == "" {
 		return false, &SessionError{
@@ -177,6 +179,16 @@ func RunSession(
 		}
 	}
 
+	if tp.allocateGate != nil {
+		if err = tp.allocateGate.Wait(ctx); err != nil {
+			return false, &SessionError{
+				Type:    SessionErrorFatal,
+				Address: turnAddr,
+				Err:     fmt.Errorf("ожидание TURN Allocate: %w", err),
+			}
+		}
+	}
+
 	relay, err := tc.Allocate()
 	if err != nil {
 		errStr := err.Error()
@@ -219,7 +231,19 @@ func RunSession(
 
 	getStreamCache(cacheStreamID).errorCount.Store(0)
 
-	log.Printf("[СЕССИЯ #%d] Relay: %s", sessionID, relay.LocalAddr())
+	log.Printf("[СЕССИЯ #%d] Relay: %s; peer: %s", sessionID, relay.LocalAddr(), peer)
+
+	// Establish and verify the TURN permission before DTLS starts. Otherwise a
+	// failure is hidden inside the relay-writing goroutine and looks like a DTLS
+	// or password timeout.
+	if err = tc.CreatePermission(peer); err != nil {
+		return false, &SessionError{
+			Type:    SessionErrorAddressDead,
+			Address: turnAddr,
+			Err:     fmt.Errorf("TURN CreatePermission для %s: %w", peer, err),
+		}
+	}
+	log.Printf("[СЕССИЯ #%d] TURN permission OK для %s", sessionID, peer)
 
 	pipeA, pipeB := connutil.AsyncPacketPipe()
 
@@ -277,6 +301,9 @@ func RunSession(
 			if readErr != nil {
 				return
 			}
+			if firstWireRead.CompareAndSwap(false, true) {
+				log.Printf("[СЕССИЯ #%d] Первый пакет получен через TURN от %s (%d байт)", sessionID, peer, n)
+			}
 			payload := buf[:n]
 			if useWrap {
 				if !obfsIsRTPPacket(payload) {
@@ -300,6 +327,7 @@ func RunSession(
 		defer relayWg.Done()
 		defer sessCancel()
 		b := make([]byte, readBufSize)
+		firstWrite := true
 		for {
 			n, _, readErr := pipeA.ReadFrom(b)
 			if readErr != nil {
@@ -316,9 +344,15 @@ func RunSession(
 					out = wrapped
 				}
 			}
-			if _, writeErr := relay.WriteTo(out, peer); writeErr != nil {
+			written, writeErr := relay.WriteTo(out, peer)
+			if writeErr != nil {
+				log.Printf("[СЕССИЯ #%d] TURN relay WriteTo %s: %v", sessionID, peer, writeErr)
 				return
 			}
+			if firstWrite {
+				log.Printf("[СЕССИЯ #%d] Первый пакет отправлен через TURN к %s (%d/%d байт)", sessionID, peer, written, len(out))
+				firstWrite = false
+			}
 		}
 	}()
 
@@ -368,11 +402,20 @@ func RunSession(
 
 	if err != nil {
 		errStr := strings.ToLower(err.Error())
-		if useWrap && (strings.Contains(errStr, "deadline") || strings.Contains(errStr, "timeout")) {
-			return false, &SessionError{
-				Type:    SessionErrorWrapTimeout,
-				Address: turnAddr,
-				Err:     fmt.Errorf("DTLS timeout, пароль/WRAP не подтверждён"),
+		if strings.Contains(errStr, "deadline") || strings.Contains(errStr, "timeout") {
+			if !firstWireRead.Load() {
+				return false, &SessionError{
+					Type:    SessionErrorAddressDead,
+					Address: turnAddr,
+					Err:     fmt.Errorf("TURN relay молчит: ни один пакет от peer не вернулся"),
+				}
+			}
+			if useWrap {
+				return false, &SessionError{
+					Type:    SessionErrorWrapTimeout,
+					Address: turnAddr,
+					Err:     fmt.Errorf("peer ответил, но WRAP/DTLS не подтверждён"),
+				}
 			}
 		}
 		return false, &SessionError{
@@ -488,7 +531,7 @@ func RunSession(
 				if ne, ok := readErr.(net.Error); ok && ne.Timeout() {
 					continue
 				}
-				log.Printf("[ВОРКЕР #%d] Ошибка Reader: %v", sessionID, readErr)
+				log.Printf("[ВОРКЕР #%d] [RECONNECT] Reader завершён: %v", sessionID, readErr)
 				return
 			}