fix(relay): восстановление потоков при отказах turn и занятой квоте

.gitignore

@@ -1,6 +1,7 @@
 .idea
 .claude
 CLAUDE.md
+AGENTS.md
 *.exe
 *.so
 *.tmp

internal/provider/vk/internal/vkauth/cache.go

@@ -80,10 +80,9 @@ func IsAuthError(err error) bool {
 	}
 	// Ответ TURN-сервера приходит типизированным - код берём из него, а не из текста.
 	if turnErr, ok := errors.AsType[*stun.TurnError](err); ok {
+		// 486 означает занятую квоту, а не невалидные реквизиты.
 		switch turnErr.ErrorCodeAttr.Code {
-		// 486 - квота аллокаций: креды живы, но новую сессию по ним не поднять.
-		case stun.CodeUnauthorized, stun.CodeWrongCredentials,
-			stun.CodeStaleNonce, stun.CodeAllocQuotaReached:
+		case stun.CodeUnauthorized, stun.CodeWrongCredentials, stun.CodeStaleNonce:
 			return true
 		default:
 			return false

internal/provider/vk/internal/vkauth/client_test.go

@@ -56,7 +56,8 @@ func TestIsAuthErrorTurnCode(t *testing.T) {
 	t.Parallel()
 
 	codes := map[stun.ErrorCode]bool{
-		stun.CodeAllocQuotaReached:    true,
+		// 486 - занята квота по валидным реквизитам, менять их незачем.
+		stun.CodeAllocQuotaReached:    false,
 		stun.CodeUnauthorized:         true,
 		stun.CodeStaleNonce:           true,
 		stun.CodeWrongCredentials:     true,

internal/proxy/tcprelay/session_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/samosvalishe/free-turn-proxy/internal/provider"
 	"github.com/samosvalishe/free-turn-proxy/internal/proxy/allocpace"
 	"github.com/samosvalishe/free-turn-proxy/internal/stats"
+	"github.com/samosvalishe/free-turn-proxy/internal/transport/turndial"
 )
 
 type fakeAuth struct {
@@ -22,6 +23,15 @@ type fakeAuth struct {
 	dropped      atomic.Int32
 }
 
+func TestRetryDelayAllocationQuota(t *testing.T) {
+	for range 100 {
+		delay := retryDelay(&fakeAuth{}, errors.Join(errors.New("allocate"), turndial.ErrAllocQuota))
+		if delay < 15*time.Second || delay >= 30*time.Second {
+			t.Fatalf("quota retry delay = %s", delay)
+		}
+	}
+}
+
 func (a *fakeAuth) IsAuthError(err error) bool {
 	return a.authErr != nil && errors.Is(err, a.authErr)
 }

internal/proxy/tcprelay/tcprelay.go

@@ -21,6 +21,7 @@ import (
 	"github.com/samosvalishe/free-turn-proxy/internal/stats"
 	"github.com/samosvalishe/free-turn-proxy/internal/transport/dtlsdial"
 	"github.com/samosvalishe/free-turn-proxy/internal/transport/kcpmux"
+	"github.com/samosvalishe/free-turn-proxy/internal/transport/turndial"
 	"github.com/samosvalishe/free-turn-proxy/internal/wire"
 	"github.com/samosvalishe/free-turn-proxy/internal/wire/shape"
 	"github.com/xtaci/smux"
@@ -324,6 +325,9 @@ func maintainSession(ctx context.Context, deps *Deps, params *Params, peer *net.
 // retryDelay: пауза провайдера важнее собственной - ретрай в её середине только продлевает
 // локаут и жжёт персону. Джиттер разводит одновременный отказ всех сессий пула.
 func retryDelay(auth AuthHandler, err error) time.Duration {
+	if errors.Is(err, turndial.ErrAllocQuota) {
+		return time.Duration(15+randx.Intn(15)) * time.Second
+	}
 	if errors.Is(err, provider.ErrBackoffActive) {
 		if until := auth.BackoffUntilUnix(); until > 0 {
 			if d := time.Until(time.Unix(until, 0)); d > 0 {

internal/proxy/udprelay/loop.go

@@ -13,6 +13,7 @@ import (
 	"github.com/samosvalishe/free-turn-proxy/internal/clientsdb"
 	"github.com/samosvalishe/free-turn-proxy/internal/provider"
 	"github.com/samosvalishe/free-turn-proxy/internal/randx"
+	"github.com/samosvalishe/free-turn-proxy/internal/transport/turndial"
 	"github.com/samosvalishe/free-turn-proxy/internal/wire"
 	"github.com/samosvalishe/free-turn-proxy/internal/wire/shape"
 )
@@ -21,6 +22,11 @@ import (
 // backoff тут только удлиняет простой.
 var errPairRecycled = errors.New("udprelay: stream pair recycled")
 
+const (
+	quotaBackoff       = 15 * time.Second
+	quotaBackoffJitter = 15 * time.Second
+)
+
 // streamPair связывает DTLS-сессию с аллокацией, поверх которой она поднята. Смерть
 // аллокации обязана ронять DTLS: у новой аллокации другой relayed-адрес, а миграция
 // адреса по DTLS Connection ID под obf-профилем на сервере не работает - сервер примет
@@ -88,6 +94,17 @@ func TURNLoop(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr
 					deps.fatal(err)
 					return
 				}
+				// Квота может оставаться занятой после deallocate.
+				if errors.Is(err, turndial.ErrAllocQuota) {
+					wait := quotaBackoff + time.Duration(randx.Intn(int(quotaBackoffJitter/time.Second)))*time.Second
+					deps.log().Warnf("[STREAM %d] квота аллокаций занята - пауза %s", streamID, wait)
+					select {
+					case <-ctx.Done():
+						return
+					case <-time.After(wait):
+					}
+					continue
+				}
 				if errors.Is(err, provider.ErrBackoffActive) {
 					lockoutEnd := deps.Auth.BackoffUntilUnix()
 					var sleepDuration time.Duration
@@ -161,6 +178,9 @@ func dtlsSession(dtlsctx context.Context, dtlscancel context.CancelFunc, deps *D
 	if err := clientsdb.WriteClientID(dtlsConn, params.ClientID, clientsdb.ModeUDP); err != nil {
 		return fmt.Errorf("failed to write client ID: %w", err)
 	}
+	// Аллокация без DTLS ещё не может передавать пользовательский трафик.
+	deps.ConnectedStreams.Add(1)
+	defer deps.ConnectedStreams.Add(-1)
 	if okchan != nil {
 		select {
 		case okchan <- struct{}{}:
@@ -232,7 +252,6 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
 		deps.log().Debugf("[STREAM %d] obf-timing=%s", streamID, params.ObfTiming)
 	}
 
-	deps.ConnectedStreams.Add(1)
 	deps.Auth.ResetErrors(streamID)
 
 	relayedAddr := relayConn.LocalAddr().String()
@@ -240,7 +259,6 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
 		streamID, relayedAddr, stream.ServerUDPAddr.IP)
 
 	defer func() {
-		deps.ConnectedStreams.Add(-1)
 		// Освобождение аллокации логируем всегда: недошедший deallocate держит квоту VK
 		// до конца её lifetime, и следующий Allocate ловит 486.
 		cerr := stream.Close()
@@ -275,7 +293,7 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
 		select {
 		case <-turnctx.Done():
 		case <-stream.PermDead:
-			deps.log().Warnf("[STREAM %d] TURN channel-bind умер - рецикл allocation", streamID)
+			deps.log().Warnf("[STREAM %d] TURN refresh failed - recycle allocation", streamID)
 			turncancel()
 		}
 		// conn2 молчит, пока приложение не шлёт: без дедлайна его читатель досидел бы

internal/proxy/udprelay/pair_test.go

@@ -53,6 +53,9 @@ func TestOneDTLSPairCancelRecycles(t *testing.T) {
 	if pair.pipe == nil || pair.cancel == nil {
 		t.Fatal("incomplete pair")
 	}
+	if got := deps.ConnectedStreams.Load(); got != 0 {
+		t.Fatalf("unfinished handshake counted as connected: %d", got)
+	}
 
 	pair.cancel()
 

internal/transport/turndial/permwatch.go

@@ -10,7 +10,9 @@ import (
 const (
 	permFailMarker = "Failed to bind channel"
 	permOKMarker   = "Channel binding successful"
-	turncScope     = "turnc"
+	// Refresh сообщает об отказе раньше очередного ChannelBind.
+	allocFailMarker = "Failed to refresh allocation"
+	turncScope      = "turnc"
 	// permFailThreshold - число последовательных провалов ChannelBind refresh до срабатывания onDead.
 	permFailThreshold = 2
 )
@@ -40,6 +42,14 @@ type permWatchLogger struct {
 
 func (l *permWatchLogger) note(msg string) {
 	switch {
+	case strings.Contains(msg, allocFailMarker):
+		l.mu.Lock()
+		fire := !l.fired
+		l.fired = true
+		l.mu.Unlock()
+		if fire && l.f.onDead != nil {
+			l.f.onDead()
+		}
 	case strings.Contains(msg, permFailMarker):
 		l.mu.Lock()
 		l.fails++

internal/transport/turndial/permwatch_test.go

@@ -64,8 +64,36 @@ func TestPermWatchIgnoresUnrelatedMessages(t *testing.T) {
 	log := f.NewLogger(turncScope)
 	log.Debug("Started refresh permission timer")
 	log.Debug("No permission to refresh")
-	log.Warnf("Failed to refresh allocation: %s", "x")
+	log.Debug("Refresh permissions successful")
 	if fired.Load() != 0 {
 		t.Fatalf("fired on unrelated message: %d", fired.Load())
 	}
 }
+
+// Провал Refresh мимо порога ChannelBind: аллокация уже мертва.
+func TestPermWatchFiresOnAllocRefreshFail(t *testing.T) {
+	f, fired := newTestWatch(2)
+	log := f.NewLogger(turncScope)
+
+	log.Warnf(allocFailMarker+": %s", "all retransmissions failed")
+	if fired.Load() != 1 {
+		t.Fatalf("expected immediate fire, got %d", fired.Load())
+	}
+	log.Warnf(allocFailMarker + ": again")
+	if fired.Load() != 1 {
+		t.Fatalf("fired more than once: %d", fired.Load())
+	}
+}
+
+// Успешный ChannelBind не отменяет мёртвую аллокацию.
+func TestPermWatchAllocFailNotResetByBindOK(t *testing.T) {
+	f, fired := newTestWatch(2)
+	log := f.NewLogger(turncScope)
+
+	log.Warnf(allocFailMarker + ": x")
+	log.Debug(permOKMarker)
+	log.Warnf(allocFailMarker + ": x")
+	if fired.Load() != 1 {
+		t.Fatalf("expected exactly 1 fire, got %d", fired.Load())
+	}
+}

internal/transport/turndial/turndial.go

@@ -3,11 +3,13 @@ package turndial
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"sync"
 	"time"
 
+	"github.com/pion/stun/v3"
 	"github.com/pion/turn/v5"
 	"github.com/samosvalishe/free-turn-proxy/internal/logx"
 	"github.com/samosvalishe/free-turn-proxy/internal/netconn"
@@ -15,6 +17,10 @@ import (
 	"github.com/samosvalishe/free-turn-proxy/internal/randx"
 )
 
+// ErrAllocQuota - TURN-код 486: реквизиты валидны, но по ним уже висит предельное число
+// аллокаций. Лечится ожиданием, а не сменой реквизитов (см. IsAuthError в vkauth).
+var ErrAllocQuota = errors.New("turndial: allocation quota reached")
+
 // Config задаёт параметры подключения к TURN-серверу.
 type Config struct {
 	HostOverride string
@@ -34,6 +40,11 @@ type Stream struct {
 	close    func() error
 }
 
+func isQuotaError(err error) bool {
+	turnErr, ok := errors.AsType[*stun.TurnError](err)
+	return ok && turnErr.ErrorCodeAttr.Code == stun.CodeAllocQuotaReached
+}
+
 // Close освобождает аллокацию, TURN-клиент и транспортное соединение.
 func (s *Stream) Close() error {
 	if s == nil || s.close == nil {
@@ -141,6 +152,9 @@ func Open(ctx context.Context, cfg Config, peer *net.UDPAddr, user, pass, rawAdd
 		if cerr := closeConn(); cerr != nil {
 			err = fmt.Errorf("%w (close: %v)", err, cerr)
 		}
+		if isQuotaError(err) {
+			return nil, fmt.Errorf("%w: %w", ErrAllocQuota, err)
+		}
 		return nil, fmt.Errorf("TURN allocate: %w", err)
 	}
 

internal/transport/turndial/turndial_test.go

@@ -2,12 +2,30 @@ package turndial
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"net"
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/pion/stun/v3"
 )
 
+func TestIsQuotaError(t *testing.T) {
+	quota := &stun.TurnError{ErrorCodeAttr: stun.ErrorCodeAttribute{Code: stun.CodeAllocQuotaReached}}
+	if !isQuotaError(fmt.Errorf("allocate: %w", quota)) {
+		t.Fatal("wrapped 486 not recognized")
+	}
+	if isQuotaError(errors.New("486 Allocation Quota Reached")) || isQuotaError(nil) {
+		t.Fatal("untyped error treated as quota")
+	}
+	unauthorized := &stun.TurnError{ErrorCodeAttr: stun.ErrorCodeAttribute{Code: stun.CodeUnauthorized}}
+	if isQuotaError(unauthorized) {
+		t.Fatal("401 treated as quota")
+	}
+}
+
 func TestOpen_BadAddress(t *testing.T) {
 	peer := &net.UDPAddr{IP: net.ParseIP("1.2.3.4"), Port: 1}
 	_, err := Open(context.Background(), Config{}, peer, "u", "p", "not-a-host-port")

mobile/api.go

@@ -33,7 +33,6 @@ const (
 )
 
 const (
-	connectTimeout     = 15 * time.Second
 	stopTimeout        = 5 * time.Second
 	tunnelCloseTimeout = 3 * time.Second
 )
@@ -304,8 +303,8 @@ func startLocked(configJSON string, tunFD int, withTunnel bool) error {
 		Solver:        solver,
 		CaptchaActive: captchaActive.Load,
 		Options: session.Options{
-			ConnectTimeout: connectTimeout,
-			Traffic:        true,
+			// Сеть и квота TURN восстанавливаются дольше 15 с; остановку решает пользователь.
+			Traffic: true,
 		},
 	}
 	if parts != nil {