From 40000fde5d82023f8552e2a0255b6ca50324b8ed Mon Sep 17 00:00:00 2001 From: samosvalishe Date: Sun, 7 Jun 2026 23:25:35 +0300 Subject: [PATCH] =?UTF-8?q?feat(turn):=20=D1=84=D0=BE=D0=BB=D0=BB=D0=B1?= =?UTF-8?q?=D1=8D=D0=BA=20=D0=BF=D0=BE=20=D0=BD=D0=B5=D1=81=D0=BA=D0=BE?= =?UTF-8?q?=D0=BB=D1=8C=D0=BA=D0=B8=D0=BC=20relay-=D0=B0=D0=B4=D1=80=D0=B5?= =?UTF-8?q?=D1=81=D0=B0=D0=BC=20=D0=BF=D1=80=D0=B8=20allocate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit VK отдаёт несколько relay-адресов, но стрим использовал ровно один (streamID%n). Если этот IP душит DPI/RST — allocate падает без альтернативы. Credentials несёт список адресов (primary-first по streamID), DialTURN пробует кандидатов по очереди и берёт первый успешный allocate. Балансировка relay-IP по стримам сохранена, остальные адреса — фоллбэк. HostOverride (-turn) схлопывает список до одного адреса. --- cmd/client/main.go | 6 +-- internal/provider/provider.go | 13 +++--- .../provider/vk/internal/vkauth/client.go | 38 +++++++++++----- .../vk/internal/vkauth/client_test.go | 10 ++--- .../provider/vk/internal/vkauth/order_test.go | 37 +++++++++++++++ internal/provider/vk/vk.go | 4 +- internal/proxy/common/common.go | 45 ++++++++++++++----- internal/proxy/common/common_test.go | 19 ++++++++ 8 files changed, 134 insertions(+), 38 deletions(-) create mode 100644 internal/provider/vk/internal/vkauth/order_test.go create mode 100644 internal/proxy/common/common_test.go diff --git a/cmd/client/main.go b/cmd/client/main.go index 15dc8f9..ca01e3b 100644 --- a/cmd/client/main.go +++ b/cmd/client/main.go @@ -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 { diff --git a/internal/provider/provider.go b/internal/provider/provider.go index 81dea39..0a575fb 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -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. diff --git a/internal/provider/vk/internal/vkauth/client.go b/internal/provider/vk/internal/vkauth/client.go index 4ff0435..d649550 100644 --- a/internal/provider/vk/internal/vkauth/client.go +++ b/internal/provider/vk/internal/vkauth/client.go @@ -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 увеличивает счётчик ошибок аутентификации кэша потока и diff --git a/internal/provider/vk/internal/vkauth/client_test.go b/internal/provider/vk/internal/vkauth/client_test.go index e31122c..7e7fc92 100644 --- a/internal/provider/vk/internal/vkauth/client_test.go +++ b/internal/provider/vk/internal/vkauth/client_test.go @@ -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]) } } diff --git a/internal/provider/vk/internal/vkauth/order_test.go b/internal/provider/vk/internal/vkauth/order_test.go new file mode 100644 index 0000000..2a5336b --- /dev/null +++ b/internal/provider/vk/internal/vkauth/order_test.go @@ -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) + } +} diff --git a/internal/provider/vk/vk.go b/internal/provider/vk/vk.go index a5fa76c..bb4f553 100644 --- a/internal/provider/vk/vk.go +++ b/internal/provider/vk/vk.go @@ -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. diff --git a/internal/proxy/common/common.go b/internal/proxy/common/common.go index dc17c25..bdfdd6f 100644 --- a/internal/proxy/common/common.go +++ b/internal/proxy/common/common.go @@ -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 нужной длины, diff --git a/internal/proxy/common/common_test.go b/internal/proxy/common/common_test.go new file mode 100644 index 0000000..0a08a20 --- /dev/null +++ b/internal/proxy/common/common_test.go @@ -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") + } +}