fix(vkauth): сбрасывать реквизиты при неподтверждённом deallocate

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 {

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)
 	}

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
 }

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")
 }

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()
 }

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")
+	}
+}

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" }

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)
 		}
 	}()
 

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
 }