feat(turn): фоллбэк по нескольким relay-адресам при allocate

VK отдаёт несколько relay-адресов, но стрим использовал ровно один
(streamID%n). Если этот IP душит DPI/RST — allocate падает без альтернативы.

Credentials несёт список адресов (primary-first по streamID), DialTURN
пробует кандидатов по очереди и берёт первый успешный allocate. Балансировка
relay-IP по стримам сохранена, остальные адреса — фоллбэк. HostOverride (-turn)
схлопывает список до одного адреса.
This commit is contained in:
samosvalishe
2026-06-07 23:25:35 +03:00
parent 106a48533c
commit 40000fde5d
8 changed files with 134 additions and 38 deletions
+3 -3
View File
@@ -140,12 +140,12 @@ func main() {
}
logger.Infof("provider=%s", prov.Name())
getCreds := func(ctx context.Context, streamID int) (string, string, string, error) {
getCreds := func(ctx context.Context, streamID int) (string, string, []string, error) {
c, err := prov.GetCredentials(ctx, streamID)
if err != nil {
return "", "", "", err
return "", "", nil, err
}
return c.User, c.Pass, c.ServerAddr, nil
return c.User, c.Pass, c.ServerAddrs, nil
}
if cfg.Proxy.Mode != config.ProxyModeUDP {
+7 -6
View File
@@ -17,13 +17,14 @@ import (
)
// Credentials — выданные провайдером TURN-реквизиты для одного стрима.
// Адрес уже разрешён до конкретного host:port; pipeline не делает round-robin
// по нескольким адресам — это ответственность провайдера.
// ServerAddrs — кандидаты host:port в порядке предпочтения (первый —
// предпочтительный для этого streamID). Pipeline пробует их по очереди при
// allocate: если первый не проходит (DPI-дроп/RST), берёт следующий.
type Credentials struct {
User string
Pass string
ServerAddr string // host:port
ExpiresAt time.Time // 0 если провайдер не сообщает дедлайн
User string
Pass string
ServerAddrs []string // host:port, primary-first
ExpiresAt time.Time // 0 если провайдер не сообщает дедлайн
}
// Provider — источник TURN-реквизитов для клиентского pipeline.
+27 -11
View File
@@ -101,9 +101,11 @@ func New(cfg Config) *Client {
return c
}
// GetCredentials возвращает (username, password, server-addr) для TURN-allocate,
// обращаясь к VK (с throttle + кэшем) только при необходимости.
func (c *Client) GetCredentials(ctx context.Context, link string, streamID int) (string, string, string, error) {
// GetCredentials возвращает (username, password, server-addrs) для TURN-allocate,
// обращаясь к VK (с throttle + кэшем) только при необходимости. Адреса отдаются
// в порядке предпочтения для streamID (предпочтительный первым) — pipeline
// пробует их по очереди при allocate.
func (c *Client) GetCredentials(ctx context.Context, link string, streamID int) (string, string, []string, error) {
cache := c.store.Get(streamID)
cacheID := c.store.CacheID(streamID)
@@ -111,10 +113,10 @@ func (c *Client) GetCredentials(ctx context.Context, link string, streamID int)
if cache.creds.Link == link && time.Now().Before(cache.creds.ExpiresAt) && len(cache.creds.ServerAddrs) > 0 {
expires := time.Until(cache.creds.ExpiresAt)
u, p := cache.creds.Username, cache.creds.Password
addr := cache.creds.ServerAddrs[streamID%len(cache.creds.ServerAddrs)]
addrs := orderAddrs(cache.creds.ServerAddrs, streamID)
cache.mutex.RUnlock()
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
c.log.Debugf("[STREAM %d] [VK Auth] Using cached credentials (cache=%d, expires in %v, server=%s)", streamID, cacheID, expires, addrs[0])
return u, p, addrs, nil
}
cache.mutex.RUnlock()
@@ -122,13 +124,12 @@ func (c *Client) GetCredentials(ctx context.Context, link string, streamID int)
defer cache.mutex.Unlock()
if cache.creds.Link == link && time.Now().Before(cache.creds.ExpiresAt) && len(cache.creds.ServerAddrs) > 0 {
addr := cache.creds.ServerAddrs[streamID%len(cache.creds.ServerAddrs)]
return cache.creds.Username, cache.creds.Password, addr, nil
return cache.creds.Username, cache.creds.Password, orderAddrs(cache.creds.ServerAddrs, streamID), nil
}
user, pass, addrs, err := c.fetchSerialized(ctx, link, streamID)
if err != nil {
return "", "", "", err
return "", "", nil, err
}
cache.creds = TurnCredentials{
@@ -138,8 +139,23 @@ func (c *Client) GetCredentials(ctx context.Context, link string, streamID int)
ExpiresAt: time.Now().Add(CredentialLifetime - CacheSafetyMargin),
Link: link,
}
addr := addrs[streamID%len(addrs)]
return user, pass, addr, nil
return user, pass, orderAddrs(addrs, streamID), nil
}
// orderAddrs возвращает копию addrs, ротированную так, что предпочтительный
// для streamID адрес стоит первым, остальные — следом (сохраняя порядок).
// Раскидывает primary по стримам (балансировка relay-IP), оставляя остальные
// как фоллбэк при DPI-дропе.
func orderAddrs(addrs []string, streamID int) []string {
n := len(addrs)
if n <= 1 {
return append([]string(nil), addrs...)
}
k := streamID % n
out := make([]string, 0, n)
out = append(out, addrs[k:]...)
out = append(out, addrs[:k]...)
return out
}
// HandleAuthError увеличивает счётчик ошибок аутентификации кэша потока и
@@ -109,8 +109,8 @@ func TestFetchFallsBackThroughCredentialsList(t *testing.T) {
if err != nil {
t.Fatalf("GetCredentials: %v", err)
}
if u != "user-c" || p != "pass-c" || addr != "server-c:443" {
t.Fatalf("unexpected creds: u=%q p=%q addr=%q", u, p, addr)
if u != "user-c" || p != "pass-c" || len(addr) != 1 || addr[0] != "server-c:443" {
t.Fatalf("unexpected creds: u=%q p=%q addr=%v", u, p, addr)
}
if calls.Load() != 3 {
t.Errorf("expected to walk all 3 creds, called %d times", calls.Load())
@@ -156,7 +156,7 @@ func TestGetCredentialsCacheHit(t *testing.T) {
t.Errorf("expected single fetch, got %d", calls.Load())
}
// Sibling-stream addr round-robin: stream 0 -> addr[0], stream 1 -> addr[1].
// Sibling-stream primary round-robin: stream 0 -> addr[0], stream 1 -> addr[1].
_, _, addr0, err := c.GetCredentials(context.Background(), "L", 0)
if err != nil {
t.Fatal(err)
@@ -165,8 +165,8 @@ func TestGetCredentialsCacheHit(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if addr0 == addr1 {
t.Errorf("expected round-robin addrs, both = %q", addr0)
if addr0[0] == addr1[0] {
t.Errorf("expected round-robin primary, both = %q", addr0[0])
}
}
@@ -0,0 +1,37 @@
package vkauth
import (
"slices"
"testing"
)
func TestOrderAddrs(t *testing.T) {
t.Parallel()
addrs := []string{"a", "b", "c"}
cases := []struct {
streamID int
want []string
}{
{0, []string{"a", "b", "c"}},
{1, []string{"b", "c", "a"}},
{2, []string{"c", "a", "b"}},
{3, []string{"a", "b", "c"}}, // wrap
}
for _, tc := range cases {
got := orderAddrs(addrs, tc.streamID)
if !slices.Equal(got, tc.want) {
t.Errorf("orderAddrs(streamID=%d) = %v, want %v", tc.streamID, got, tc.want)
}
}
// Не мутирует вход.
if !slices.Equal(addrs, []string{"a", "b", "c"}) {
t.Errorf("orderAddrs mutated input: %v", addrs)
}
// Один адрес — копия из одного элемента.
if got := orderAddrs([]string{"x"}, 5); !slices.Equal(got, []string{"x"}) {
t.Errorf("single addr: got %v", got)
}
}
+2 -2
View File
@@ -83,11 +83,11 @@ func New(cfg Config, solver ManualSolverFunc) (*Provider, error) {
// GetCredentials реализует provider.Provider.
func (p *Provider) GetCredentials(ctx context.Context, streamID int) (provider.Credentials, error) {
user, pass, addr, err := p.auth.GetCredentials(ctx, p.link, streamID)
user, pass, addrs, err := p.auth.GetCredentials(ctx, p.link, streamID)
if err != nil {
return provider.Credentials{}, err
}
return provider.Credentials{User: user, Pass: pass, ServerAddr: addr}, nil
return provider.Credentials{User: user, Pass: pass, ServerAddrs: addrs}, nil
}
// IsAuthError реализует provider.Provider.
+34 -11
View File
@@ -6,6 +6,7 @@ package common
import (
"context"
"errors"
"fmt"
"net"
@@ -15,22 +16,44 @@ import (
// GetCredsFunc разрешает TURN-реквизиты для streamID. Реализуется provider'ом
// (см. internal/provider): provider держит идентификатор сессии (link/room/key)
// внутри, pipeline передаёт только streamID.
type GetCredsFunc func(ctx context.Context, streamID int) (user, pass, rawURL string, err error)
// внутри, pipeline передаёт только streamID. rawURLs — кандидаты host:port в
// порядке предпочтения.
type GetCredsFunc func(ctx context.Context, streamID int) (user, pass string, rawURLs []string, err error)
// DialTURN получает реквизиты и открывает TURN-поток. Вызывающий отвечает
// за закрытие потока и политику retry при auth-ошибке (udprelay)
// или перезапуска сессии (tcpfwd).
// DialTURN получает реквизиты и открывает TURN-поток, пробуя кандидатов по
// очереди: если allocate на первом не проходит (DPI-дроп/RST на relay-IP),
// берёт следующий. Возвращает первый успешный Stream. Вызывающий отвечает за
// закрытие потока и политику retry при auth-ошибке (udprelay) или перезапуска
// сессии (tcpfwd).
func DialTURN(ctx context.Context, host, port string, udp bool, peer *net.UDPAddr, streamID int, getCreds GetCredsFunc) (*turndial.Stream, error) {
user, pass, rawURL, err := getCreds(ctx, streamID)
user, pass, rawURLs, err := getCreds(ctx, streamID)
if err != nil {
return nil, fmt.Errorf("get TURN creds: %w", err)
}
return turndial.Open(ctx, turndial.Config{
HostOverride: host,
PortOverride: port,
TransportUDP: udp,
}, peer, user, pass, rawURL)
if len(rawURLs) == 0 {
return nil, fmt.Errorf("no TURN candidates")
}
// HostOverride (-turn) принудительно задаёт host → все кандидаты резолвятся
// в одну цель, гонять их нет смысла; пробуем только первого.
if host != "" {
rawURLs = rawURLs[:1]
}
var errs []error
for _, rawURL := range rawURLs {
stream, derr := turndial.Open(ctx, turndial.Config{
HostOverride: host,
PortOverride: port,
TransportUDP: udp,
}, peer, user, pass, rawURL)
if derr == nil {
return stream, nil
}
errs = append(errs, fmt.Errorf("%s: %w", rawURL, derr))
if ctx.Err() != nil {
break
}
}
return nil, fmt.Errorf("all TURN candidates failed: %w", errors.Join(errs...))
}
// NewClientObf возвращает клиентский rtpopus.Conn если key нужной длины,
+19
View File
@@ -0,0 +1,19 @@
package common
import (
"context"
"net"
"testing"
)
func TestDialTURNNoCandidates(t *testing.T) {
t.Parallel()
getCreds := func(context.Context, int) (string, string, []string, error) {
return "u", "p", nil, nil
}
peer := &net.UDPAddr{IP: net.IPv4(1, 2, 3, 4), Port: 1}
_, err := DialTURN(context.Background(), "", "", false, peer, 0, getCreds)
if err == nil {
t.Fatal("expected error on empty candidate list")
}
}