diff --git a/core/allocation_gate.go b/core/allocation_gate.go new file mode 100644 index 00000000..b97a9fb8 --- /dev/null +++ b/core/allocation_gate.go @@ -0,0 +1,44 @@ +package core + +import ( + "context" + "sync" + "time" +) + +const allocateGateInterval = 100 * time.Millisecond + +// allocationGate spaces TURN Allocate attempts. A Core owns one gate, so its +// workers share a quota without unrelated Core instances blocking each other. +type allocationGate struct { + mu sync.Mutex + next time.Time + interval time.Duration +} + +func newAllocationGate(interval time.Duration) *allocationGate { + return &allocationGate{interval: interval} +} + +func (g *allocationGate) Wait(ctx context.Context) error { + g.mu.Lock() + due := time.Now() + if g.next.After(due) { + due = g.next + } + g.next = due.Add(g.interval) + g.mu.Unlock() + + wait := time.Until(due) + if wait <= 0 { + return nil + } + t := time.NewTimer(wait) + defer t.Stop() + select { + case <-t.C: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} diff --git a/core/allocation_gate_test.go b/core/allocation_gate_test.go new file mode 100644 index 00000000..2e2fcc4e --- /dev/null +++ b/core/allocation_gate_test.go @@ -0,0 +1,59 @@ +package core + +import ( + "context" + "errors" + "sort" + "sync" + "testing" + "time" +) + +func TestAllocationGatePacesConcurrentAttempts(t *testing.T) { + interval := 25 * time.Millisecond + g := newAllocationGate(interval) + start := make(chan struct{}) + attempts := make(chan time.Time, 3) + var wg sync.WaitGroup + for range 3 { + wg.Add(1) + go func() { + defer wg.Done() + <-start + if err := g.Wait(context.Background()); err != nil { + t.Error(err) + return + } + attempts <- time.Now() + }() + } + close(start) + wg.Wait() + close(attempts) + + got := make([]time.Time, 0, 3) + for at := range attempts { + got = append(got, at) + } + if len(got) != 3 { + t.Fatalf("completed attempts = %d, want 3", len(got)) + } + sort.Slice(got, func(i, j int) bool { return got[i].Before(got[j]) }) + for i := 1; i < len(got); i++ { + if elapsed := got[i].Sub(got[i-1]); elapsed < interval/2 { + t.Fatalf("attempts %d and %d were not paced: %v", i-1, i, elapsed) + } + } +} + +func TestAllocationGateWaitCancels(t *testing.T) { + g := newAllocationGate(time.Second) + if err := g.Wait(context.Background()); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := g.Wait(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("Wait() error = %v, want context.Canceled", err) + } +} diff --git a/core/main.go b/core/main.go index c5607f5f..bb8b2571 100644 --- a/core/main.go +++ b/core/main.go @@ -80,9 +80,10 @@ type Core struct { captchaMode atomic.Value vkAuthMode atomic.Value events chan Event - startOnce sync.Once // guard от двойного Start() - startErr atomic.Value // *startFailure — ошибка первого Start(), если он провалился - once sync.Once // guard от двойного Stop() + allocateGate *allocationGate + startOnce sync.Once // guard от двойного Start() + startErr atomic.Value // *startFailure — ошибка первого Start(), если он провалился + once sync.Once // guard от двойного Stop() } // startFailure — обёртка для atomic.Value (чтобы разные error-типы не паниковали). @@ -129,6 +130,7 @@ func New(cfg Config) *Core { cancel: cancel, CaptchaResultChan: make(chan string, 1), events: make(chan Event, 1024), // увеличен буфер для логов + allocateGate: newAllocationGate(allocateGateInterval), } c.captchaMode.Store(normalizeCaptchaMode(cfg.CaptchaMode)) c.vkAuthMode.Store(normalizeVKAuthMode("vkcalls")) @@ -228,14 +230,7 @@ func (c *Core) start() error { } // Нормализация воркеров - n := c.cfg.Workers - if n > 108 { - n = 108 - } - if n < workersPerGroup { - n = workersPerGroup - } - n = (n / workersPerGroup) * workersPerGroup + n := normalizeWorkers(c.cfg.Workers) tp := &TurnParams{ Host: c.cfg.TurnHost, @@ -244,6 +239,7 @@ func (c *Core) start() error { WrapKey: wrapKey, ObfsMode: c.cfg.ObfsMode, TCPTransport: c.cfg.TurnTCP, + allocateGate: c.allocateGate, } // Локальный UDP сокет @@ -573,7 +569,8 @@ func classifyLevel(msg string) string { switch { case strings.Contains(low, "fatal") || strings.Contains(low, "ошибка") || strings.Contains(low, "error"): return "ERROR" - case strings.Contains(low, "warn") || strings.Contains(low, "не удалось") || strings.Contains(low, "retry"): + case strings.Contains(low, "warn") || strings.Contains(low, "не удалось") || strings.Contains(low, "retry") || + strings.Contains(low, "[failover]") || strings.Contains(low, "[reconnect]"): return "WARN" default: return "INFO" diff --git a/core/session.go b/core/session.go index 63adf2e1..77abc8e3 100644 --- a/core/session.go +++ b/core/session.go @@ -8,6 +8,7 @@ import ( "net" "strings" "sync" + "sync/atomic" "time" "github.com/cbeuw/connutil" @@ -104,6 +105,7 @@ func RunSession( stats *Stats, ) (bool, *SessionError) { configDelivered := false + var firstWireRead atomic.Bool if turnAddr == "" { return false, &SessionError{ @@ -177,6 +179,16 @@ func RunSession( } } + if tp.allocateGate != nil { + if err = tp.allocateGate.Wait(ctx); err != nil { + return false, &SessionError{ + Type: SessionErrorFatal, + Address: turnAddr, + Err: fmt.Errorf("ожидание TURN Allocate: %w", err), + } + } + } + relay, err := tc.Allocate() if err != nil { errStr := err.Error() @@ -219,7 +231,19 @@ func RunSession( getStreamCache(cacheStreamID).errorCount.Store(0) - log.Printf("[СЕССИЯ #%d] Relay: %s", sessionID, relay.LocalAddr()) + log.Printf("[СЕССИЯ #%d] Relay: %s; peer: %s", sessionID, relay.LocalAddr(), peer) + + // Establish and verify the TURN permission before DTLS starts. Otherwise a + // failure is hidden inside the relay-writing goroutine and looks like a DTLS + // or password timeout. + if err = tc.CreatePermission(peer); err != nil { + return false, &SessionError{ + Type: SessionErrorAddressDead, + Address: turnAddr, + Err: fmt.Errorf("TURN CreatePermission для %s: %w", peer, err), + } + } + log.Printf("[СЕССИЯ #%d] TURN permission OK для %s", sessionID, peer) pipeA, pipeB := connutil.AsyncPacketPipe() @@ -277,6 +301,9 @@ func RunSession( if readErr != nil { return } + if firstWireRead.CompareAndSwap(false, true) { + log.Printf("[СЕССИЯ #%d] Первый пакет получен через TURN от %s (%d байт)", sessionID, peer, n) + } payload := buf[:n] if useWrap { if !obfsIsRTPPacket(payload) { @@ -300,6 +327,7 @@ func RunSession( defer relayWg.Done() defer sessCancel() b := make([]byte, readBufSize) + firstWrite := true for { n, _, readErr := pipeA.ReadFrom(b) if readErr != nil { @@ -316,9 +344,15 @@ func RunSession( out = wrapped } } - if _, writeErr := relay.WriteTo(out, peer); writeErr != nil { + written, writeErr := relay.WriteTo(out, peer) + if writeErr != nil { + log.Printf("[СЕССИЯ #%d] TURN relay WriteTo %s: %v", sessionID, peer, writeErr) return } + if firstWrite { + log.Printf("[СЕССИЯ #%d] Первый пакет отправлен через TURN к %s (%d/%d байт)", sessionID, peer, written, len(out)) + firstWrite = false + } } }() @@ -368,11 +402,20 @@ func RunSession( if err != nil { errStr := strings.ToLower(err.Error()) - if useWrap && (strings.Contains(errStr, "deadline") || strings.Contains(errStr, "timeout")) { - return false, &SessionError{ - Type: SessionErrorWrapTimeout, - Address: turnAddr, - Err: fmt.Errorf("DTLS timeout, пароль/WRAP не подтверждён"), + if strings.Contains(errStr, "deadline") || strings.Contains(errStr, "timeout") { + if !firstWireRead.Load() { + return false, &SessionError{ + Type: SessionErrorAddressDead, + Address: turnAddr, + Err: fmt.Errorf("TURN relay молчит: ни один пакет от peer не вернулся"), + } + } + if useWrap { + return false, &SessionError{ + Type: SessionErrorWrapTimeout, + Address: turnAddr, + Err: fmt.Errorf("peer ответил, но WRAP/DTLS не подтверждён"), + } } } return false, &SessionError{ @@ -488,7 +531,7 @@ func RunSession( if ne, ok := readErr.(net.Error); ok && ne.Timeout() { continue } - log.Printf("[ВОРКЕР #%d] Ошибка Reader: %v", sessionID, readErr) + log.Printf("[ВОРКЕР #%d] [RECONNECT] Reader завершён: %v", sessionID, readErr) return } diff --git a/core/worker_group.go b/core/worker_group.go index 556095fe..8146bfa8 100644 --- a/core/worker_group.go +++ b/core/worker_group.go @@ -31,6 +31,10 @@ const ( // firstBatchSize — сколько воркеров запускаем в первой волне firstBatchSize = 5 + + // configRacerLimit bounds concurrent initial config requests. The first + // racers are assigned different TURN URLs whenever enough URLs exist. + configRacerLimit = 3 ) func WorkerGroup( @@ -113,6 +117,7 @@ func WorkerGroup( credStreamID := groupID * 100 var credsMu sync.RWMutex + var refreshMu sync.Mutex var curUser, curPass string var curURLs []string var credGen uint64 @@ -187,7 +192,6 @@ func WorkerGroup( } } - var configRequestInFlight int32 var wg sync.WaitGroup // Запускаем воркеры с поэтапной задержкой @@ -205,7 +209,7 @@ func WorkerGroup( startDelay = time.Duration(firstBatchSize)*workerStartDelay + workerBatchDelay + time.Duration(secondWaveIdx)*workerStartDelay } - go func(wid int, delay time.Duration) { + go func(wid, workerIndex int, delay time.Duration) { // Ждём свою задержку перед стартом select { case <-time.After(delay): @@ -217,8 +221,10 @@ func WorkerGroup( defer wg.Done() shouldGetConfig := getConfig - // Счётчик неудач на текущий адрес + // Retry state is local to this worker, avoiding synchronized storms. addressAttempts := 0 + relayOffset := 0 + lastTurnAddr := "" // Храним текущий ObfsMode для этого воркера (может меняться при WRAP_TIMEOUT) workerObfsMode := tp.ObfsMode @@ -251,16 +257,10 @@ func WorkerGroup( continue } - // Фильтруем забаненные адреса - var available []string - for _, u := range urls { - if !GlobalBlacklist.IsBanned(u) { - available = append(available, u) - } - } - - // Если все адреса забанены — спим и пробуем заново - if len(available) == 0 { + // Spread workers deterministically over relays and advance after a + // failed relay instead of converging on urls[0]. + turnAddr, ok := selectTurnURL(urls, wid, relayOffset, GlobalBlacklist.IsBanned) + if !ok { log.Printf("[ВОРКЕР #%d] Все TURN-адреса забанены (всего %d), спим %v", wid, len(urls), sleepWhenAllBanned) select { @@ -270,36 +270,34 @@ func WorkerGroup( continue } - // Берём первый доступный адрес - turnAddr := available[0] - - getConf := false - if shouldGetConfig && atomic.LoadInt32(&configSent) == 0 { - getConf = atomic.CompareAndSwapInt32(&configRequestInFlight, 0, 1) + if turnAddr != lastTurnAddr { + addressAttempts = 0 + lastTurnAddr = turnAddr } + + // A few workers race for config on round-robin-selected relays. This + // prevents one silent relay from blocking startup without starting a + // config request from every worker. + getConf := shouldRaceForConfig(shouldGetConfig, workerIndex, atomic.LoadInt32(&configSent)) var cc chan<- string if getConf { cc = configCh } - // Передаём в RunSession конкретный адрес и креды - configDelivered, sessErr := RunSession( - ctx, tp, peer, d, localPort, - getConf, cc, wid, - turnAddr, // конкретный TURN-адрес - wUser, // username из кредов - wPass, // password из кредов - workerObfsMode, - credStreamID, // для handleAuthError - deviceID, password, stats, - ) - - if getConf { - if configDelivered { - atomic.StoreInt32(&configSent, 1) - } else { - atomic.StoreInt32(&configRequestInFlight, 0) - } + // Передаём в RunSession конкретный адрес и креды + configDelivered, sessErr := RunSession( + ctx, tp, peer, d, localPort, + getConf, cc, wid, + turnAddr, // конкретный TURN-адрес + wUser, // username из кредов + wPass, // password из кредов + workerObfsMode, + credStreamID, // для handleAuthError + deviceID, password, stats, + ) + + if getConf && configDelivered { + atomic.StoreInt32(&configSent, 1) } // Обработка ошибок @@ -308,7 +306,7 @@ func WorkerGroup( return } - log.Printf("[ВОРКЕР #%d] Ошибка сессии: %v (адрес=%s, тип=%s)", + log.Printf("[ВОРКЕР #%d] [FAILOVER] Сбой сессии: %v (адрес=%s, тип=%s)", wid, sessErr.Err, sessErr.Address, sessErr.Type) switch sessErr.Type { @@ -319,6 +317,8 @@ func WorkerGroup( GlobalBlacklist.Ban(sessErr.Address) log.Printf("[ВОРКЕР #%d] TURN-адрес %s забанен на 5 минут", wid, sessErr.Address) } + relayOffset++ + addressAttempts = 0 // Небольшая пауза перед переходом к следующему адресу select { case <-time.After(sleepAfterAddressDead): @@ -353,6 +353,7 @@ func WorkerGroup( wid, sessErr.Address, addressAttempts) } addressAttempts = 0 + relayOffset++ // Продолжаем цикл — возьмём следующий адрес continue } @@ -364,16 +365,21 @@ func WorkerGroup( // Креды протухли (401) — обновляем и продолжаем. // Обновляет только один воркер: остальные увидят новый credGen log.Printf("[ВОРКЕР #%d] Ошибка авторизации TURN, обновляю креды", wid) - credsMu.Lock() - if credGen == gen { + refreshMu.Lock() + credsMu.RLock() + currentGen := credGen + credsMu.RUnlock() + if currentGen == gen { if err := fetchCreds(); err == nil { + credsMu.Lock() credGen++ + credsMu.Unlock() log.Printf("[ВОРКЕР #%d] Креды обновлены", wid) } else { log.Printf("[ВОРКЕР #%d] Не удалось обновить креды: %v", wid, err) } } - credsMu.Unlock() + refreshMu.Unlock() select { case <-time.After(2 * time.Second): case <-ctx.Done(): @@ -407,7 +413,7 @@ func WorkerGroup( return } } - }(wid, startDelay) + }(wid, idx, startDelay) } signalSpawnOnce() @@ -416,6 +422,38 @@ func WorkerGroup( log.Printf("[ГРУППА #%d] Все воркеры группы завершились.", groupID) } +func shouldRaceForConfig(enabled bool, workerIndex int, configSent int32) bool { + return enabled && configSent == 0 && workerIndex < configRacerLimit +} + +func normalizeWorkers(n int) int { + if n > 108 { + n = 108 + } + if n < workersPerGroup { + n = workersPerGroup + } + return (n / workersPerGroup) * workersPerGroup +} + +// selectTurnURL assigns workers round-robin and skips currently banned relays. +func selectTurnURL(urls []string, workerID, offset int, isBanned func(string) bool) (string, bool) { + if len(urls) == 0 { + return "", false + } + start := (workerID - 1 + offset) % len(urls) + if start < 0 { + start += len(urls) + } + for i := 0; i < len(urls); i++ { + u := urls[(start+i)%len(urls)] + if !isBanned(u) { + return u, true + } + } + return "", false +} + func ParseHashes(raw string) []string { var result []string seen := make(map[string]struct{}) @@ -460,4 +498,5 @@ type TurnParams struct { WrapKey []byte ObfsMode string TCPTransport bool -} \ No newline at end of file + allocateGate *allocationGate +} diff --git a/core/worker_group_test.go b/core/worker_group_test.go new file mode 100644 index 00000000..4885aae0 --- /dev/null +++ b/core/worker_group_test.go @@ -0,0 +1,47 @@ +package core + +import "testing" + +func TestSelectTurnURLRoundRobinAndFallback(t *testing.T) { + urls := []string{"a:1", "b:1", "c:1"} + noneBanned := func(string) bool { return false } + for worker, want := range []string{"a:1", "b:1", "c:1", "a:1"} { + got, ok := selectTurnURL(urls, worker+1, 0, noneBanned) + if !ok || got != want { + t.Fatalf("worker %d: got %q,%v want %q,true", worker+1, got, ok, want) + } + } + + got, ok := selectTurnURL(urls, 1, 0, func(s string) bool { return s == "a:1" }) + if !ok || got != "b:1" { + t.Fatalf("fallback: got %q,%v want b:1,true", got, ok) + } + got, _ = selectTurnURL(urls, 1, 1, noneBanned) + if got != "b:1" { + t.Fatalf("retry rotation: got %q want b:1", got) + } +} + +func TestConfigRacersAreBounded(t *testing.T) { + for i := 0; i < 9; i++ { + want := i < configRacerLimit + if got := shouldRaceForConfig(true, i, 0); got != want { + t.Errorf("worker %d races = %v, want %v", i, got, want) + } + } + if shouldRaceForConfig(true, 0, 1) { + t.Error("config racers continue after config delivery") + } + if shouldRaceForConfig(false, 0, 0) { + t.Error("config racers started when config is disabled") + } +} + +func TestNormalizeWorkers(t *testing.T) { + cases := map[int]int{-1: 9, 4: 9, 9: 9, 16: 9, 18: 18, 999: 108} + for in, want := range cases { + if got := normalizeWorkers(in); got != want { + t.Errorf("normalizeWorkers(%d)=%d, want %d", in, got, want) + } + } +}