diff --git a/internal/provider/multi/multi.go b/internal/provider/multi/multi.go index e9968e0..5ff0215 100644 --- a/internal/provider/multi/multi.go +++ b/internal/provider/multi/multi.go @@ -46,6 +46,11 @@ func (m *Provider) ResetErrors(streamID int) { p.ResetErrors(innerID) } +func (m *Provider) DropCredentials(streamID int) { + p, innerID := m.providerFor(streamID) + p.DropCredentials(innerID) +} + func (m *Provider) BackoffUntilUnix() int64 { var until int64 for _, p := range m.providers { diff --git a/internal/provider/multi/multi_test.go b/internal/provider/multi/multi_test.go index 5ada231..e9b86e5 100644 --- a/internal/provider/multi/multi_test.go +++ b/internal/provider/multi/multi_test.go @@ -17,6 +17,7 @@ type fakeProvider struct { gotCreds []int gotHandle []int gotReset []int + gotDrop []int authResult bool } @@ -27,6 +28,7 @@ func (f *fakeProvider) GetCredentials(_ context.Context, streamID int) (provider func (f *fakeProvider) IsAuthError(error) bool { return f.authResult } func (f *fakeProvider) HandleAuthError(s int) bool { f.gotHandle = append(f.gotHandle, s); return true } func (f *fakeProvider) ResetErrors(s int) { f.gotReset = append(f.gotReset, s) } +func (f *fakeProvider) DropCredentials(s int) { f.gotDrop = append(f.gotDrop, s) } func (f *fakeProvider) BackoffUntilUnix() int64 { return f.backoff } func (f *fakeProvider) Name() string { return fmt.Sprintf("fake%d", f.idx) } @@ -93,6 +95,10 @@ func TestHandleAndResetRouting(t *testing.T) { if len(fs[0].gotReset) != 1 || fs[0].gotReset[0] != 3 { t.Fatalf("ResetErrors routed wrong: %v", fs[0].gotReset) } + m.DropCredentials(5) + if len(fs[0].gotDrop) != 1 || fs[0].gotDrop[0] != 3 { + t.Fatalf("DropCredentials routed wrong: %v", fs[0].gotDrop) + } if len(fs[1].gotHandle) != 0 { t.Fatalf("provider 1 should not be touched: %v", fs[1].gotHandle) } diff --git a/internal/provider/provider.go b/internal/provider/provider.go index 2d5f961..6f9eb20 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -21,6 +21,7 @@ type Provider interface { IsAuthError(err error) bool HandleAuthError(streamID int) bool ResetErrors(streamID int) + DropCredentials(streamID int) BackoffUntilUnix() int64 Name() string } diff --git a/internal/provider/vk/internal/vkauth/cache.go b/internal/provider/vk/internal/vkauth/cache.go index 5053b22..360fc9b 100644 --- a/internal/provider/vk/internal/vkauth/cache.go +++ b/internal/provider/vk/internal/vkauth/cache.go @@ -60,13 +60,15 @@ func (s *Store) Get(streamID int) *StreamCredentialsCache { return cache } -func (c *StreamCredentialsCache) Invalidate() { +func (c *StreamCredentialsCache) Invalidate() bool { c.mutex.Lock() + had := c.creds.Username != "" c.creds = TurnCredentials{} c.mutex.Unlock() c.errorCount.Store(0) c.lastErrorTime.Store(0) + return had } func IsAuthError(err error) bool { @@ -79,5 +81,5 @@ func IsAuthError(err error) bool { strings.Contains(s, "authentication") || strings.Contains(s, "invalid credential") || strings.Contains(s, "stale nonce") || - strings.Contains(s, "486") + strings.Contains(s, "Allocation Quota") } diff --git a/internal/provider/vk/internal/vkauth/client.go b/internal/provider/vk/internal/vkauth/client.go index ffb95b1..9cad956 100644 --- a/internal/provider/vk/internal/vkauth/client.go +++ b/internal/provider/vk/internal/vkauth/client.go @@ -192,6 +192,14 @@ func (c *Client) ResetErrors(streamID int) { c.store.Get(streamID).errorCount.Store(0) } +func (c *Client) DropCredentials(streamID int) { + if !c.store.Get(streamID).Invalidate() { + return + } + c.log.Warnf("[STREAM %d] [VK Auth] Deallocate unconfirmed - credentials dropped (cache=%d)", + streamID, c.store.CacheID(streamID)) +} + func (c *Client) LockoutUntilUnix() int64 { return c.lockout.Load() } diff --git a/internal/provider/vk/internal/vkauth/client_test.go b/internal/provider/vk/internal/vkauth/client_test.go index 8834a88..ad5db6e 100644 --- a/internal/provider/vk/internal/vkauth/client_test.go +++ b/internal/provider/vk/internal/vkauth/client_test.go @@ -31,14 +31,15 @@ func TestIsAuthError(t *testing.T) { t.Parallel() cases := map[string]bool{ - "401 Unauthorized": true, - "alloc failed: 401": true, - "stale nonce": true, - "invalid credential": true, - "authentication required": true, - "486: Allocation Quota Reached": true, - "connection refused": false, - "": false, + "401 Unauthorized": true, + "alloc failed: 401": true, + "stale nonce": true, + "invalid credential": true, + "authentication required": true, + "486: Allocation Quota Reached": true, + "connection refused": false, + "dial tcp 10.0.0.1:48612: i/o timeout": false, + "": false, } for msg, want := range cases { got := IsAuthError(errors.New(msg)) @@ -222,3 +223,26 @@ func TestThrottleHonorsContextCancel(t *testing.T) { t.Fatalf("expected context.Canceled, got %v", err) } } + +func TestDropCredentialsClearsCacheOnce(t *testing.T) { + t.Parallel() + c := newTestClient(t, nil) + + cache := c.store.Get(1) + cache.creds = TurnCredentials{ + Username: "u", + Password: "p", + ServerAddrs: []string{"turn:1.2.3.4:19302"}, + ExpiresAt: time.Now().Add(time.Minute), + Link: "l", + } + + c.DropCredentials(1) + if cache.creds.Username != "" { + t.Fatal("credentials survived DropCredentials") + } + // Второй стрим того же кэша не должен повторять сброс: иначе один обрыв даёт N строк. + if cache.Invalidate() { + t.Fatal("empty cache reported as invalidated") + } +} diff --git a/internal/provider/vk/vk.go b/internal/provider/vk/vk.go index c497f6e..3e17b30 100644 --- a/internal/provider/vk/vk.go +++ b/internal/provider/vk/vk.go @@ -75,6 +75,8 @@ func (p *Provider) HandleAuthError(streamID int) bool { return p.auth.HandleAuth func (p *Provider) ResetErrors(streamID int) { p.auth.ResetErrors(streamID) } +func (p *Provider) DropCredentials(streamID int) { p.auth.DropCredentials(streamID) } + func (p *Provider) BackoffUntilUnix() int64 { return p.auth.BackoffUntilUnix() } func (*Provider) Name() string { return "vk" } diff --git a/internal/proxy/udprelay/loop.go b/internal/proxy/udprelay/loop.go index 70236c3..784e349 100644 --- a/internal/proxy/udprelay/loop.go +++ b/internal/proxy/udprelay/loop.go @@ -252,9 +252,8 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, cerr := stream.Close() deps.log().Infof("[STREAM %d] TURN allocation released: relayed=%s deallocate=%v", streamID, relayedAddr, cerr) - // Ошибку закрытия не ставим поверх причины выхода - она уже сказала, почему стрим упал. - if cerr != nil && err == nil { - err = fmt.Errorf("failed to close TURN stream: %w", cerr) + if cerr != nil { + deps.Auth.DropCredentials(streamID) } }() diff --git a/internal/proxy/udprelay/run.go b/internal/proxy/udprelay/run.go index 155de83..6df612c 100644 --- a/internal/proxy/udprelay/run.go +++ b/internal/proxy/udprelay/run.go @@ -19,6 +19,7 @@ type AuthHandler interface { IsAuthError(err error) bool HandleAuthError(streamID int) bool ResetErrors(streamID int) + DropCredentials(streamID int) BackoffUntilUnix() int64 }